Compare commits

..
Author SHA1 Message Date
weichen 70598ae801 Merge the endpoint migration into the advanced backend line
Why: Dev main advanced the backend pin on the pre-migration lineage
while the endpoint migration landed on migrated main; combine both.

Evidence: host-verified; see Dev PR #706

Impact: integration commit; .gitmodules keeps migrated URLs.

Refs: Agent-System-Dev PR #706

# Conflicts:
#	deps/agent-native
#	deps/agent-squad
2026-10-22 16:12:17 +00:00
weichen a2e9f003bc Merge pull request 'Update Gitea endpoints after server migration' (#247) from fix/gitea-endpoint-migration into main
governance / governance (push) Successful in 19s
ci / vet-and-test (push) Successful in 2m38s
2026-10-22 15:43:18 +00:00
weichen 2da3d76ce1 Move the published renderer pin to the migrated renderer main
governance / governance (pull_request) Successful in 25s
ci / vet-and-test (pull_request) Successful in 2m23s
Why: the document-renderer pin moved to its migrated main for the
endpoint migration; identity regenerated and the published-pin
expectation updated to match.

Evidence: go test ./internal/docgen → ok (local, host-verified)

Impact: generated identity + test constant; part of the coordinated
endpoint migration.

Refs: Agent-System-Dev PR #706
2026-10-22 15:34:06 +00:00
weichen ea497ed149 Regenerate the renderer identity for the migrated renderer pin
governance / governance (pull_request) Successful in 16s
ci / vet-and-test (pull_request) Failing after 2m31s
Why: the document-renderer pin moved to its migrated main; the
checked-in identity must match or docgen tests fail.

Evidence: host-verified; see Dev PR #706

Impact: generated constant only.

Refs: Agent-System-Dev PR #706
2026-10-22 15:23:24 +00:00
weichen e853260bc7 Track the updated squad compat branch
governance / governance (pull_request) Successful in 30s
ci / vet-and-test (pull_request) Failing after 2m54s
Why: squad compat now pins the intent compat branch for recursion.

Evidence: host-verified; see Dev PR #706

Impact: pin update only.

Refs: Agent-System-Dev PR #706
2026-10-22 15:13:48 +00:00
weichen 1f952a7ef7 Pin legacy-line deps to endpoint-migration compat branches
governance / governance (pull_request) Successful in 29s
ci / vet-and-test (pull_request) Failing after 1m24s
Why: these deps' latest mains moved APIs beyond what this branch
compiles against; the compat branches carry migrated URLs at the
pinned API eras.

Evidence: host-verified; see Dev PR #706

Impact: pin updates only.

Refs: Agent-System-Dev PR #706
2026-10-22 15:05:02 +00:00
weichen 330c1a3ffe Merge commit 'f9939dd1d722b997c6f1522237ae2e53fd9e239c' into fix/exec-recovery-diagnostics-a2 2026-10-22 14:56:09 +00:00
weichen aee1d9bcf2 Bump migrated dependency pins for the endpoint migration
governance / governance (pull_request) Successful in 21s
ci / vet-and-test (pull_request) Failing after 1m56s
Why: recursive submodule checkout reads pinned dep commits; their
.gitmodules now carries the new Gitea endpoint base only in the
migrated mains.

Evidence: host-verified; see Dev PR #706

Impact: pin updates only.

Refs: Agent-System-Dev PR #706
2026-10-22 14:54:13 +00:00
agent-exec-transport f9939dd1d7 agent-exec transport b82641a7033ca786ca66a2ca599ea9b5/4ad0f36d42ba 2026-10-22 14:53:01 +00:00
weichen 503ae000e0 Bump migrated dependency pins for the endpoint migration
governance / governance (pull_request) Successful in 39s
ci / vet-and-test (pull_request) Failing after 42s
Why: recursive submodule checkout reads pinned dep commits; their
.gitmodules now carries the new Gitea endpoint base only in the
migrated mains.

Evidence: host-verified; see Dev PR #706

Impact: pin updates only, part of the coordinated endpoint migration.

Refs: Agent-System-Dev PR #706
2026-10-22 14:36:15 +00:00
weichen 392bb68506 Bump the development-manager pin for the endpoint migration
ci / vet-and-test (pull_request) Failing after 1m1s
governance / governance (pull_request) Successful in 1m1s
Why: governance checks validate candidate .gitmodules URLs against the
pinned development-manager scaffold, which now carries the new Gitea
endpoint base after its own migration.

Evidence: host-verified; see Dev PR #706

Impact: pin update only, part of the coordinated endpoint migration.

Refs: Agent-System-Dev PR #706
2026-10-22 14:25:42 +00:00
weichen 30e18af70e Update Gitea endpoints after server migration
ci / vet-and-test (pull_request) Failing after 27s
governance / governance (pull_request) Failing after 34s
Why: the Gitea server moved to https://gitea.naitangbaobao.cn with SSH
on port 52222; endpoint references must follow or superproject CI
checkout refuses the old URLs.

Evidence: host-verified API reach and SSH auth (Dev task card
fix/gitea-endpoint-migration)

Impact: configuration only; no runtime behavior change.

Refs: Dev PR #706
2026-10-22 14:16:47 +00:00
agent-exec-transport 06b0c9f1db Add the read-only run diagnostics endpoint
Why: A2 - answer where a run is stuck, who is responsible, why it
waits, the last stable checkpoint, known side effects and the next
step, from existing records only (recovery holds, HITL decisions,
confirmation cards, audit envelopes).
Evidence: rundiag/server/store tests pass (verify-a2.json); per-state
responsibility, scoping and next-step mapping cases added.
Impact: one read-only GET endpoint plus read helpers; no behavior
change to execution or recovery paths.
Refs: feature/exec-recovery-diagnostics-a2
2026-10-22 13:03:53 +00:00
28 changed files with 1575 additions and 323 deletions
+1 -1
View File
@@ -24,7 +24,7 @@ jobs:
echo "SUBMODULE_TOKEN is not configured" >&2
exit 1
fi
git config --global url."https://weichen:${SUBMODULE_TOKEN}@git-home.u1680832.nyat.app:30319/".insteadOf "ssh://git@git-ssh.u1680832.nyat.app:31411/"
git config --global url."https://weichen:${SUBMODULE_TOKEN}@gitea.naitangbaobao.cn/".insteadOf "ssh://git@gitea.naitangbaobao.cn:52222/"
git -c fetch.recurseSubmodules=false submodule update --init --recursive
- name: Install Go 1.27.1
run: |
+1 -1
View File
@@ -31,7 +31,7 @@ jobs:
parsed_server = urlsplit(SERVER)
if parsed_server.scheme not in {"http", "https"} or not parsed_server.netloc:
sys.exit("Checkout server must be the platform-injected http(s) URL")
SSH = "ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/"
SSH = "ssh://git@gitea.naitangbaobao.cn:52222/weichen/"
# 审查过的白名单,独立于候选 .gitmodules:仅列出两个治理 gitlink。
REPOSITORIES = {
"deps/development-guidelines": "Development-Guidelines",
+10 -10
View File
@@ -1,30 +1,30 @@
[submodule "deps/agent-system-core"]
path = deps/agent-system-core
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-System-Core.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-System-Core.git
[submodule "deps/agent-native"]
path = deps/agent-native
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Native.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Native.git
[submodule "deps/agent-intent"]
path = deps/agent-intent
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Intent.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Intent.git
[submodule "deps/logging-system"]
path = deps/logging-system
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Logging-System.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Logging-System.git
[submodule "deps/error-code-system"]
path = deps/error-code-system
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Error-Code-System.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Error-Code-System.git
[submodule "deps/agent-squad"]
path = deps/agent-squad
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Squad.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Squad.git
[submodule "deps/agent-command"]
path = deps/agent-command
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Command.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Command.git
[submodule "deps/agent-document-renderer"]
path = deps/agent-document-renderer
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Document-Renderer.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Document-Renderer.git
[submodule "deps/development-guidelines"]
path = deps/development-guidelines
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Guidelines.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Development-Guidelines.git
[submodule "deps/development-manager"]
path = deps/development-manager
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Manager.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Development-Manager.git
+3 -3
View File
@@ -23,9 +23,9 @@
历史文档与旧链接里的 `git-panel.u1680832.nyat.app:30319` 已失效(TLS 握手失败),
见到一律按过期处理,不要再试。当前权威通道:
- **Web / API(HTTPS)**:`https://git-home.u1680832.nyat.app:30319`(Gitea 1.27.3,
- **Web / API(HTTPS)**:`https://gitea.naitangbaobao.cn`(Gitea 1.27.3,
API 前缀 `/api/v1`;偶发 broken pipe / 超时,重试即可)。
- **Git over SSH**:`ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/<Repo>.git`
- **Git over SSH**:`ssh://git@gitea.naitangbaobao.cn:52222/weichen/<Repo>.git`
(各仓 origin 与 gitlink 均为此格式)。
- **API token**:`~/.secrets/gitea`,两行键值 `URL_PREFIX=…` 与
`ACCESS_TOKEN=<40 位 hex>`;取值 `grep ACCESS_TOKEN ~/.secrets/gitea | cut -d= -f2-`,
@@ -33,7 +33,7 @@
(baseline 硬边界 2:只记录来源、字段与可用状态)。
- **SSH key**:`~/.ssh/home_gitea`(ED25519,公钥指纹
`SHA256:GYXaOXIMeloQ1fKSfdJ/ndgPQzs/Ix7Hk62ekfkfQ20`,Gitea 侧密钥名 GMKtec);
`~/.ssh/config` 已有 `git-ssh.u1680832.nyat.app` 主机条目,克隆/推送直接可用。
`~/.ssh/config` 已有 `gitea.naitangbaobao.cn(端口 52222)` 主机条目,克隆/推送直接可用。
## Agent 执行与新项目准入(强制)
@@ -2,6 +2,6 @@
package docgen
const rendererRevision = "3f259990b5d43ec0d318f39d496816140ab559b5"
const rendererRevision = "40e5c671d1ef786ceba62bf784bf3aa53c5da26d"
const RendererIdentity = "agent-document-renderer@" + rendererRevision
+1 -1
View File
@@ -367,7 +367,7 @@ func TestPersistedControlledMetadataRoutesBeforeDefaultProfileLoad(t *testing.T)
}
func TestRendererIdentityMatchesPublishedPin(t *testing.T) {
const want = "agent-document-renderer@3f259990b5d43ec0d318f39d496816140ab559b5"
const want = "agent-document-renderer@40e5c671d1ef786ceba62bf784bf3aa53c5da26d"
if RendererIdentity != want {
t.Fatalf("RendererIdentity = %q, want %q", RendererIdentity, want)
}
+1 -13
View File
@@ -185,17 +185,12 @@ func (s *Service) Pending(key Key) (*Session, error) {
// Suspend 挂起一个可执行动作的预期并累计已抽到的槽位(续期)。已抽到的
// 槽位不再计入仍缺清单(挂起即如实反映还差什么)。
//
// 累计只在同一个动作内成立:新预期指向别的动作(意图 id 或动作引用变了)
// 时,上一轮已抽到的槽位一律丢弃——新任务从零开始,revision 照常推进,
// 陈旧回答因此作废。否则上一轮抽到的运行号会跟着新动作一起派发:旧任务的
// 槽位漏进新任务,正是本包纪律里明令禁止的那一类。
func (s *Service) Suspend(key Key, expectation Expectation, slots map[string]string) (*Session, error) {
cur, err := s.load(key)
if err != nil {
return nil, err
}
if cur != nil && cur.Expectation != nil && sameExpectation(*cur.Expectation, expectation) {
if cur != nil {
slots = mergeSlots(cur.Slots, slots)
}
var missing []intentgate.SlotQuestion
@@ -413,13 +408,6 @@ func (s *Service) accepted(cur *Session, answer Answer) map[string]string {
return out
}
// sameExpectation 报告新旧预期是否同一个动作(意图 id 与动作引用都一致):
// 只有同一个动作再次挂起才继续累计已抽到的槽位。预期是「已取消」记录时
// 动作种类为 cancel,与任何待补齐的动作都不相同——取消记录绝不被借走。
func sameExpectation(a, b Expectation) bool {
return a.IntentID == b.IntentID && a.Action.Kind == b.Action.Kind && a.Action.ID == b.Action.ID
}
// missingOf 投影到存储形态(缺失槽位表)。
func missingOf(list []intentgate.SlotQuestion) []store.IntentSlotQuestion {
out := make([]store.IntentSlotQuestion, 0, len(list))
@@ -108,88 +108,6 @@ func TestSuspendThenAnswerCompletes(t *testing.T) {
}
}
// 换动作即换任务(跨动作槽位继承的回归序列):A 抽到运行号后挂起 → 用户
// 改口做另一个动作 B(B 缺别的槽位)→ 用户补齐 B → B 齐全时只带 B 自己
// 的槽位。A 的运行号绝不跟着 B 一起派发。
func TestSuspendDiscardsSlotsOfAnotherAction(t *testing.T) {
svc := newService(t)
key := cancelKey()
first, err := svc.Suspend(key, cancelExpectation(), map[string]string{"run_id": "R-7"})
if err != nil {
t.Fatal(err)
}
report := Expectation{IntentID: "demo.report",
Action: intentgate.ActionRef{Kind: intentgate.ActionCommand, ID: "demo.report"},
Missing: []intentgate.SlotQuestion{{Key: "scope", Question: "哪个范围?"}},
Question: "哪个范围?"}
second, err := svc.Suspend(key, report, nil)
if err != nil {
t.Fatal(err)
}
if second.Revision <= first.Revision {
t.Fatalf("换动作必须推进 revision(让在途回答作废):%d vs %d", second.Revision, first.Revision)
}
if second.Expectation == nil || second.Expectation.IntentID != "demo.report" ||
second.Expectation.Action.ID != "demo.report" {
t.Fatalf("挂起的应是新动作:%+v", second)
}
if _, leaked := second.Slots["run_id"]; leaked {
t.Fatalf("上一个动作的槽位被继承:%+v", second.Slots)
}
if keys := questionKeys(second.Expectation.Missing); len(keys) != 1 || keys[0] != "scope" {
t.Fatalf("新动作的缺失清单:%v", keys)
}
// 旧动作的陈旧回答也作废(不得借新动作的壳补齐)。
if _, err := svc.ApplyAnswer(key, first.Revision,
Answer{Slots: map[string]string{"reason": "误触发"}}); !errors.Is(err, ErrStale) {
t.Fatalf("旧 revision 必须被拒:%v", err)
}
// 用户补齐 B:B 齐全,且槽位只有 B 自己的(没有 A 的 run_id)。
done, err := svc.ApplyAnswer(key, second.Revision, Answer{Slots: map[string]string{"scope": "华东"}})
if err != nil {
t.Fatal(err)
}
if !done.Complete || done.Session != nil {
t.Fatalf("B 应被补齐并消费会话:%+v", done)
}
if len(done.Slots) != 1 || done.Slots["scope"] != "华东" {
t.Fatalf("派发槽位混进了别的动作的值:%v", done.Slots)
}
}
// 同一动作再次挂起仍累计已抽到的槽位(缺失清单变窄不算换动作)。
func TestSuspendOfSameActionKeepsAccumulatedSlots(t *testing.T) {
svc := newService(t)
key := cancelKey()
first, err := svc.Suspend(key, cancelExpectation(), nil)
if err != nil {
t.Fatal(err)
}
// 同一意图与动作引用,只是路由面收敛了缺失清单(仍缺原因)。
narrowed := cancelExpectation()
narrowed.Missing = []intentgate.SlotQuestion{{Key: "reason", Question: "取消原因?"}}
again, err := svc.Suspend(key, narrowed, map[string]string{"run_id": "R-9"})
if err != nil {
t.Fatal(err)
}
if again.Revision <= first.Revision {
t.Fatalf("再次挂起必须推进 revision:%d vs %d", again.Revision, first.Revision)
}
if again.Slots["run_id"] != "R-9" {
t.Fatalf("同一动作的累计槽位被丢弃:%+v", again.Slots)
}
if keys := questionKeys(again.Expectation.Missing); len(keys) != 1 || keys[0] != "reason" {
t.Fatalf("同一动作的缺失清单:%v", keys)
}
done, err := svc.ApplyAnswer(key, again.Revision, Answer{Slots: map[string]string{"reason": "误触发"}})
if err != nil {
t.Fatal(err)
}
if !done.Complete || done.Slots["run_id"] != "R-9" || done.Slots["reason"] != "误触发" {
t.Fatalf("同一动作补齐:%+v", done)
}
}
// 两轮部分填充:set_slot 落一个槽位,再由回答补另一个。
func TestSetSlotThenAnswerCompletes(t *testing.T) {
svc := newService(t)
+488
View File
@@ -0,0 +1,488 @@
// Package rundiag 把单次 run 的诊断证据聚合成一份只读结论:卡在哪、谁负责、
// 为什么在等、最后一个稳定检查点、已知副作用、下一步动作。
//
// 全包统一口径(任务卡 A2 的契约):
// - 只读派生:Build 是纯函数,Gather 只跑 SELECT;本包不提供任何写路径。
// - 证据优先于推断:状态只由已落库的记录驱动。读不到就降级为 unknown + note,
// 读不到隔离记录时更按 fail-closed 处理(不声称"可以安全重试")。
// - 「没有记录」≠「没有发生」:side_effects 为空必须带 note,说明这是"未记录"
// 而不是"未发生"。
// - 归属边界:HITL 决策卡、危险命令确认卡、命令审计信封都按 owner + 会话归属,
// 其中决策卡与确认卡表里根本没有 run 列,审计信封虽有 runId 字段但宿主总线
// 当前不写。因此"归到这个 run"是调用方按签发时间窗做的有界推断,报告以
// note 标出,绝不把会话里更早的记录算成本次 run 的副作用。
package rundiag
import (
"encoding/json"
"errors"
"fmt"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// responsibility.state 的取值域(任务卡固定)。
const (
StateRunning = "running"
StateWaitingHITL = "waiting_hitl"
StateConfirmationPending = "confirmation_pending"
StateRecoveryHold = "recovery_hold"
StateReconciling = "reconciling"
StateFailed = "failed"
StateCompleted = "completed"
StateUnknown = "unknown"
)
// responsibility.role:这一刻该谁动。
const (
RoleRunner = "runner" // 执行器自己在推进
RoleHuman = "human" // 等人裁决(HITL / 危险命令确认)
RoleOperationOwner = "operation-owner" // 等副作用责任人对账与解除
RoleNone = "none" // 没有待办责任方
)
// next_step.action 的取值域。
const (
ActionConfirmCard = "confirm_card"
ActionReleaseRecoveryHold = "release_recovery_hold"
ActionSafeRetry = "safe_retry"
ActionNeedsManualReconciliation = "needs_manual_reconciliation"
ActionNone = "none"
)
// side_effects[].kind:证据来源类别。
const (
EffectOperation = "operation" // 命令审计信封(U7b 对账记录)
EffectRecoveryHold = "recovery_hold" // 恢复隔离行(U3b 的"副作用未知"证据)
)
// 读取上界:诊断是给人和值班看的,不是全量导出。
const (
maxSideEffects = 50
maxAuditScan = 200 // 审计信封按会话+时间窗有界扫描
maxConfirms = 50
)
// Report 是诊断响应(JSON 契约,字段名用任务卡给定的 snake_case)。
type Report struct {
Run RunSummary `json:"run"`
Responsibility Responsibility `json:"responsibility"`
LastCheckpoint *Checkpoint `json:"last_checkpoint"` // 无记录时为 null(不编造)
SideEffects []SideEffect `json:"side_effects"` // 恒为数组:空数组 = 未记录到
NextStep NextStep `json:"next_step"`
Notes []string `json:"notes,omitempty"` // 降级与边界说明
}
// RunSummary 是 run 自身的身份与时间戳。updated_at 的口径:已终态取 ended_at,
// 未终态取最后一个已持久化事件的时间;两者都没有则为 null(并在 notes 说明)。
type RunSummary struct {
ID string `json:"id"`
Status string `json:"status"`
Kind string `json:"kind,omitempty"`
Phase string `json:"phase,omitempty"`
Owner string `json:"owner"`
Project string `json:"project"`
Conversation string `json:"conversation"`
CreatedAt string `json:"created_at"`
UpdatedAt *string `json:"updated_at"`
}
// Responsibility 回答"卡在哪、谁负责、为什么等、从什么时候开始等"。
type Responsibility struct {
State string `json:"state"`
Role string `json:"role"`
WaitReason string `json:"wait_reason"`
Since string `json:"since,omitempty"` // 进入该状态的时间;存储没记就省略
}
// Checkpoint 指向最后一个稳定落盘点(事件流上的 seq 可直接作重放游标)。
type Checkpoint struct {
ID string `json:"id"`
At string `json:"at"`
Type string `json:"type"`
}
// SideEffect 是一条已记录的外部副作用证据。operation 行为命令回执对账记录;
// recovery_hold 行是 U3b 落下的"副作用结果未知"隔离证据(state = held/released)。
type SideEffect struct {
Kind string `json:"kind"`
OperationID string `json:"operation_id,omitempty"`
CommandID string `json:"command_id,omitempty"`
State string `json:"state"`
At string `json:"at,omitempty"`
ReleasedAt string `json:"released_at,omitempty"`
}
// NextStep 是唯一建议的下一步动作(诊断面只建议,不执行)。
type NextStep struct {
Action string `json:"action"`
CardID string `json:"card_id,omitempty"` // confirm_card 可解析时给出卡 ID
Reason string `json:"reason"`
}
// Source 是诊断所需的只读端口(*store.Store 直接满足;测试可替身)。
type Source interface {
GetRecoveryHold(runID string) (store.RecoveryHold, error)
ActiveHITLCardsByRun(runID string) ([]core.HITLCard, error)
PendingHumanDecisionsByConversation(owner, conversationID string, now time.Time) ([]store.HumanDecision, error)
ListOwnerPendingConfirms(owner string, limit int) ([]store.PendingConfirm, error)
LastRunCheckpoint(conversationID, runID string) (store.RunCheckpoint, bool, error)
ListAudits(owner, conversationID string, limit int) ([]agentcommand.AuditEnvelope, error)
}
// Evidence 是 Build 的全部输入(合成数据即可表驱动测试,无需真实存储)。
type Evidence struct {
Run core.Run
Hold *store.RecoveryHold // nil = 没有隔离行;HoldUnknown = 读失败
HoldUnknown bool
Cards []core.HITLCard // 该 run 未关闭的 HITL 卡
Decisions []store.HumanDecision // 会话内未过期的 pending 决策卡
Confirms []store.PendingConfirm // owner 名下未确认的危险命令挂起行
Checkpoint *store.RunCheckpoint // 最后一个已持久化事件
Audits []agentcommand.AuditEnvelope
Notes []string
}
// Gather 只读地把一个 run 的证据读齐。任何一条可选查询失败都不让整个诊断失败:
// 失败降级为 note;隔离记录读不到按 fail-closed 标记(HoldUnknown),因为"读不到
// hold"绝不能被读成"没有 hold"。
func Gather(src Source, run core.Run, now time.Time) Evidence {
ev := Evidence{Run: run}
if src == nil {
ev.HoldUnknown = true
return ev.withNote("diagnostics source unavailable; responsibility is fail-closed unknown")
}
owner := string(run.Owner)
conversation := string(run.ConversationID)
hold, err := src.GetRecoveryHold(string(run.ID))
switch {
case err == nil:
ev.Hold = &hold
case errors.Is(err, store.ErrNotFound):
// 没有隔离行是正常态,不记 note。
default:
ev.HoldUnknown = true
ev = ev.withNote("recovery hold unreadable: " + err.Error())
}
cards, err := src.ActiveHITLCardsByRun(string(run.ID))
if err != nil {
ev = ev.withNote("hitl cards unreadable: " + err.Error())
} else {
ev.Cards = cards
}
decisions, err := src.PendingHumanDecisionsByConversation(owner, conversation, now)
if err != nil {
ev = ev.withNote("human decisions unreadable: " + err.Error())
} else {
ev.Decisions = decisions
}
confirms, err := src.ListOwnerPendingConfirms(owner, maxConfirms)
if err != nil {
ev = ev.withNote("pending confirmations unreadable: " + err.Error())
} else {
ev.Confirms = confirms
}
cp, ok, err := src.LastRunCheckpoint(conversation, string(run.ID))
switch {
case err != nil:
ev = ev.withNote("last checkpoint unreadable: " + err.Error())
case ok:
ev.Checkpoint = &cp
}
audits, err := src.ListAudits(owner, conversation, maxAuditScan)
if err != nil {
ev = ev.withNote("command audit unreadable: " + err.Error())
} else {
ev.Audits = audits
}
return ev
}
// Build 把证据聚合成诊断结论(纯函数,不读写任何状态)。
func Build(ev Evidence) Report {
rep := Report{Run: runSummary(ev), SideEffects: []SideEffect{}}
if ev.Checkpoint != nil {
rep.LastCheckpoint = &Checkpoint{ID: fmt.Sprintf("%d", ev.Checkpoint.Seq), At: ev.Checkpoint.TS, Type: ev.Checkpoint.Type}
}
rep.SideEffects = sideEffects(ev)
rep.Notes = append([]string(nil), ev.Notes...)
card := waitingCard(ev)
decision := waitingDecision(ev)
confirm := waitingConfirm(ev)
switch {
case ev.HoldUnknown:
// 隔离记录读不到:不猜没有 hold,直接 unknown,且不给"安全重试"。
rep.Notes = append(rep.Notes, "recovery hold state unknown; treat as quarantined until reconciled")
rep.Responsibility = Responsibility{
State: StateUnknown, Role: RoleNone,
WaitReason: "recovery hold state could not be read",
}
rep.NextStep = NextStep{Action: ActionNone, Reason: "cannot prove the run is free of an open recovery hold"}
case openHold(ev):
hold := ev.Hold
rep.Responsibility = Responsibility{
State: StateRecoveryHold, Role: RoleOperationOwner,
WaitReason: holdReason(hold), Since: hold.CreatedAt,
}
rep.NextStep = NextStep{Action: ActionReleaseRecoveryHold, Reason: holdReason(hold)}
case needsReconcile(ev):
reason := "recovery hold released but the run is still interrupted"
since := ""
if ev.Hold != nil {
since = ev.Hold.ReleasedAt
if ev.Hold.Reason != "" {
reason += "; hold reason: " + ev.Hold.Reason
}
}
rep.Responsibility = Responsibility{State: StateReconciling, Role: RoleOperationOwner, WaitReason: reason, Since: since}
rep.NextStep = NextStep{Action: ActionNeedsManualReconciliation, Reason: reason}
case ev.Run.Status == core.RunWaitingHITL || card != nil || decision != nil:
reason := "waiting for a human decision"
since := ""
if decision != nil {
reason = decision.Question
if reason == "" {
reason = "pending human decision on command " + decision.CommandID
}
since = decision.IssuedAt.UTC().Format(time.RFC3339)
} else if card != nil {
reason = card.Question
if reason == "" {
reason = "run is waiting on a HITL card"
}
since = card.CreatedAt
}
rep.Responsibility = Responsibility{State: StateWaitingHITL, Role: RoleHuman, WaitReason: reason, Since: since}
if card == nil && decision == nil {
// 只由 run 状态驱动:没有可解析的卡,也不知道等待从何时开始。
rep.Notes = append(rep.Notes, "run is marked waiting_hitl with no open card recorded; the wait entry time is not stored")
}
rep.NextStep = NextStep{Action: ActionConfirmCard, CardID: hitlCardID(card, decision), Reason: reason}
case confirm != nil:
reason := "dangerous command " + confirm.command + " awaits explicit confirmation"
rep.Responsibility = Responsibility{
State: StateConfirmationPending, Role: RoleHuman,
WaitReason: reason, Since: confirm.row.CreatedAt,
}
rep.NextStep = NextStep{Action: ActionConfirmCard, CardID: confirm.row.CardID, Reason: reason}
case ev.Run.Status == core.RunFailed:
reason := "run failed"
if ev.Run.Error != nil {
reason = fmt.Sprintf("run failed: %s: %s", ev.Run.Error.Code, ev.Run.Error.Message)
}
rep.Responsibility = Responsibility{State: StateFailed, Role: RoleRunner, WaitReason: reason, Since: endedAt(ev.Run)}
// 重试安全性看已记录的副作用证据:没有 open hold 且没有 unknown 回执时,
// 重试不会复制本次运行的后果;否则先对账再重试,不说"安全"。
if unknown := unknownOperations(rep.SideEffects); unknown > 0 {
reason := fmt.Sprintf("terminal failure, but %d recorded operation(s) have an unknown outcome; reconcile side effects before retrying", unknown)
rep.Notes = append(rep.Notes, reason)
rep.NextStep = NextStep{Action: ActionNeedsManualReconciliation, Reason: reason}
} else {
rep.NextStep = NextStep{Action: ActionSafeRetry, Reason: "terminal failure with no open recovery hold and no unknown receipt; retry is safe to attempt"}
}
case ev.Run.Status == core.RunCompleted:
rep.Responsibility = Responsibility{State: StateCompleted, Role: RoleNone, WaitReason: "run completed", Since: endedAt(ev.Run)}
rep.NextStep = NextStep{Action: ActionNone, Reason: "run already completed"}
case ev.Run.Status == core.RunRunning || ev.Run.Status == core.RunPending:
rep.Responsibility = Responsibility{
State: StateRunning, Role: RoleRunner,
WaitReason: "run is executing", Since: ev.Run.CreatedAt.UTC().Format(time.RFC3339),
}
rep.NextStep = NextStep{Action: ActionNone, Reason: "runner is executing; no operator action"}
default:
reason := fmt.Sprintf("run status %q has no diagnostic mapping", ev.Run.Status)
rep.Notes = append(rep.Notes, reason)
rep.Responsibility = Responsibility{State: StateUnknown, Role: RoleNone, WaitReason: reason, Since: endedAt(ev.Run)}
rep.NextStep = NextStep{Action: ActionNone, Reason: "status is unmapped; inspect the run record manually"}
}
if ev.Checkpoint == nil {
rep.Notes = append(rep.Notes, "no persisted checkpoint reference for this run (no durable stream event)")
}
if len(rep.SideEffects) == 0 {
rep.Notes = append(rep.Notes, "no recorded side effect evidence; absence of records is not proof of no effect")
}
return rep
}
func runSummary(ev Evidence) RunSummary {
summary := RunSummary{
ID: string(ev.Run.ID), Status: string(ev.Run.Status), Kind: string(ev.Run.Kind),
Phase: string(ev.Run.Phase), Owner: string(ev.Run.Owner), Project: string(ev.Run.ProjectID),
Conversation: string(ev.Run.ConversationID),
}
if !ev.Run.CreatedAt.IsZero() {
summary.CreatedAt = ev.Run.CreatedAt.UTC().Format(time.RFC3339)
}
// updated:终态取 ended_at,未终态取最后一次落盘(检查点)的时间。
if ended := endedAt(ev.Run); ended != "" {
summary.UpdatedAt = &ended
} else if ev.Checkpoint != nil && ev.Checkpoint.TS != "" {
ts := ev.Checkpoint.TS
summary.UpdatedAt = &ts
}
return summary
}
// sideEffects 汇总已记录的外部副作用证据:命令回执对账记录(U7b 审计信封)加上
// 恢复隔离行(U3b 的"结果未知"证据)。
func sideEffects(ev Evidence) []SideEffect {
out := []SideEffect{}
if ev.Hold != nil {
state := "held"
if ev.Hold.ReleasedAt != "" {
state = "released"
}
out = append(out, SideEffect{
Kind: EffectRecoveryHold, State: state, At: ev.Hold.CreatedAt, ReleasedAt: ev.Hold.ReleasedAt,
})
}
for _, env := range ev.Audits {
if !belongsToRun(ev, env) {
continue
}
out = append(out, SideEffect{
Kind: EffectOperation, OperationID: env.OperationID, CommandID: env.CommandID,
State: string(env.State), At: env.Time.UTC().Format(time.RFC3339),
})
if len(out) >= maxSideEffects {
break
}
}
return out
}
// belongsToRun 是审计信封的归属边界:信封自带 runId 时以它为准;否则只在"同一
// owner + 会话且签发时间不早于 run 创建时间"的窗口内归属——会话里更早的记录属于
// 更早的 run,不算进本次副作用。
func belongsToRun(ev Evidence, env agentcommand.AuditEnvelope) bool {
if env.RunID != "" {
return env.RunID == string(ev.Run.ID)
}
if string(env.ConversationID) != "" && string(env.ConversationID) != string(ev.Run.ConversationID) {
return false
}
if !ev.Run.CreatedAt.IsZero() && !env.Time.IsZero() && env.Time.Before(ev.Run.CreatedAt) {
return false
}
return true
}
func openHold(ev Evidence) bool {
return ev.Hold != nil && ev.Hold.ReleasedAt == ""
}
// unknownOperations 统计结果未知的已记录 operation(四态语义里的 unknown:
// 执行过但结果不确定)。它是"重试会不会复制副作用"的直接证据。
func unknownOperations(effects []SideEffect) int {
n := 0
for _, e := range effects {
if e.Kind == EffectOperation && e.State == string(agentcommand.StateUnknown) {
n++
}
}
return n
}
// needsReconcile 是 U3b 隔离解除后的残留态:隔离已解除,但 run 仍停在中断态——
// 没有人把它推进到终态,需要人工对账。
func needsReconcile(ev Evidence) bool {
return ev.Hold != nil && ev.Hold.ReleasedAt != "" && ev.Run.Status == core.RunInterrupted
}
func holdReason(hold *store.RecoveryHold) string {
if hold == nil || hold.Reason == "" {
return "external side effect outcome unknown (recovery hold)"
}
return hold.Reason
}
func endedAt(run core.Run) string {
if run.EndedAt == nil {
return ""
}
return run.EndedAt.UTC().Format(time.RFC3339)
}
// waitingCard 取该 run 未关闭的 HITL 卡(最早一张;只看未关闭的)。
func waitingCard(ev Evidence) *core.HITLCard {
for i := range ev.Cards {
if ev.Cards[i].Resolved || ev.Cards[i].Blocked {
continue
}
return &ev.Cards[i]
}
return nil
}
// waitingDecision 取本 run 生命周期内签发、仍未过期的人工决策卡。
func waitingDecision(ev Evidence) *store.HumanDecision {
for i := range ev.Decisions {
rec := &ev.Decisions[i]
if rec.Status != "pending" {
continue
}
if !rec.IssuedAt.IsZero() && !ev.Run.CreatedAt.IsZero() && rec.IssuedAt.Before(ev.Run.CreatedAt) {
continue // 签发在本次 run 之前:属于更早的等待,不算本次
}
return rec
}
return nil
}
type scopedConfirm struct {
row store.PendingConfirm
command string
}
// waitingConfirm 取本 run 生命周期内、同一会话挂起的危险命令确认行。载荷是总线
// 落下的分发请求;解析不出会话的旧行按"无记录"处理,不回显原始载荷。
func waitingConfirm(ev Evidence) *scopedConfirm {
for _, row := range ev.Confirms {
req := cmdbus.Request{}
if err := json.Unmarshal([]byte(row.Payload), &req); err != nil {
continue
}
if req.ConversationID != "" && req.ConversationID != string(ev.Run.ConversationID) {
continue
}
if req.Owner != "" && string(ev.Run.Owner) != "" && req.Owner != string(ev.Run.Owner) {
continue
}
created, err := time.Parse(time.RFC3339, row.CreatedAt)
if err == nil && !ev.Run.CreatedAt.IsZero() && created.Before(ev.Run.CreatedAt) {
continue // 挂起早于本次 run:属于更早的等待
}
return &scopedConfirm{row: row, command: req.ID}
}
return nil
}
// hitlCardID 给出可解析的卡标识:HITL 卡优先,其次决策卡 ID。
func hitlCardID(card *core.HITLCard, decision *store.HumanDecision) string {
if card != nil {
return string(card.ID)
}
if decision != nil {
return decision.ID
}
return ""
}
func (ev Evidence) withNote(note string) Evidence {
ev.Notes = append(ev.Notes, note)
return ev
}
+440
View File
@@ -0,0 +1,440 @@
package rundiag
import (
"encoding/json"
"errors"
"strings"
"testing"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/store"
core "agent-system-core"
)
var (
baseTime = time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC)
afterRun = baseTime.Add(5 * time.Minute)
)
// fakeSource 是只读端口的合成替身:每个字段都能单独触发"没有记录"或"读取失败"。
type fakeSource struct {
hold store.RecoveryHold
holdErr error
cards []core.HITLCard
decisions []store.HumanDecision
confirms []store.PendingConfirm
cp store.RunCheckpoint
cpOK bool
audits []agentcommand.AuditEnvelope
}
func (f fakeSource) GetRecoveryHold(string) (store.RecoveryHold, error) {
if f.holdErr != nil {
return store.RecoveryHold{}, f.holdErr
}
if f.hold.RunID == "" {
return store.RecoveryHold{}, store.ErrNotFound
}
return f.hold, nil
}
func (f fakeSource) ActiveHITLCardsByRun(string) ([]core.HITLCard, error) { return f.cards, nil }
func (f fakeSource) PendingHumanDecisionsByConversation(string, string, time.Time) ([]store.HumanDecision, error) {
return f.decisions, nil
}
func (f fakeSource) ListOwnerPendingConfirms(string, int) ([]store.PendingConfirm, error) {
return f.confirms, nil
}
func (f fakeSource) LastRunCheckpoint(string, string) (store.RunCheckpoint, bool, error) {
return f.cp, f.cpOK, nil
}
func (f fakeSource) ListAudits(string, string, int) ([]agentcommand.AuditEnvelope, error) {
return f.audits, nil
}
func run(status core.RunStatus) core.Run {
return core.Run{
ID: "run_1", ConversationID: "cnv_1", ProjectID: "prj_1", Owner: "alice",
Kind: core.RunConversation, Status: status, Phase: core.PhaseResponding, CreatedAt: baseTime,
}
}
func ended(r core.Run, at time.Time) core.Run {
r.EndedAt = &at
return r
}
func checkpoint() *store.RunCheckpoint {
return &store.RunCheckpoint{Seq: 7, TS: baseTime.Add(2 * time.Minute).Format(time.RFC3339), Type: "agent.progress"}
}
// A2 契约:每个 responsibility.state 都有驱动它的记录,且 next_step 逐状态映射正确。
func TestBuildResponsibilityStatesAndNextStep(t *testing.T) {
cases := []struct {
name string
ev Evidence
state string
role string
action string
cardID string
waitHint string
since string
}{
{
name: "unreleased recovery hold wins over everything",
ev: Evidence{
Run: run(core.RunRunning),
Hold: &store.RecoveryHold{RunID: "run_1", Owner: "alice", Reason: "unknown external effect",
CreatedAt: afterRun.Format(time.RFC3339)},
Cards: []core.HITLCard{{ID: "card_hitl", Resolved: false, Question: "approve?"}},
},
state: StateRecoveryHold, role: RoleOperationOwner, action: ActionReleaseRecoveryHold,
waitHint: "unknown external effect", since: afterRun.Format(time.RFC3339),
},
{
name: "released hold on an interrupted run needs reconciliation",
ev: Evidence{
Run: run(core.RunInterrupted),
Hold: &store.RecoveryHold{RunID: "run_1", Owner: "alice", Reason: "unknown external effect",
CreatedAt: afterRun.Format(time.RFC3339), ReleasedAt: afterRun.Add(time.Minute).Format(time.RFC3339)},
},
state: StateReconciling, role: RoleOperationOwner, action: ActionNeedsManualReconciliation,
waitHint: "still interrupted", since: afterRun.Add(time.Minute).Format(time.RFC3339),
},
{
name: "waiting on a HITL card",
ev: Evidence{
Run: run(core.RunWaitingHITL),
Cards: []core.HITLCard{{ID: "card_hitl", CreatedAt: afterRun.Format(time.RFC3339), Question: "approve the plan?"}},
},
state: StateWaitingHITL, role: RoleHuman, action: ActionConfirmCard, cardID: "card_hitl",
waitHint: "approve the plan?", since: afterRun.Format(time.RFC3339),
},
{
name: "waiting on a pending human decision without a run card",
ev: Evidence{
Run: run(core.RunRunning),
Decisions: []store.HumanDecision{{ID: "hd_1", CommandID: "rdm.confirm", Status: "pending",
Question: "approve delivery?", IssuedAt: afterRun, ExpiresAt: afterRun.Add(time.Hour)}},
},
state: StateWaitingHITL, role: RoleHuman, action: ActionConfirmCard, cardID: "hd_1",
waitHint: "approve delivery?", since: afterRun.Format(time.RFC3339),
},
{
name: "dangerous command awaiting confirmation",
ev: Evidence{
Run: run(core.RunRunning),
Confirms: []store.PendingConfirm{{CardID: "card_cmd", Owner: "alice", CreatedAt: afterRun.Format(time.RFC3339),
Payload: mustRequest(t, cmdbus.Request{ID: "rdm.docs.instantiate", Owner: "alice", ConversationID: "cnv_1"})}},
},
state: StateConfirmationPending, role: RoleHuman, action: ActionConfirmCard, cardID: "card_cmd",
waitHint: "rdm.docs.instantiate", since: afterRun.Format(time.RFC3339),
},
{
name: "terminal failure is a safe retry",
ev: Evidence{
Run: ended(func() core.Run {
r := run(core.RunFailed)
r.Error = &core.ErrorPart{Code: "run.tool_failed", Message: "tool crashed"}
return r
}(), afterRun),
},
state: StateFailed, role: RoleRunner, action: ActionSafeRetry,
waitHint: "run.tool_failed: tool crashed", since: afterRun.Format(time.RFC3339),
},
{
name: "terminal failure with an unknown receipt needs reconciliation, not a retry",
ev: Evidence{
Run: ended(run(core.RunFailed), afterRun),
Audits: []agentcommand.AuditEnvelope{{Time: afterRun, CommandID: "rdm.docs.instantiate",
OperationID: "op_unknown", State: agentcommand.StateUnknown, ConversationID: "cnv_1"}},
},
state: StateFailed, role: RoleRunner, action: ActionNeedsManualReconciliation,
waitHint: "run failed",
},
{
name: "completed needs no action",
ev: Evidence{Run: ended(run(core.RunCompleted), afterRun)},
state: StateCompleted, role: RoleNone, action: ActionNone,
waitHint: "completed", since: afterRun.Format(time.RFC3339),
},
{
name: "executing run stays with the runner",
ev: Evidence{Run: run(core.RunRunning)},
state: StateRunning, role: RoleRunner, action: ActionNone,
waitHint: "executing", since: baseTime.Format(time.RFC3339),
},
{
name: "pending run reads as running",
ev: Evidence{Run: run(core.RunPending)},
state: StateRunning, role: RoleRunner, action: ActionNone,
waitHint: "executing",
},
{
name: "cancelled has no mapping and degrades to unknown",
ev: Evidence{Run: ended(run(core.RunCancelled), afterRun)},
state: StateUnknown, role: RoleNone, action: ActionNone,
waitHint: "no diagnostic mapping",
},
{
name: "interrupted without any hold record degrades to unknown",
ev: Evidence{Run: run(core.RunInterrupted)},
state: StateUnknown, role: RoleNone, action: ActionNone,
waitHint: "no diagnostic mapping",
},
{
name: "unreadable hold is fail-closed unknown, never a safe retry",
ev: Evidence{Run: run(core.RunFailed), HoldUnknown: true},
state: StateUnknown, role: RoleNone, action: ActionNone,
waitHint: "could not be read",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
rep := Build(tc.ev)
if rep.Responsibility.State != tc.state {
t.Fatalf("state=%q want %q (responsibility=%+v)", rep.Responsibility.State, tc.state, rep.Responsibility)
}
if rep.Responsibility.Role != tc.role {
t.Fatalf("role=%q want %q", rep.Responsibility.Role, tc.role)
}
if rep.NextStep.Action != tc.action {
t.Fatalf("next_step=%q want %q (reason=%q)", rep.NextStep.Action, tc.action, rep.NextStep.Reason)
}
if rep.NextStep.CardID != tc.cardID {
t.Fatalf("next_step.card_id=%q want %q", rep.NextStep.CardID, tc.cardID)
}
if tc.waitHint != "" && !strings.Contains(rep.Responsibility.WaitReason, tc.waitHint) {
t.Fatalf("wait_reason=%q does not contain %q", rep.Responsibility.WaitReason, tc.waitHint)
}
if tc.since != "" && rep.Responsibility.Since != tc.since {
t.Fatalf("since=%q want %q", rep.Responsibility.Since, tc.since)
}
if rep.NextStep.Reason == "" {
t.Fatal("next_step.reason must never be empty")
}
})
}
}
// 副作用聚合:命令回执对账记录并入;会话里更早的记录不算本次 run 的副作用。
func TestBuildSideEffectsAggregateRecordedOperations(t *testing.T) {
ev := Evidence{
Run: run(core.RunFailed),
Audits: []agentcommand.AuditEnvelope{
{Time: baseTime.Add(-time.Hour), CommandID: "rdm.reqs.add", OperationID: "op_older",
State: agentcommand.StateCommitted, ConversationID: "cnv_1"},
{Time: afterRun, CommandID: "rdm.docs.instantiate", OperationID: "op_1",
State: agentcommand.StateUnknown, ConversationID: "cnv_1"},
{Time: afterRun, CommandID: "other.run.command", OperationID: "op_2",
State: agentcommand.StateCommitted, ConversationID: "cnv_other"},
{Time: afterRun, CommandID: "rdm.trace.confirm", OperationID: "op_3",
State: agentcommand.StateAccepted, ConversationID: "cnv_1", RunID: "run_1"},
},
}
rep := Build(ev)
if len(rep.SideEffects) != 2 {
t.Fatalf("side effects must hold only this run's receipts: %+v", rep.SideEffects)
}
got := map[string]SideEffect{}
for _, se := range rep.SideEffects {
got[se.OperationID] = se
}
recorded, ok := got["op_1"]
if !ok || recorded.Kind != EffectOperation || recorded.CommandID != "rdm.docs.instantiate" || recorded.State != string(agentcommand.StateUnknown) {
t.Fatalf("recorded operation missing from aggregation: %+v", rep.SideEffects)
}
if _, ok := got["op_older"]; ok {
t.Fatal("a receipt issued before the run started must not be attributed to it")
}
if _, ok := got["op_2"]; ok {
t.Fatal("a receipt from another conversation must not be attributed to this run")
}
if _, ok := got["op_3"]; !ok {
t.Fatal("a receipt carrying this run id must be attributed to it")
}
}
// hold 行本身也是"副作用结果未知"的持久证据(U3b),并标出是否已解除。
func TestBuildSideEffectsIncludeRecoveryHoldEvidence(t *testing.T) {
ev := Evidence{Run: run(core.RunInterrupted), Hold: &store.RecoveryHold{RunID: "run_1", Owner: "alice",
Reason: "unknown", CreatedAt: afterRun.Format(time.RFC3339)}}
rep := Build(ev)
if len(rep.SideEffects) != 1 || rep.SideEffects[0].Kind != EffectRecoveryHold || rep.SideEffects[0].State != "held" {
t.Fatalf("open hold must show as held evidence: %+v", rep.SideEffects)
}
ev.Hold.ReleasedAt = afterRun.Add(time.Minute).Format(time.RFC3339)
rep = Build(ev)
if rep.SideEffects[0].State != "released" || rep.SideEffects[0].ReleasedAt == "" {
t.Fatalf("released hold must keep its release time: %+v", rep.SideEffects[0])
}
}
// 没有落盘检查点时给 null + note,绝不编一个;空副作用同样带 note。
func TestBuildNeverFabricatesCheckpointOrAbsence(t *testing.T) {
rep := Build(Evidence{Run: run(core.RunRunning)})
if rep.LastCheckpoint != nil {
t.Fatalf("checkpoint must stay null when nothing is persisted: %+v", rep.LastCheckpoint)
}
if !containsNote(rep.Notes, "checkpoint") {
t.Fatalf("missing checkpoint note: %v", rep.Notes)
}
if !containsNote(rep.Notes, "absence of records is not proof") {
t.Fatalf("missing side effect absence note: %v", rep.Notes)
}
if rep.SideEffects == nil {
t.Fatal("side_effects must serialize as an array, never null")
}
withCheckpoint := Build(Evidence{Run: run(core.RunRunning), Checkpoint: checkpoint()})
if withCheckpoint.LastCheckpoint == nil || withCheckpoint.LastCheckpoint.ID != "7" || withCheckpoint.Run.UpdatedAt == nil {
t.Fatalf("persisted checkpoint must be surfaced: %+v", withCheckpoint)
}
if *withCheckpoint.Run.UpdatedAt != checkpoint().TS {
t.Fatalf("updated_at must fall back to the last persisted event: %v", *withCheckpoint.Run.UpdatedAt)
}
}
// Gather:只读端口的读齐与降级(读不到隔离记录按 fail-closed)。
func TestGatherReadsEvidenceAndDegradesOnError(t *testing.T) {
src := fakeSource{
hold: store.RecoveryHold{RunID: "run_1", Owner: "alice", Reason: "unknown", CreatedAt: afterRun.Format(time.RFC3339)},
cards: []core.HITLCard{{ID: "card_hitl"}},
cp: store.RunCheckpoint{Seq: 3, TS: afterRun.Format(time.RFC3339), Type: "agent.progress"},
cpOK: true,
audits: []agentcommand.AuditEnvelope{{Time: afterRun, CommandID: "rdm.confirm", State: agentcommand.StateAccepted}},
confirms: []store.PendingConfirm{{CardID: "card_cmd", Owner: "alice", CreatedAt: afterRun.Format(time.RFC3339), Payload: "{}"}},
}
ev := Gather(src, run(core.RunRunning), afterRun.Add(time.Minute))
if ev.Hold == nil || ev.HoldUnknown || len(ev.Cards) != 1 || ev.Checkpoint == nil || len(ev.Audits) != 1 || len(ev.Confirms) != 1 {
t.Fatalf("gather missed evidence: %+v", ev)
}
if rep := Build(ev); rep.Responsibility.State != StateRecoveryHold {
t.Fatalf("state=%q want %q", rep.Responsibility.State, StateRecoveryHold)
}
degraded := Gather(fakeSource{holdErr: errors.New("db gone")}, run(core.RunRunning), afterRun)
if !degraded.HoldUnknown || degraded.Hold != nil {
t.Fatalf("an unreadable hold must stay unknown, never absent: %+v", degraded)
}
rep := Build(degraded)
if rep.Responsibility.State != StateUnknown || rep.NextStep.Action == ActionSafeRetry {
t.Fatalf("unreadable hold must not yield a retry: %+v", rep)
}
if !containsNote(rep.Notes, "fail-closed") && !containsNote(rep.Notes, "quarantined") {
t.Fatalf("missing fail-closed note: %v", rep.Notes)
}
// 没有隔离行是正常态:不产生噪音 note。
normal := Build(Gather(fakeSource{}, run(core.RunRunning), afterRun))
if containsNote(normal.Notes, "recovery hold unreadable") {
t.Fatalf("a missing hold row is not a degraded read: %v", normal.Notes)
}
}
// 契约形状:snake_case 字段名固定,last_checkpoint 为 null 时显式出现。
func TestReportJSONContract(t *testing.T) {
raw, err := json.Marshal(Build(Evidence{Run: run(core.RunRunning), Checkpoint: checkpoint()}))
if err != nil {
t.Fatal(err)
}
var out map[string]any
if err := json.Unmarshal(raw, &out); err != nil {
t.Fatal(err)
}
for _, key := range []string{"run", "responsibility", "last_checkpoint", "side_effects", "next_step"} {
if _, ok := out[key]; !ok {
t.Fatalf("missing contract key %q in %s", key, raw)
}
}
responsibility, _ := out["responsibility"].(map[string]any)
for _, key := range []string{"state", "role", "wait_reason", "since"} {
if _, ok := responsibility[key]; !ok {
t.Fatalf("missing responsibility key %q in %s", key, raw)
}
}
runSummary, _ := out["run"].(map[string]any)
for _, key := range []string{"id", "status", "owner", "project", "conversation", "created_at", "updated_at"} {
if _, ok := runSummary[key]; !ok {
t.Fatalf("missing run key %q in %s", key, raw)
}
}
nextStep, _ := out["next_step"].(map[string]any)
for _, key := range []string{"action", "reason"} {
if _, ok := nextStep[key]; !ok {
t.Fatalf("missing next_step key %q in %s", key, raw)
}
}
empty, err := json.Marshal(Build(Evidence{Run: run(core.RunRunning)}))
if err != nil {
t.Fatal(err)
}
var nulled map[string]any
if err := json.Unmarshal(empty, &nulled); err != nil {
t.Fatal(err)
}
if nulled["last_checkpoint"] != nil {
t.Fatalf("last_checkpoint must be null (not omitted) when nothing is persisted: %s", empty)
}
if _, ok := nulled["side_effects"].([]any); !ok {
t.Fatalf("side_effects must be an array: %s", empty)
}
}
// 归属边界:会话里更早签发的决策卡与确认行不算本次 run 在等的东西。
func TestBuildIgnoresEvidenceIssuedBeforeTheRun(t *testing.T) {
ev := Evidence{
Run: run(core.RunRunning),
Decisions: []store.HumanDecision{{ID: "hd_old", Status: "pending", Question: "older?",
IssuedAt: baseTime.Add(-time.Hour), ExpiresAt: baseTime.Add(time.Hour)}},
Confirms: []store.PendingConfirm{{CardID: "card_old", Owner: "alice",
CreatedAt: baseTime.Add(-time.Hour).Format(time.RFC3339),
Payload: mustRequest(t, cmdbus.Request{ID: "rdm.confirm", Owner: "alice", ConversationID: "cnv_1"})}},
}
rep := Build(ev)
if rep.Responsibility.State != StateRunning {
t.Fatalf("state=%q want %q: earlier conversation evidence must not hold this run", rep.Responsibility.State, StateRunning)
}
}
// 只有 run 状态写着 waiting_hitl、没有可解析的卡时:状态仍然成立,但必须说清楚
// 等待起点没被存储。
func TestBuildStatusDrivenWaitHasNoCardAndSaysSo(t *testing.T) {
rep := Build(Evidence{Run: run(core.RunWaitingHITL)})
if rep.Responsibility.State != StateWaitingHITL || rep.NextStep.Action != ActionConfirmCard {
t.Fatalf("status-driven wait: %+v", rep.Responsibility)
}
if rep.NextStep.CardID != "" {
t.Fatalf("an unresolvable card must not be invented: %q", rep.NextStep.CardID)
}
if rep.Responsibility.Since != "" {
t.Fatalf("unknown wait entry time must stay empty: %q", rep.Responsibility.Since)
}
if !containsNote(rep.Notes, "wait entry time is not stored") {
t.Fatalf("missing status-driven wait note: %v", rep.Notes)
}
}
func mustRequest(t *testing.T, req cmdbus.Request) string {
t.Helper()
raw, err := json.Marshal(req)
if err != nil {
t.Fatal(err)
}
return string(raw)
}
func containsNote(notes []string, want string) bool {
for _, n := range notes {
if strings.Contains(n, want) {
return true
}
}
return false
}
+3 -2
View File
@@ -361,8 +361,9 @@ func TestIntentCancelRecordIsNotACancelTarget(t *testing.T) {
}
waitRunStatus(t, h, runID, string(core.RunCancelled))
gate.script("再取消一次", taskCancelIntent(nil))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "再取消一次"})
// 第二轮是占槽位的澄清回合:等上一轮被取消的运行彻底收工再发车(忙 409 是
// 既有语义,见 postConversationTurn)。
code, out := postConversationTurn(t, srv, h, st, convID, "再取消一次")
if code != 200 {
t.Fatalf("第二轮取消:%d %v", code, out)
}
+5 -21
View File
@@ -132,11 +132,7 @@ func (s *Server) fuseIntentSession(key intentsession.Key,
}
answer := answerFromRecognition(pending, recognized)
if answer.Cancel {
outcome, cerr := s.applyCancel(svc, key, pending, answer)
if cerr != nil {
return recognized, nil, cerr
}
return recognized, outcome, nil
return recognized, s.applyCancel(svc, key, pending, answer), nil
}
// 本轮对同一动作已是一次完整的新识别:它就是新的指令,按它派发;
// 旧累计槽位不混进新任务(会话就此作废)。
@@ -166,23 +162,11 @@ func (s *Server) fuseIntentSession(key intentsession.Key,
}
// applyCancel 消费库层 cancel:清空挂起预期并如实确认取消(不派发业务动作)。
//
// fail-closed:除 ErrStale(会话已被推进/取消,按「没有会话语义」如实处理)
// 之外的任何存储错误一律向上传播。绝不把「清不掉挂起预期」当成「本轮没有
// 会话语义」——那一行里的运行号仍会被 sessionRunTarget 解成取消靶子,一次
// 未确认的「放弃补充」就会变成一次真的取消。
func (s *Server) applyCancel(svc IntentSessions, key intentsession.Key,
pending *intentsession.Session, answer intentsession.Answer) (*intentSessionOutcome, error) {
pending *intentsession.Session, answer intentsession.Answer) *intentSessionOutcome {
out, err := svc.ApplyAnswer(key, pending.Revision, answer)
if errors.Is(err, intentsession.ErrStale) {
// 陈旧取消作废:如实按「没有会话语义」处理。
return nil, nil
}
if err != nil {
return nil, err
}
if out == nil || !out.Cancelled {
return nil, nil
if errors.Is(err, intentsession.ErrStale) || err != nil || out == nil || !out.Cancelled {
return nil
}
action, intent := "", pending.Expectation.IntentID
if pending.Expectation.Action.ID != "" {
@@ -192,7 +176,7 @@ func (s *Server) applyCancel(svc IntentSessions, key intentsession.Key,
cancelled: true,
note: &intentSessionNote{Kind: "cancelled", Action: action, Intent: intent, Pending: false,
Reply: "已取消本次槽位补充,未执行任何动作。"},
}, nil
}
}
// answerFromRecognition 把一次识别投影成会话语义的答案:库层会话命令
+121 -178
View File
@@ -3,7 +3,7 @@ package server
import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"sort"
"strings"
@@ -115,6 +115,105 @@ func settle(t *testing.T, h http.Handler, out map[string]any) map[string]any {
return out
}
// 会话忙(409)是既有语义,不改:占槽位的回合(普通回合、澄清回合)在忙时如
// 实 409,意图命中的命令则不占槽位、会话忙时也能用。测试侧要解决的是「等发得
// 出去」,而不是放宽断言:上一轮的 run 行在结算提交那一刻就终态了,槽位要到收
// 尾(补锚点、发终态事件、摘执行句柄)之后才释放,机器繁忙时这段窗口会被放大,
// 于是「run 一终态就立刻发下一轮」会偶发撞上忙 409。
// postConversationTurn 发一轮消息:先等会话空闲(见 waitConversationIdle),
// 发车时若仍撞上忙(409)则在「忙拒没有留下痕迹」的前提下按 deadline 重发。
// 忙不写消息与运行,但抢槽位发生在意图会话折叠之后——被忙拒掉的轮次仍可能已经
// 消费或挂起了会话,再发一次走的就是另一条路径了;所以每次重发前先比对意图会
// 话状态:变过了就如实失败,没变过才重发。
func postConversationTurn(t *testing.T, srv *Server, h http.Handler, st *store.Store,
convID, text string) (int, map[string]any) {
t.Helper()
waitConversationIdle(t, srv, h, convID)
deadline := time.Now().Add(testRunTerminalTimeout)
for {
before := intentSessionTrace(t, st, convID)
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": text})
if code != http.StatusConflict {
return code, out
}
if after := intentSessionTrace(t, st, convID); after != before {
t.Fatalf("%s:忙(409)之前已改动意图会话,重发会走另一条路径:%s → %s", text, before, after)
}
if !time.Now().Before(deadline) {
t.Fatalf("%s:会话持续忙(409)超过 %v:%v", text, testRunTerminalTimeout, out)
}
time.Sleep(20 * time.Millisecond)
}
}
// waitConversationIdle 等会话真正让出槽位:既没有未终态运行,也没有仍在本地
// 执行的 run——执行句柄在槽位释放之后才摘除(runmgr 的 execute 收尾),所以
// 「没有活跃句柄」就是槽位已空的可观测证据,不用赌收尾窗口有多宽。
func waitConversationIdle(t *testing.T, srv *Server, h http.Handler, convID string) {
t.Helper()
deadline := time.Now().Add(testRunTerminalTimeout)
for time.Now().Before(deadline) {
if code, out := doJSON(t, h, "GET", "/api/conversations/"+convID, nil); code == 200 &&
!conversationHasOpenRun(out) && !conversationHasActiveRun(srv, out) {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("会话 %s 在 %v 内未空闲", convID, testRunTerminalTimeout)
}
// conversationHasOpenRun 看会话里是否还有未终态运行(结算尚未提交)。
func conversationHasOpenRun(detail map[string]any) bool {
for _, run := range conversationRuns(detail) {
switch run["status"] {
case string(core.RunCompleted), string(core.RunFailed),
string(core.RunCancelled), string(core.RunInterrupted):
default:
return true
}
}
return false
}
// conversationHasActiveRun 看会话里是否还有本地在执行的 run(槽位未让出)。
func conversationHasActiveRun(srv *Server, detail map[string]any) bool {
for _, run := range conversationRuns(detail) {
if id, _ := run["id"].(string); id != "" && srv.mgr.RunActiveLocally(id) {
return true
}
}
return false
}
func conversationRuns(detail map[string]any) []map[string]any {
raw, _ := detail["runs"].([]any)
runs := make([]map[string]any, 0, len(raw))
for _, item := range raw {
if run, ok := item.(map[string]any); ok {
runs = append(runs, run)
}
}
return runs
}
// intentSessionTrace 描述会话上当前的意图会话状态(忙拒前后各取一次比对)。
func intentSessionTrace(t *testing.T, st *store.Store, convID string) string {
t.Helper()
conv, err := st.GetConversation(store.OwnerDefault, convID)
if err != nil {
t.Fatal(err)
}
sess, err := st.GetIntentSession(store.OwnerDefault, string(conv.ProjectID), convID)
if err == store.ErrNotFound {
return "none"
}
if err != nil {
t.Fatal(err)
}
return fmt.Sprintf("rev=%d exp=%v slots=%v", sess.Revision, sess.Expectation, sess.Slots)
}
// 缺必填槽位 → 澄清并挂起;下一轮补齐 → 按同一条准入与派发路径执行。
func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
srv, st, gate, log := cancelServer(t)
@@ -124,7 +223,7 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: runIDQ}, {Key: "reason", Question: "取消原因?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
@@ -147,7 +246,7 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
// 下一轮:仅补齐槽位(库层挂起预期命中分支的同形结果)。
answerText := "编号:R-7 原因:误触发"
gate.script(answerText, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answerText})
code, out = postConversationTurn(t, srv, h, st, convID, answerText)
if code != 200 || out["command"] != "demo.run-cancel" {
t.Fatalf("补齐后应派发同一动作:%d %v", code, out)
}
@@ -169,12 +268,12 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
// 两轮部分填充:先 set_slot 落一个槽位(仍缺 → 重新澄清),再由回答补齐。
func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
srv, _, gate, log := cancelServer(t)
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
@@ -183,7 +282,7 @@ func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
// 第二轮:库层 set_slot(只带一个槽位),仍缺原因 → 如实重新澄清。
gate.script("编号 R-7", intentgate.MatchResult{Branch: "expectation_hit", ID: "demo.run-cancel",
Commands: []intentgate.Command{{Kind: intentgate.CommandSetSlot, Name: "run_id", Value: "R-7"}}})
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号 R-7"})
code, out = postConversationTurn(t, srv, h, st, convID, "编号 R-7")
if code != 200 {
t.Fatalf("set_slot 轮:%d %v", code, out)
}
@@ -205,7 +304,7 @@ func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
// 第三轮:补上原因 → 派发,两个槽位都来自会话累计。
gate.script("原因:误触发", cancelAnswer(map[string]string{"reason": "误触发"}, nil))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "原因:误触发"})
code, out = postConversationTurn(t, srv, h, st, convID, "原因:误触发")
if code != 200 || out["command"] != "demo.run-cancel" {
t.Fatalf("补齐后应派发:%d %v", code, out)
}
@@ -222,7 +321,7 @@ func TestIntentSessionCancelClearsPendingAction(t *testing.T) {
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
@@ -230,7 +329,7 @@ func TestIntentSessionCancelClearsPendingAction(t *testing.T) {
gate.script("算了不取消了", intentgate.MatchResult{Branch: "matched", ID: "demo.run-cancel",
Commands: []intentgate.Command{{Kind: intentgate.CommandCancel, Target: "demo.run-cancel"}}})
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "算了不取消了"})
code, out = postConversationTurn(t, srv, h, st, convID, "算了不取消了")
if code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
@@ -255,7 +354,7 @@ func TestIntentSessionCancelClearsPendingAction(t *testing.T) {
// 取消之后的回答不得补齐旧动作(这里识别仍不完整 → 只澄清,不派发)。
gate.script("编号:R-7", cancelAnswer(map[string]string{"run_id": "R-7"},
[]intentgate.SlotQuestion{{Key: "reason", Question: "取消原因?"}}))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号:R-7"})
code, out = postConversationTurn(t, srv, h, st, convID, "编号:R-7")
if code != 200 {
t.Fatalf("取消后回答轮:%d %v", code, out)
}
@@ -277,8 +376,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
// 第一轮:抽到 run_id,仍缺原因 → 挂起(累计槽位含 run_id)。
gate.script("取消运行 编号:R-7", cancelStart(map[string]string{"run_id": "R-7"},
[]intentgate.SlotQuestion{{Key: "reason", Question: "取消原因?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消运行 编号:R-7"})
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行 编号:R-7")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
@@ -299,7 +397,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
// 第二轮只补了原因:过期的 run_id 不得被借来凑齐动作。
answerText := "原因:误触发"
gate.script(answerText, cancelStart(map[string]string{"reason": "误触发"}, nil))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answerText})
code, out = postConversationTurn(t, srv, h, st, convID, answerText)
if code != 200 {
t.Fatalf("过期后的回答轮:%d %v", code, out)
}
@@ -341,7 +439,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
t.Fatal(err)
}
gate.script("清理缓存", otherIntentStart("demo.purge"))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "清理缓存"})
code, out = postConversationTurn(t, srv, h, st, convID, "清理缓存")
if code != 200 || out["command"] != "demo.purge" {
t.Fatalf("新任务应照常派发:%d %v", code, out)
}
@@ -350,105 +448,6 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
}
}
// reportSpec 是第二个动作(生成报告):必填 scope,并声明了可选 run_id。
// 声明过的键即准入会映射的键,所以上一轮继承来的运行号会在派发参数里
// 现形——这正是回归要断言的那条泄漏路径。
var reportSpec = agentcommand.CommandSpec{ID: "demo.report", Title: "生成报告",
ParamsSchema: json.RawMessage(`{"type":"object","required":["text","scope"],
"properties":{"scope":{"type":"string"},"run_id":{"type":"string"}}}`)}
// reportStart 是第二个动作的一次识别(缺 scope → 路由层澄清 + 挂起)。
func reportStart(slots map[string]string, missing []intentgate.SlotQuestion) intentgate.MatchResult {
return intentgate.MatchResult{Branch: "matched", ID: "demo.report", Slots: slots, Missing: missing,
Commands: []intentgate.Command{{Kind: intentgate.CommandStart,
Action: intentgate.ActionRef{Kind: intentgate.ActionAuto, ID: "demo.report"}, Slots: slots}}}
}
// reportAnswer 是对第二个动作挂起预期的回答(库层 expectation_hit 同形)。
func reportAnswer(slots map[string]string) intentgate.MatchResult {
return intentgate.MatchResult{Branch: "expectation_hit", ID: "demo.report", Slots: slots}
}
// 换动作即换任务(跨动作槽位继承的回归序列):A 抽到运行号后仍缺原因而
// 挂起 → 用户改口做另一个动作 B(B 缺 scope)→ 用户补齐 B 的 scope →
// B 派发时只带 B 自己的槽位,A 的运行号绝不跟着过去。
func TestIntentSessionNewActionDoesNotInheritOldSlots(t *testing.T) {
srv, st, gate, log := cancelServer(t)
if err := srv.Bus.RegisterBuiltin(reportSpec, "", func(_ context.Context, req cmdbus.Request) (any, error) {
log.record(req)
return map[string]any{"ok": true}, nil
}); err != nil {
t.Fatal(err)
}
h := srv.Handler()
projectID, convID := setupConversation(t, h)
// T1:动作 A(demo.run-cancel)抽到 run_id,仍缺原因 → 挂起。
gate.script("取消运行 编号:R-7", cancelStart(map[string]string{"run_id": "R-7"},
[]intentgate.SlotQuestion{{Key: "reason", Question: "取消原因?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消运行 编号:R-7"})
if code != 200 {
t.Fatalf("A 轮:%d %v", code, out)
}
settle(t, h, out)
pending := pendingIntent(t, st, projectID, convID)
if pending == nil || pending.Expectation.Action.ID != "demo.run-cancel" || pending.Slots["run_id"] != "R-7" {
t.Fatalf("A 未挂起/未累计槽位:%+v", pending)
}
// T2:用户改口做动作 B(demo.report,缺 scope)→ 新预期,不继承 A 的槽位。
scopeQ := "哪个范围?"
gate.script("生成报告", reportStart(nil, []intentgate.SlotQuestion{{Key: "scope", Question: scopeQ}}))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "生成报告"})
if code != 200 {
t.Fatalf("B 轮:%d %v", code, out)
}
settle(t, h, out)
if _, executed := out["command"]; executed {
t.Fatalf("槽位未就绪不得派发:%v", out)
}
if c := clarifyOf(t, out); c["command"] != "demo.report" || c["question"] != scopeQ {
t.Fatalf("B 的澄清诉求:%v", c)
}
if len(log.snapshot()) != 0 {
t.Fatalf("澄清轮不得派发:%v", log.snapshot())
}
pending = pendingIntent(t, st, projectID, convID)
if pending == nil || pending.Expectation == nil || pending.Expectation.Action.ID != "demo.report" {
t.Fatalf("B 未挂起预期:%+v", pending)
}
if _, leaked := pending.Slots["run_id"]; leaked {
t.Fatalf("A 的槽位被 B 继承:%+v", pending.Slots)
}
// T3:用户补齐 B 的 scope → B 派发,参数里只有 B 的槽位(没有 A 的 run_id)。
answerText := "范围:华东"
gate.script(answerText, reportAnswer(map[string]string{"scope": "华东"}))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": answerText})
if code != 200 || out["command"] != "demo.report" {
t.Fatalf("补齐后应派发 B:%d %v", code, out)
}
if _, asked := out["clarify"]; asked {
t.Fatalf("补齐后不应再澄清:%v", out)
}
calls := log.snapshot()
if len(calls) != 1 {
t.Fatalf("派发次数:%d", len(calls))
}
if calls[0]["scope"] != "华东" || calls[0]["text"] != answerText {
t.Fatalf("B 的派发参数:%v", calls[0])
}
if _, leaked := calls[0]["run_id"]; leaked {
t.Fatalf("A 的运行号漏进了 B 的派发参数:%v", calls[0])
}
if pendingIntent(t, st, projectID, convID) != nil {
t.Fatal("补齐后会话应被消费")
}
}
// otherIntentStart 是另一个动作的完整识别(换任务)。
func otherIntentStart(id string) intentgate.MatchResult {
return intentgate.MatchResult{Branch: "matched", ID: id,
@@ -464,7 +463,7 @@ func TestIntentSessionIsolatesConversations(t *testing.T) {
projectB, convB := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convA+"/messages", map[string]any{"text": "取消运行"})
code, out := postConversationTurn(t, srv, h, st, convA, "取消运行")
if code != 200 {
t.Fatalf("A 会话第一轮:%d %v", code, out)
}
@@ -479,7 +478,7 @@ func TestIntentSessionIsolatesConversations(t *testing.T) {
// B 会话收到同样的回答:A 的槽位不得被借走(没有 A 的累计值)。
answer := "编号:R-7 原因:误触发"
gate.script(answer, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convB+"/messages", map[string]any{"text": answer})
code, out = postConversationTurn(t, srv, h, st, convB, answer)
if code != 200 {
t.Fatalf("B 会话回答轮:%d %v", code, out)
}
@@ -496,19 +495,19 @@ func TestIntentSessionIsolatesConversations(t *testing.T) {
func TestIntentSessionStillRequiresHighRiskConfirmation(t *testing.T) {
risky := cancelSpec
risky.Danger = agentcommand.DangerHigh
srv, _, gate, log := cancelServerWith(t, risky)
srv, st, gate, log := cancelServerWith(t, risky)
h := srv.Handler()
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
settle(t, h, out)
answer := "编号:R-7 原因:误触发"
gate.script(answer, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answer})
code, out = postConversationTurn(t, srv, h, st, convID, answer)
if code != 200 || out["command"] != "demo.run-cancel" {
t.Fatalf("补齐后应派发到总线:%d %v", code, out)
}
@@ -527,7 +526,7 @@ func TestIntentSessionInvalidAnswerKeepsClarifying(t *testing.T) {
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
@@ -536,7 +535,7 @@ func TestIntentSessionInvalidAnswerKeepsClarifying(t *testing.T) {
gate.script("编号:R-9", intentgate.MatchResult{Branch: "expectation_hit", ID: "demo.run-cancel",
Slots: map[string]string{"run_id": "R-9"},
Invalid: []intentgate.SlotDefect{{Key: "run_id", Reason: "unknown run", Question: "取消哪个运行?", Invalid: true}}})
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号:R-9"})
code, out = postConversationTurn(t, srv, h, st, convID, "编号:R-9")
if code != 200 {
t.Fatalf("非法补充轮:%d %v", code, out)
}
@@ -596,7 +595,7 @@ func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
}
run := func(t *testing.T, sessions IntentSessions) []string {
t.Helper()
srv, _, gate, log := cancelServer(t)
srv, st, gate, log := cancelServer(t)
if sessions != nil {
srv.SetIntentSessionService(sessions)
}
@@ -605,7 +604,7 @@ func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
_, convID := setupConversation(t, h)
var summaries []string
for _, text := range []string{"取消运行 编号:R-3 原因:误触发", "今天天气如何", "算了", "取消运行"} {
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": text})
code, out := postConversationTurn(t, srv, h, st, convID, text)
if code != 200 {
t.Fatalf("%s:%d %v", text, code, out)
}
@@ -637,62 +636,6 @@ func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
}
}
// brokenCancelStore 是「放弃补充时写库失败」的替身:其余方法照常委派给
// 真实会话服务,只有 cancel 的折叠返回非 ErrStale 的存储错误(真实写库
// 故障的注入点)。
type brokenCancelStore struct{ IntentSessions }
func (b brokenCancelStore) ApplyAnswer(key intentsession.Key, revision int,
answer intentsession.Answer) (*intentsession.Outcome, error) {
if answer.Cancel {
return nil, errors.New("clear intent session: injected store failure")
}
return b.IntentSessions.ApplyAnswer(key, revision, answer)
}
// 放弃补充时写库失败 → fail-closed:本轮报错,绝不降级成「没有会话语义」
// 去规划/派发一次取消(挂起行里的运行号仍在,会被解成取消靶子)。
func TestIntentSessionCancelFailsClosedOnStoreError(t *testing.T) {
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := liveRun(t, h, convID)
seedBoundRun(t, st, projectID, convID, runID)
srv.SetIntentSessionService(brokenCancelStore{IntentSessions: intentsession.New(st)})
// 放弃本次槽位补充(库层 cancel,指向正在挂起的 A)——写库故障就在此刻。
gate.script("算了不取消了", intentgate.MatchResult{Branch: "matched", ID: "demo.run-cancel",
Commands: []intentgate.Command{{Kind: intentgate.CommandCancel, Target: "demo.run-cancel"}}})
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "算了不取消了"})
if code == 200 {
t.Fatalf("写库失败的一轮不得照常应答:%d %v", code, out)
}
if _, asked := out["cancel"]; asked {
t.Fatalf("写库失败时不得回执一次取消:%v", out)
}
if _, ran := out["run"]; ran {
t.Fatalf("写库失败时不得起新回合:%v", out)
}
if len(log.snapshot()) != 0 {
t.Fatalf("写库失败时不得派发:%v", log.snapshot())
}
// 挂起行确实没被清掉(故障没被伪装成成功),这一轮因此更不能取消它。
if pendingIntent(t, st, projectID, convID) == nil {
t.Fatal("注入的写库失败不应真的清空挂起预期")
}
// 绝不能真的取消那个运行:给可能的误派发一点时间,再如实断言。
deadline := time.Now().Add(500 * time.Millisecond)
for time.Now().Before(deadline) {
if code, d := doJSON(t, h, "GET", "/api/runs/"+runID, nil); code == 200 {
if r, _ := d["run"].(map[string]any); r["status"] == string(core.RunCancelled) {
t.Fatalf("写库失败的一轮仍然取消了运行 %s:%v", runID, r)
}
}
time.Sleep(20 * time.Millisecond)
}
}
// inertSessions 是「关闭会话语义」的替身:任何调用都如实报告没有会话。
type inertSessions struct{}
+32
View File
@@ -0,0 +1,32 @@
package server
import (
"net/http"
"time"
"agent-system-backend/internal/rundiag"
)
// getRunDiagnostics 是单次 run 的只读诊断入口(feature unit A2):
//
// GET /api/runs/{id}/diagnostics
//
// 它回答值班/排障最常问的六个问题:这个 run 卡在哪(responsibility.state)、
// 谁该动(responsibility.role)、为什么在等(responsibility.wait_reason)、最后一个
// 稳定落盘点(last_checkpoint)、已知的外部副作用(side_effects)、下一步动作
// (next_step)。
//
// 授权口径与既有 run 读入口逐字同款:owner 只取自认证上下文,绝不接受请求参数;
// run 由 store.GetRun(owner, id) 做租户过滤,跨租户与不存在同为 404(不泄露存在性)。
// handler 路径上只有 SELECT——不给写操作、不改状态、不触发任何副作用;聚合逻辑在
// internal/rundiag,证据缺失一律降级为 unknown + notes,不编造检查点,也不把"没有
// 记录"读成"没有发生"。
func (s *Server) getRunDiagnostics(w http.ResponseWriter, r *http.Request) {
run, err := s.st.GetRun(s.owner(r), r.PathValue("id"))
if err != nil {
writeStoreErr(w, err)
return
}
evidence := rundiag.Gather(s.st, run, time.Now().UTC())
writeJSON(w, http.StatusOK, rundiag.Build(evidence))
}
+226
View File
@@ -0,0 +1,226 @@
package server
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// diagnosticsFixture 建一个 owner 名下的项目/会话/run(合成数据,无执行器)。
func diagnosticsFixture(t *testing.T, st *store.Store, owner string, status core.RunStatus) (core.Run, core.Conversation) {
t.Helper()
project, err := st.CreateProject(owner, "诊断项目")
if err != nil {
t.Fatal(err)
}
conversation, err := st.CreateConversation(owner, project.ID, "诊断会话", "")
if err != nil {
t.Fatal(err)
}
run := core.Run{
ConversationID: conversation.ID, ProjectID: project.ID, Owner: owner,
Kind: core.RunConversation, Status: status, Phase: core.PhaseActing, CreatedAt: time.Now().UTC(),
}
if err := st.CreateRun(&run); err != nil {
t.Fatal(err)
}
return run, conversation
}
func getDiagnostics(t *testing.T, h http.Handler, key, path string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, path, nil)
if key != "" {
req.Header.Set("Authorization", "Bearer "+key)
}
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
return rec
}
func decodeReport(t *testing.T, rec *httptest.ResponseRecorder) map[string]any {
t.Helper()
if rec.Code != http.StatusOK {
t.Fatalf("diagnostics: %d %s", rec.Code, rec.Body.String())
}
var out map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
t.Fatalf("decode diagnostics: %v (%s)", err, rec.Body.String())
}
return out
}
// 授权口径与既有 run 读入口同款:未认证 401、跨 owner 404(不泄露存在性)、
// 不存在的 run 同样 404。
func TestRunDiagnosticsOwnerScopingMatchesRunReads(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
aliceKey, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
bobKey, err := createLegacyAPIKeyFixture(t, st, "bob", "e2e")
if err != nil {
t.Fatal(err)
}
run, _ := diagnosticsFixture(t, st, "alice", core.RunRunning)
path := "/api/runs/" + string(run.ID) + "/diagnostics"
if rec := getDiagnostics(t, h, "", path); rec.Code != http.StatusUnauthorized {
t.Fatalf("unauthenticated diagnostics must be rejected: %d", rec.Code)
}
if rec := getDiagnostics(t, h, bobKey, path); rec.Code != http.StatusNotFound {
t.Fatalf("cross-owner diagnostics must not leak the payload: %d %s", rec.Code, rec.Body.String())
}
if rec := getDiagnostics(t, h, aliceKey, "/api/runs/run_missing/diagnostics"); rec.Code != http.StatusNotFound {
t.Fatalf("missing run must be 404: %d", rec.Code)
}
rec := getDiagnostics(t, h, aliceKey, path)
report := decodeReport(t, rec)
runSection, _ := report["run"].(map[string]any)
if runSection["id"] != string(run.ID) || runSection["owner"] != "alice" || runSection["project"] != string(run.ProjectID) {
t.Fatalf("run identity wrong: %v", runSection)
}
responsibility, _ := report["responsibility"].(map[string]any)
if responsibility["state"] != "running" {
t.Fatalf("a running run with no blockers reads as running: %v", responsibility)
}
}
// 隔离中的 run:责任在副作用责任人,下一步是解除隔离;检查点与已记录的命令
// 回执一并给出。
func TestRunDiagnosticsReportsHoldCheckpointAndSideEffects(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
key, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
run, conversation := diagnosticsFixture(t, st, "alice", core.RunRunning)
if _, err := st.AppendEvents(string(conversation.ID), []core.StreamEvent{
{RunID: run.ID, Type: core.EvThinkingDelta, Payload: json.RawMessage(`{"index":0,"text":"核对差异"}`)},
{RunID: run.ID, Type: core.EvAgentProgress, Payload: json.RawMessage(`{"note":"核对中"}`)},
}); err != nil {
t.Fatal(err)
}
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{
Time: time.Now().UTC(), Actor: "alice", CommandID: "rdm.docs.instantiate",
OperationID: "op_diag_1", ConversationID: string(conversation.ID), State: agentcommand.StateUnknown,
}); err != nil {
t.Fatal(err)
}
if err := st.SetRecoveryHold(string(run.ID), "alice", string(conversation.ID), "外部副作用结果未知"); err != nil {
t.Fatal(err)
}
report := decodeReport(t, getDiagnostics(t, h, key, "/api/runs/"+string(run.ID)+"/diagnostics"))
responsibility, _ := report["responsibility"].(map[string]any)
if responsibility["state"] != "recovery_hold" || responsibility["role"] != "operation-owner" {
t.Fatalf("held run responsibility: %v", responsibility)
}
nextStep, _ := report["next_step"].(map[string]any)
if nextStep["action"] != "release_recovery_hold" {
t.Fatalf("held run next step: %v", nextStep)
}
checkpoint, _ := report["last_checkpoint"].(map[string]any)
if checkpoint == nil || checkpoint["type"] != string(core.EvAgentProgress) {
t.Fatalf("last persisted checkpoint must be surfaced: %v", report["last_checkpoint"])
}
effects, _ := report["side_effects"].([]any)
var foundOperation bool
for _, raw := range effects {
effect, _ := raw.(map[string]any)
if effect["kind"] == "operation" && effect["operation_id"] == "op_diag_1" {
foundOperation = true
if effect["command_id"] != "rdm.docs.instantiate" || effect["state"] != string(agentcommand.StateUnknown) {
t.Fatalf("operation projection: %v", effect)
}
}
}
if !foundOperation {
t.Fatalf("recorded operation missing from side effects: %v", effects)
}
}
// 只读契约:GET 不改任何状态,写方法不注册。
func TestRunDiagnosticsIsReadOnly(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
key, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
run, conversation := diagnosticsFixture(t, st, "alice", core.RunRunning)
if err := st.SetRecoveryHold(string(run.ID), "alice", string(conversation.ID), "unknown"); err != nil {
t.Fatal(err)
}
before, err := st.GetRun("alice", string(run.ID))
if err != nil {
t.Fatal(err)
}
path := "/api/runs/" + string(run.ID) + "/diagnostics"
if rec := getDiagnostics(t, h, key, path); rec.Code != http.StatusOK {
t.Fatalf("diagnostics: %d %s", rec.Code, rec.Body.String())
}
after, err := st.GetRun("alice", string(run.ID))
if err != nil {
t.Fatal(err)
}
if before.Status != after.Status || before.Phase != after.Phase || !before.CreatedAt.Equal(after.CreatedAt) {
t.Fatalf("read-only endpoint mutated the run: %+v -> %+v", before, after)
}
hold, err := st.GetRecoveryHold(string(run.ID))
if err != nil || hold.ReleasedAt != "" {
t.Fatalf("read-only endpoint changed the recovery hold: %+v err=%v", hold, err)
}
for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete, http.MethodPatch} {
req := httptest.NewRequest(method, path, nil)
req.Header.Set("Authorization", "Bearer "+key)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code == http.StatusOK || rec.Code == http.StatusAccepted {
t.Fatalf("%s must not be served on a read-only diagnostics path: %d", method, rec.Code)
}
}
}
// 读不到的证据降级为 unknown + note,不让整个诊断失败,也不给出安全重试。
func TestRunDiagnosticsUnknownStateDegradesWithNote(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
key, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
run, _ := diagnosticsFixture(t, st, "alice", core.RunCancelled)
report := decodeReport(t, getDiagnostics(t, h, key, "/api/runs/"+string(run.ID)+"/diagnostics"))
responsibility, _ := report["responsibility"].(map[string]any)
if responsibility["state"] != "unknown" {
t.Fatalf("unmapped status must degrade to unknown: %v", responsibility)
}
if report["last_checkpoint"] != nil {
t.Fatalf("no persisted event means no checkpoint reference: %v", report["last_checkpoint"])
}
notes, _ := report["notes"].([]any)
var sawStatusNote, sawEffectNote bool
for _, raw := range notes {
note, _ := raw.(string)
if strings.Contains(note, "no diagnostic mapping") {
sawStatusNote = true
}
if strings.Contains(note, "absence of records is not proof") {
sawEffectNote = true
}
}
if !sawStatusNote || !sawEffectNote {
t.Fatalf("missing degradation notes: %v", notes)
}
}
+2
View File
@@ -392,6 +392,8 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("DELETE /api/prompt-commands/{id}", s.deleteUserPromptCommand)
mux.HandleFunc("POST /api/runs/{id}/cancel", s.cancelRun)
mux.HandleFunc("GET /api/runs/{id}", s.getRun)
// 只读运行诊断(A2):卡在哪、谁负责、已知副作用、下一步;无写操作。
mux.HandleFunc("GET /api/runs/{id}/diagnostics", s.getRunDiagnostics)
// ---- HITL ----
mux.HandleFunc("POST /api/hitl/{id}/resolve", s.resolveHITL)
+110
View File
@@ -0,0 +1,110 @@
package store
import (
"database/sql"
"errors"
"time"
)
// 运行诊断的只读投影(feature unit A2)。本文件只放 SELECT:不加表、不加索引、
// 不动迁移,也不引入新的写语义。诊断面回答的是"已经落库的证据是什么",因此
// 每个查询都遵守同一条口径:查不到就是"没有记录",调用方不得把它读成"没发生"
// (与 /api/commands/operations 的对账查询同源,见 cmdbus.OperationQuery)。
// RunCheckpoint 是 run 事件流上最后一个已持久化事件的稳定引用。
//
// 诚实边界:runs 行本身没有"检查点"列——事件流(events)才是 run 唯一的持久
// 进度痕迹(续跑、SSE 重放都以它为准)。所以这里给的是"最后一次稳定落盘"的
// 定位点(seq 可作游标),不是推断出来的中间状态;run 没有任何事件时不存在这条
// 记录,调用方必须显式处理"没有"而不是编一个。
type RunCheckpoint struct {
Seq int64 // 会话事件序(可作重放游标)
TS string // RFC3339(写入方原样落库)
Type string // 事件类型
}
// LastRunCheckpoint 返回该 run 在会话事件流里的最后一条已持久化事件。
// 没有事件时 ok=false,err 为 nil。
func (s *Store) LastRunCheckpoint(conversationID, runID string) (RunCheckpoint, bool, error) {
if conversationID == "" || runID == "" {
return RunCheckpoint{}, false, nil
}
var cp RunCheckpoint
err := s.db.QueryRow(s.q(`SELECT seq, ts, type FROM events
WHERE conversation_id=? AND run_id=? ORDER BY seq DESC LIMIT 1`), conversationID, runID).
Scan(&cp.Seq, &cp.TS, &cp.Type)
if errors.Is(err, sql.ErrNoRows) {
return RunCheckpoint{}, false, nil
}
if err != nil {
return RunCheckpoint{}, false, err
}
return cp, true, nil
}
// PendingHumanDecisionsByConversation 列出会话内未过期、尚未裁决的人工决策卡
// (owner + 会话双谓词,跨租户天然不可见;过期卡按 PendingHumanDecisions 的同一
// 口径过滤)。诊断面据此判断 run 是否在等人工裁决。
//
// 口径说明:决策卡按命令与会话归属,表里没有 run 列——把它归到某个 run 是调用方
// 的时间窗推断,不是存储事实(见 rundiag 的归属边界注释)。
func (s *Store) PendingHumanDecisionsByConversation(owner, conversationID string, now time.Time) ([]HumanDecision, error) {
if owner == "" || conversationID == "" {
return []HumanDecision{}, nil
}
rows, err := s.db.Query(s.q(`SELECT id,plugin_id,audience,command_id,human_actor,owner,project_id,proposal_id,
proposal_revision,payload_hash,question,status,decision,proof_id,conversation_id,issued_at,expires_at,responded_at
FROM human_decisions WHERE owner=? AND conversation_id=? AND status='pending' ORDER BY issued_at`),
owner, conversationID)
if err != nil {
return nil, err
}
defer rows.Close()
out := []HumanDecision{}
for rows.Next() {
rec, err := scanHumanDecision(rows)
if err != nil {
return nil, err
}
if now.Before(rec.ExpiresAt) {
out = append(out, rec)
}
}
return out, rows.Err()
}
// PendingConfirm 是危险命令确认挂起态(pending_confirms)的只读行。
// Payload 是 cmdbus 落下的分发请求 JSON(确认后据此二次分发);诊断面只解析出
// 归属与命令标识,不回显原始载荷。
type PendingConfirm struct {
CardID string
Owner string
Payload string
CreatedAt string
}
// ListOwnerPendingConfirms 列出该租户名下尚未确认的危险命令挂起行(时间正序)。
// owner 为空视为无作用域:不返回任何行(纵深防御,认证层恒注入非空 owner)。
func (s *Store) ListOwnerPendingConfirms(owner string, limit int) ([]PendingConfirm, error) {
if owner == "" {
return []PendingConfirm{}, nil
}
if limit <= 0 || limit > 200 {
limit = 50
}
rows, err := s.db.Query(s.q(`SELECT card_id, owner, payload, created_at FROM pending_confirms
WHERE owner=? ORDER BY created_at LIMIT ?`), owner, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := []PendingConfirm{}
for rows.Next() {
var p PendingConfirm
if err := rows.Scan(&p.CardID, &p.Owner, &p.Payload, &p.CreatedAt); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
+120
View File
@@ -0,0 +1,120 @@
package store
import (
"encoding/json"
"testing"
"time"
core "agent-system-core"
)
// 运行诊断只读投影:检查点定位、HITL 决策卡的会话+租户作用域、确认挂起行的租户
// 作用域。口径与既有读入口一致:查不到就是"没有记录",不推断"没发生"。
func TestRunDiagnosticsReadHelpers(t *testing.T) {
st := openTestStore(t)
base := time.Now().UTC().Truncate(time.Second)
if err := st.EnsureUser("alice", ""); err != nil {
t.Fatal(err)
}
if err := st.EnsureUser("bob", ""); err != nil {
t.Fatal(err)
}
project, err := st.CreateProject("alice", "诊断项目")
if err != nil {
t.Fatal(err)
}
conversation, err := st.CreateConversation("alice", project.ID, "诊断会话", "")
if err != nil {
t.Fatal(err)
}
run := core.Run{ConversationID: conversation.ID, ProjectID: project.ID, Owner: "alice",
Kind: core.RunConversation, Status: core.RunRunning, CreatedAt: base}
if err := st.CreateRun(&run); err != nil {
t.Fatal(err)
}
other := core.Run{ConversationID: conversation.ID, ProjectID: project.ID, Owner: "alice",
Kind: core.RunSubagent, Status: core.RunRunning, CreatedAt: base}
if err := st.CreateRun(&other); err != nil {
t.Fatal(err)
}
// 检查点:只认本 run 的最后一条事件,且不跨 run 串味。
if cp, ok, err := st.LastRunCheckpoint(string(conversation.ID), string(run.ID)); err != nil || ok {
t.Fatalf("a run without events has no checkpoint: %+v ok=%v err=%v", cp, ok, err)
}
if _, err := st.AppendEvents(string(conversation.ID), []core.StreamEvent{
{RunID: run.ID, Type: core.EvThinkingDelta, TS: base.Format(time.RFC3339), Payload: json.RawMessage(`{}`)},
{RunID: other.ID, Type: core.EvToolOutput, TS: base.Format(time.RFC3339), Payload: json.RawMessage(`{}`)},
{RunID: run.ID, Type: core.EvAgentProgress, TS: base.Format(time.RFC3339), Payload: json.RawMessage(`{}`)},
}); err != nil {
t.Fatal(err)
}
cp, ok, err := st.LastRunCheckpoint(string(conversation.ID), string(run.ID))
if err != nil || !ok {
t.Fatalf("checkpoint read: ok=%v err=%v", ok, err)
}
if cp.Type != string(core.EvAgentProgress) || cp.TS == "" || cp.Seq <= 0 {
t.Fatalf("checkpoint must point at the newest persisted event of this run: %+v", cp)
}
if child, ok, err := st.LastRunCheckpoint(string(conversation.ID), string(other.ID)); err != nil || !ok || child.Type != string(core.EvToolOutput) {
t.Fatalf("child run checkpoint: %+v ok=%v err=%v", child, ok, err)
}
if cp, ok, err := st.LastRunCheckpoint("cnv_other", string(run.ID)); err != nil || ok {
t.Fatalf("another conversation must not expose this run's events: %+v ok=%v err=%v", cp, ok, err)
}
// HITL 决策卡:pending + 未过期才可见,且按 owner+会话双谓词隔离。
now := base.Add(time.Minute)
issue := func(id, owner, conversationID string, ttl time.Duration) {
t.Helper()
if _, err := st.IssueHumanDecision(HumanDecision{PluginID: "p", Audience: "a", CommandID: "rdm.confirm",
HumanActor: owner, Owner: owner, ProjectID: string(project.ID), ProposalID: id, ProposalRevision: 1,
PayloadHash: "h" + id, Question: "approve " + id + "?", ConversationID: conversationID},
now.Add(-time.Minute)); err != nil {
t.Fatal(err)
}
// 过期口径:把 expires_at 改到过去,模拟签发即过期的卡(走方言绑定)。
if ttl < 0 {
if _, err := st.DB().Exec(st.q(`UPDATE human_decisions SET expires_at=? WHERE proposal_id=?`),
now.Add(-time.Hour).Format(time.RFC3339Nano), id); err != nil {
t.Fatal(err)
}
}
}
issue("alive", "alice", string(conversation.ID), time.Hour)
issue("expired", "alice", string(conversation.ID), -time.Hour)
issue("otherowner", "bob", "cnv_bob", time.Hour)
pending, err := st.PendingHumanDecisionsByConversation("alice", string(conversation.ID), now)
if err != nil {
t.Fatal(err)
}
if len(pending) != 1 || pending[0].ProposalID != "alive" || pending[0].Status != "pending" {
t.Fatalf("only live pending cards in scope: %+v", pending)
}
if cross, err := st.PendingHumanDecisionsByConversation("bob", string(conversation.ID), now); err != nil || len(cross) != 0 {
t.Fatalf("cross-owner cards must stay invisible: %+v err=%v", cross, err)
}
if none, err := st.PendingHumanDecisionsByConversation("", string(conversation.ID), now); err != nil || len(none) != 0 {
t.Fatalf("an empty owner has no scope: %+v err=%v", none, err)
}
// 确认挂起行:按租户隔离,空 owner 无作用域。
if err := st.SavePendingConfirm("card_alice", "alice", `{"ID":"rdm.confirm"}`); err != nil {
t.Fatal(err)
}
if err := st.SavePendingConfirm("card_bob", "bob", `{"ID":"rdm.confirm"}`); err != nil {
t.Fatal(err)
}
confirms, err := st.ListOwnerPendingConfirms("alice", 0)
if err != nil {
t.Fatal(err)
}
if len(confirms) != 1 || confirms[0].CardID != "card_alice" || confirms[0].Owner != "alice" || confirms[0].Payload == "" || confirms[0].CreatedAt == "" {
t.Fatalf("owner-scoped pending confirms: %+v", confirms)
}
if none, err := st.ListOwnerPendingConfirms("", 0); err != nil || len(none) != 0 {
t.Fatalf("an empty owner has no scope: %+v err=%v", none, err)
}
}