Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
35c6995546 |
@@ -24,7 +24,7 @@ jobs:
|
||||
echo "SUBMODULE_TOKEN is not configured" >&2
|
||||
exit 1
|
||||
fi
|
||||
git config --global url."https://weichen:${SUBMODULE_TOKEN}@gitea.naitangbaobao.cn/".insteadOf "ssh://git@gitea.naitangbaobao.cn:52222/"
|
||||
git config --global url."https://weichen:${SUBMODULE_TOKEN}@git-home.u1680832.nyat.app:30319/".insteadOf "ssh://git@git-ssh.u1680832.nyat.app:31411/"
|
||||
git -c fetch.recurseSubmodules=false submodule update --init --recursive
|
||||
- name: Install Go 1.27.1
|
||||
run: |
|
||||
|
||||
@@ -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@gitea.naitangbaobao.cn:52222/weichen/"
|
||||
SSH = "ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/"
|
||||
# 审查过的白名单,独立于候选 .gitmodules:仅列出两个治理 gitlink。
|
||||
REPOSITORIES = {
|
||||
"deps/development-guidelines": "Development-Guidelines",
|
||||
|
||||
+10
-10
@@ -1,30 +1,30 @@
|
||||
[submodule "deps/agent-system-core"]
|
||||
path = deps/agent-system-core
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-System-Core.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-System-Core.git
|
||||
[submodule "deps/agent-native"]
|
||||
path = deps/agent-native
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Native.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Native.git
|
||||
[submodule "deps/agent-intent"]
|
||||
path = deps/agent-intent
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Intent.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Intent.git
|
||||
[submodule "deps/logging-system"]
|
||||
path = deps/logging-system
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Logging-System.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Logging-System.git
|
||||
[submodule "deps/error-code-system"]
|
||||
path = deps/error-code-system
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Error-Code-System.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Error-Code-System.git
|
||||
[submodule "deps/agent-squad"]
|
||||
path = deps/agent-squad
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Squad.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Squad.git
|
||||
[submodule "deps/agent-command"]
|
||||
path = deps/agent-command
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Command.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Command.git
|
||||
[submodule "deps/agent-document-renderer"]
|
||||
path = deps/agent-document-renderer
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Document-Renderer.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Document-Renderer.git
|
||||
[submodule "deps/development-guidelines"]
|
||||
path = deps/development-guidelines
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Development-Guidelines.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Guidelines.git
|
||||
[submodule "deps/development-manager"]
|
||||
path = deps/development-manager
|
||||
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Development-Manager.git
|
||||
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Manager.git
|
||||
|
||||
@@ -23,9 +23,9 @@
|
||||
历史文档与旧链接里的 `git-panel.u1680832.nyat.app:30319` 已失效(TLS 握手失败),
|
||||
见到一律按过期处理,不要再试。当前权威通道:
|
||||
|
||||
- **Web / API(HTTPS)**:`https://gitea.naitangbaobao.cn`(Gitea 1.27.3,
|
||||
- **Web / API(HTTPS)**:`https://git-home.u1680832.nyat.app:30319`(Gitea 1.27.3,
|
||||
API 前缀 `/api/v1`;偶发 broken pipe / 超时,重试即可)。
|
||||
- **Git over SSH**:`ssh://git@gitea.naitangbaobao.cn:52222/weichen/<Repo>.git`
|
||||
- **Git over SSH**:`ssh://git@git-ssh.u1680832.nyat.app:31411/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` 已有 `gitea.naitangbaobao.cn(端口 52222)` 主机条目,克隆/推送直接可用。
|
||||
`~/.ssh/config` 已有 `git-ssh.u1680832.nyat.app` 主机条目,克隆/推送直接可用。
|
||||
|
||||
## Agent 执行与新项目准入(强制)
|
||||
|
||||
|
||||
Vendored
+1
-1
Submodule deps/agent-command updated: b0b4f39959...fe7987dc03
Vendored
+1
-1
Submodule deps/agent-document-renderer updated: 40e5c671d1...3f259990b5
Vendored
+1
-1
Submodule deps/agent-intent updated: 70b39a9f70...765a7e7177
Vendored
+1
-1
Submodule deps/agent-native updated: 6cc0a86615...f2fb9ffa6d
Vendored
+1
-1
Submodule deps/agent-squad updated: 9bb77c1ab7...b6d87638cc
Vendored
+1
-1
Submodule deps/agent-system-core updated: 614dcbff4b...0cf03446c4
Vendored
+1
-1
Submodule deps/development-guidelines updated: 3ed982138c...2688fd2be6
Vendored
+1
-1
Submodule deps/development-manager updated: 24e8a5c8ba...67784f43eb
Vendored
+1
-1
Submodule deps/error-code-system updated: 00c03035b5...a3ee67bba2
Vendored
+1
-1
Submodule deps/logging-system updated: 7005493077...49b5d8be60
@@ -2,6 +2,6 @@
|
||||
|
||||
package docgen
|
||||
|
||||
const rendererRevision = "40e5c671d1ef786ceba62bf784bf3aa53c5da26d"
|
||||
const rendererRevision = "3f259990b5d43ec0d318f39d496816140ab559b5"
|
||||
|
||||
const RendererIdentity = "agent-document-renderer@" + rendererRevision
|
||||
|
||||
@@ -367,7 +367,7 @@ func TestPersistedControlledMetadataRoutesBeforeDefaultProfileLoad(t *testing.T)
|
||||
}
|
||||
|
||||
func TestRendererIdentityMatchesPublishedPin(t *testing.T) {
|
||||
const want = "agent-document-renderer@40e5c671d1ef786ceba62bf784bf3aa53c5da26d"
|
||||
const want = "agent-document-renderer@3f259990b5d43ec0d318f39d496816140ab559b5"
|
||||
if RendererIdentity != want {
|
||||
t.Fatalf("RendererIdentity = %q, want %q", RendererIdentity, want)
|
||||
}
|
||||
|
||||
@@ -185,12 +185,17 @@ 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 {
|
||||
if cur != nil && cur.Expectation != nil && sameExpectation(*cur.Expectation, expectation) {
|
||||
slots = mergeSlots(cur.Slots, slots)
|
||||
}
|
||||
var missing []intentgate.SlotQuestion
|
||||
@@ -408,6 +413,13 @@ 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,6 +108,88 @@ 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)
|
||||
|
||||
@@ -1,488 +0,0 @@
|
||||
// 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
|
||||
}
|
||||
@@ -1,440 +0,0 @@
|
||||
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
|
||||
}
|
||||
@@ -361,9 +361,8 @@ func TestIntentCancelRecordIsNotACancelTarget(t *testing.T) {
|
||||
}
|
||||
waitRunStatus(t, h, runID, string(core.RunCancelled))
|
||||
gate.script("再取消一次", taskCancelIntent(nil))
|
||||
// 第二轮是占槽位的澄清回合:等上一轮被取消的运行彻底收工再发车(忙 409 是
|
||||
// 既有语义,见 postConversationTurn)。
|
||||
code, out := postConversationTurn(t, srv, h, st, convID, "再取消一次")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
|
||||
map[string]any{"text": "再取消一次"})
|
||||
if code != 200 {
|
||||
t.Fatalf("第二轮取消:%d %v", code, out)
|
||||
}
|
||||
|
||||
@@ -132,7 +132,11 @@ func (s *Server) fuseIntentSession(key intentsession.Key,
|
||||
}
|
||||
answer := answerFromRecognition(pending, recognized)
|
||||
if answer.Cancel {
|
||||
return recognized, s.applyCancel(svc, key, pending, answer), nil
|
||||
outcome, cerr := s.applyCancel(svc, key, pending, answer)
|
||||
if cerr != nil {
|
||||
return recognized, nil, cerr
|
||||
}
|
||||
return recognized, outcome, nil
|
||||
}
|
||||
// 本轮对同一动作已是一次完整的新识别:它就是新的指令,按它派发;
|
||||
// 旧累计槽位不混进新任务(会话就此作废)。
|
||||
@@ -162,11 +166,23 @@ 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 {
|
||||
pending *intentsession.Session, answer intentsession.Answer) (*intentSessionOutcome, error) {
|
||||
out, err := svc.ApplyAnswer(key, pending.Revision, answer)
|
||||
if errors.Is(err, intentsession.ErrStale) || err != nil || out == nil || !out.Cancelled {
|
||||
return nil
|
||||
if errors.Is(err, intentsession.ErrStale) {
|
||||
// 陈旧取消作废:如实按「没有会话语义」处理。
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if out == nil || !out.Cancelled {
|
||||
return nil, nil
|
||||
}
|
||||
action, intent := "", pending.Expectation.IntentID
|
||||
if pending.Expectation.Action.ID != "" {
|
||||
@@ -176,7 +192,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 把一次识别投影成会话语义的答案:库层会话命令
|
||||
|
||||
@@ -3,7 +3,7 @@ package server
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"errors"
|
||||
"net/http"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -115,105 +115,6 @@ 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)
|
||||
@@ -223,7 +124,7 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
|
||||
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
|
||||
{Key: "run_id", Question: runIDQ}, {Key: "reason", Question: "取消原因?"}}))
|
||||
|
||||
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
|
||||
if code != 200 {
|
||||
t.Fatalf("第一轮:%d %v", code, out)
|
||||
}
|
||||
@@ -246,7 +147,7 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
|
||||
// 下一轮:仅补齐槽位(库层挂起预期命中分支的同形结果)。
|
||||
answerText := "编号:R-7 原因:误触发"
|
||||
gate.script(answerText, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
|
||||
code, out = postConversationTurn(t, srv, h, st, convID, answerText)
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answerText})
|
||||
if code != 200 || out["command"] != "demo.run-cancel" {
|
||||
t.Fatalf("补齐后应派发同一动作:%d %v", code, out)
|
||||
}
|
||||
@@ -268,12 +169,12 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
|
||||
|
||||
// 两轮部分填充:先 set_slot 落一个槽位(仍缺 → 重新澄清),再由回答补齐。
|
||||
func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
|
||||
srv, st, gate, log := cancelServer(t)
|
||||
srv, _, 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 := postConversationTurn(t, srv, h, st, convID, "取消运行")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
|
||||
if code != 200 {
|
||||
t.Fatalf("第一轮:%d %v", code, out)
|
||||
}
|
||||
@@ -282,7 +183,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 = postConversationTurn(t, srv, h, st, convID, "编号 R-7")
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号 R-7"})
|
||||
if code != 200 {
|
||||
t.Fatalf("set_slot 轮:%d %v", code, out)
|
||||
}
|
||||
@@ -304,7 +205,7 @@ func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
|
||||
|
||||
// 第三轮:补上原因 → 派发,两个槽位都来自会话累计。
|
||||
gate.script("原因:误触发", cancelAnswer(map[string]string{"reason": "误触发"}, nil))
|
||||
code, out = postConversationTurn(t, srv, h, st, convID, "原因:误触发")
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "原因:误触发"})
|
||||
if code != 200 || out["command"] != "demo.run-cancel" {
|
||||
t.Fatalf("补齐后应派发:%d %v", code, out)
|
||||
}
|
||||
@@ -321,7 +222,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 := postConversationTurn(t, srv, h, st, convID, "取消运行")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
|
||||
if code != 200 {
|
||||
t.Fatalf("第一轮:%d %v", code, out)
|
||||
}
|
||||
@@ -329,7 +230,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 = postConversationTurn(t, srv, h, st, convID, "算了不取消了")
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "算了不取消了"})
|
||||
if code != 200 {
|
||||
t.Fatalf("取消轮:%d %v", code, out)
|
||||
}
|
||||
@@ -354,7 +255,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 = postConversationTurn(t, srv, h, st, convID, "编号:R-7")
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号:R-7"})
|
||||
if code != 200 {
|
||||
t.Fatalf("取消后回答轮:%d %v", code, out)
|
||||
}
|
||||
@@ -376,7 +277,8 @@ 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 := postConversationTurn(t, srv, h, st, convID, "取消运行 编号:R-7")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
|
||||
map[string]any{"text": "取消运行 编号:R-7"})
|
||||
if code != 200 {
|
||||
t.Fatalf("第一轮:%d %v", code, out)
|
||||
}
|
||||
@@ -397,7 +299,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
|
||||
// 第二轮只补了原因:过期的 run_id 不得被借来凑齐动作。
|
||||
answerText := "原因:误触发"
|
||||
gate.script(answerText, cancelStart(map[string]string{"reason": "误触发"}, nil))
|
||||
code, out = postConversationTurn(t, srv, h, st, convID, answerText)
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answerText})
|
||||
if code != 200 {
|
||||
t.Fatalf("过期后的回答轮:%d %v", code, out)
|
||||
}
|
||||
@@ -439,7 +341,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
gate.script("清理缓存", otherIntentStart("demo.purge"))
|
||||
code, out = postConversationTurn(t, srv, h, st, convID, "清理缓存")
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "清理缓存"})
|
||||
if code != 200 || out["command"] != "demo.purge" {
|
||||
t.Fatalf("新任务应照常派发:%d %v", code, out)
|
||||
}
|
||||
@@ -448,6 +350,105 @@ 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,
|
||||
@@ -463,7 +464,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 := postConversationTurn(t, srv, h, st, convA, "取消运行")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convA+"/messages", map[string]any{"text": "取消运行"})
|
||||
if code != 200 {
|
||||
t.Fatalf("A 会话第一轮:%d %v", code, out)
|
||||
}
|
||||
@@ -478,7 +479,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 = postConversationTurn(t, srv, h, st, convB, answer)
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convB+"/messages", map[string]any{"text": answer})
|
||||
if code != 200 {
|
||||
t.Fatalf("B 会话回答轮:%d %v", code, out)
|
||||
}
|
||||
@@ -495,19 +496,19 @@ func TestIntentSessionIsolatesConversations(t *testing.T) {
|
||||
func TestIntentSessionStillRequiresHighRiskConfirmation(t *testing.T) {
|
||||
risky := cancelSpec
|
||||
risky.Danger = agentcommand.DangerHigh
|
||||
srv, st, gate, log := cancelServerWith(t, risky)
|
||||
srv, _, 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 := postConversationTurn(t, srv, h, st, convID, "取消运行")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
|
||||
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 = postConversationTurn(t, srv, h, st, convID, answer)
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answer})
|
||||
if code != 200 || out["command"] != "demo.run-cancel" {
|
||||
t.Fatalf("补齐后应派发到总线:%d %v", code, out)
|
||||
}
|
||||
@@ -526,7 +527,7 @@ func TestIntentSessionInvalidAnswerKeepsClarifying(t *testing.T) {
|
||||
_, convID := setupConversation(t, h)
|
||||
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
|
||||
{Key: "run_id", Question: "取消哪个运行?"}}))
|
||||
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
|
||||
if code != 200 {
|
||||
t.Fatalf("第一轮:%d %v", code, out)
|
||||
}
|
||||
@@ -535,7 +536,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 = postConversationTurn(t, srv, h, st, convID, "编号:R-9")
|
||||
code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号:R-9"})
|
||||
if code != 200 {
|
||||
t.Fatalf("非法补充轮:%d %v", code, out)
|
||||
}
|
||||
@@ -595,7 +596,7 @@ func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
|
||||
}
|
||||
run := func(t *testing.T, sessions IntentSessions) []string {
|
||||
t.Helper()
|
||||
srv, st, gate, log := cancelServer(t)
|
||||
srv, _, gate, log := cancelServer(t)
|
||||
if sessions != nil {
|
||||
srv.SetIntentSessionService(sessions)
|
||||
}
|
||||
@@ -604,7 +605,7 @@ func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
|
||||
_, convID := setupConversation(t, h)
|
||||
var summaries []string
|
||||
for _, text := range []string{"取消运行 编号:R-3 原因:误触发", "今天天气如何", "算了", "取消运行"} {
|
||||
code, out := postConversationTurn(t, srv, h, st, convID, text)
|
||||
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": text})
|
||||
if code != 200 {
|
||||
t.Fatalf("%s:%d %v", text, code, out)
|
||||
}
|
||||
@@ -636,6 +637,62 @@ 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{}
|
||||
|
||||
|
||||
@@ -1,32 +0,0 @@
|
||||
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))
|
||||
}
|
||||
@@ -1,226 +0,0 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -392,8 +392,6 @@ 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)
|
||||
|
||||
@@ -1,110 +0,0 @@
|
||||
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()
|
||||
}
|
||||
@@ -1,120 +0,0 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user