Compare commits

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

Evidence: host-verified; see Dev PR #706

Impact: integration commit; .gitmodules keeps migrated URLs.

Refs: Agent-System-Dev PR #706

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

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

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

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

Evidence: host-verified; see Dev PR #706

Impact: generated constant only.

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

Evidence: host-verified; see Dev PR #706

Impact: pin update only.

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

Evidence: host-verified; see Dev PR #706

Impact: pin updates only.

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

Evidence: host-verified; see Dev PR #706

Impact: pin updates only.

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

Evidence: host-verified; see Dev PR #706

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

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

Evidence: host-verified; see Dev PR #706

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

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

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

Impact: configuration only; no runtime behavior change.

Refs: Dev PR #706
2026-10-22 14:16:47 +00:00
agent-exec-transport 06b0c9f1db Add the read-only run diagnostics endpoint
Why: A2 - answer where a run is stuck, who is responsible, why it
waits, the last stable checkpoint, known side effects and the next
step, from existing records only (recovery holds, HITL decisions,
confirmation cards, audit envelopes).
Evidence: rundiag/server/store tests pass (verify-a2.json); per-state
responsibility, scoping and next-step mapping cases added.
Impact: one read-only GET endpoint plus read helpers; no behavior
change to execution or recovery paths.
Refs: feature/exec-recovery-diagnostics-a2
2026-10-22 13:03:53 +00:00
weichen 9ee752c380 Add intent_session to the storage-migrate manifest
Why: the A1 session table joined the schema but the backup tool's
copyOrder list lagged; the coverage guard caught it.
Evidence: storage-migrate tests pass (verify-a1sr.json, exit=0).
Impact: backup tool manifest only; existing table semantics unchanged.
Refs: feature/intent-multi-turn-a1
2026-10-22 10:41:58 +00:00
weichen c2e3104970 Route the cancel intent through cmdbus with explicit targets
Why: A1 - the cancel library command was declined as host-unconsumed,
so "取消刚才的任务" could not act on a real run.
Evidence: route/server/intentsession tests pass (verify-a1c.json);
explicit-id, session-bound, ambiguous and cross-owner cases added;
confirmation card semantics unchanged.
Impact: cancel decisions dispatch via the existing host path; other
intents unchanged.
Refs: feature/intent-multi-turn-a1
2026-10-22 10:04:23 +00:00
weichen 0fb64c488e Add conversation-scoped intent sessions for multi-turn closure
Why: A1 - the host declined set_slot/change_slot/cancel and never kept
pending expectations, so multi-turn clarifications could not complete
an action.
Evidence: intentsession/store/server tests pass (verify-a1b.json);
TTL, revision, isolation, expiry and legacy-behavior cases added.
Impact: new table + service; chat flow suspends/fills expectations;
cmdbus confirmation semantics untouched (U-A1c routes cancel next).
Refs: feature/intent-multi-turn-a1
2026-10-22 09:30:25 +00:00
weichen a75783fd2d Merge commit '7bcbbdb37684' into HEAD 2026-10-22 07:10:53 +00:00
weichen 92c2b4eb12 Pin the nested agent-squad to the followups tip
Why: composition consistency - Dev pins agent-squad at the nested
native M5 bump; the backend container's nested copy must match
(check-pins).
Evidence: pin only.
Impact: nested gitlink pointer change.
Refs: feature/audit-followups-pin-gate-m5
2026-10-22 06:48:37 +00:00
weichen 7bcbbdb376 Merge pull request 'Qualify the role permission version upsert and wait for the second HITL card' (#246) from fix/pg-role-policy-version-and-hitl-wait into main
governance / governance (push) Successful in 21s
ci / vet-and-test (push) Successful in 1m29s
2026-10-22 06:42:24 +00:00
weichenandCursor 8cd4dd0ab7 Qualify the role permission version upsert and wait for the second HITL card
governance / governance (pull_request) Successful in 16s
ci / vet-and-test (pull_request) Successful in 1m27s
Why: Dev composite test-pg failed: PostgreSQL rejects the unqualified
permission_version self-reference in the role_permissions upsert
(SQLSTATE 42702), and TestResolveHITLSecondRoundKeepsTerminalStatus read
the round-2 card before it was persisted.

Evidence: go test ./internal/store ./internal/runmgr pass and the HITL
test passes 20 consecutive runs locally on SQLite (SELF); PostgreSQL is
verified by the Dev composite test-pg job.

Impact: role_permissions.permission_version+1 on conflict; test-only
wait mirrors round 1. No other behavior change.

Refs: Agent-System-Backend#240, Agent-System-Dev#691 (test-pg run 4841);
agent-exec receipt u7-run (pi, success)

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-22 06:40:12 +00:00
weichen 3ab7cd71a4 Pin the nested agent-native to the M5 fix
Why: composition consistency - Dev pins agent-native at the M5
preflight tip; the backend container's nested copy must match
(check-pins).
Evidence: pin only; adapters/native tests green on the new pin.
Impact: nested gitlink pointer change.
Refs: feature/audit-followups-pin-gate-m5
2026-10-22 06:26:13 +00:00
weichen fd69932c7d Merge pull request 'Land agent-stack audit batch-2 on main' (#245) from fix/agent-stack-audit-b2 into main
governance / governance (push) Successful in 17s
ci / vet-and-test (push) Successful in 1m34s
2026-10-22 05:46:42 +00:00
weichen 443e7af1b8 Merge branch 'main' into fix/agent-stack-audit-b2
governance / governance (pull_request) Successful in 18s
ci / vet-and-test (pull_request) Successful in 1m22s
2026-10-22 05:44:47 +00:00
weichen 029917e843 Merge pull request 'Add isolated scratch sessions (no-project general-agent chats)' (#241) from feature/scratch-session-general-agent into main
governance / governance (push) Successful in 17s
ci / vet-and-test (push) Failing after 1m18s
2026-10-22 05:14:44 +00:00
weichen 9b40595728 Pin the nested agent-squad to the repaired batch-2 tip
governance / governance (pull_request) Successful in 21s
ci / vet-and-test (pull_request) Successful in 1m35s
Why: composition consistency - the squad container advanced with the
frame-order repair; backend's nested copy (and its nested native) must
match so the composed chain carries the new event fields.
Evidence: pin only; squad and native verified per-unit.
Impact: nested gitlink pointer changes.
Refs: feature/agent-stack-audit-remediation-b2
2026-10-22 04:18:21 +00:00
weichen 7b8a34d1aa Synchronize the shared tool-run state behind the direct path
Why: the U11C concurrent event plumbing exposed a data race between
runToolCtx writes and callback reads; -race failed the package on
fc0a050 but passed on its parent.
Evidence: full-package race run green (verify-u11cr.json, exit=0).
Impact: synchronization only; no behavior change.
Refs: feature/agent-stack-audit-remediation-b2 U11c repair
2026-10-22 04:17:55 +00:00
weichen fc0a0505b3 Honor tool-call identity and real status in squad runner events
Why: F14 backend half - tool events used the tool name as the call id
and hardcoded status ok, so same-named tools could not be paired and
failures displayed as success; parallel callbacks shared artifact state.
Evidence: squad runner event tests pass (verify-u11c.json); pairing,
error mapping and legacy passthrough cases added; artifact list locked.
Impact: legacy empty-field events keep byte-identical output.
Refs: feature/agent-stack-audit-remediation-b2 U11c
2026-10-22 03:49:37 +00:00
weichen f8e9e9f2d2 Pin nested agent-native and agent-squad to the batch-2 tips
Why: U11C consumes the new ProcessEvent CallID/Status/ErrorReason
fields; the backend container's nested copies must match the Dev pins
for U11c and check-pins.
Evidence: pin only; both libraries verified per-unit.
Impact: nested gitlink pointer changes.
Refs: feature/agent-stack-audit-remediation-b2
2026-10-22 03:25:54 +00:00
44 changed files with 5624 additions and 129 deletions
+1 -1
View File
@@ -24,7 +24,7 @@ jobs:
echo "SUBMODULE_TOKEN is not configured" >&2
exit 1
fi
git config --global url."https://weichen:${SUBMODULE_TOKEN}@git-home.u1680832.nyat.app:30319/".insteadOf "ssh://git@git-ssh.u1680832.nyat.app:31411/"
git config --global url."https://weichen:${SUBMODULE_TOKEN}@gitea.naitangbaobao.cn/".insteadOf "ssh://git@gitea.naitangbaobao.cn:52222/"
git -c fetch.recurseSubmodules=false submodule update --init --recursive
- name: Install Go 1.27.1
run: |
+1 -1
View File
@@ -31,7 +31,7 @@ jobs:
parsed_server = urlsplit(SERVER)
if parsed_server.scheme not in {"http", "https"} or not parsed_server.netloc:
sys.exit("Checkout server must be the platform-injected http(s) URL")
SSH = "ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/"
SSH = "ssh://git@gitea.naitangbaobao.cn:52222/weichen/"
# 审查过的白名单,独立于候选 .gitmodules:仅列出两个治理 gitlink。
REPOSITORIES = {
"deps/development-guidelines": "Development-Guidelines",
+10 -10
View File
@@ -1,30 +1,30 @@
[submodule "deps/agent-system-core"]
path = deps/agent-system-core
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-System-Core.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-System-Core.git
[submodule "deps/agent-native"]
path = deps/agent-native
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Native.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Native.git
[submodule "deps/agent-intent"]
path = deps/agent-intent
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Intent.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Intent.git
[submodule "deps/logging-system"]
path = deps/logging-system
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Logging-System.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Logging-System.git
[submodule "deps/error-code-system"]
path = deps/error-code-system
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Error-Code-System.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Error-Code-System.git
[submodule "deps/agent-squad"]
path = deps/agent-squad
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Squad.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Squad.git
[submodule "deps/agent-command"]
path = deps/agent-command
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Command.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Command.git
[submodule "deps/agent-document-renderer"]
path = deps/agent-document-renderer
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Agent-Document-Renderer.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Agent-Document-Renderer.git
[submodule "deps/development-guidelines"]
path = deps/development-guidelines
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Guidelines.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Development-Guidelines.git
[submodule "deps/development-manager"]
path = deps/development-manager
url = ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/Development-Manager.git
url = ssh://git@gitea.naitangbaobao.cn:52222/weichen/Development-Manager.git
+3 -3
View File
@@ -23,9 +23,9 @@
历史文档与旧链接里的 `git-panel.u1680832.nyat.app:30319` 已失效(TLS 握手失败),
见到一律按过期处理,不要再试。当前权威通道:
- **Web / API(HTTPS)**:`https://git-home.u1680832.nyat.app:30319`(Gitea 1.27.3,
- **Web / API(HTTPS)**:`https://gitea.naitangbaobao.cn`(Gitea 1.27.3,
API 前缀 `/api/v1`;偶发 broken pipe / 超时,重试即可)。
- **Git over SSH**:`ssh://git@git-ssh.u1680832.nyat.app:31411/weichen/<Repo>.git`
- **Git over SSH**:`ssh://git@gitea.naitangbaobao.cn:52222/weichen/<Repo>.git`
(各仓 origin 与 gitlink 均为此格式)。
- **API token**:`~/.secrets/gitea`,两行键值 `URL_PREFIX=…` 与
`ACCESS_TOKEN=<40 位 hex>`;取值 `grep ACCESS_TOKEN ~/.secrets/gitea | cut -d= -f2-`,
@@ -33,7 +33,7 @@
(baseline 硬边界 2:只记录来源、字段与可用状态)。
- **SSH key**:`~/.ssh/home_gitea`(ED25519,公钥指纹
`SHA256:GYXaOXIMeloQ1fKSfdJ/ndgPQzs/Ix7Hk62ekfkfQ20`,Gitea 侧密钥名 GMKtec);
`~/.ssh/config` 已有 `git-ssh.u1680832.nyat.app` 主机条目,克隆/推送直接可用。
`~/.ssh/config` 已有 `gitea.naitangbaobao.cn(端口 52222)` 主机条目,克隆/推送直接可用。
## Agent 执行与新项目准入(强制)
+4
View File
@@ -307,6 +307,10 @@ var copyOrder = []string{
"exec_env_requests", "exec_env_bindings",
// U8:活动服务锁的库内标记(无外键;迁移须保留以免目标库丢掉锁根)
"instance_guard",
// v52 intent_session(A1/U-A1b):会话维度的挂起动作预期 + 已收集槽位
// (无外键,纯数据表;键含 owner+project+conversation,随 conversations
// 之后迁移即可,无需额外排序)
"intent_session",
}
// schemaSeededSingletons:schema 迁移在每个新库上预置的单行表。它们不是目标
@@ -2,6 +2,6 @@
package docgen
const rendererRevision = "3f259990b5d43ec0d318f39d496816140ab559b5"
const rendererRevision = "40e5c671d1ef786ceba62bf784bf3aa53c5da26d"
const RendererIdentity = "agent-document-renderer@" + rendererRevision
+1 -1
View File
@@ -367,7 +367,7 @@ func TestPersistedControlledMetadataRoutesBeforeDefaultProfileLoad(t *testing.T)
}
func TestRendererIdentityMatchesPublishedPin(t *testing.T) {
const want = "agent-document-renderer@3f259990b5d43ec0d318f39d496816140ab559b5"
const want = "agent-document-renderer@40e5c671d1ef786ceba62bf784bf3aa53c5da26d"
if RendererIdentity != want {
t.Fatalf("RendererIdentity = %q, want %q", RendererIdentity, want)
}
+469
View File
@@ -0,0 +1,469 @@
// Package intentsession 是宿主侧的「会话内意图会话」服务(A1/U-A1b)。
//
// 它把一条可执行动作的预期(动作引用 + 仍缺的槽位 + 已收集的槽位)留在
// 会话里,等用户后续轮次把槽位补齐,而不是把同一次识别丢掉重来:
//
// Pending 读当前挂起预期(无会话/已过期即没有,过期行读侧惰性删除)
// Suspend 识别出「动作可执行但槽位未就绪」后挂起预期
// ApplyAnswer 把一轮用户回答折叠进会话、推进 revision、重新判定齐全性
// ApplyCommand 消费库层命令:set_slot/change_slot 改槽位、cancel 清预期
// Binding 读会话当前绑定的运行上下文(A1/U-A1c:取消意图的靶子来源)
// RecordCancel 记一次已派发的取消:清预期,留一条短期「已取消某运行」记录
//
// 纪律:
// - 会话语义只携带槽位与预期,绝不携带授权:高危动作的确认仍由 cmdbus
// 收口,会话补齐不会绕过任何确认门禁。
// - 键是 (owner, project, conversation):两个用户、两个项目的会话在构造上
// 互不可见。
// - revision 单调递增:陈旧回答(期间已被别的轮次推进或取消)一律拒绝
// (ErrStale),绝不允许旧槽位漏进新任务。
// - 库层 Stage-1 语义不变:本服务只消费库层产出的 set_slot/change_slot/
// cancel/expectation 命令,不改动识别分支语义。
// - 「已取消」记录不是待补齐的动作:它不再吸收槽位、不再被回答折叠,
// 只供紧接着的追问(如「确认取消」)如实回话(CancelRecordTTL 到期即
// 惰性删除)。
package intentsession
import (
"errors"
"strings"
"time"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/store"
)
// DefaultTTL 是挂起预期的默认存活时间(30 分钟;每次更新续期)。
const DefaultTTL = 30 * time.Minute
// CancelRecordTTL 是「已取消」记录的留存时间:只够紧接着的追问读到,
// 绝不跨会话长期驻留。
const CancelRecordTTL = 5 * time.Minute
// CancelActionKind 标记会话行的预期是「一次已派发的取消」记录,而不是待补齐
// 的动作。宿主据此不再把本行当挂起动作折叠,只把它当如实回话的依据。
const CancelActionKind = "cancel"
// cancelledRunSlot 是取消记录里被取消运行号所在的槽位键(宿主不透明)。
const cancelledRunSlot = "cancelled_run_id"
// runSlotKeys 是识别结果里承载运行号的槽位键(按优先级;固定序,结果确定)。
var runSlotKeys = []string{"run_id", "runId", "run"}
// ErrStale 表示调用方持有的 revision 已过期(期间会话已被推进或取消)。
// 调用方遇到它必须丢弃这次折叠,不得据此执行任何动作。
var ErrStale = errors.New("intentsession: stale session revision")
// SlotValidator 按动作声明的 schema 校验一个槽位值(宿主注入:命令总线侧的
// DECLARED schema)。返回错误即不采纳该值——非法值永远进不了会话槽位表。
// 动作不可解析(未注册/非命令)时宿主返回 nil 表示不校验。
type SlotValidator func(actionID, slot, value string) error
// Service 是意图会话服务。用 New 构造;零值不可用。
type Service struct {
st *store.Store
validate SlotValidator
clock func() time.Time
ttlValue time.Duration
}
// New 构造意图会话服务(挂起期默认 DefaultTTL,即每次更新后 30 分钟)。
func New(st *store.Store) *Service {
return &Service{st: st, clock: time.Now, ttlValue: DefaultTTL}
}
// SetSlotValidator 接线动作声明 schema 的槽位校验(启动期调用)。
func (s *Service) SetSlotValidator(v SlotValidator) { s.validate = v }
// SetTTL 覆盖挂起期(<=0 回落 DefaultTTL;仅启动期与测试期调用)。
func (s *Service) SetTTL(d time.Duration) {
if d <= 0 {
d = DefaultTTL
}
s.ttlValue = d
}
func (s *Service) ttl() time.Duration {
if s.ttlValue <= 0 {
return DefaultTTL
}
return s.ttlValue
}
func (s *Service) now() time.Time {
if s.clock == nil {
return time.Now()
}
return s.clock()
}
// Key 是一个会话的标识。Owner 与 Conversation 必填:owner 缺失即 fail-closed,
// 绝不会退化成跨用户共享的会话状态。
type Key struct {
Owner string
Project string
Conversation string
}
func (k Key) valid() bool { return k.Owner != "" && k.Conversation != "" }
// Expectation 是挂起的可执行动作预期。Action 是权威动作引用(命令/工作流/
// 小队/auto),Missing 是仍缺的槽位及建议提问。
type Expectation struct {
IntentID string
Action intentgate.ActionRef
Missing []intentgate.SlotQuestion
Question string
}
// Session 是一个会话当前的意图会话状态。
type Session struct {
Key Key
// Expectation 为 nil 表示没有挂起动作(已取消或已补齐)。
Expectation *Expectation
// Cancelled 非 nil 表示这一行是一次已派发的取消记录(不是待补齐动作)。
Cancelled *CancelRecord
Slots map[string]string
Revision int
ExpiresAt time.Time
}
// CancelRecord 是一次已派发的取消的短期记录:被取消的运行与其来源动作。
type CancelRecord struct {
RunID string
Intent string
Action string
}
// Binding 是会话当前绑定的运行上下文——「取消」意图的靶子来源(A1/U-A1c)。
//
// 它只声明「这个会话当时对着哪个运行」,不是授权:派发与确认仍在宿主侧的
// 既有取消路径上收口。
type Binding struct {
RunID string
Intent string
Action string
// Cancelled 为真时这条绑定来自一次已派发的取消(而不是待补齐动作):
// 宿主不应把它再当成可取消的靶子。
Cancelled bool
}
// Answer 是一轮用户回答折叠后的结果(由宿主从识别结果与库层命令投影)。
type Answer struct {
// Slots 是本轮补上的槽位;Invalid 是本轮给出但没通过权威校验的槽位
// (key→原因:值不采纳,仍按非法追问)。
Slots map[string]string
Invalid map[string]string
// Cancel 为真时本轮是取消:不折叠任何槽位,只清空挂起预期。
Cancel bool
}
// Outcome 是一次折叠的结果。nil 结果(error==nil)表示本轮没有任何会话
// 语义:宿主保持既有行为,不改响应。
type Outcome struct {
// Session 是折叠后的会话(取消或补齐被消费时为 nil)。
Session *Session
// Complete 表示动作已齐全:宿主可按与一次完整识别完全相同的准入与
// 派发路径执行(确认语义仍由 cmdbus 收口)。
Complete bool
// Cancelled 表示用户取消了挂起的补充(不得派发任何业务动作)。
Cancelled bool
// Slots 是累计后的全部槽位(Complete 时用于派发)。
Slots map[string]string
// Remaining/Invalid 是仍缺的槽位与本轮未通过的槽位(Complete=false 时
// 用于如实重新澄清)。
Remaining []intentgate.SlotQuestion
Invalid []intentgate.SlotDefect
Question string
}
// Pending 返回仍然有效的挂起预期。没有行、没有预期或已过期都返回 nil;
// 过期行在读侧惰性删除(过期会话绝不影响新的一轮)。
func (s *Service) Pending(key Key) (*Session, error) {
return s.load(key)
}
// Suspend 挂起一个可执行动作的预期并累计已抽到的槽位(续期)。已抽到的
// 槽位不再计入仍缺清单(挂起即如实反映还差什么)。
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 {
slots = mergeSlots(cur.Slots, slots)
}
var missing []intentgate.SlotQuestion
for _, q := range expectation.Missing {
if _, ok := slots[q.Key]; !ok {
missing = append(missing, q)
}
}
stored, err := s.st.PutIntentSession(key.Owner, key.Project, key.Conversation,
storeExpectation(Expectation{IntentID: expectation.IntentID, Action: expectation.Action,
Missing: missing, Question: expectation.Question}), slots, s.ttl())
if err != nil {
return nil, err
}
return fromStore(key, stored), nil
}
// ApplyAnswer 把一轮用户回答折叠进挂起的会话:合并槽位、推进 revision、
// 重新判定齐全性。revision 不匹配时返回 ErrStale(不改任何状态)。
func (s *Service) ApplyAnswer(key Key, revision int, answer Answer) (*Outcome, error) {
if answer.Cancel {
return s.ApplyCommand(key, revision, string(intentgate.CommandCancel), "", "")
}
cur, err := s.current(key, revision)
if err != nil || cur == nil {
return nil, err
}
if cur.Expectation == nil || cur.Cancelled != nil {
return nil, nil
}
if len(answer.Slots) == 0 && len(answer.Invalid) == 0 {
return nil, nil // 本轮没有槽位信息:不消耗 revision
}
merged := mergeSlots(cur.Slots, s.accepted(cur, answer))
expectation := *cur.Expectation
remaining, invalid := s.recheck(&expectation, merged, answer)
out := &Outcome{Slots: merged, Remaining: remaining, Invalid: invalid, Question: expectation.Question}
if len(remaining) == 0 && len(invalid) == 0 {
// 补齐即消费:会话状态归零,宿主按既有准入与派发路径执行。
if err := s.st.ClearIntentSession(key.Owner, key.Project, key.Conversation); err != nil {
return nil, err
}
out.Complete = true
return out, nil
}
stored, err := s.st.PutIntentSession(key.Owner, key.Project, key.Conversation,
storeExpectation(Expectation{IntentID: expectation.IntentID, Action: expectation.Action,
Missing: remaining, Question: expectation.Question}), merged, s.ttl())
if err != nil {
return nil, err
}
out.Session = fromStore(key, stored)
return out, nil
}
// ApplyCommand 消费库层会话命令:
// - set_slot / change_slot 写槽位(动作声明可解析时按声明校验,非法值不采纳);
// - cancel 清空挂起预期(保留已收集槽位,revision +1 让在途回答作废);
// - 其它种类(start/clarify/chitchat/handoff…)一律忽略,如实返回 nil。
func (s *Service) ApplyCommand(key Key, revision int, kind, name, value string) (*Outcome, error) {
switch kind {
case string(intentgate.CommandSetSlot), string(intentgate.CommandChangeSlot):
if strings.TrimSpace(name) == "" {
return nil, nil
}
return s.ApplyAnswer(key, revision, Answer{Slots: map[string]string{name: value}})
case string(intentgate.CommandCancel):
cur, err := s.current(key, revision)
if err != nil || cur == nil {
return nil, err
}
if cur.Expectation == nil || cur.Cancelled != nil {
// 没有挂起动作可放弃;已取消记录更不是「待补齐的补充」,
// 如实报告「本会话没有会话语义」,绝不假装又取消了一次。
return nil, nil
}
if _, err := s.st.ClearIntentSessionExpectation(key.Owner, key.Project, key.Conversation); err != nil {
if errors.Is(err, store.ErrNotFound) {
return nil, nil
}
return nil, err
}
return &Outcome{Cancelled: true, Slots: cur.Slots}, nil
default:
return nil, nil // 未知命令种类:忽略,绝不假装已消费
}
}
// Drop 清空整个会话状态(动作已派发或用户明确放弃时用;幂等)。
func (s *Service) Drop(key Key) error {
if !key.valid() {
return nil
}
return s.st.ClearIntentSession(key.Owner, key.Project, key.Conversation)
}
// Binding 返回会话当前绑定的运行上下文:
// - 挂起预期里已经抽到的运行号(连同其意图与动作);
// - 或上一次取消留下的短期记录(Cancelled=true)。
//
// 没有绑定(无会话/已过期/抽到的槽位里没有运行号)一律返回 nil:宿主据此
// 澄清,绝不臆造一个靶子。读取失败向上传播,宿主一律 fail-closed。
func (s *Service) Binding(key Key) (*Binding, error) {
cur, err := s.load(key)
if err != nil || cur == nil || cur.Expectation == nil {
return nil, err
}
if cur.Cancelled != nil {
return &Binding{RunID: cur.Cancelled.RunID, Intent: cur.Cancelled.Intent,
Action: cur.Cancelled.Action, Cancelled: true}, nil
}
runID := runIDOf(cur.Slots)
if runID == "" {
return nil, nil
}
return &Binding{RunID: runID, Intent: cur.Expectation.IntentID, Action: cur.Expectation.Action.ID}, nil
}
// RecordCancel 记录一次已派发的取消:清空挂起预期(连带作废在途回答,
// revision +1),留下一条短期可读的「已取消某运行」记录,供紧接着的追问
// (如「确认取消」)如实回话。没有运行号或键不完整即不记录(fail-closed)。
func (s *Service) RecordCancel(key Key, b Binding) (*Session, error) {
if !key.valid() || strings.TrimSpace(b.RunID) == "" {
return nil, nil
}
slots := map[string]string{cancelledRunSlot: strings.TrimSpace(b.RunID)}
stored, err := s.st.PutIntentSession(key.Owner, key.Project, key.Conversation,
storeExpectation(Expectation{IntentID: b.Intent,
Action: intentgate.ActionRef{Kind: CancelActionKind, ID: b.Action}}),
slots, CancelRecordTTL)
if err != nil {
return nil, err
}
return fromStore(key, stored), nil
}
// current 读出会话并校验 revision:过期行惰性删除返回 nil;revision 不匹配
// 返回 ErrStale。
func (s *Service) current(key Key, revision int) (*Session, error) {
cur, err := s.load(key)
if err != nil || cur == nil {
return nil, err
}
if revision > 0 && revision != cur.Revision {
return nil, ErrStale
}
return cur, nil
}
// load 读出会话;过期行惰性删除后返回 nil。
func (s *Service) load(key Key) (*Session, error) {
if !key.valid() {
return nil, nil
}
cur, err := s.st.GetIntentSession(key.Owner, key.Project, key.Conversation)
if errors.Is(err, store.ErrNotFound) {
return nil, nil
}
if err != nil {
return nil, err
}
if !s.live(cur) {
if err := s.st.ClearIntentSession(key.Owner, key.Project, key.Conversation); err != nil {
return nil, err
}
return nil, nil
}
return fromStore(key, cur), nil
}
// live 报告会话是否仍在有效期内。没有挂起预期的行(已取消/已补齐)对宿主
// 等于「没有会话」。
func (s *Service) live(cur store.IntentSession) bool {
if cur.Expectation == nil {
return false
}
if cur.ExpiresAt.IsZero() {
return true
}
return s.now().Before(cur.ExpiresAt)
}
// recheck 折叠后重算仍缺的槽位与本轮未通过校验的槽位(确定性顺序:沿用挂起
// 时的缺失序;未通过的槽位保留其原提问并带上权威原因)。
func (s *Service) recheck(expectation *Expectation, slots map[string]string, answer Answer) ([]intentgate.SlotQuestion, []intentgate.SlotDefect) {
var remaining []intentgate.SlotQuestion
var invalid []intentgate.SlotDefect
for _, q := range expectation.Missing {
if reason, bad := answer.Invalid[q.Key]; bad {
invalid = append(invalid, intentgate.SlotDefect{Key: q.Key, Reason: reason, Question: q.Question, Invalid: true})
continue
}
if _, ok := slots[q.Key]; !ok {
remaining = append(remaining, q)
}
}
return remaining, invalid
}
// accepted 采纳本轮通过校验的槽位:非法值不采纳(仍按非法追问),其余按
// 动作声明校验(宿主注入;未接线即不校验)。
func (s *Service) accepted(cur *Session, answer Answer) map[string]string {
out := map[string]string{}
for k, v := range answer.Slots {
if _, bad := answer.Invalid[k]; bad {
continue
}
if s.validate != nil && cur.Expectation != nil && cur.Expectation.Action.ID != "" {
if err := s.validate(cur.Expectation.Action.ID, k, v); err != nil {
continue
}
}
out[k] = v
}
return out
}
// missingOf 投影到存储形态(缺失槽位表)。
func missingOf(list []intentgate.SlotQuestion) []store.IntentSlotQuestion {
out := make([]store.IntentSlotQuestion, 0, len(list))
for _, q := range list {
out = append(out, store.IntentSlotQuestion{Key: q.Key, Question: q.Question})
}
return out
}
// storeExpectation 投影到存储形态。
func storeExpectation(e Expectation) *store.IntentExpectation {
return &store.IntentExpectation{
IntentID: e.IntentID, ActionKind: e.Action.Kind, ActionID: e.Action.ID,
Missing: missingOf(e.Missing), Question: e.Question,
}
}
// fromStore 从存储行投影出宿主视图。
func fromStore(key Key, cur store.IntentSession) *Session {
out := &Session{Key: key, Slots: cur.Slots, Revision: cur.Revision, ExpiresAt: cur.ExpiresAt}
if cur.Expectation != nil {
exp := &Expectation{IntentID: cur.Expectation.IntentID, Question: cur.Expectation.Question,
Action: intentgate.ActionRef{Kind: cur.Expectation.ActionKind, ID: cur.Expectation.ActionID}}
for _, q := range cur.Expectation.Missing {
exp.Missing = append(exp.Missing, intentgate.SlotQuestion{Key: q.Key, Question: q.Question})
}
out.Expectation = exp
if exp.Action.Kind == CancelActionKind {
out.Cancelled = &CancelRecord{RunID: strings.TrimSpace(cur.Slots[cancelledRunSlot]),
Intent: exp.IntentID, Action: exp.Action.ID}
}
}
if out.Slots == nil {
out.Slots = map[string]string{}
}
return out
}
// runIDOf 从槽位表里取运行号(固定键序,结果与 map 迭代序无关;无则空)。
func runIDOf(slots map[string]string) string {
for _, key := range runSlotKeys {
if v := strings.TrimSpace(slots[key]); v != "" {
return v
}
}
return ""
}
// mergeSlots 合并两组槽位(后者覆盖前者,返回新表:不改调用方的 map)。
func mergeSlots(base, add map[string]string) map[string]string {
out := map[string]string{}
for k, v := range base {
out[k] = v
}
for k, v := range add {
out[k] = v
}
return out
}
@@ -0,0 +1,506 @@
package intentsession
import (
"errors"
"path/filepath"
"testing"
"time"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/store"
)
// A1/U-A1b:会话内意图会话。合成数据 + 嵌入式存储(PG 档由 store 包双档
// 测试覆盖同一张表)。命令动作(run-cancel 形态)先行,取消路由归 U-A1c。
func newService(t *testing.T) *Service {
t.Helper()
st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "intent-session.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { st.Close() })
return New(st)
}
func cancelKey() Key { return Key{Owner: "alice", Project: "prj_1", Conversation: "cnv_1"} }
// cancelExpectation 是首个动作(取消运行):动作引用 + 仍缺槽位 + 提问。
func cancelExpectation() Expectation {
return Expectation{
IntentID: "demo.run-cancel",
Action: intentgate.ActionRef{Kind: intentgate.ActionCommand, ID: "demo.run-cancel"},
Missing: []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"},
{Key: "reason", Question: "取消原因?"},
},
Question: "取消哪个运行?",
}
}
func questionKeys(list []intentgate.SlotQuestion) []string {
out := make([]string, 0, len(list))
for _, q := range list {
out = append(out, q.Key)
}
return out
}
// 无会话时 Pending 就是「没有」,且不产生任何行。
func TestPendingWithoutSession(t *testing.T) {
svc := newService(t)
got, err := svc.Pending(cancelKey())
if err != nil || got != nil {
t.Fatalf("无会话时应为空:%v %v", got, err)
}
// 归属不完整即 fail-closed(不会退化成跨用户共享状态)。
if got, err := svc.Pending(Key{Conversation: "cnv_1"}); err != nil || got != nil {
t.Fatalf("无 owner 的键必须被拒:%v %v", got, err)
}
}
// 缺槽位动作 → 挂起 → 回答补齐:齐全即消费会话(宿主按既有准入派发)。
func TestSuspendThenAnswerCompletes(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, cancelExpectation(), map[string]string{"run_id": "R-1"})
if err != nil {
t.Fatal(err)
}
if suspended.Revision != 1 || suspended.Expectation == nil || suspended.Expectation.Action.ID != "demo.run-cancel" {
t.Fatalf("挂起结果:%+v", suspended)
}
if keys := questionKeys(suspended.Expectation.Missing); len(keys) != 1 || keys[0] != "reason" {
t.Fatalf("已抽到的槽位不应再算缺失:%v", keys)
}
if !suspended.ExpiresAt.After(time.Now()) {
t.Fatalf("到期时间未设:%v", suspended.ExpiresAt)
}
// 部分回答:仍缺原因 → 仍挂起、revision 递增、到期时间续期。
partial, err := svc.ApplyAnswer(key, suspended.Revision, Answer{Slots: map[string]string{"run_id": "R-2"}})
if err != nil {
t.Fatal(err)
}
if partial.Complete || partial.Cancelled {
t.Fatalf("部分回答不得视为完成:%+v", partial)
}
if keys := questionKeys(partial.Remaining); len(keys) != 1 || keys[0] != "reason" {
t.Fatalf("剩余缺失:%v", keys)
}
if partial.Session.Revision != 2 || !partial.Session.ExpiresAt.After(suspended.ExpiresAt) {
t.Fatalf("revision/续期:%+v", partial.Session)
}
// 补齐最后一个:Complete=true,会话被消费。
done, err := svc.ApplyAnswer(key, partial.Session.Revision, Answer{Slots: map[string]string{"reason": "误触发"}})
if err != nil {
t.Fatal(err)
}
if !done.Complete || done.Session != nil {
t.Fatalf("补齐应完成并消费会话:%+v", done)
}
if done.Slots["run_id"] != "R-2" || done.Slots["reason"] != "误触发" {
t.Fatalf("累计槽位:%v", done.Slots)
}
if got, err := svc.Pending(key); err != nil || got != nil {
t.Fatalf("完成后仍挂起:%v %v", got, err)
}
}
// 两轮部分填充:set_slot 落一个槽位,再由回答补另一个。
func TestSetSlotThenAnswerCompletes(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, cancelExpectation(), nil)
if err != nil {
t.Fatal(err)
}
set, err := svc.ApplyCommand(key, suspended.Revision, string(intentgate.CommandSetSlot), "run_id", "R-7")
if err != nil {
t.Fatal(err)
}
if set.Complete || set.Session.Slots["run_id"] != "R-7" || set.Session.Revision != 2 {
t.Fatalf("set_slot 后:%+v", set)
}
if keys := questionKeys(set.Remaining); len(keys) != 1 || keys[0] != "reason" {
t.Fatalf("剩余缺失:%v", keys)
}
change, err := svc.ApplyCommand(key, set.Session.Revision, string(intentgate.CommandChangeSlot), "run_id", "R-8")
if err != nil {
t.Fatal(err)
}
if change.Session.Slots["run_id"] != "R-8" {
t.Fatalf("change_slot 未改写:%+v", change.Session)
}
done, err := svc.ApplyAnswer(key, change.Session.Revision, Answer{Slots: map[string]string{"reason": "重复提交"}})
if err != nil {
t.Fatal(err)
}
if !done.Complete || done.Slots["run_id"] != "R-8" || done.Slots["reason"] != "重复提交" {
t.Fatalf("两轮补齐:%+v", done)
}
}
// 取消清空挂起预期:确认取消、不假装成功,陈旧回答也作废。
func TestCancelClearsPendingExpectation(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, cancelExpectation(), map[string]string{"run_id": "R-1"})
if err != nil {
t.Fatal(err)
}
out, err := svc.ApplyCommand(key, suspended.Revision, string(intentgate.CommandCancel), "", "")
if err != nil {
t.Fatal(err)
}
if !out.Cancelled || out.Complete || out.Session != nil {
t.Fatalf("取消结果:%+v", out)
}
if got, err := svc.Pending(key); err != nil || got != nil {
t.Fatalf("取消后仍有挂起预期:%v %v", got, err)
}
if _, err := svc.ApplyAnswer(key, suspended.Revision, Answer{Slots: map[string]string{"reason": "误触发"}}); err != nil {
t.Fatalf("取消后不应再有会话可折叠:%v", err)
}
}
// 陈旧回答一律拒绝:期间 revision 已被别的轮次推进过。
func TestStaleRevisionRejected(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, cancelExpectation(), nil)
if err != nil {
t.Fatal(err)
}
moved, err := svc.ApplyCommand(key, suspended.Revision, string(intentgate.CommandSetSlot), "run_id", "R-7")
if err != nil {
t.Fatal(err)
}
if _, err := svc.ApplyAnswer(key, suspended.Revision, Answer{Slots: map[string]string{"reason": "旧回答"}}); !errors.Is(err, ErrStale) {
t.Fatalf("陈旧回答必须被拒:%v", err)
}
if _, err := svc.ApplyCommand(key, suspended.Revision, string(intentgate.CommandCancel), "", ""); !errors.Is(err, ErrStale) {
t.Fatalf("陈旧取消必须被拒:%v", err)
}
// 状态未被陈旧回答改动:仍是推进后的那一版。
cur, err := svc.Pending(key)
if err != nil || cur.Revision != moved.Session.Revision || cur.Slots["run_id"] != "R-7" || cur.Slots["reason"] != "" {
t.Fatalf("陈旧回答改动了状态:%+v %v", cur, err)
}
}
// 过期会话既不补齐旧动作,也不影响新的挂起:读侧惰性删除过期行。
func TestExpiredSessionNeverCompletesOldAction(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, cancelExpectation(), nil)
if err != nil {
t.Fatal(err)
}
// 时钟前移越过到期时间:过期会话对任何新轮次都不存在。
svc.clock = func() time.Time { return suspended.ExpiresAt.Add(time.Second) }
got, err := svc.Pending(key)
if err != nil || got != nil {
t.Fatalf("过期会话仍可见:%v %v", got, err)
}
if out, err := svc.ApplyAnswer(key, suspended.Revision, Answer{Slots: map[string]string{"run_id": "R-1", "reason": "误触发"}}); err != nil || out != nil {
t.Fatalf("过期回答不得补齐旧动作:%+v %v", out, err)
}
// 过期行已删除 → 新任务重新挂起时从干净状态开始。
fresh, err := svc.Suspend(key, Expectation{IntentID: "demo.purge",
Action: intentgate.ActionRef{Kind: intentgate.ActionCommand, ID: "demo.purge"},
Missing: []intentgate.SlotQuestion{{Key: "target", Question: "清理哪个对象?"}},
Question: "清理哪个对象?"}, nil)
if err != nil {
t.Fatal(err)
}
if fresh.Revision != 1 || fresh.Slots["run_id"] != "" {
t.Fatalf("新挂起被旧会话污染:%+v", fresh)
}
}
// 两个用户、两个项目在构造上互不可见。
func TestSessionsAreIsolatedByOwnerAndProject(t *testing.T) {
svc := newService(t)
alice := Key{Owner: "alice", Project: "prj_1", Conversation: "cnv_1"}
bob := Key{Owner: "bob", Project: "prj_1", Conversation: "cnv_1"}
other := Key{Owner: "alice", Project: "prj_2", Conversation: "cnv_1"}
otherConv := Key{Owner: "alice", Project: "prj_1", Conversation: "cnv_2"}
suspended, err := svc.Suspend(alice, cancelExpectation(), map[string]string{"run_id": "R-alice"})
if err != nil {
t.Fatal(err)
}
for _, key := range []Key{bob, other, otherConv} {
got, err := svc.Pending(key)
if err != nil || got != nil {
t.Fatalf("%+v 读到了别人的会话:%v %v", key, got, err)
}
// 陈旧的答案也绝不命中别人的会话。
out, err := svc.ApplyAnswer(key, suspended.Revision, Answer{Slots: map[string]string{"run_id": "R-x", "reason": "x"}})
if err != nil || out != nil {
t.Fatalf("%+v 的回答折叠进了别人的会话:%+v %v", key, out, err)
}
if err := svc.Drop(key); err != nil {
t.Fatal(err)
}
}
cur, err := svc.Pending(alice)
if err != nil || cur == nil || cur.Slots["run_id"] != "R-alice" || cur.Revision != 1 {
t.Fatalf("属主会话被影响:%+v %v", cur, err)
}
}
// 槽位值按动作声明 schema 校验:非法值不采纳(仍按缺失追问),
// 合法值正常折叠。
func TestAnswerSlotsValidatedAgainstDeclaredSchema(t *testing.T) {
svc := newService(t)
key := cancelKey()
var validated []string
svc.SetSlotValidator(func(actionID, slot, value string) error {
if actionID != "demo.run-cancel" || slot != "count" {
return nil
}
validated = append(validated, value)
if value == "abc" {
return errors.New("不是整数")
}
return nil
})
expectation := Expectation{IntentID: "demo.run-cancel",
Action: intentgate.ActionRef{Kind: intentgate.ActionCommand, ID: "demo.run-cancel"},
Missing: []intentgate.SlotQuestion{{Key: "count", Question: "几次?"}, {Key: "reason", Question: "原因?"}},
Question: "几次?"}
suspended, err := svc.Suspend(key, expectation, nil)
if err != nil {
t.Fatal(err)
}
bad, err := svc.ApplyAnswer(key, suspended.Revision, Answer{Slots: map[string]string{"count": "abc"}})
if err != nil {
t.Fatal(err)
}
if bad.Complete || bad.Session.Slots["count"] != "" {
t.Fatalf("非法值被采纳:%+v", bad.Session)
}
if keys := questionKeys(bad.Remaining); len(keys) != 2 {
t.Fatalf("非法值应仍按缺失追问:%v", keys)
}
if len(validated) != 1 || validated[0] != "abc" {
t.Fatalf("声明校验未生效:%v", validated)
}
good, err := svc.ApplyAnswer(key, bad.Session.Revision, Answer{Slots: map[string]string{"count": "3", "reason": "误触发"}})
if err != nil {
t.Fatal(err)
}
if !good.Complete || good.Slots["count"] != "3" {
t.Fatalf("合法值未补齐:%+v", good)
}
}
// 本轮给出但没通过权威校验的值:如实按非法追问(值不采纳)。
func TestInvalidAnswerIsReportedNotAdopted(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, Expectation{IntentID: "demo.run-cancel",
Action: intentgate.ActionRef{Kind: intentgate.ActionCommand, ID: "demo.run-cancel"},
Missing: []intentgate.SlotQuestion{{Key: "mode", Question: "要哪种模式?"}},
Question: "要哪种模式?"}, nil)
if err != nil {
t.Fatal(err)
}
out, err := svc.ApplyAnswer(key, suspended.Revision, Answer{
Slots: map[string]string{"mode": "随手写"},
Invalid: map[string]string{"mode": "value not in enum"},
})
if err != nil {
t.Fatal(err)
}
if out.Complete || out.Session.Slots["mode"] != "" {
t.Fatalf("非法值被采纳:%+v", out)
}
if len(out.Invalid) != 1 || out.Invalid[0].Key != "mode" || out.Invalid[0].Reason != "value not in enum" {
t.Fatalf("非法原因未透传:%+v", out.Invalid)
}
if len(out.Remaining) != 0 {
t.Fatalf("非法不应重复计入缺失:%+v", out.Remaining)
}
}
// 未知命令种类如实忽略:不改状态、不假装已消费。
func TestUnknownCommandsAreIgnored(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, cancelExpectation(), nil)
if err != nil {
t.Fatal(err)
}
for _, kind := range []string{string(intentgate.CommandStart), string(intentgate.CommandClarify),
string(intentgate.CommandChitchat), string(intentgate.CommandHandoff), "made-up"} {
out, err := svc.ApplyCommand(key, suspended.Revision, kind, "run_id", "R-9")
if err != nil || out != nil {
t.Fatalf("%s 不得被当成会话命令:%+v %v", kind, out, err)
}
}
cur, err := svc.Pending(key)
if err != nil || cur.Revision != suspended.Revision || cur.Slots["run_id"] != "" {
t.Fatalf("未知命令改动了状态:%+v %v", cur, err)
}
// 没有槽位名时同样忽略(不写入空键)。
if out, err := svc.ApplyCommand(key, suspended.Revision, string(intentgate.CommandSetSlot), " ", "R-9"); err != nil || out != nil {
t.Fatalf("空槽位名必须被忽略:%+v %v", out, err)
}
}
// Drop 幂等清空;之后重新挂起从 revision 1 开始。
func TestDropIsIdempotent(t *testing.T) {
svc := newService(t)
key := cancelKey()
if _, err := svc.Suspend(key, cancelExpectation(), nil); err != nil {
t.Fatal(err)
}
for i := 0; i < 2; i++ {
if err := svc.Drop(key); err != nil {
t.Fatalf("drop %d:%v", i, err)
}
}
if err := svc.Drop(Key{Conversation: "cnv_1"}); err != nil {
t.Fatalf("不完整的键必须安全无操作:%v", err)
}
again, err := svc.Suspend(key, cancelExpectation(), nil)
if err != nil {
t.Fatal(err)
}
if again.Revision != 1 || again.Slots["run_id"] != "" {
t.Fatalf("drop 后重新挂起:%+v", again)
}
}
// ---- A1/U-A1c:取消的靶子来源(Binding)与已取消记录(RecordCancel)----
// 挂起预期里已经抽到的运行号就是取消的靶子;连同意图与动作一并透出。
func TestBindingComesFromSuspendedSlots(t *testing.T) {
svc := newService(t)
key := cancelKey()
// 没有会话 → 没有靶子(绝不臆造)。
if got, err := svc.Binding(key); err != nil || got != nil {
t.Fatalf("无会话时必须有 binding:%v %v", got, err)
}
// 槽位里没有运行号 → 同样没有靶子。
if _, err := svc.Suspend(key, cancelExpectation(), nil); err != nil {
t.Fatal(err)
}
if got, err := svc.Binding(key); err != nil || got != nil {
t.Fatalf("无运行号时必须没有靶子:%v %v", got, err)
}
// 抽到运行号 → 靶子就绪(取消意图据此可无澄清地派发)。
svc.Drop(key)
if _, err := svc.Suspend(key, cancelExpectation(), map[string]string{"run_id": "R-7"}); err != nil {
t.Fatal(err)
}
got, err := svc.Binding(key)
if err != nil || got == nil {
t.Fatalf("有运行号时必须解出靶子:%v %v", got, err)
}
if got.RunID != "R-7" || got.Intent != "demo.run-cancel" || got.Action != "demo.run-cancel" || got.Cancelled {
t.Fatalf("靶子上下文:%+v", got)
}
// runId/run 两种写法同样识别(键序固定,结果确定)。
svc.Drop(key)
for _, slot := range []string{"runId", "run"} {
if _, err := svc.Suspend(key, cancelExpectation(), map[string]string{slot: "R-8", "reason": "误触发"}); err != nil {
t.Fatal(err)
}
if got, _ := svc.Binding(key); got == nil || got.RunID != "R-8" {
t.Fatalf("槽位 %s 未被识别为运行号:%+v", slot, got)
}
svc.Drop(key)
}
}
// 已派发的取消:清挂起预期 + 留短期记录;记录不可再被回答折叠,也不作靶子。
func TestRecordCancelClearsExpectationAndKeepsShortLivedRecord(t *testing.T) {
svc := newService(t)
key := cancelKey()
suspended, err := svc.Suspend(key, cancelExpectation(), map[string]string{"run_id": "R-7"})
if err != nil {
t.Fatal(err)
}
recorded, err := svc.RecordCancel(key, Binding{RunID: "R-7", Intent: "demo.task-cancel", Action: "demo.run-cancel"})
if err != nil {
t.Fatal(err)
}
if recorded == nil || recorded.Cancelled == nil || recorded.Cancelled.RunID != "R-7" {
t.Fatalf("取消记录:%+v", recorded)
}
if recorded.Cancelled.Intent != "demo.task-cancel" || recorded.Cancelled.Action != "demo.run-cancel" {
t.Fatalf("取消记录上下文:%+v", recorded.Cancelled)
}
if recorded.Revision <= suspended.Revision {
t.Fatalf("取消必须推进 revision(让在途回答作废):%d vs %d", recorded.Revision, suspended.Revision)
}
if len(recorded.Expectation.Missing) != 0 || recorded.Slots["run_id"] != "" {
t.Fatalf("取消后不得还留着待补齐的槽位/缺失:%+v", recorded)
}
// 记录仍可读:宿主据此如实回答「确认取消」这类追问。
got, err := svc.Binding(key)
if err != nil || got == nil || got.RunID != "R-7" || !got.Cancelled {
t.Fatalf("取消记录应可读且标记为已取消:%+v %v", got, err)
}
// 取消记录不是待补齐的动作:回答与 cancel 命令都不再折叠进它。
if out, err := svc.ApplyAnswer(key, recorded.Revision, Answer{Slots: map[string]string{"reason": "误触发"}}); err != nil || out != nil {
t.Fatalf("取消记录不得吸收回答:%+v %v", out, err)
}
if out, err := svc.ApplyCommand(key, recorded.Revision, string(intentgate.CommandCancel), "", ""); err != nil || out != nil {
t.Fatalf("取消记录不得再被取消一次:%+v %v", out, err)
}
if cur, err := svc.Pending(key); err != nil || cur == nil || cur.Cancelled == nil {
t.Fatalf("取消记录不得被当成挂起动作改写:%+v %v", cur, err)
}
// 记录短期留存:越过 CancelRecordTTL 后对任何新轮次都不存在。
svc.clock = func() time.Time { return recorded.ExpiresAt.Add(time.Second) }
if got, err := svc.Binding(key); err != nil || got != nil {
t.Fatalf("过期记录仍可见:%+v %v", got, err)
}
if _, err := svc.Pending(key); err != nil {
t.Fatal(err)
}
if _, err := svc.st.GetIntentSession(key.Owner, key.Project, key.Conversation); !errors.Is(err, store.ErrNotFound) {
t.Fatalf("过期行未被惰性删除:%v", err)
}
}
// 缺键或没有运行号时如实不记录(fail-closed,绝不写半条取消记录)。
func TestRecordCancelIsFailClosed(t *testing.T) {
svc := newService(t)
key := cancelKey()
if got, err := svc.RecordCancel(Key{Conversation: "cnv_1"}, Binding{RunID: "R-7"}); err != nil || got != nil {
t.Fatalf("缺 owner 的键必须无操作:%+v %v", got, err)
}
if got, err := svc.RecordCancel(key, Binding{RunID: " "}); err != nil || got != nil {
t.Fatalf("没有运行号必须无操作:%+v %v", got, err)
}
if got, err := svc.Binding(key); err != nil || got != nil {
t.Fatalf("无操作不得留下任何行:%+v %v", got, err)
}
}
// 两个用户的绑定互不可见:别人的运行号绝不会成为本会话的靶子。
func TestBindingIsIsolatedByOwner(t *testing.T) {
svc := newService(t)
alice := Key{Owner: "alice", Project: "prj_1", Conversation: "cnv_1"}
bob := Key{Owner: "bob", Project: "prj_1", Conversation: "cnv_1"}
if _, err := svc.Suspend(alice, cancelExpectation(), map[string]string{"run_id": "R-alice"}); err != nil {
t.Fatal(err)
}
if _, err := svc.RecordCancel(bob, Binding{RunID: "R-bob"}); err != nil {
t.Fatal(err)
}
if got, _ := svc.Binding(bob); got == nil || got.RunID != "R-bob" {
t.Fatalf("bob 的绑定:%+v", got)
}
if got, _ := svc.Binding(alice); got == nil || got.RunID != "R-alice" || got.Cancelled {
t.Fatalf("alice 的绑定被污染:%+v", got)
}
}
+122 -12
View File
@@ -18,9 +18,18 @@
// 当前版本的已知边界(如实记录,不静默):
// - squad 决策需要 runmgr 的 per-run entry 覆盖才能真正执行(独立设计项);
// 本版本调用方按普通回合处理并保留 Decision 供审计。
// - SetSlot/ChangeSlot/Cancel/Handoff 由库层产出,但 Backend 尚无会话槽位
// 状态机与语义层:它们记入 Unsupported 并回落普通回合,绝不假装已支持
// (例如用户说「取消」时不会把同一句话里的 Start 当成业务动作派发)。
// - Handoff 仍由宿主不消费,记 Unsupported 并回落普通回合。
// - Cancel 已是宿主可执行的既有动作(A1/U-A1c):本包只解出「取消哪一个
// 运行」——原文点名的运行号优先,其次当前会话绑定动作的运行上下文
// (Env.SessionRun,宿主从意图会话注入)。解不出唯一靶子时 Decision
// 如实留空 RunID 并写明原因,由宿主澄清,绝不替用户猜一个。
// 本包不派发取消、也不认识取消的实现(收件箱/命令总线的确认语义都在
// 宿主侧),更不自动确认。
// - SetSlot/ChangeSlot 已被宿主消费,但消费的语义不在本包:宿主 chat 入口
// 把它们折叠进会话里的挂起预期(internal/intentsession 的会话状态机,
// A1/U-A1b)——落槽与改槽是「同一动作的补充说明」,不是可派发的动作。
// 本包仍不把它们当动作派发:命中即记 Unsupported 并回落普通回合,
// 同时绝不把同一句话里的 Start 当作业务动作派发。
package route
import (
@@ -38,8 +47,44 @@ const (
KindWorkflow Kind = "workflow"
KindSquad Kind = "squad"
KindTurn Kind = "turn"
// KindCancel 是「取消」意图的裁决:取消是既有动作(取消一个运行),
// 不是可以顺带启动的动作。靶子解出与否都在 Decision.Cancel 里,
// 派发与确认由宿主按既有取消路径完成。
KindCancel Kind = "cancel"
)
// CancelSource 说明取消靶子的来源(宿主据此如实回话;不是授权凭据)。
type CancelSource string
const (
// CancelSourceExplicit:用户这一轮点名了运行(命令值或 run_id 槽位)。
CancelSourceExplicit CancelSource = "explicit"
// CancelSourceSession:当前会话绑定动作的运行上下文(意图会话)。
CancelSourceSession CancelSource = "session"
)
// CancelTarget 是「取消」的裁决靶子。
//
// 纪律:RunID 为空就是「没有唯一靶子」——宿主必须如实澄清,绝不猜一个
// 「大概是它」;带靶子的裁决也只声明「要取消谁」,不代表已获授权。
type CancelTarget struct {
// RunID 是要取消的运行(空 = 待宿主澄清)。
RunID string
Source CancelSource
// Intent/Action 是靶子的来源上下文(会话绑定的意图与动作;审计与回话用)。
Intent string
Action string
// Candidates 是库层随取消命令给出的候选(如有):如实透出,不做取舍。
Candidates []string
}
// SessionRun 是会话(意图会话)里绑定的运行上下文,由宿主注入给 Env。
type SessionRun struct {
RunID string
Intent string
Action string
}
// Clarification 是「业务动作未派发,需要向用户澄清」的诉求。
// 槽位缺失与非法都带结构化原因(key + reason + question),
// 宿主据此渲染澄清话术或转成回合提示,不再自行猜测。
@@ -62,8 +107,10 @@ type Decision struct {
Missing []intentgate.SlotQuestion
Invalid []intentgate.SlotDefect
Risk string // 权威意图声明的风险档(low/medium/high)
Reason string // 回落 turn 的原因(显式声明不可解析、准入未过等)
StartFound bool // 是否存在 Start 命令(runmgr 转轨判定用)
Reason string // 回落 turn 的原因(显式声明不可解析、准入未过、取消无唯一靶子等)
StartFound bool // 本轮是否会启动动作(runmgr 转轨判定用;取消裁决下恒为 false——取消与启动互斥)
// Cancel 非 nil 时本轮是「取消」意图的裁决(Kind=KindCancel)。
Cancel *CancelTarget
// RequiresConfirm 是意图声明的显式确认要求(与命令自身的 danger 分级正交,
// 本包只透传,不替代 cmdbus 的高危确认收口)。
RequiresConfirm bool
@@ -78,25 +125,34 @@ type Env struct {
HasCommand func(id string) bool
WorkflowFor func(intentID string) (core.WorkflowDef, bool)
SquadReady bool
// SessionRun 返回当前会话绑定动作的运行上下文(取消靶子的会话来源;
// 宿主从意图会话注入)。nil 或 false = 会话没有给出靶子——绝不臆造。
SessionRun func() (SessionRun, bool)
}
// unconsumedCommandKinds 是宿主 v1 不消费的库层命令种类(无会话槽位状态机)。
// 命中即不派发业务动作:把「取消」当成「启动」是比不响更糟的错配。
// unconsumedCommandKinds 是本包不派发的库层命令种类。
//
// 命中即不派发业务动作:把「改槽」当成「启动」是比不响更糟的错配。
// set_slot/change_slot 已由宿主的会话状态机消费(internal/intentsession,
// A1/U-A1b)——它们改的是会话里挂起动作的槽位;cancel 自 U-A1c 起由宿主
// 按「取消既有动作」消费(解靶子 → 宿主派发既有取消路径),故不在本表。
// chitchat/handoff 由宿主自行处理。
var unconsumedCommandKinds = map[string]bool{
string(intentgate.CommandSetSlot): true,
string(intentgate.CommandChangeSlot): true,
string(intentgate.CommandCancel): true,
string(intentgate.CommandHandoff): true,
string(intentgate.CommandChitchat): true,
}
// Decide 裁决一次识别结果。规则:
// 1. 非 matched(fallback/guard/ambiguous 经闸门折叠后)→ turn。
// 2. 出现宿主不消费的库层命令(cancel/set_slot/…)→ turn,并如实记录
// 2. 出现 cancel 命令 → 取消裁决(KindCancel):解出唯一靶子,宿主据此派发
// 既有取消动作;同句的 Start 一并不派发(取消与启动互斥)。
// 3. 出现宿主不消费的库层命令(set_slot/handoff/…)→ turn,并如实记录
// Unsupported 与 Reason(不假装支持,也不派发同句的 Start)。
// 3. 找到首个 Start 命令;没有时按 TopID(或显式 action 声明)走 auto
// 4. 找到首个 Start 命令;没有时按 TopID(或显式 action 声明)走 auto
// (兼容兜底,与现网一致)。
// 4. 按 ActionRef.Kind 解析:command→命令总线(先过槽位准入);
// 5. 按 ActionRef.Kind 解析:command→命令总线(先过槽位准入);
// workflow→工作流(草稿,缺槽照常启动);blueprint→squad 决策;
// handler/auto→命令总线 > 工作流 > turn。
func Decide(res intentgate.MatchResult, env Env) Decision {
@@ -106,7 +162,7 @@ func Decide(res intentgate.MatchResult, env Env) Decision {
return d
}
d.Unsupported = unconsumedKinds(res.Commands)
var start, clarify *intentgate.Command
var start, clarify, cancel *intentgate.Command
for i := range res.Commands {
switch res.Commands[i].Kind {
case intentgate.CommandStart:
@@ -117,8 +173,18 @@ func Decide(res intentgate.MatchResult, env Env) Decision {
if clarify == nil {
clarify = &res.Commands[i]
}
case intentgate.CommandCancel:
if cancel == nil {
cancel = &res.Commands[i]
}
}
}
if cancel != nil {
// 取消优先于同句的其它命令:用户这一轮要的是「取消」,不是「顺便
// 启动」。StartFound 刻意不置位——本轮不会启动任何动作,runmgr 的
// 转轨判定因此保持与取消前一致。
return d.decideCancel(*cancel, res, env)
}
if len(d.Unsupported) > 0 {
d.Reason = "宿主未消费库层命令:" + strings.Join(d.Unsupported, "、")
if clarify != nil {
@@ -168,6 +234,50 @@ func Decide(res intentgate.MatchResult, env Env) Decision {
return d
}
// decideCancel 裁决「取消」:解出唯一靶子,把「取消哪一个运行」交给宿主。
//
// 靶子优先级(确定性,用户点名优先于会话绑定):
// 1. 取消命令自身的值 / 本轮槽位里的运行号——用户明确指认;
// 2. 当前会话绑定动作的运行上下文(Env.SessionRun,宿主从意图会话读)。
//
// 解不出唯一靶子时 RunID 留空并写明原因:宿主据此如实澄清,绝不猜。
// 本包不派发、不确认:执行授权始终由宿主的既有取消路径收口。
// 同句里其余未消费的库层命令(如 set_slot)仍如实记在 Unsupported 里供审计,
// 但既不派发也不参与靶子解算——取消与启动互斥。
func (d Decision) decideCancel(cmd intentgate.Command, res intentgate.MatchResult, env Env) Decision {
d.Kind = KindCancel
t := &CancelTarget{Intent: res.ID, Candidates: append([]string(nil), cmd.Candidates...)}
if runID := explicitRunID(res, cmd); runID != "" {
t.RunID, t.Source = runID, CancelSourceExplicit
} else if env.SessionRun != nil {
if bound, ok := env.SessionRun(); ok && strings.TrimSpace(bound.RunID) != "" {
t.RunID, t.Source, t.Action = strings.TrimSpace(bound.RunID), CancelSourceSession, bound.Action
if t.Intent == "" {
t.Intent = bound.Intent
}
}
}
if t.RunID == "" {
d.Reason = "取消没有唯一靶子:原文未指明运行,当前会话也没有绑定运行"
}
d.Cancel = t
return d
}
// explicitRunID 取消命令里点名的运行:命令自身的值优先,其次本轮槽位里的
// 运行号(键序固定,结果与 map 迭代序无关)。
func explicitRunID(res intentgate.MatchResult, cmd intentgate.Command) string {
if v := strings.TrimSpace(cmd.Value); v != "" {
return v
}
for _, key := range []string{"run_id", "runId", "run"} {
if v := strings.TrimSpace(res.Slots[key]); v != "" {
return v
}
}
return ""
}
// dispatchCommand 派发可执行业务动作前的最后一道闸:槽位必须齐全且通过权威校验。
// 未就绪时不碰业务动作,改为带结构化原因的澄清诉求(Kind 留在 turn)。
func (d Decision) dispatchCommand(id string, res intentgate.MatchResult, clarify *intentgate.Command) Decision {
+166 -7
View File
@@ -153,21 +153,22 @@ func TestWorkflowDraftStartsWhileGatheringSlots(t *testing.T) {
}
}
// 显式的取消/改槽命令不得被静默当成已支持:回落普通回合,如实记录。
// 显式的取消/改槽命令不得被静默当成已支持:改槽回落普通回合,如实记录;
// 取消已经是宿主可执行的既有动作(U-A1c),但解不出靶子时绝不假装已取消。
func TestUnsupportedLibraryCommandsAreNotClaimed(t *testing.T) {
in := intentgate.MatchResult{Branch: "matched", ID: "demo.brief", Commands: []intentgate.Command{
{Kind: intentgate.CommandStart, Action: intentgate.ActionRef{Kind: intentgate.ActionAuto, ID: "demo.brief"}},
{Kind: intentgate.CommandCancel, Target: "demo.brief"},
}}
d := Decide(in, hasCommandEnv())
if d.Kind != KindTurn || d.CommandID != "" {
t.Fatalf("未消费的库层命令不得派发业务动作:%+v", d)
if d.Kind != KindCancel || d.CommandID != "" {
t.Fatalf("取消不得被当成新业务动作派发:%+v", d)
}
if len(d.Unsupported) != 1 || d.Unsupported[0] != string(intentgate.CommandCancel) {
t.Fatalf("未消费命令种类未如实记录:%+v", d.Unsupported)
if len(d.Unsupported) != 0 {
t.Fatalf("取消已由宿主消费,不得再记为未消费:%+v", d.Unsupported)
}
if d.Reason == "" {
t.Fatalf("回落原因缺失:%+v", d)
if d.Cancel == nil || d.Cancel.RunID != "" || d.Reason == "" {
t.Fatalf("没有唯一靶子时必须如实留空并说明原因:%+v", d)
}
// 只有改槽命令时同样回落,且不得声称支持。
only := intentgate.MatchResult{Branch: "matched", ID: "demo.brief", Commands: []intentgate.Command{
@@ -202,3 +203,161 @@ func TestLegacyAutoStillNeedsCompleteSlots(t *testing.T) {
t.Fatalf("旧路径也必须过槽位准入:%+v", d)
}
}
// A1/U-A1b:set_slot/change_slot 由宿主的会话状态机消费(internal/intentsession),
// 但本包的公开契约不变——落槽不是可派发的动作。
func TestSessionCommandsStayUnsupportedForDispatch(t *testing.T) {
start := intentgate.Command{Kind: intentgate.CommandStart,
Action: intentgate.ActionRef{Kind: intentgate.ActionAuto, ID: "demo.brief"},
Slots: map[string]string{"topic": "核能"}}
sessionKinds := []struct {
command intentgate.Command
kind string
}{
{intentgate.Command{Kind: intentgate.CommandSetSlot, Name: "topic", Value: "电机"}, string(intentgate.CommandSetSlot)},
{intentgate.Command{Kind: intentgate.CommandChangeSlot, Name: "topic", Value: "电机"}, string(intentgate.CommandChangeSlot)},
}
for _, item := range sessionKinds {
in := intentgate.MatchResult{Branch: "matched", ID: "demo.brief",
Commands: []intentgate.Command{start, item.command}}
d := Decide(in, hasCommandEnv())
if d.Kind != KindTurn || d.CommandID != "" {
t.Fatalf("%s 不得被当成动作派发:%+v", item.kind, d)
}
if len(d.Unsupported) != 1 || d.Unsupported[0] != item.kind || d.Reason == "" {
t.Fatalf("%s 未如实记录为宿主已消费但不派发:%+v", item.kind, d)
}
}
// 取消是既有动作(U-A1c):它派发的是「取消某个运行」,不是同句里的
// demo.brief;靶子缺失时留给宿主澄清,绝不假装已取消。
cancel := intentgate.MatchResult{Branch: "matched", ID: "demo.brief",
Commands: []intentgate.Command{start, {Kind: intentgate.CommandCancel, Target: "demo.brief"}}}
if d := Decide(cancel, hasCommandEnv()); d.Kind != KindCancel || d.CommandID != "" || d.Cancel == nil {
t.Fatalf("取消必须走取消裁决,不得派发别的新动作:%+v", d)
}
if d := Decide(cancel, hasCommandEnv()); d.StartFound {
t.Fatalf("取消轮不得声称会启动动作(runmgr 转轨判定依赖此位):%+v", d)
}
}
// ---- A1/U-A1c:取消路由——解出唯一靶子,否则如实留给宿主澄清 ----
// sessionEnv 是「会话里绑定了一个运行」的路由环境。
func sessionEnv(runID string) Env {
return Env{SessionRun: func() (SessionRun, bool) {
return SessionRun{RunID: runID, Intent: "demo.run-cancel", Action: "demo.run-cancel"}, runID != ""
}}
}
func cancelResult(slots map[string]string, cancel intentgate.Command) intentgate.MatchResult {
return intentgate.MatchResult{Branch: "matched", ID: "demo.task-cancel", Slots: slots,
Commands: []intentgate.Command{cancel}}
}
func TestCancelResolvesTargetFromTextAndSession(t *testing.T) {
cases := []struct {
name string
in intentgate.MatchResult
env Env
runID string
source CancelSource
}{
{
name: "命令点名",
in: cancelResult(nil, intentgate.Command{Kind: intentgate.CommandCancel, Value: "run_7"}),
env: hasCommandEnv(),
runID: "run_7",
source: CancelSourceExplicit,
},
{
name: "槽位里的运行号",
in: cancelResult(map[string]string{"run_id": "run_8"}, intentgate.Command{Kind: intentgate.CommandCancel}),
env: hasCommandEnv(),
runID: "run_8",
source: CancelSourceExplicit,
},
{
name: "会话绑定的运行",
in: cancelResult(nil, intentgate.Command{Kind: intentgate.CommandCancel}),
env: sessionEnv("run_9"),
runID: "run_9",
source: CancelSourceSession,
},
{
// 用户点名优先于会话绑定(确定性:显式意图高于隐式上下文)。
name: "显式优先于会话",
in: cancelResult(map[string]string{"run_id": "run_8"},
intentgate.Command{Kind: intentgate.CommandCancel}),
env: sessionEnv("run_9"),
runID: "run_8",
source: CancelSourceExplicit,
},
}
for _, c := range cases {
d := Decide(c.in, c.env)
if d.Kind != KindCancel || d.Cancel == nil {
t.Fatalf("%s:取消未走取消裁决:%+v", c.name, d)
}
if d.Cancel.RunID != c.runID || d.Cancel.Source != c.source {
t.Fatalf("%s:靶子/来源 = %q/%q;want %q/%q", c.name, d.Cancel.RunID, d.Cancel.Source, c.runID, c.source)
}
if d.CommandID != "" || d.Reason != "" || d.Clarification.Needed {
t.Fatalf("%s:解出靶子后不得再回落/再澄清:%+v", c.name, d)
}
if d.Cancel.Intent != "demo.task-cancel" {
t.Fatalf("%s:意图 id 未透传:%+v", c.name, d.Cancel)
}
}
}
// 没有唯一靶子(无绑定、无点名)时如实留空,绝不替用户猜一个。
func TestCancelWithoutTargetAsksForClarification(t *testing.T) {
envs := map[string]Env{
"无环境": {},
"无绑定": sessionEnv(""),
"绑定为空运行": {SessionRun: func() (SessionRun, bool) { return SessionRun{}, true }},
}
for name, env := range envs {
d := Decide(cancelResult(nil, intentgate.Command{Kind: intentgate.CommandCancel}), env)
if d.Kind != KindCancel {
t.Fatalf("%s:没有靶子也应给出取消裁决(宿主据此澄清):%+v", name, d)
}
if d.Cancel.RunID != "" || d.Reason == "" {
t.Fatalf("%s:无靶子必须如实留空并写明原因:%+v", name, d)
}
if d.CommandID != "" {
t.Fatalf("%s:取消轮不得派发别的新动作:%+v", name, d)
}
}
}
// 库层随取消给出的候选如实透出(不在路由层做取舍)。
func TestCancelKeepsLibraryCandidates(t *testing.T) {
d := Decide(cancelResult(nil, intentgate.Command{Kind: intentgate.CommandCancel,
Candidates: []string{"run_1", "run_2"}}), Env{})
if d.Cancel == nil || len(d.Cancel.Candidates) != 2 {
t.Fatalf("候选未透传:%+v", d.Cancel)
}
if d.Cancel.RunID != "" {
t.Fatalf("多个候选时不得替用户选一个:%+v", d.Cancel)
}
}
// 非 matched 分支的取消不产生任何靶子(与既有回落逐字节一致)。
func TestCancelIgnoredOnNonMatched(t *testing.T) {
for _, branch := range []string{"fallback", "guard_hit", "ambiguous", "expectation_hit"} {
d := Decide(intentgate.MatchResult{Branch: branch, ID: "demo.task-cancel",
Commands: []intentgate.Command{{Kind: intentgate.CommandCancel, Value: "run_7"}}}, sessionEnv("run_9"))
if d.Kind != KindTurn || d.Cancel != nil {
t.Fatalf("%s:非命中不得产生取消裁决:%+v", branch, d)
}
}
}
// Env 未接线会话(nil hook)时行为与 U-A1b 一致:没有靶子,如实澄清。
func TestCancelWithoutSessionHookKeepsLegacyContract(t *testing.T) {
d := Decide(cancelResult(nil, intentgate.Command{Kind: intentgate.CommandCancel}), Env{})
if d.Cancel == nil || d.Cancel.RunID != "" {
t.Fatalf("无会话接线不得臆造靶子:%+v", d)
}
}
+488
View File
@@ -0,0 +1,488 @@
// Package rundiag 把单次 run 的诊断证据聚合成一份只读结论:卡在哪、谁负责、
// 为什么在等、最后一个稳定检查点、已知副作用、下一步动作。
//
// 全包统一口径(任务卡 A2 的契约):
// - 只读派生:Build 是纯函数,Gather 只跑 SELECT;本包不提供任何写路径。
// - 证据优先于推断:状态只由已落库的记录驱动。读不到就降级为 unknown + note,
// 读不到隔离记录时更按 fail-closed 处理(不声称"可以安全重试")。
// - 「没有记录」≠「没有发生」:side_effects 为空必须带 note,说明这是"未记录"
// 而不是"未发生"。
// - 归属边界:HITL 决策卡、危险命令确认卡、命令审计信封都按 owner + 会话归属,
// 其中决策卡与确认卡表里根本没有 run 列,审计信封虽有 runId 字段但宿主总线
// 当前不写。因此"归到这个 run"是调用方按签发时间窗做的有界推断,报告以
// note 标出,绝不把会话里更早的记录算成本次 run 的副作用。
package rundiag
import (
"encoding/json"
"errors"
"fmt"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// responsibility.state 的取值域(任务卡固定)。
const (
StateRunning = "running"
StateWaitingHITL = "waiting_hitl"
StateConfirmationPending = "confirmation_pending"
StateRecoveryHold = "recovery_hold"
StateReconciling = "reconciling"
StateFailed = "failed"
StateCompleted = "completed"
StateUnknown = "unknown"
)
// responsibility.role:这一刻该谁动。
const (
RoleRunner = "runner" // 执行器自己在推进
RoleHuman = "human" // 等人裁决(HITL / 危险命令确认)
RoleOperationOwner = "operation-owner" // 等副作用责任人对账与解除
RoleNone = "none" // 没有待办责任方
)
// next_step.action 的取值域。
const (
ActionConfirmCard = "confirm_card"
ActionReleaseRecoveryHold = "release_recovery_hold"
ActionSafeRetry = "safe_retry"
ActionNeedsManualReconciliation = "needs_manual_reconciliation"
ActionNone = "none"
)
// side_effects[].kind:证据来源类别。
const (
EffectOperation = "operation" // 命令审计信封(U7b 对账记录)
EffectRecoveryHold = "recovery_hold" // 恢复隔离行(U3b 的"副作用未知"证据)
)
// 读取上界:诊断是给人和值班看的,不是全量导出。
const (
maxSideEffects = 50
maxAuditScan = 200 // 审计信封按会话+时间窗有界扫描
maxConfirms = 50
)
// Report 是诊断响应(JSON 契约,字段名用任务卡给定的 snake_case)。
type Report struct {
Run RunSummary `json:"run"`
Responsibility Responsibility `json:"responsibility"`
LastCheckpoint *Checkpoint `json:"last_checkpoint"` // 无记录时为 null(不编造)
SideEffects []SideEffect `json:"side_effects"` // 恒为数组:空数组 = 未记录到
NextStep NextStep `json:"next_step"`
Notes []string `json:"notes,omitempty"` // 降级与边界说明
}
// RunSummary 是 run 自身的身份与时间戳。updated_at 的口径:已终态取 ended_at,
// 未终态取最后一个已持久化事件的时间;两者都没有则为 null(并在 notes 说明)。
type RunSummary struct {
ID string `json:"id"`
Status string `json:"status"`
Kind string `json:"kind,omitempty"`
Phase string `json:"phase,omitempty"`
Owner string `json:"owner"`
Project string `json:"project"`
Conversation string `json:"conversation"`
CreatedAt string `json:"created_at"`
UpdatedAt *string `json:"updated_at"`
}
// Responsibility 回答"卡在哪、谁负责、为什么等、从什么时候开始等"。
type Responsibility struct {
State string `json:"state"`
Role string `json:"role"`
WaitReason string `json:"wait_reason"`
Since string `json:"since,omitempty"` // 进入该状态的时间;存储没记就省略
}
// Checkpoint 指向最后一个稳定落盘点(事件流上的 seq 可直接作重放游标)。
type Checkpoint struct {
ID string `json:"id"`
At string `json:"at"`
Type string `json:"type"`
}
// SideEffect 是一条已记录的外部副作用证据。operation 行为命令回执对账记录;
// recovery_hold 行是 U3b 落下的"副作用结果未知"隔离证据(state = held/released)。
type SideEffect struct {
Kind string `json:"kind"`
OperationID string `json:"operation_id,omitempty"`
CommandID string `json:"command_id,omitempty"`
State string `json:"state"`
At string `json:"at,omitempty"`
ReleasedAt string `json:"released_at,omitempty"`
}
// NextStep 是唯一建议的下一步动作(诊断面只建议,不执行)。
type NextStep struct {
Action string `json:"action"`
CardID string `json:"card_id,omitempty"` // confirm_card 可解析时给出卡 ID
Reason string `json:"reason"`
}
// Source 是诊断所需的只读端口(*store.Store 直接满足;测试可替身)。
type Source interface {
GetRecoveryHold(runID string) (store.RecoveryHold, error)
ActiveHITLCardsByRun(runID string) ([]core.HITLCard, error)
PendingHumanDecisionsByConversation(owner, conversationID string, now time.Time) ([]store.HumanDecision, error)
ListOwnerPendingConfirms(owner string, limit int) ([]store.PendingConfirm, error)
LastRunCheckpoint(conversationID, runID string) (store.RunCheckpoint, bool, error)
ListAudits(owner, conversationID string, limit int) ([]agentcommand.AuditEnvelope, error)
}
// Evidence 是 Build 的全部输入(合成数据即可表驱动测试,无需真实存储)。
type Evidence struct {
Run core.Run
Hold *store.RecoveryHold // nil = 没有隔离行;HoldUnknown = 读失败
HoldUnknown bool
Cards []core.HITLCard // 该 run 未关闭的 HITL 卡
Decisions []store.HumanDecision // 会话内未过期的 pending 决策卡
Confirms []store.PendingConfirm // owner 名下未确认的危险命令挂起行
Checkpoint *store.RunCheckpoint // 最后一个已持久化事件
Audits []agentcommand.AuditEnvelope
Notes []string
}
// Gather 只读地把一个 run 的证据读齐。任何一条可选查询失败都不让整个诊断失败:
// 失败降级为 note;隔离记录读不到按 fail-closed 标记(HoldUnknown),因为"读不到
// hold"绝不能被读成"没有 hold"。
func Gather(src Source, run core.Run, now time.Time) Evidence {
ev := Evidence{Run: run}
if src == nil {
ev.HoldUnknown = true
return ev.withNote("diagnostics source unavailable; responsibility is fail-closed unknown")
}
owner := string(run.Owner)
conversation := string(run.ConversationID)
hold, err := src.GetRecoveryHold(string(run.ID))
switch {
case err == nil:
ev.Hold = &hold
case errors.Is(err, store.ErrNotFound):
// 没有隔离行是正常态,不记 note。
default:
ev.HoldUnknown = true
ev = ev.withNote("recovery hold unreadable: " + err.Error())
}
cards, err := src.ActiveHITLCardsByRun(string(run.ID))
if err != nil {
ev = ev.withNote("hitl cards unreadable: " + err.Error())
} else {
ev.Cards = cards
}
decisions, err := src.PendingHumanDecisionsByConversation(owner, conversation, now)
if err != nil {
ev = ev.withNote("human decisions unreadable: " + err.Error())
} else {
ev.Decisions = decisions
}
confirms, err := src.ListOwnerPendingConfirms(owner, maxConfirms)
if err != nil {
ev = ev.withNote("pending confirmations unreadable: " + err.Error())
} else {
ev.Confirms = confirms
}
cp, ok, err := src.LastRunCheckpoint(conversation, string(run.ID))
switch {
case err != nil:
ev = ev.withNote("last checkpoint unreadable: " + err.Error())
case ok:
ev.Checkpoint = &cp
}
audits, err := src.ListAudits(owner, conversation, maxAuditScan)
if err != nil {
ev = ev.withNote("command audit unreadable: " + err.Error())
} else {
ev.Audits = audits
}
return ev
}
// Build 把证据聚合成诊断结论(纯函数,不读写任何状态)。
func Build(ev Evidence) Report {
rep := Report{Run: runSummary(ev), SideEffects: []SideEffect{}}
if ev.Checkpoint != nil {
rep.LastCheckpoint = &Checkpoint{ID: fmt.Sprintf("%d", ev.Checkpoint.Seq), At: ev.Checkpoint.TS, Type: ev.Checkpoint.Type}
}
rep.SideEffects = sideEffects(ev)
rep.Notes = append([]string(nil), ev.Notes...)
card := waitingCard(ev)
decision := waitingDecision(ev)
confirm := waitingConfirm(ev)
switch {
case ev.HoldUnknown:
// 隔离记录读不到:不猜没有 hold,直接 unknown,且不给"安全重试"。
rep.Notes = append(rep.Notes, "recovery hold state unknown; treat as quarantined until reconciled")
rep.Responsibility = Responsibility{
State: StateUnknown, Role: RoleNone,
WaitReason: "recovery hold state could not be read",
}
rep.NextStep = NextStep{Action: ActionNone, Reason: "cannot prove the run is free of an open recovery hold"}
case openHold(ev):
hold := ev.Hold
rep.Responsibility = Responsibility{
State: StateRecoveryHold, Role: RoleOperationOwner,
WaitReason: holdReason(hold), Since: hold.CreatedAt,
}
rep.NextStep = NextStep{Action: ActionReleaseRecoveryHold, Reason: holdReason(hold)}
case needsReconcile(ev):
reason := "recovery hold released but the run is still interrupted"
since := ""
if ev.Hold != nil {
since = ev.Hold.ReleasedAt
if ev.Hold.Reason != "" {
reason += "; hold reason: " + ev.Hold.Reason
}
}
rep.Responsibility = Responsibility{State: StateReconciling, Role: RoleOperationOwner, WaitReason: reason, Since: since}
rep.NextStep = NextStep{Action: ActionNeedsManualReconciliation, Reason: reason}
case ev.Run.Status == core.RunWaitingHITL || card != nil || decision != nil:
reason := "waiting for a human decision"
since := ""
if decision != nil {
reason = decision.Question
if reason == "" {
reason = "pending human decision on command " + decision.CommandID
}
since = decision.IssuedAt.UTC().Format(time.RFC3339)
} else if card != nil {
reason = card.Question
if reason == "" {
reason = "run is waiting on a HITL card"
}
since = card.CreatedAt
}
rep.Responsibility = Responsibility{State: StateWaitingHITL, Role: RoleHuman, WaitReason: reason, Since: since}
if card == nil && decision == nil {
// 只由 run 状态驱动:没有可解析的卡,也不知道等待从何时开始。
rep.Notes = append(rep.Notes, "run is marked waiting_hitl with no open card recorded; the wait entry time is not stored")
}
rep.NextStep = NextStep{Action: ActionConfirmCard, CardID: hitlCardID(card, decision), Reason: reason}
case confirm != nil:
reason := "dangerous command " + confirm.command + " awaits explicit confirmation"
rep.Responsibility = Responsibility{
State: StateConfirmationPending, Role: RoleHuman,
WaitReason: reason, Since: confirm.row.CreatedAt,
}
rep.NextStep = NextStep{Action: ActionConfirmCard, CardID: confirm.row.CardID, Reason: reason}
case ev.Run.Status == core.RunFailed:
reason := "run failed"
if ev.Run.Error != nil {
reason = fmt.Sprintf("run failed: %s: %s", ev.Run.Error.Code, ev.Run.Error.Message)
}
rep.Responsibility = Responsibility{State: StateFailed, Role: RoleRunner, WaitReason: reason, Since: endedAt(ev.Run)}
// 重试安全性看已记录的副作用证据:没有 open hold 且没有 unknown 回执时,
// 重试不会复制本次运行的后果;否则先对账再重试,不说"安全"。
if unknown := unknownOperations(rep.SideEffects); unknown > 0 {
reason := fmt.Sprintf("terminal failure, but %d recorded operation(s) have an unknown outcome; reconcile side effects before retrying", unknown)
rep.Notes = append(rep.Notes, reason)
rep.NextStep = NextStep{Action: ActionNeedsManualReconciliation, Reason: reason}
} else {
rep.NextStep = NextStep{Action: ActionSafeRetry, Reason: "terminal failure with no open recovery hold and no unknown receipt; retry is safe to attempt"}
}
case ev.Run.Status == core.RunCompleted:
rep.Responsibility = Responsibility{State: StateCompleted, Role: RoleNone, WaitReason: "run completed", Since: endedAt(ev.Run)}
rep.NextStep = NextStep{Action: ActionNone, Reason: "run already completed"}
case ev.Run.Status == core.RunRunning || ev.Run.Status == core.RunPending:
rep.Responsibility = Responsibility{
State: StateRunning, Role: RoleRunner,
WaitReason: "run is executing", Since: ev.Run.CreatedAt.UTC().Format(time.RFC3339),
}
rep.NextStep = NextStep{Action: ActionNone, Reason: "runner is executing; no operator action"}
default:
reason := fmt.Sprintf("run status %q has no diagnostic mapping", ev.Run.Status)
rep.Notes = append(rep.Notes, reason)
rep.Responsibility = Responsibility{State: StateUnknown, Role: RoleNone, WaitReason: reason, Since: endedAt(ev.Run)}
rep.NextStep = NextStep{Action: ActionNone, Reason: "status is unmapped; inspect the run record manually"}
}
if ev.Checkpoint == nil {
rep.Notes = append(rep.Notes, "no persisted checkpoint reference for this run (no durable stream event)")
}
if len(rep.SideEffects) == 0 {
rep.Notes = append(rep.Notes, "no recorded side effect evidence; absence of records is not proof of no effect")
}
return rep
}
func runSummary(ev Evidence) RunSummary {
summary := RunSummary{
ID: string(ev.Run.ID), Status: string(ev.Run.Status), Kind: string(ev.Run.Kind),
Phase: string(ev.Run.Phase), Owner: string(ev.Run.Owner), Project: string(ev.Run.ProjectID),
Conversation: string(ev.Run.ConversationID),
}
if !ev.Run.CreatedAt.IsZero() {
summary.CreatedAt = ev.Run.CreatedAt.UTC().Format(time.RFC3339)
}
// updated:终态取 ended_at,未终态取最后一次落盘(检查点)的时间。
if ended := endedAt(ev.Run); ended != "" {
summary.UpdatedAt = &ended
} else if ev.Checkpoint != nil && ev.Checkpoint.TS != "" {
ts := ev.Checkpoint.TS
summary.UpdatedAt = &ts
}
return summary
}
// sideEffects 汇总已记录的外部副作用证据:命令回执对账记录(U7b 审计信封)加上
// 恢复隔离行(U3b 的"结果未知"证据)。
func sideEffects(ev Evidence) []SideEffect {
out := []SideEffect{}
if ev.Hold != nil {
state := "held"
if ev.Hold.ReleasedAt != "" {
state = "released"
}
out = append(out, SideEffect{
Kind: EffectRecoveryHold, State: state, At: ev.Hold.CreatedAt, ReleasedAt: ev.Hold.ReleasedAt,
})
}
for _, env := range ev.Audits {
if !belongsToRun(ev, env) {
continue
}
out = append(out, SideEffect{
Kind: EffectOperation, OperationID: env.OperationID, CommandID: env.CommandID,
State: string(env.State), At: env.Time.UTC().Format(time.RFC3339),
})
if len(out) >= maxSideEffects {
break
}
}
return out
}
// belongsToRun 是审计信封的归属边界:信封自带 runId 时以它为准;否则只在"同一
// owner + 会话且签发时间不早于 run 创建时间"的窗口内归属——会话里更早的记录属于
// 更早的 run,不算进本次副作用。
func belongsToRun(ev Evidence, env agentcommand.AuditEnvelope) bool {
if env.RunID != "" {
return env.RunID == string(ev.Run.ID)
}
if string(env.ConversationID) != "" && string(env.ConversationID) != string(ev.Run.ConversationID) {
return false
}
if !ev.Run.CreatedAt.IsZero() && !env.Time.IsZero() && env.Time.Before(ev.Run.CreatedAt) {
return false
}
return true
}
func openHold(ev Evidence) bool {
return ev.Hold != nil && ev.Hold.ReleasedAt == ""
}
// unknownOperations 统计结果未知的已记录 operation(四态语义里的 unknown:
// 执行过但结果不确定)。它是"重试会不会复制副作用"的直接证据。
func unknownOperations(effects []SideEffect) int {
n := 0
for _, e := range effects {
if e.Kind == EffectOperation && e.State == string(agentcommand.StateUnknown) {
n++
}
}
return n
}
// needsReconcile 是 U3b 隔离解除后的残留态:隔离已解除,但 run 仍停在中断态——
// 没有人把它推进到终态,需要人工对账。
func needsReconcile(ev Evidence) bool {
return ev.Hold != nil && ev.Hold.ReleasedAt != "" && ev.Run.Status == core.RunInterrupted
}
func holdReason(hold *store.RecoveryHold) string {
if hold == nil || hold.Reason == "" {
return "external side effect outcome unknown (recovery hold)"
}
return hold.Reason
}
func endedAt(run core.Run) string {
if run.EndedAt == nil {
return ""
}
return run.EndedAt.UTC().Format(time.RFC3339)
}
// waitingCard 取该 run 未关闭的 HITL 卡(最早一张;只看未关闭的)。
func waitingCard(ev Evidence) *core.HITLCard {
for i := range ev.Cards {
if ev.Cards[i].Resolved || ev.Cards[i].Blocked {
continue
}
return &ev.Cards[i]
}
return nil
}
// waitingDecision 取本 run 生命周期内签发、仍未过期的人工决策卡。
func waitingDecision(ev Evidence) *store.HumanDecision {
for i := range ev.Decisions {
rec := &ev.Decisions[i]
if rec.Status != "pending" {
continue
}
if !rec.IssuedAt.IsZero() && !ev.Run.CreatedAt.IsZero() && rec.IssuedAt.Before(ev.Run.CreatedAt) {
continue // 签发在本次 run 之前:属于更早的等待,不算本次
}
return rec
}
return nil
}
type scopedConfirm struct {
row store.PendingConfirm
command string
}
// waitingConfirm 取本 run 生命周期内、同一会话挂起的危险命令确认行。载荷是总线
// 落下的分发请求;解析不出会话的旧行按"无记录"处理,不回显原始载荷。
func waitingConfirm(ev Evidence) *scopedConfirm {
for _, row := range ev.Confirms {
req := cmdbus.Request{}
if err := json.Unmarshal([]byte(row.Payload), &req); err != nil {
continue
}
if req.ConversationID != "" && req.ConversationID != string(ev.Run.ConversationID) {
continue
}
if req.Owner != "" && string(ev.Run.Owner) != "" && req.Owner != string(ev.Run.Owner) {
continue
}
created, err := time.Parse(time.RFC3339, row.CreatedAt)
if err == nil && !ev.Run.CreatedAt.IsZero() && created.Before(ev.Run.CreatedAt) {
continue // 挂起早于本次 run:属于更早的等待
}
return &scopedConfirm{row: row, command: req.ID}
}
return nil
}
// hitlCardID 给出可解析的卡标识:HITL 卡优先,其次决策卡 ID。
func hitlCardID(card *core.HITLCard, decision *store.HumanDecision) string {
if card != nil {
return string(card.ID)
}
if decision != nil {
return decision.ID
}
return ""
}
func (ev Evidence) withNote(note string) Evidence {
ev.Notes = append(ev.Notes, note)
return ev
}
+440
View File
@@ -0,0 +1,440 @@
package rundiag
import (
"encoding/json"
"errors"
"strings"
"testing"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/store"
core "agent-system-core"
)
var (
baseTime = time.Date(2026, 10, 20, 10, 0, 0, 0, time.UTC)
afterRun = baseTime.Add(5 * time.Minute)
)
// fakeSource 是只读端口的合成替身:每个字段都能单独触发"没有记录"或"读取失败"。
type fakeSource struct {
hold store.RecoveryHold
holdErr error
cards []core.HITLCard
decisions []store.HumanDecision
confirms []store.PendingConfirm
cp store.RunCheckpoint
cpOK bool
audits []agentcommand.AuditEnvelope
}
func (f fakeSource) GetRecoveryHold(string) (store.RecoveryHold, error) {
if f.holdErr != nil {
return store.RecoveryHold{}, f.holdErr
}
if f.hold.RunID == "" {
return store.RecoveryHold{}, store.ErrNotFound
}
return f.hold, nil
}
func (f fakeSource) ActiveHITLCardsByRun(string) ([]core.HITLCard, error) { return f.cards, nil }
func (f fakeSource) PendingHumanDecisionsByConversation(string, string, time.Time) ([]store.HumanDecision, error) {
return f.decisions, nil
}
func (f fakeSource) ListOwnerPendingConfirms(string, int) ([]store.PendingConfirm, error) {
return f.confirms, nil
}
func (f fakeSource) LastRunCheckpoint(string, string) (store.RunCheckpoint, bool, error) {
return f.cp, f.cpOK, nil
}
func (f fakeSource) ListAudits(string, string, int) ([]agentcommand.AuditEnvelope, error) {
return f.audits, nil
}
func run(status core.RunStatus) core.Run {
return core.Run{
ID: "run_1", ConversationID: "cnv_1", ProjectID: "prj_1", Owner: "alice",
Kind: core.RunConversation, Status: status, Phase: core.PhaseResponding, CreatedAt: baseTime,
}
}
func ended(r core.Run, at time.Time) core.Run {
r.EndedAt = &at
return r
}
func checkpoint() *store.RunCheckpoint {
return &store.RunCheckpoint{Seq: 7, TS: baseTime.Add(2 * time.Minute).Format(time.RFC3339), Type: "agent.progress"}
}
// A2 契约:每个 responsibility.state 都有驱动它的记录,且 next_step 逐状态映射正确。
func TestBuildResponsibilityStatesAndNextStep(t *testing.T) {
cases := []struct {
name string
ev Evidence
state string
role string
action string
cardID string
waitHint string
since string
}{
{
name: "unreleased recovery hold wins over everything",
ev: Evidence{
Run: run(core.RunRunning),
Hold: &store.RecoveryHold{RunID: "run_1", Owner: "alice", Reason: "unknown external effect",
CreatedAt: afterRun.Format(time.RFC3339)},
Cards: []core.HITLCard{{ID: "card_hitl", Resolved: false, Question: "approve?"}},
},
state: StateRecoveryHold, role: RoleOperationOwner, action: ActionReleaseRecoveryHold,
waitHint: "unknown external effect", since: afterRun.Format(time.RFC3339),
},
{
name: "released hold on an interrupted run needs reconciliation",
ev: Evidence{
Run: run(core.RunInterrupted),
Hold: &store.RecoveryHold{RunID: "run_1", Owner: "alice", Reason: "unknown external effect",
CreatedAt: afterRun.Format(time.RFC3339), ReleasedAt: afterRun.Add(time.Minute).Format(time.RFC3339)},
},
state: StateReconciling, role: RoleOperationOwner, action: ActionNeedsManualReconciliation,
waitHint: "still interrupted", since: afterRun.Add(time.Minute).Format(time.RFC3339),
},
{
name: "waiting on a HITL card",
ev: Evidence{
Run: run(core.RunWaitingHITL),
Cards: []core.HITLCard{{ID: "card_hitl", CreatedAt: afterRun.Format(time.RFC3339), Question: "approve the plan?"}},
},
state: StateWaitingHITL, role: RoleHuman, action: ActionConfirmCard, cardID: "card_hitl",
waitHint: "approve the plan?", since: afterRun.Format(time.RFC3339),
},
{
name: "waiting on a pending human decision without a run card",
ev: Evidence{
Run: run(core.RunRunning),
Decisions: []store.HumanDecision{{ID: "hd_1", CommandID: "rdm.confirm", Status: "pending",
Question: "approve delivery?", IssuedAt: afterRun, ExpiresAt: afterRun.Add(time.Hour)}},
},
state: StateWaitingHITL, role: RoleHuman, action: ActionConfirmCard, cardID: "hd_1",
waitHint: "approve delivery?", since: afterRun.Format(time.RFC3339),
},
{
name: "dangerous command awaiting confirmation",
ev: Evidence{
Run: run(core.RunRunning),
Confirms: []store.PendingConfirm{{CardID: "card_cmd", Owner: "alice", CreatedAt: afterRun.Format(time.RFC3339),
Payload: mustRequest(t, cmdbus.Request{ID: "rdm.docs.instantiate", Owner: "alice", ConversationID: "cnv_1"})}},
},
state: StateConfirmationPending, role: RoleHuman, action: ActionConfirmCard, cardID: "card_cmd",
waitHint: "rdm.docs.instantiate", since: afterRun.Format(time.RFC3339),
},
{
name: "terminal failure is a safe retry",
ev: Evidence{
Run: ended(func() core.Run {
r := run(core.RunFailed)
r.Error = &core.ErrorPart{Code: "run.tool_failed", Message: "tool crashed"}
return r
}(), afterRun),
},
state: StateFailed, role: RoleRunner, action: ActionSafeRetry,
waitHint: "run.tool_failed: tool crashed", since: afterRun.Format(time.RFC3339),
},
{
name: "terminal failure with an unknown receipt needs reconciliation, not a retry",
ev: Evidence{
Run: ended(run(core.RunFailed), afterRun),
Audits: []agentcommand.AuditEnvelope{{Time: afterRun, CommandID: "rdm.docs.instantiate",
OperationID: "op_unknown", State: agentcommand.StateUnknown, ConversationID: "cnv_1"}},
},
state: StateFailed, role: RoleRunner, action: ActionNeedsManualReconciliation,
waitHint: "run failed",
},
{
name: "completed needs no action",
ev: Evidence{Run: ended(run(core.RunCompleted), afterRun)},
state: StateCompleted, role: RoleNone, action: ActionNone,
waitHint: "completed", since: afterRun.Format(time.RFC3339),
},
{
name: "executing run stays with the runner",
ev: Evidence{Run: run(core.RunRunning)},
state: StateRunning, role: RoleRunner, action: ActionNone,
waitHint: "executing", since: baseTime.Format(time.RFC3339),
},
{
name: "pending run reads as running",
ev: Evidence{Run: run(core.RunPending)},
state: StateRunning, role: RoleRunner, action: ActionNone,
waitHint: "executing",
},
{
name: "cancelled has no mapping and degrades to unknown",
ev: Evidence{Run: ended(run(core.RunCancelled), afterRun)},
state: StateUnknown, role: RoleNone, action: ActionNone,
waitHint: "no diagnostic mapping",
},
{
name: "interrupted without any hold record degrades to unknown",
ev: Evidence{Run: run(core.RunInterrupted)},
state: StateUnknown, role: RoleNone, action: ActionNone,
waitHint: "no diagnostic mapping",
},
{
name: "unreadable hold is fail-closed unknown, never a safe retry",
ev: Evidence{Run: run(core.RunFailed), HoldUnknown: true},
state: StateUnknown, role: RoleNone, action: ActionNone,
waitHint: "could not be read",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
rep := Build(tc.ev)
if rep.Responsibility.State != tc.state {
t.Fatalf("state=%q want %q (responsibility=%+v)", rep.Responsibility.State, tc.state, rep.Responsibility)
}
if rep.Responsibility.Role != tc.role {
t.Fatalf("role=%q want %q", rep.Responsibility.Role, tc.role)
}
if rep.NextStep.Action != tc.action {
t.Fatalf("next_step=%q want %q (reason=%q)", rep.NextStep.Action, tc.action, rep.NextStep.Reason)
}
if rep.NextStep.CardID != tc.cardID {
t.Fatalf("next_step.card_id=%q want %q", rep.NextStep.CardID, tc.cardID)
}
if tc.waitHint != "" && !strings.Contains(rep.Responsibility.WaitReason, tc.waitHint) {
t.Fatalf("wait_reason=%q does not contain %q", rep.Responsibility.WaitReason, tc.waitHint)
}
if tc.since != "" && rep.Responsibility.Since != tc.since {
t.Fatalf("since=%q want %q", rep.Responsibility.Since, tc.since)
}
if rep.NextStep.Reason == "" {
t.Fatal("next_step.reason must never be empty")
}
})
}
}
// 副作用聚合:命令回执对账记录并入;会话里更早的记录不算本次 run 的副作用。
func TestBuildSideEffectsAggregateRecordedOperations(t *testing.T) {
ev := Evidence{
Run: run(core.RunFailed),
Audits: []agentcommand.AuditEnvelope{
{Time: baseTime.Add(-time.Hour), CommandID: "rdm.reqs.add", OperationID: "op_older",
State: agentcommand.StateCommitted, ConversationID: "cnv_1"},
{Time: afterRun, CommandID: "rdm.docs.instantiate", OperationID: "op_1",
State: agentcommand.StateUnknown, ConversationID: "cnv_1"},
{Time: afterRun, CommandID: "other.run.command", OperationID: "op_2",
State: agentcommand.StateCommitted, ConversationID: "cnv_other"},
{Time: afterRun, CommandID: "rdm.trace.confirm", OperationID: "op_3",
State: agentcommand.StateAccepted, ConversationID: "cnv_1", RunID: "run_1"},
},
}
rep := Build(ev)
if len(rep.SideEffects) != 2 {
t.Fatalf("side effects must hold only this run's receipts: %+v", rep.SideEffects)
}
got := map[string]SideEffect{}
for _, se := range rep.SideEffects {
got[se.OperationID] = se
}
recorded, ok := got["op_1"]
if !ok || recorded.Kind != EffectOperation || recorded.CommandID != "rdm.docs.instantiate" || recorded.State != string(agentcommand.StateUnknown) {
t.Fatalf("recorded operation missing from aggregation: %+v", rep.SideEffects)
}
if _, ok := got["op_older"]; ok {
t.Fatal("a receipt issued before the run started must not be attributed to it")
}
if _, ok := got["op_2"]; ok {
t.Fatal("a receipt from another conversation must not be attributed to this run")
}
if _, ok := got["op_3"]; !ok {
t.Fatal("a receipt carrying this run id must be attributed to it")
}
}
// hold 行本身也是"副作用结果未知"的持久证据(U3b),并标出是否已解除。
func TestBuildSideEffectsIncludeRecoveryHoldEvidence(t *testing.T) {
ev := Evidence{Run: run(core.RunInterrupted), Hold: &store.RecoveryHold{RunID: "run_1", Owner: "alice",
Reason: "unknown", CreatedAt: afterRun.Format(time.RFC3339)}}
rep := Build(ev)
if len(rep.SideEffects) != 1 || rep.SideEffects[0].Kind != EffectRecoveryHold || rep.SideEffects[0].State != "held" {
t.Fatalf("open hold must show as held evidence: %+v", rep.SideEffects)
}
ev.Hold.ReleasedAt = afterRun.Add(time.Minute).Format(time.RFC3339)
rep = Build(ev)
if rep.SideEffects[0].State != "released" || rep.SideEffects[0].ReleasedAt == "" {
t.Fatalf("released hold must keep its release time: %+v", rep.SideEffects[0])
}
}
// 没有落盘检查点时给 null + note,绝不编一个;空副作用同样带 note。
func TestBuildNeverFabricatesCheckpointOrAbsence(t *testing.T) {
rep := Build(Evidence{Run: run(core.RunRunning)})
if rep.LastCheckpoint != nil {
t.Fatalf("checkpoint must stay null when nothing is persisted: %+v", rep.LastCheckpoint)
}
if !containsNote(rep.Notes, "checkpoint") {
t.Fatalf("missing checkpoint note: %v", rep.Notes)
}
if !containsNote(rep.Notes, "absence of records is not proof") {
t.Fatalf("missing side effect absence note: %v", rep.Notes)
}
if rep.SideEffects == nil {
t.Fatal("side_effects must serialize as an array, never null")
}
withCheckpoint := Build(Evidence{Run: run(core.RunRunning), Checkpoint: checkpoint()})
if withCheckpoint.LastCheckpoint == nil || withCheckpoint.LastCheckpoint.ID != "7" || withCheckpoint.Run.UpdatedAt == nil {
t.Fatalf("persisted checkpoint must be surfaced: %+v", withCheckpoint)
}
if *withCheckpoint.Run.UpdatedAt != checkpoint().TS {
t.Fatalf("updated_at must fall back to the last persisted event: %v", *withCheckpoint.Run.UpdatedAt)
}
}
// Gather:只读端口的读齐与降级(读不到隔离记录按 fail-closed)。
func TestGatherReadsEvidenceAndDegradesOnError(t *testing.T) {
src := fakeSource{
hold: store.RecoveryHold{RunID: "run_1", Owner: "alice", Reason: "unknown", CreatedAt: afterRun.Format(time.RFC3339)},
cards: []core.HITLCard{{ID: "card_hitl"}},
cp: store.RunCheckpoint{Seq: 3, TS: afterRun.Format(time.RFC3339), Type: "agent.progress"},
cpOK: true,
audits: []agentcommand.AuditEnvelope{{Time: afterRun, CommandID: "rdm.confirm", State: agentcommand.StateAccepted}},
confirms: []store.PendingConfirm{{CardID: "card_cmd", Owner: "alice", CreatedAt: afterRun.Format(time.RFC3339), Payload: "{}"}},
}
ev := Gather(src, run(core.RunRunning), afterRun.Add(time.Minute))
if ev.Hold == nil || ev.HoldUnknown || len(ev.Cards) != 1 || ev.Checkpoint == nil || len(ev.Audits) != 1 || len(ev.Confirms) != 1 {
t.Fatalf("gather missed evidence: %+v", ev)
}
if rep := Build(ev); rep.Responsibility.State != StateRecoveryHold {
t.Fatalf("state=%q want %q", rep.Responsibility.State, StateRecoveryHold)
}
degraded := Gather(fakeSource{holdErr: errors.New("db gone")}, run(core.RunRunning), afterRun)
if !degraded.HoldUnknown || degraded.Hold != nil {
t.Fatalf("an unreadable hold must stay unknown, never absent: %+v", degraded)
}
rep := Build(degraded)
if rep.Responsibility.State != StateUnknown || rep.NextStep.Action == ActionSafeRetry {
t.Fatalf("unreadable hold must not yield a retry: %+v", rep)
}
if !containsNote(rep.Notes, "fail-closed") && !containsNote(rep.Notes, "quarantined") {
t.Fatalf("missing fail-closed note: %v", rep.Notes)
}
// 没有隔离行是正常态:不产生噪音 note。
normal := Build(Gather(fakeSource{}, run(core.RunRunning), afterRun))
if containsNote(normal.Notes, "recovery hold unreadable") {
t.Fatalf("a missing hold row is not a degraded read: %v", normal.Notes)
}
}
// 契约形状:snake_case 字段名固定,last_checkpoint 为 null 时显式出现。
func TestReportJSONContract(t *testing.T) {
raw, err := json.Marshal(Build(Evidence{Run: run(core.RunRunning), Checkpoint: checkpoint()}))
if err != nil {
t.Fatal(err)
}
var out map[string]any
if err := json.Unmarshal(raw, &out); err != nil {
t.Fatal(err)
}
for _, key := range []string{"run", "responsibility", "last_checkpoint", "side_effects", "next_step"} {
if _, ok := out[key]; !ok {
t.Fatalf("missing contract key %q in %s", key, raw)
}
}
responsibility, _ := out["responsibility"].(map[string]any)
for _, key := range []string{"state", "role", "wait_reason", "since"} {
if _, ok := responsibility[key]; !ok {
t.Fatalf("missing responsibility key %q in %s", key, raw)
}
}
runSummary, _ := out["run"].(map[string]any)
for _, key := range []string{"id", "status", "owner", "project", "conversation", "created_at", "updated_at"} {
if _, ok := runSummary[key]; !ok {
t.Fatalf("missing run key %q in %s", key, raw)
}
}
nextStep, _ := out["next_step"].(map[string]any)
for _, key := range []string{"action", "reason"} {
if _, ok := nextStep[key]; !ok {
t.Fatalf("missing next_step key %q in %s", key, raw)
}
}
empty, err := json.Marshal(Build(Evidence{Run: run(core.RunRunning)}))
if err != nil {
t.Fatal(err)
}
var nulled map[string]any
if err := json.Unmarshal(empty, &nulled); err != nil {
t.Fatal(err)
}
if nulled["last_checkpoint"] != nil {
t.Fatalf("last_checkpoint must be null (not omitted) when nothing is persisted: %s", empty)
}
if _, ok := nulled["side_effects"].([]any); !ok {
t.Fatalf("side_effects must be an array: %s", empty)
}
}
// 归属边界:会话里更早签发的决策卡与确认行不算本次 run 在等的东西。
func TestBuildIgnoresEvidenceIssuedBeforeTheRun(t *testing.T) {
ev := Evidence{
Run: run(core.RunRunning),
Decisions: []store.HumanDecision{{ID: "hd_old", Status: "pending", Question: "older?",
IssuedAt: baseTime.Add(-time.Hour), ExpiresAt: baseTime.Add(time.Hour)}},
Confirms: []store.PendingConfirm{{CardID: "card_old", Owner: "alice",
CreatedAt: baseTime.Add(-time.Hour).Format(time.RFC3339),
Payload: mustRequest(t, cmdbus.Request{ID: "rdm.confirm", Owner: "alice", ConversationID: "cnv_1"})}},
}
rep := Build(ev)
if rep.Responsibility.State != StateRunning {
t.Fatalf("state=%q want %q: earlier conversation evidence must not hold this run", rep.Responsibility.State, StateRunning)
}
}
// 只有 run 状态写着 waiting_hitl、没有可解析的卡时:状态仍然成立,但必须说清楚
// 等待起点没被存储。
func TestBuildStatusDrivenWaitHasNoCardAndSaysSo(t *testing.T) {
rep := Build(Evidence{Run: run(core.RunWaitingHITL)})
if rep.Responsibility.State != StateWaitingHITL || rep.NextStep.Action != ActionConfirmCard {
t.Fatalf("status-driven wait: %+v", rep.Responsibility)
}
if rep.NextStep.CardID != "" {
t.Fatalf("an unresolvable card must not be invented: %q", rep.NextStep.CardID)
}
if rep.Responsibility.Since != "" {
t.Fatalf("unknown wait entry time must stay empty: %q", rep.Responsibility.Since)
}
if !containsNote(rep.Notes, "wait entry time is not stored") {
t.Fatalf("missing status-driven wait note: %v", rep.Notes)
}
}
func mustRequest(t *testing.T, req cmdbus.Request) string {
t.Helper()
raw, err := json.Marshal(req)
if err != nil {
t.Fatal(err)
}
return string(raw)
}
func containsNote(notes []string, want string) bool {
for _, n := range notes {
if strings.Contains(n, want) {
return true
}
}
return false
}
+18
View File
@@ -217,6 +217,24 @@ func TestResolveHITLSecondRoundKeepsTerminalStatus(t *testing.T) {
case <-time.After(10 * time.Second):
t.Fatal("runner did not reach the second HITL question")
}
// asked 只保证 runner 已经发出提问,不保证 waiting_hitl 行与卡片都已落库;
// 与第 1 轮同一序列:等 run 处于 waiting_hitl 且存在未解决卡片后再读取。
waitUntilWithin(t, 10*time.Second, func() bool {
got, err := st.GetRunByID(run.ID)
if err != nil || got.Status != core.RunWaitingHITL {
return false
}
cards, err := st.ActiveHITLCards(conv.Owner, conv.ID)
if err != nil {
return false
}
for _, c := range cards {
if !c.Resolved {
return true
}
}
return false
})
second := pendingCard(t, st, conv, 1)
if got, err := st.GetRunByID(run.ID); err != nil || got.Status != core.RunWaitingHITL {
t.Fatalf("round 2 status=%s err=%v (want waiting_hitl)", got.Status, err)
+19 -2
View File
@@ -7,6 +7,7 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"agent-native/adapters/native"
"agent-native/runtime/tools"
@@ -25,14 +26,30 @@ type producedSink interface {
add(ref *core.ArtifactRef)
}
// artifactList 是本次运行已登记的工件表。宿主工具可能被内核并行调用,
// 因此写入与读取都在同一把锁下:并发调用既不丢条目也不读到半成品。
type artifactList struct {
mu sync.Mutex
items []*core.ArtifactRef
}
func (a *artifactList) add(ref *core.ArtifactRef) {
if ref != nil {
a.items = append(a.items, ref)
if ref == nil {
return
}
a.mu.Lock()
defer a.mu.Unlock()
a.items = append(a.items, ref)
}
// snapshot 返回当前工件的副本,供终局检查与部件装配使用。
func (a *artifactList) snapshot() []*core.ArtifactRef {
if a == nil {
return nil
}
a.mu.Lock()
defer a.mu.Unlock()
return append([]*core.ArtifactRef(nil), a.items...)
}
// bridgeHostTools 把画像允许的产品工具包成内核宿主工具。
+104 -34
View File
@@ -110,6 +110,13 @@ type toolEvidence struct {
VerifyText string // 给验证器的尾部摘要,比界面上的摘要长
}
// toolSlot 是一次在途工具调用的过程状态。
//
// 并行工具执行下槽位有两个并发访问方:工具 goroutine 回写结果字段,
// 内核事件回调(onEvent 的 tool_execution_end)在 toolMu 下读它们装部件。
// 字段一律经 directRun.writeSlot 在 toolMu 下访问,直接写即数据竞争。
// claimed 只在 toolMu 下访问:并行且无 ctx 注入的调用(宿主 UnknownTool
// 直调)按认领取槽位,两个 goroutine 不会写同一个槽位、结果不会互相覆盖。
type toolSlot struct {
call agentllm.ToolCall
start time.Time
@@ -117,6 +124,7 @@ type toolSlot struct {
verifyText string
status string
truncated bool
claimed bool // 已被某次工具执行认领(无注入路径据此避免撞槽)
artifact *core.ArtifactRef
}
@@ -565,9 +573,11 @@ func (l *LLMRunner) runDirect(rc RunCtx, stream agentllm.StreamFunc, model strin
// 摘要/状态/验证文本在这里补记,保证过程面板与终轮验证器看到同样的证据。
func (st *directRun) noteKernelResult(ctx context.Context, res agenttools.Result) {
call, ok := nruntime.ToolCallFromContext(ctx)
slot := st.slotFor("")
if ok {
var slot *toolSlot
if ok && call.ID != "" {
slot = st.slotFor(call.ID)
} else {
slot = st.slotFor("")
}
if slot == nil {
return
@@ -576,28 +586,74 @@ func (st *directRun) noteKernelResult(ctx context.Context, res agenttools.Result
if res.IsError {
summary = "error: " + summary
}
slot.summary = summary
slot.status = statusOf(nil)
status := statusOf(nil)
if res.IsError {
slot.status = "error"
status = "error"
}
slot.verifyText = tailRunes(res.Content, 600)
verifyText := tailRunes(res.Content, 600)
st.writeSlot(slot, func(s *toolSlot) {
s.summary = summary
s.status = status
s.verifyText = verifyText
})
}
// slotFor 按调用 ID 取在途槽位;ID 为空且恰有一个槽位时取之(兼容无注入
// 的调用路径)。返回的指针仅供本次工具执行内写回,不跨事件持有。
// writeSlot 在 toolMu 下回写在途槽位:工具 goroutine 写、事件回调读的是同一
// 块内存,只有同一把锁才能给出 happens-before。
func (st *directRun) writeSlot(slot *toolSlot, apply func(*toolSlot)) {
if slot == nil {
return
}
st.toolMu.Lock()
defer st.toolMu.Unlock()
apply(slot)
}
// appendFinalPart 在 toolMu 下追加最终部件:并行工具执行(工件门面卡、卡片
// 工具)与事件回调共享同一片 finalParts。
func (st *directRun) appendFinalPart(parts ...core.Part) {
st.toolMu.Lock()
defer st.toolMu.Unlock()
st.finalParts = append(st.finalParts, parts...)
}
// markExplicitCard / cardSuppressed 记「模型已自己发过卡片」:写方是卡片工具
// (保序执行,仍与并行工具 goroutine 同时在跑),读方是工件门面卡分支。
func (st *directRun) markExplicitCard() {
st.toolMu.Lock()
defer st.toolMu.Unlock()
st.explicitCard = true
}
func (st *directRun) cardSuppressed() bool {
st.toolMu.Lock()
defer st.toolMu.Unlock()
return st.explicitCard
}
// slotFor 按调用 ID 取在途槽位;ID 为空时认领一个尚未被认领的槽位(兼容无
// 注入的调用路径)。返回的指针仅供本次工具执行内写回,不跨事件持有。
func (st *directRun) slotFor(callID string) *toolSlot {
st.toolMu.Lock()
defer st.toolMu.Unlock()
if callID != "" {
return st.inflight[callID]
slot := st.inflight[callID]
if slot != nil {
slot.claimed = true
}
return slot
}
if len(st.inflight) == 1 {
for _, slot := range st.inflight {
return slot
var free *toolSlot
for _, slot := range st.inflight {
if !slot.claimed {
free = slot
break
}
}
return nil
if free != nil {
free.claimed = true
}
return free
}
// noteToolDelta 把模型正在编写的工具参数立刻转成会话事件。
@@ -725,18 +781,22 @@ func (st *directRun) onEvent(ev nruntime.Event) {
st.toolMu.Unlock()
return
}
// 锁内快照结果字段:工具回写(并行工具 goroutine)与本回调分属两个
// goroutine,解锁后再读槽位字段即数据竞争,事件只带快照走。
call, start := slot.call, slot.start
status, summary, truncated := slot.status, slot.summary, slot.truncated
end := st.l.now()
st.finalParts = append(st.finalParts,
core.Part{Type: core.PartToolCall, Process: true, ToolCall: &core.ToolCallPart{ID: slot.call.ID, Name: slot.call.Name, Arguments: slot.call.Arguments, Duration: durationMs(slot.start, end)}},
core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: slot.call.ID, Status: slot.status, Summary: slot.summary}},
core.Part{Type: core.PartToolCall, Process: true, ToolCall: &core.ToolCallPart{ID: call.ID, Name: call.Name, Arguments: call.Arguments, Duration: durationMs(start, end)}},
core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: call.ID, Status: status, Summary: summary}},
)
if slot.artifact != nil {
st.deliverables = append(st.deliverables, slot.artifact)
}
st.evidence = append(st.evidence, toolEvidence{Name: slot.call.Name, Status: slot.status, Summary: slot.summary, VerifyText: slot.verifyText})
st.evidence = append(st.evidence, toolEvidence{Name: call.Name, Status: status, Summary: summary, VerifyText: slot.verifyText})
st.toolMu.Unlock()
st.noteFailureOutcome(slot.status)
_ = st.rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: slot.call.ID, Status: slot.status, Summary: slot.summary, Truncated: slot.truncated, DurationMs: durationMs(slot.start, end)})
st.noteFailureOutcome(status)
_ = st.rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: call.ID, Status: status, Summary: summary, Truncated: truncated, DurationMs: durationMs(start, end)})
}
}
@@ -1129,17 +1189,17 @@ func (st *directRun) runToolCtx(ctx context.Context, name, args string) agenttoo
}
}
if res.Artifact != nil && slot != nil {
slot.artifact = &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind}
if st.cardflow && res.Truncated && !st.explicitCard {
st.writeSlot(slot, func(s *toolSlot) {
s.artifact = &core.ArtifactRef{ArtifactID: res.Artifact.ID, Name: res.Artifact.Name, Kind: res.Artifact.Kind}
})
if st.cardflow && res.Truncated && !st.cardSuppressed() {
fc, ferr := st.rc.SubmitCard(artifactFacadeCard(call.Name, *res.Artifact, summarize(out, nil)))
if ferr == nil {
st.finalParts = append(st.finalParts, core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: fc.ID}})
st.appendFinalPart(core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: fc.ID}})
}
}
}
if slot != nil {
slot.truncated = res.Truncated
}
st.writeSlot(slot, func(s *toolSlot) { s.truncated = res.Truncated })
}
summary := summarize(out, nil)
@@ -1148,11 +1208,12 @@ func (st *directRun) runToolCtx(ctx context.Context, name, args string) agenttoo
summary = "error: " + err.Error()
out = summary
}
if slot != nil {
slot.summary = summary
slot.status = status
slot.verifyText = tailRunes(out, 600)
}
verifyText := tailRunes(out, 600)
st.writeSlot(slot, func(s *toolSlot) {
s.summary = summary
s.status = status
s.verifyText = verifyText
})
if r := []rune(out); len(r) > 16000 {
out = string(r[:16000]) + "…(超长截断)"
}
@@ -1257,8 +1318,8 @@ func (st *directRun) submitCard(args string) string {
if cerr != nil {
return "submit_card 失败: " + cerr.Error()
}
st.finalParts = append(st.finalParts, core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: card.ID}})
st.explicitCard = true
st.appendFinalPart(core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: card.ID}})
st.markExplicitCard()
where := "对话"
if a.Scope == "project" {
where = "项目"
@@ -1311,18 +1372,21 @@ func (st *directRun) commitCard() string {
if cerr != nil {
return "card_commit 失败: " + cerr.Error()
}
st.finalParts = append(st.finalParts, core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: card.ID}})
st.explicitCard = true
st.appendFinalPart(core.Part{Type: core.PartCard, Card: &core.CardPart{Kind: string(core.CardKindResult), CardID: card.ID}})
st.markExplicitCard()
return fmt.Sprintf("已作为卡片发出(%s,card:%s)。对话中保留简短结论即可,不要重复卡片全文。", card.Title, card.ID)
}
func (st *directRun) pinSubagentArtifacts(text string) {
st.toolMu.Lock()
seen := map[string]bool{}
for _, ref := range st.deliverables {
if ref != nil {
seen[ref.ArtifactID] = true
}
}
st.toolMu.Unlock()
pinned := make([]*core.ArtifactRef, 0, 4)
for _, id := range gates.ArtifactIDPattern.FindAllString(text, -1) {
if seen[id] {
continue
@@ -1332,8 +1396,14 @@ func (st *directRun) pinSubagentArtifacts(text string) {
continue
}
seen[id] = true
st.deliverables = append(st.deliverables, &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind})
pinned = append(pinned, &core.ArtifactRef{ArtifactID: art.ID, Name: art.Name, Kind: art.Kind})
}
if len(pinned) == 0 {
return
}
st.toolMu.Lock()
st.deliverables = append(st.deliverables, pinned...)
st.toolMu.Unlock()
}
func errString(err error) string {
+30 -2
View File
@@ -27,6 +27,7 @@ type fakeRunCtx struct {
cardflow bool
events []emittedEvent
eventsMu sync.Mutex
stateMu sync.Mutex // 并行工具执行下会写回的假状态(计数/卡片/用量/后台)
assistant []core.Part
cards []core.ActivityCard
streamCardID core.ID // 活跃流式卡(Begin/Append/Commit fake 语义)
@@ -80,8 +81,15 @@ func (f *fakeRunCtx) emittedEvents() []emittedEvent {
defer f.eventsMu.Unlock()
return append([]emittedEvent(nil), f.events...)
}
func (f *fakeRunCtx) SetAssistantParts(parts []core.Part) error { f.assistant = parts; return nil }
func (f *fakeRunCtx) SetAssistantParts(parts []core.Part) error {
f.stateMu.Lock()
defer f.stateMu.Unlock()
f.assistant = parts
return nil
}
func (f *fakeRunCtx) AddUsage(u core.Usage) {
f.stateMu.Lock()
defer f.stateMu.Unlock()
f.usageTotal.PromptTokens += u.PromptTokens
f.usageTotal.CompletionTokens += u.CompletionTokens
f.usageTotal.TotalTokens += u.TotalTokens
@@ -113,6 +121,8 @@ func (f *fakeRunCtx) WriteArtifact(name, kind, mime string, content []byte) (cor
}
func (f *fakeRunCtx) Retrieve(q core.RetrievalQuery) ([]core.RetrievalHit, error) { return nil, nil }
func (f *fakeRunCtx) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
f.stateMu.Lock()
defer f.stateMu.Unlock()
f.toolRuns++
res := core.ToolCallResult{Status: "ok", Output: "done", Truncated: f.execTruncated}
if f.execStatus != "" {
@@ -133,6 +143,8 @@ func (f *fakeRunCtx) History() []core.Message {
return []core.Message{{ID: "m1", Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "hi"}}}}
}
func (f *fakeRunCtx) SpawnBackground(agentID, name, task, hint string) (core.ID, error) {
f.stateMu.Lock()
defer f.stateMu.Unlock()
f.bgCalls++
if f.bgErr != nil {
return "", f.bgErr
@@ -174,6 +186,14 @@ func (f *fakeRunCtx) Attachments() []core.ParsedDocument { return f.attachments
func (f *fakeRunCtx) Tools() []core.ToolDesc { return nil }
func (f *fakeRunCtx) Subtask() string { return "" }
func (f *fakeRunCtx) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
f.stateMu.Lock()
defer f.stateMu.Unlock()
return f.createCardLocked(card)
}
// createCardLocked 建卡(调用方须持有 stateMu);卡片工具与工件门面卡
// 可能从并行工具 goroutine 同时进来。
func (f *fakeRunCtx) createCardLocked(card core.ActivityCard) (core.ActivityCard, error) {
card.ID = core.ID(fmt.Sprintf("card_%d", len(f.cards)+1))
f.cards = append(f.cards, card)
return card, nil
@@ -199,7 +219,9 @@ func (f *fakeRunCtx) SubmitCard(card core.ActivityCard) (core.ActivityCard, erro
// runmgr/cardsvc 语义,供工具编排层断言。
func (f *fakeRunCtx) BeginStreamCard(card core.ActivityCard) (core.ActivityCard, error) {
card.Status = core.CardActive
created, err := f.CreateCard(card)
f.stateMu.Lock()
defer f.stateMu.Unlock()
created, err := f.createCardLocked(card)
if err != nil {
return created, err
}
@@ -208,6 +230,8 @@ func (f *fakeRunCtx) BeginStreamCard(card core.ActivityCard) (core.ActivityCard,
}
func (f *fakeRunCtx) AppendStreamCard(text string) error {
f.stateMu.Lock()
defer f.stateMu.Unlock()
if f.streamCardID == "" {
return errors.New("card: no active stream card (card_begin first)")
}
@@ -226,6 +250,8 @@ func (f *fakeRunCtx) AppendStreamCard(text string) error {
}
func (f *fakeRunCtx) CommitStreamCard() (core.ActivityCard, error) {
f.stateMu.Lock()
defer f.stateMu.Unlock()
if f.streamCardID == "" {
return core.ActivityCard{}, errors.New("card: no active stream card (card_begin first)")
}
@@ -240,6 +266,8 @@ func (f *fakeRunCtx) CommitStreamCard() (core.ActivityCard, error) {
return core.ActivityCard{}, errors.New("card: not found")
}
func (f *fakeRunCtx) Card(id string) *core.ActivityCard {
f.stateMu.Lock()
defer f.stateMu.Unlock()
for i := range f.cards {
if string(f.cards[i].ID) == id {
return &f.cards[i]
+188
View File
@@ -0,0 +1,188 @@
package runtime
import (
"strings"
"sync"
"agent-squad/squad"
core "agent-system-core"
)
// toolResultSummaryLen 是工具结果摘要的截断长度(与既有展示一致)。
const toolResultSummaryLen = 160
// toolEventID 是宿主侧的工具调用身份:优先用上游 CallID(并行同名工具据此配对,
// 而不是按到达顺序);空串 = legacy/未知,退回工具名,保持既有展示不变。
func toolEventID(ev squad.ProcessEvent) string {
if id := strings.TrimSpace(ev.CallID); id != "" {
return id
}
return ev.Tool
}
// toolEventOutcome 把上游 tool_execution_end 翻译成宿主展示的真实结局。
//
// 取值域与上游一致(ok | error | denied | canceled),不新增、不改写上游状态:
// - Status 非空:原样透传;
// - Status 为空且没有失败信号(ErrorReason 也空):legacy/未知,沿用旧的 "ok"
// 与 Detail 摘要(字节一致,不臆造结局);
// - Status 为空但有 ErrorReason:有失败信号而状态未知,报 error 而不是 ok;
// - 非 ok 时摘要取 ErrorReason 原文(截断),上游没给原因才回退 Detail。
func toolEventOutcome(ev squad.ProcessEvent) (status, summary string) {
status = strings.TrimSpace(ev.Status)
if status == "" {
if strings.TrimSpace(ev.ErrorReason) == "" {
return squad.ToolStatusOK, truncateRun(ev.Detail, toolResultSummaryLen)
}
status = squad.ToolStatusError
}
if status == squad.ToolStatusOK {
return status, truncateRun(ev.Detail, toolResultSummaryLen)
}
if strings.TrimSpace(ev.ErrorReason) != "" {
return status, truncateRun(ev.ErrorReason, toolResultSummaryLen)
}
return status, truncateRun(ev.Detail, toolResultSummaryLen)
}
// toolFailure 是一次未被恢复的工具失败(用于终局文本一致性检查,不覆盖工具部件)。
type toolFailure struct {
Tool string
CallID string
Status string
Summary string
}
// squadEventLog 是小队过程帧在宿主侧的聚合点。
//
// 并行工具调用会从多个 goroutine 回调进来(内核 executeParallel 逐个派发),
// 因此流式文本、工具部件、用量与 pending 配对表全部在同一把锁下更新:
// 既不丢条目,也不会把 A 调用的 start 与 B 调用的 end 交错写坏。
// 事件发射与部件追加在同一临界区内,保持到达顺序与最终部件顺序一致。
type squadEventLog struct {
rc RunCtx
mu sync.Mutex
streamed strings.Builder
textIdx int
parts []core.Part
pending map[string]string // callID → 工具名(start/end 配对)
failure *toolFailure // 最近一次失败;同名工具后续再起即视为已恢复
}
// newSquadEventLog 绑定一次运行的过程帧回调。
func newSquadEventLog(rc RunCtx) *squadEventLog {
return &squadEventLog{rc: rc, pending: map[string]string{}}
}
// onEvent 是小队过程帧回调(并发安全)。
func (l *squadEventLog) onEvent(ev squad.ProcessEvent) {
l.mu.Lock()
defer l.mu.Unlock()
switch ev.Type {
case "text_delta":
if ev.Text == "" {
return
}
l.streamed.WriteString(ev.Text)
_ = l.rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: l.textIdx, Text: ev.Text})
case "reasoning_delta":
if ev.Text != "" {
_ = l.rc.Emit(core.EvThinkingDelta, core.ThinkingDeltaPayload{Text: ev.Text})
}
case "tool_execution_start":
id := toolEventID(ev)
if id != "" {
l.pending[id] = ev.Tool
}
l.noteRecovered(ev.Tool)
_ = l.rc.Emit(core.EvToolCall, core.ToolCallPayload{CallID: id, Name: ev.Tool, Arguments: ev.Detail})
l.parts = append(l.parts, core.Part{Type: core.PartToolCall, Process: true, ToolCall: &core.ToolCallPart{ID: id, Name: ev.Tool, Arguments: ev.Detail}})
case "tool_execution_end":
id := toolEventID(ev)
if tool, ok := l.pending[id]; ok && ev.Tool == "" {
ev.Tool = tool
}
delete(l.pending, id)
status, summary := toolEventOutcome(ev)
// 发射与部件同源:前端过程行与落库部件说的是同一件事。
_ = l.rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: id, Status: status, Summary: summary})
l.parts = append(l.parts, core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: id, Status: status, Summary: summary}})
if status == squad.ToolStatusOK {
l.noteRecovered(ev.Tool)
} else {
l.failure = &toolFailure{Tool: ev.Tool, CallID: id, Status: status, Summary: summary}
}
case "usage":
if sink, ok := l.rc.(interface{ AddUsage(core.Usage) }); ok {
sink.AddUsage(core.Usage{
PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens,
LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens,
CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens,
})
}
}
}
// noteRecovered 把同名工具的后续进展(再次发起或调用成功)记为已恢复:
// 失败部件仍留在过程部件里,只是不再让收尾文本替它宣布成功。
func (l *squadEventLog) noteRecovered(tool string) {
if l.failure != nil && l.failure.Tool == tool {
l.failure = nil
}
}
// reply 是本次运行聚合到的模型文本(不含任何宿主补写)。
func (l *squadEventLog) reply() string {
l.mu.Lock()
defer l.mu.Unlock()
return l.streamed.String()
}
// parts 返回过程部件的快照(复制,避免把内部切片交给最终部件)。
func (l *squadEventLog) toolParts() []core.Part {
l.mu.Lock()
defer l.mu.Unlock()
return append([]core.Part(nil), l.parts...)
}
// unresolvedFailure 返回终局时仍未恢复的工具失败,没有则 nil。
func (l *squadEventLog) unresolvedFailure() *toolFailure {
l.mu.Lock()
defer l.mu.Unlock()
if l.failure == nil {
return nil
}
f := *l.failure
return &f
}
// finalReply 决定终局文本。模型文本原样保留(不代写、不改写);只有宿主兜底的
// 「已完成。」在仍有未恢复的工具失败时会被换成如实说明,避免收尾文本与失败部件
// 自相矛盾。legacy 事件(无 CallID/Status)走不到这里,展示与旧行为一致。
func (l *squadEventLog) finalReply(modelText, summary, fallback string) string {
reply := strings.TrimSpace(modelText)
if reply == "" {
reply = strings.TrimSpace(summary)
}
if reply != "" {
return reply
}
if f := l.unresolvedFailure(); f != nil {
return failedToolLine(f)
}
return fallback
}
// failedToolLine 如实描述未恢复的工具失败,不声称成功,也不编造原因。
func failedToolLine(f *toolFailure) string {
name := strings.TrimSpace(f.Tool)
if name == "" {
name = "工具"
}
reason := strings.TrimSpace(f.Summary)
if reason == "" {
return name + " 未完成(" + f.Status + ")"
}
return name + " 未完成(" + f.Status + "):" + reason
}
+443
View File
@@ -0,0 +1,443 @@
package runtime
import (
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
agentllm "agent-native/runtime/llm"
"agent-squad/squad"
core "agent-system-core"
)
// eventProbe 记录小队过程帧回调发射出去的事件(顺序与内容都要能断言)。
// Emit 由 squadEventLog 的临界区内调用,因此这里无需再自锁。
type eventProbe struct {
*entryFake
rec *squadEventLog
toolCall []core.ToolCallPayload
result []core.ToolResultPayload
}
func newEventProbe() *eventProbe {
p := &eventProbe{entryFake: &entryFake{}}
p.rec = newSquadEventLog(p)
return p
}
func (p *eventProbe) Emit(typ core.StreamEventType, payload any) error {
switch v := payload.(type) {
case core.ToolCallPayload:
p.toolCall = append(p.toolCall, v)
case core.ToolResultPayload:
p.result = append(p.result, v)
}
return p.entryFake.Emit(typ, payload)
}
func (p *eventProbe) onEvent(ev squad.ProcessEvent) { p.rec.onEvent(ev) }
func (p *eventProbe) parts() []core.Part { return p.rec.toolParts() }
func (p *eventProbe) calls() []core.ToolCallPayload {
return append([]core.ToolCallPayload(nil), p.toolCall...)
}
func (p *eventProbe) results() []core.ToolResultPayload {
return append([]core.ToolResultPayload(nil), p.result...)
}
func toolCallsOf(parts []core.Part) []core.ToolCallPart {
var out []core.ToolCallPart
for _, p := range parts {
if p.ToolCall != nil {
out = append(out, *p.ToolCall)
}
}
return out
}
func toolResultsOf(parts []core.Part) []core.ToolResultPart {
var out []core.ToolResultPart
for _, p := range parts {
if p.ToolResult != nil {
out = append(out, *p.ToolResult)
}
}
return out
}
// TestSquadEventsPairByCallID:两个同名工具的 start/end 必须按 CallID 配对,
// 不得因名字相同而互相对调(这是并行同名工具显示错乱的根因)。
func TestSquadEventsPairByCallID(t *testing.T) {
p := newEventProbe()
p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: "call_a", Detail: `{"q":"a"}`})
p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: "call_b", Detail: `{"q":"b"}`})
// 交错到达:B 先结束。
p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "call_b", Status: squad.ToolStatusOK, Detail: "命中 B"})
p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "call_a", Status: squad.ToolStatusError, ErrorReason: "路径不存在"})
parts := p.parts()
calls := toolCallsOf(parts)
if len(calls) != 2 || calls[0].ID != "call_a" || calls[1].ID != "call_b" {
t.Fatalf("tool call parts = %+v", calls)
}
emittedCalls := p.calls()
if len(emittedCalls) != 2 || emittedCalls[0].CallID != "call_a" || emittedCalls[1].CallID != "call_b" {
t.Fatalf("emitted tool calls = %+v", emittedCalls)
}
results := toolResultsOf(parts)
if len(results) != 2 {
t.Fatalf("tool result parts = %+v", results)
}
if results[0].CallID != "call_b" || results[0].Status != squad.ToolStatusOK || results[0].Summary != "命中 B" {
t.Fatalf("result B = %+v", results[0])
}
if results[1].CallID != "call_a" || results[1].Status != squad.ToolStatusError || results[1].Summary != "路径不存在" {
t.Fatalf("result A = %+v", results[1])
}
if p.rec.unresolvedFailure() == nil || p.rec.unresolvedFailure().CallID != "call_a" {
t.Fatalf("unresolved failure = %+v", p.rec.unresolvedFailure())
}
}
// TestSquadEventsStatusIsTruthful:失败/取消/拒绝必须原样透传,不得一律报 ok,
// 摘要取真实原因(截断),部件与发射载荷完全一致。
func TestSquadEventsStatusIsTruthful(t *testing.T) {
long := strings.Repeat("失败原因", 80)
cases := []struct {
name string
ev squad.ProcessEvent
status string
summary string
}{
{"error", squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "c1",
Status: squad.ToolStatusError, ErrorReason: "权限不足", Detail: "旧摘要"}, squad.ToolStatusError, "权限不足"},
{"denied", squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c2",
Status: squad.ToolStatusDenied, ErrorReason: "策略拒绝执行"}, squad.ToolStatusDenied, "策略拒绝执行"},
{"canceled", squad.ProcessEvent{Type: "tool_execution_end", Tool: "web_search", CallID: "c3",
Status: squad.ToolStatusCanceled, ErrorReason: "用户取消"}, squad.ToolStatusCanceled, "用户取消"},
{"ok", squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "c4",
Status: squad.ToolStatusOK, Detail: "命中 3 处"}, squad.ToolStatusOK, "命中 3 处"},
// 有失败信号但状态未知:不得报 ok。
{"status-less error signal", squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "c5",
ErrorReason: "连接被拒绝", Detail: "旧摘要"}, squad.ToolStatusError, "连接被拒绝"},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
status, summary := toolEventOutcome(tc.ev)
if status != tc.status || summary != tc.summary {
t.Fatalf("outcome = %q/%q, want %q/%q", status, summary, tc.status, tc.summary)
}
p := newEventProbe()
p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: tc.ev.Tool, CallID: tc.ev.CallID})
p.onEvent(tc.ev)
got := toolResultsOf(p.parts())
if len(got) != 1 {
t.Fatalf("parts = %+v", p.parts())
}
// 落库部件与发射载荷同源。
if got[0].Status != tc.status || got[0].Summary != tc.summary || got[0].CallID != tc.ev.CallID {
t.Fatalf("result part = %+v", got[0])
}
wantPayload := core.ToolResultPayload{CallID: tc.ev.CallID, Status: tc.status, Summary: tc.summary}
if gotPayloads := p.results(); len(gotPayloads) != 1 || gotPayloads[0] != wantPayload {
t.Fatalf("emitted payload = %+v", gotPayloads)
}
if gotCalls := p.calls(); len(gotCalls) != 1 || gotCalls[0].CallID != tc.ev.CallID || gotCalls[0].Name != tc.ev.Tool {
t.Fatalf("emitted tool call = %+v", gotCalls)
}
})
}
t.Run("summary truncated", func(t *testing.T) {
status, summary := toolEventOutcome(squad.ProcessEvent{Status: squad.ToolStatusError, ErrorReason: long})
if status != squad.ToolStatusError {
t.Fatalf("status = %q", status)
}
if want := truncateRun(long, toolResultSummaryLen); summary != want {
t.Fatalf("summary = %q, want truncated %q", summary, want)
}
})
}
// TestSquadEventsLegacyFramesUnchanged:字段全空的 legacy 事件必须与旧行为字节一致
// (按工具名配对、状态恒为 ok、摘要取 Detail)。
func TestSquadEventsLegacyFramesUnchanged(t *testing.T) {
p := newEventProbe()
p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", Detail: `{"q":"x"}`})
p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", Detail: strings.Repeat("d", 200)})
p.onEvent(squad.ProcessEvent{Type: "text_delta", Text: "结论"})
p.onEvent(squad.ProcessEvent{Type: "usage", PromptTokens: 3, CompletionTokens: 1, LastPromptTokens: 2, CacheReadTokens: 1})
parts := p.parts()
calls := toolCallsOf(parts)
if len(calls) != 1 {
t.Fatalf("calls = %+v", calls)
}
wantCall := core.ToolCallPart{ID: "grep", Name: "grep", Arguments: `{"q":"x"}`}
if calls[0] != wantCall {
t.Fatalf("call part = %+v, want %+v", calls[0], wantCall)
}
results := toolResultsOf(parts)
wantResult := core.ToolResultPart{CallID: "grep", Status: "ok", Summary: truncateRun(strings.Repeat("d", 200), 160)}
if len(results) != 1 || results[0] != wantResult {
t.Fatalf("result part = %+v, want %+v", results, wantResult)
}
emitted := p.results()
if len(emitted) != 1 || emitted[0].CallID != "grep" || emitted[0].Status != "ok" ||
emitted[0].Summary != truncateRun(strings.Repeat("d", 200), 160) {
t.Fatalf("emitted = %+v", emitted)
}
if p.rec.reply() != "结论" {
t.Fatalf("reply = %q", p.rec.reply())
}
if f := p.rec.unresolvedFailure(); f != nil {
t.Fatalf("legacy frames must not register a failure: %+v", f)
}
if p.rec.finalReply("", "", "已完成。") != "已完成。" {
t.Fatalf("legacy fallback text changed: %q", p.rec.finalReply("", "", "已完成。"))
}
}
// TestSquadEventsConcurrentCallbacksKeepEveryEntry:并行回调(-race)下不得丢条目、
// 不得串字段;每次调用自己的 start/end 仍成对。
func TestSquadEventsConcurrentCallbacksKeepEveryEntry(t *testing.T) {
const calls = 64
p := newEventProbe()
wantStatus := map[string]string{}
var wg sync.WaitGroup
for i := 0; i < calls; i++ {
id := fmt.Sprintf("call_%02d", i)
wantStatus[id] = squad.ToolStatusOK
failing := i%8 == 0
if failing {
wantStatus[id] = squad.ToolStatusError
}
wg.Add(1)
go func() {
defer wg.Done()
p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: id, Detail: id})
if failing {
p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: id,
Status: squad.ToolStatusError, ErrorReason: "boom " + id})
return
}
p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: id,
Status: squad.ToolStatusOK, Detail: "ok " + id})
}()
}
wg.Wait()
parts := p.parts()
if got := len(toolCallsOf(parts)); got != calls {
t.Fatalf("tool call parts = %d, want %d", got, calls)
}
results := toolResultsOf(parts)
if len(results) != calls {
t.Fatalf("tool result parts = %d, want %d", len(results), calls)
}
seenStart := map[string]bool{}
for _, c := range toolCallsOf(parts) {
if seenStart[c.ID] {
t.Fatalf("duplicate start for %s", c.ID)
}
seenStart[c.ID] = true
}
emitted := map[string]string{}
for _, r := range results {
if _, dup := emitted[r.CallID]; dup {
t.Fatalf("duplicate end for %s", r.CallID)
}
emitted[r.CallID] = r.Status
if !seenStart[r.CallID] {
t.Fatalf("end without start: %s", r.CallID)
}
}
for id, status := range emitted {
if status != wantStatus[id] {
t.Fatalf("end %s status = %q, want %q", id, status, wantStatus[id])
}
}
}
// TestSquadEventsConcurrentToolCallbacksShareArtifactList:宿主工具被并行调用时,
// 登记的工件一条都不能丢,snapshot 也不应把内部切片直接交出去。
func TestSquadEventsConcurrentToolCallbacksShareArtifactList(t *testing.T) {
const n = 128
produced := &artifactList{}
var wg sync.WaitGroup
for i := 0; i < n; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
produced.add(&core.ArtifactRef{ArtifactID: core.ID(fmt.Sprintf("art_%03d", i)), Name: "纪要.docx", Kind: "document"})
}(i)
}
wg.Wait()
got := produced.snapshot()
if len(got) != n {
t.Fatalf("artifacts = %d, want %d", len(got), n)
}
ids := map[string]bool{}
for _, ref := range got {
ids[string(ref.ArtifactID)] = true
}
for i := 0; i < n; i++ {
if !ids[fmt.Sprintf("art_%03d", i)] {
t.Fatalf("missing artifact %d", i)
}
}
got[0] = nil // 快照必须是副本:改动不得回写内部表
if produced.snapshot()[0] == nil {
t.Fatal("snapshot aliases the internal list")
}
}
// TestSquadEventsFinalTextMatchesToolOutcome:未恢复的失败不得被收尾文本改写成成功;
// 同名工具重试成功后收尾文本恢复旧行为。
func TestSquadEventsFinalTextMatchesToolOutcome(t *testing.T) {
p := newEventProbe()
p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "run_command", CallID: "c1"})
p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c1",
Status: squad.ToolStatusCanceled, ErrorReason: "用户取消"})
if got := p.rec.finalReply("", "", "已完成。"); got == "已完成。" || !strings.Contains(got, "run_command") || !strings.Contains(got, "用户取消") {
t.Fatalf("fallback text = %q", got)
}
if got := p.rec.finalReply("已按你的要求处理。", "", "已完成。"); got != "已按你的要求处理。" {
t.Fatalf("model text must be preserved: %q", got)
}
// 失败部件必须留在最终部件里,不被收尾文本覆盖。
if results := toolResultsOf(p.parts()); len(results) != 1 || results[0].Status != squad.ToolStatusCanceled {
t.Fatalf("failure part dropped: %+v", results)
}
// 同名工具重试成功 → 失败记为已恢复,兜底文本回到旧行为。
q := newEventProbe()
q.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "run_command", CallID: "c1"})
q.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c1",
Status: squad.ToolStatusError, ErrorReason: "超时"})
q.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "run_command", CallID: "c2"})
q.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c2",
Status: squad.ToolStatusOK, Detail: "done"})
if f := q.rec.unresolvedFailure(); f != nil {
t.Fatalf("retry should clear the failure: %+v", f)
}
if got := q.rec.finalReply("", "", "已完成。"); got != "已完成。" {
t.Fatalf("recovered run fallback = %q", got)
}
}
// entryProbe 在真实运行桩上记录工具事件载荷,并让指定工具真的失败,
// 用来端到端验证「工具结局 → 过程事件 → 落库部件」这条链。
type entryProbe struct {
*entryFake
mu sync.Mutex
calls []core.ToolCallPayload
results []core.ToolResultPayload
}
func (p *entryProbe) Emit(typ core.StreamEventType, payload any) error {
switch v := payload.(type) {
case core.ToolCallPayload:
p.mu.Lock()
p.calls = append(p.calls, v)
p.mu.Unlock()
case core.ToolResultPayload:
p.mu.Lock()
p.results = append(p.results, v)
p.mu.Unlock()
}
return p.entryFake.Emit(typ, payload)
}
func (p *entryProbe) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
if call.Name == "knowledge_search" {
return core.ToolCallResult{Status: "error", Output: "索引不可用"}
}
return p.entryFake.ExecuteTool(call)
}
func (p *entryProbe) resultPayloads() []core.ToolResultPayload {
p.mu.Lock()
defer p.mu.Unlock()
return append([]core.ToolResultPayload(nil), p.results...)
}
// TestSquadEntryFailedToolReachesFinalParts:真实跑一次小队入口,同名并行工具
// 必须按各自 CallID 落部件,失败必须显示为失败,且不被收尾文本改写成成功。
func TestSquadEntryFailedToolReachesFinalParts(t *testing.T) {
dir := t.TempDir()
base := baseEntry(dir)
rc := &entryProbe{entryFake: base}
rc.tools = []core.ToolDesc{{Name: "knowledge_search", Description: "查资料", Parameters: json.RawMessage(`{"type":"object"}`)}}
runner := &SquadRunner{
Router: fixedRoute{mode: squad.RouteDirect}, Checker: squad.MinimalChecker{},
Hitl: squad.HitlPolicy{Mode: squad.HitlDisabled},
Stream: scriptedCalls(t, nil, [][]agentllm.Event{
{{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, StopReason: agentllm.StopToolCalls,
ToolCalls: []agentllm.ToolCall{
{ID: "call_k1", Name: "knowledge_search", Arguments: `{"q":"a"}`},
{ID: "call_k2", Name: "knowledge_search", Arguments: `{"q":"b"}`},
}}}},
{{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, StopReason: agentllm.StopToolCalls,
ToolCalls: []agentllm.ToolCall{{ID: "w", Name: "write_file", Arguments: mustDocArgs()}}}}},
{{Kind: agentllm.KindTextDelta, Text: "写完了。"},
{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "写完了。", StopReason: agentllm.StopEnd}}},
}),
}
if err := runner.Execute(rc); err != nil {
t.Fatal(err)
}
parts := base.parts
calls := toolCallsOf(parts)
ids := map[string]bool{}
for _, c := range calls {
if c.Name != "knowledge_search" {
continue
}
if c.ID == "knowledge_search" || ids[c.ID] {
t.Fatalf("tool call part lost its call id: %+v (all %+v)", c, calls)
}
ids[c.ID] = true
}
if len(ids) != 2 {
t.Fatalf("distinct call ids = %v", ids)
}
results := toolResultsOf(parts)
search := map[string]core.ToolResultPart{}
for _, r := range results {
if ids[r.CallID] {
search[r.CallID] = r
}
}
if len(search) != 2 {
t.Fatalf("tool results for knowledge_search = %+v", results)
}
for id, r := range search {
if r.Status != squad.ToolStatusError {
t.Fatalf("%s status = %q, want error", id, r.Status)
}
if r.Summary != "索引不可用" {
t.Fatalf("%s summary = %q", id, r.Summary)
}
}
// 发射载荷与落库部件一致:前端过程行不得说 ok。
emitted := map[string]string{}
for _, pl := range rc.resultPayloads() {
if ids[pl.CallID] {
emitted[pl.CallID] = pl.Status
}
}
for id, r := range search {
if emitted[id] != r.Status {
t.Fatalf("emitted %s status = %q, part status = %q", id, emitted[id], r.Status)
}
}
// 收尾文本保留模型原话,失败部件不被覆盖。
if last := parts[len(parts)-1]; last.Type != core.PartText || last.Text != "写完了。" {
t.Fatalf("final part = %+v", last)
}
}
+8 -42
View File
@@ -30,37 +30,8 @@ func (s *SquadRunner) Execute(rc RunCtx) error {
prof := rc.Profile()
maxTurns, maxTools := loopCaps(prof.Budget)
produced := &artifactList{}
var streamed strings.Builder
var toolParts []core.Part
textIdx := 0
onEvent := func(ev squad.ProcessEvent) {
switch ev.Type {
case "text_delta":
if ev.Text == "" {
return
}
streamed.WriteString(ev.Text)
_ = rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: textIdx, Text: ev.Text})
case "reasoning_delta":
if ev.Text != "" {
_ = rc.Emit(core.EvThinkingDelta, core.ThinkingDeltaPayload{Text: ev.Text})
}
case "tool_execution_start":
_ = rc.Emit(core.EvToolCall, core.ToolCallPayload{CallID: ev.Tool, Name: ev.Tool, Arguments: ev.Detail})
toolParts = append(toolParts, core.Part{Type: core.PartToolCall, Process: true, ToolCall: &core.ToolCallPart{ID: ev.Tool, Name: ev.Tool, Arguments: ev.Detail}})
case "tool_execution_end":
_ = rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: ev.Tool, Status: "ok", Summary: truncateRun(ev.Detail, 160)})
toolParts = append(toolParts, core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: ev.Tool, Status: "ok", Summary: truncateRun(ev.Detail, 160)}})
case "usage":
if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok {
sink.AddUsage(core.Usage{
PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens,
LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens,
CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens,
})
}
}
}
// 过程帧可能来自并行的工具回调:聚合全部走 squadEventLog(内部互斥)。
evlog := newSquadEventLog(rc)
router := s.Router
if router == nil {
router = newModelRouter(rc)
@@ -73,7 +44,7 @@ func (s *SquadRunner) Execute(rc RunCtx) error {
DataDir: rc.Settings().GetString("storage.dataDir", "./data"),
Goal: entryGoal(rc),
HostTools: bridgeHostTools(rc, produced),
OnEvent: onEvent,
OnEvent: evlog.onEvent,
Stream: stream,
Router: router,
Checker: s.Checker,
@@ -90,14 +61,9 @@ func (s *SquadRunner) Execute(rc RunCtx) error {
if err != nil {
return err
}
reply := strings.TrimSpace(streamed.String())
if reply == "" {
reply = strings.TrimSpace(res.Summary)
}
if reply == "" {
reply = "已完成。"
}
if claimsFalseDeliverable(reply, produced.items, rc) {
// 收尾文本:模型文本原样保留;仍有未恢复的工具失败时不用「已完成。」兜底。
reply := evlog.finalReply(evlog.reply(), res.Summary, "已完成。")
if claimsFalseDeliverable(reply, produced.snapshot(), rc) {
return fmt.Errorf("声称已生成文件,但没有登记对应工件")
}
if strings.TrimSpace(res.Body) != "" {
@@ -105,9 +71,9 @@ func (s *SquadRunner) Execute(rc RunCtx) error {
return werr
}
}
parts := append([]core.Part{}, toolParts...)
parts := evlog.toolParts()
parts = append(parts, core.Part{Type: core.PartText, Text: reply})
for _, ref := range produced.items {
for _, ref := range produced.snapshot() {
parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: ref})
}
return rc.SetAssistantParts(parts)
+220
View File
@@ -0,0 +1,220 @@
package server
// 取消路由(A1/U-A1c):把「取消刚才的任务」这类意图派发成**既有**取消动作。
//
// 纪律:
// - 唯一入口:路由层解出唯一靶子(原文点名的运行号,或本会话绑定动作的
// 运行上下文),本文件只负责把它交给既有取消路径,绝不自己发明取消
// 语义、绝不绕过命令总线。
// - 靶子不唯一就不猜:没有绑定也没有点名时如实澄清,并把发起人自己最近
// 的运行如实列出来(别人的运行在构造上就不可见)。
// - 确认语义不变:注册了取消命令的部署走 Bus.Dispatch(与任何高后果
// 动作同款——审计信封、内联确认卡原样生效,绝不自动确认);未注册时
// 回落到 cancelRun 已有的投递(PG 档且 run 不在本实例 → 收件箱持久命令
// EnqueueCommandAccountChecked;本实例活跃 → 就地 CancelRun),与
// POST /api/runs/{id}/cancel 逐字同款。
// - 不谎报成功:投递失败、运行不可见或已终态都如实回话,不写 accepted。
import (
"net/http"
"sort"
"strings"
agentcommand "agent-command"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/intentsession"
"agent-system-backend/internal/route"
"agent-system-backend/internal/runmgr"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// intentCancelCommandID 是「取消」意图派发时优先使用的命令总线 id:注册了
// 该命令的部署走与任何其他业务动作同款的分发路径(准入、审计信封、高危
// 确认卡全部原样生效);未注册时回落到既有的收件箱投递,语义不变。
const intentCancelCommandID = "run.cancel"
// intentCancelParam 是取消命令声明里运行号的参数名(与既有取消用例一致)。
const intentCancelParam = "run_id"
// cancelCandidateLimit 是澄清时如实列出的候选运行上限(防止回话过长)。
const cancelCandidateLimit = 5
// intentCancelResult 是取消派发的如实回执(响应上的 cancel 字段;不新增 API 面)。
type intentCancelResult struct {
RunID string `json:"runId"`
Source string `json:"source"`
Intent string `json:"intent,omitempty"`
Action string `json:"action,omitempty"`
// Accepted 表示取消请求已被既有路径受理(命令已投递);高危档还需用户
// 显式确认,响应里的 card 就是那道门禁。未受理时 Reason 说明为什么。
Accepted bool `json:"accepted"`
Via string `json:"via"`
Reason string `json:"reason,omitempty"`
}
// planIntentCancel 裁决要不要真的取消:
// - 返回 runID != "":靶子唯一且对发起人可见,按既有路径派发;
// - 返回澄清诉求(runID == ""):靶子缺失/不可见/候选不唯一——如实澄清,
// 绝不猜一个。
//
// 可见性只认发起人自己的运行:GetRun(owner, id) 查不到就与「不存在」同一
// 出口(不据此泄露别人的运行是否存在)。
func (s *Server) planIntentCancel(owner string, conv core.Conversation,
res intentgate.MatchResult, dec route.Decision) (string, *clarifyPayload) {
t := dec.Cancel
if t == nil || strings.TrimSpace(t.RunID) == "" {
return "", s.cancelClarify(owner, conv, res, "没有可取消的运行:请点名运行号,或先发起一个任务")
}
if _, err := s.st.GetRun(owner, t.RunID); err != nil {
return "", s.cancelClarify(owner, conv, res, "该运行不可见或不存在")
}
return t.RunID, nil
}
// cancelClarify 组装取消的澄清诉求:问题问「取消哪一个」,候选如实列出
// (发起人自己最近的未终态运行,按新到旧)。一个候选都没有时如实说明。
func (s *Server) cancelClarify(owner string, conv core.Conversation,
res intentgate.MatchResult, reason string) *clarifyPayload {
c := &clarifyPayload{Intent: res.ID, Command: intentCancelCommandID, Question: "要取消哪一个任务?", Reason: reason}
candidates := s.cancelCandidateRuns(owner, conv.ID)
for _, r := range candidates {
c.Missing = append(c.Missing, intentgate.SlotQuestion{
Key: string(r.ID), Question: "取消运行 " + string(r.ID) + "(" + string(r.Status) + ")"})
}
if len(candidates) == 0 {
c.Question = "现在没有可取消的任务。"
}
return c
}
// cancelCandidateRuns 列出发起人自己最近的未终态运行(会话维度、按创建时间
// 倒序、上限 cancelCandidateLimit)。别人的运行与已终态的运行一律不列。
func (s *Server) cancelCandidateRuns(owner, convID string) []core.Run {
runs, err := s.st.ListRunsByConversation(convID)
if err != nil {
return nil
}
out := []core.Run{}
for _, r := range runs {
if string(r.Owner) != owner || runTerminal(r.Status) {
continue
}
out = append(out, r)
}
sort.SliceStable(out, func(i, j int) bool { return out[i].CreatedAt.After(out[j].CreatedAt) })
if len(out) > cancelCandidateLimit {
out = out[:cancelCandidateLimit]
}
return out
}
func runTerminal(status core.RunStatus) bool {
switch status {
case core.RunCompleted, core.RunFailed, core.RunCancelled, core.RunInterrupted:
return true
}
return false
}
// handleIntentCancel 派发一次取消意图(靶子已唯一且可见)。响应形态与既有
// 命令入口同源:消息落库后派发,回执/确认卡/如实回话都在响应里。
func (s *Server) handleIntentCancel(w http.ResponseWriter, r *http.Request, conv core.Conversation,
msg core.Message, target *route.CancelTarget, runID string) {
owner := s.owner(r)
result := intentCancelResult{RunID: runID, Source: string(target.Source),
Intent: target.Intent, Action: target.Action}
out := map[string]any{"message": msg}
awaitingConfirm := false
switch {
case s.Bus != nil && s.Bus.HasCommand(intentCancelCommandID):
// 已注册取消命令:走与任何其他业务动作完全相同的分发路径。高危档
// 在这里会拿到内联确认卡并返回 accepted(awaiting explicit confirm),
// 确认前处理器不执行——绝不自动确认。
resp := s.Bus.Dispatch(r.Context(), cmdbus.Request{
ID: intentCancelCommandID,
Params: map[string]any{intentCancelParam: runID, chatTextParam: strings.TrimSpace(messageText(msg))},
// chat 入口的规范 Entrance;Actor 是发起人,Owner 是归属租户。
Entrance: "chat", Actor: owner, Owner: owner, ConversationID: conv.ID,
AnchorMessageID: msg.ID, HostProjectID: string(conv.ProjectID),
})
result.Via, result.Accepted = "bus", resp.Receipt.State != agentcommand.StateRejected
awaitingConfirm = resp.Card != nil
out["command"], out["receipt"] = intentCancelCommandID, resp.Receipt
if resp.Card != nil {
out["card"] = resp.Card
}
if m, ok := resp.Payload.(map[string]any); ok {
for k, v := range m {
out[k] = v
}
}
default:
// 未注册取消命令:与 cancelRun 同款投递——PG 档且 run 不在本实例时
// 写持久命令由执行实例消费,否则就地级联取消本实例的句柄。
err := s.dispatchCancel(owner, runID)
result.Via = "local"
if s.st.Dialect() == "postgres" && !s.mgr.RunActiveLocally(runID) {
result.Via = "command"
}
if err != nil {
result.Reason = err.Error()
} else {
result.Accepted = true
}
}
out["cancel"] = result
switch {
case !result.Accepted:
// 没有真的取消:不写会话记录,也不声称已取消(原因随 cancel 字段透出)。
case awaitingConfirm:
// 已提交、等待显式确认:尚未生效,因此不记「已取消」;确认后的收尾
// 走确认卡自带的路径。
default:
// 已派发的取消记回会话:清挂起预期,留短期「已取消某运行」记录。
out["intent_session"] = s.recordIntentCancel(conv, owner, result, target, runID)
}
writeJSON(w, http.StatusOK, out)
}
// dispatchCancel 派发一次取消(与 cancelRun 同一条路径):跨实例投递走收件箱
// 持久命令,本实例活跃时就地取消句柄。投递本身经 store 的账号围栏——别人的
// 运行号在构造上不可投递。
func (s *Server) dispatchCancel(owner, runID string) error {
if s.st.Dialect() == "postgres" && !s.mgr.RunActiveLocally(runID) {
return s.enqueueCancelCommand(owner, runID)
}
return s.mgr.CancelRun(runID)
}
// enqueueCancelCommand 是 cancelRun 与取消意图共用的投递入口:写一条持久
// cancel 命令,由目标 run 的当前执行实例消费。
func (s *Server) enqueueCancelCommand(owner, runID string) error {
return s.st.EnqueueCommandAccountChecked(owner, store.NewID("cmd"), "cancel", runID, 0,
runmgr.CancelCommandPayload{Owner: owner, RunID: runID}, s.Auth == nil || !s.Auth.Enabled)
}
// recordIntentCancel 把一次已派发的取消记回会话:清挂起预期,留一条短期
// 「已取消某运行」记录,供紧接着的追问(如「确认取消」)如实回话。
func (s *Server) recordIntentCancel(conv core.Conversation, owner string,
result intentCancelResult, target *route.CancelTarget, runID string) *intentSessionNote {
reply := "已取消运行 " + runID + "。"
if result.Via == "command" {
// 跨实例投递:取消由目标 run 的执行实例收尾,这里只如实说「已提交」。
reply = "已提交取消运行 " + runID + "(由执行实例处理)。"
}
note := &intentSessionNote{Kind: "cancelled", Intent: target.Intent, Action: target.Action,
RunID: runID, Reply: reply}
svc := s.intentSessionService()
key := intentSessionKey(owner, conv)
if svc == nil || !keyUsable(key) {
return note
}
if _, err := svc.RecordCancel(key, intentsession.Binding{
RunID: runID, Intent: target.Intent, Action: target.Action}); err != nil {
// 记录失败不改变已发生的取消;回话仍如实说明取消了哪个运行。
note.Reply += "(会话记录未更新)"
}
return note
}
+434
View File
@@ -0,0 +1,434 @@
package server
// A1/U-A1c 取消路由的 chat 端到端(合成数据 + 现有 fakes):
// - 会话里绑定了一个活跃运行 → 无澄清地按既有取消路径派发(回执可见),
// 并在会话上留下「已取消某运行」的短期记录;
// - 原文点名了运行 → 点名优先于会话绑定;
// - 靶子缺失/不可见 → 如实澄清并列出自己最近的运行,绝不猜;
// - 高危取消命令 → 仍先出确认卡,确认前不执行(会话不绕过确认门禁)。
import (
"context"
"encoding/json"
"net/http"
"strings"
"testing"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/intentsession"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// taskCancelIntent 是「取消刚才的任务」的一次识别(库层 cancel 命令;与
// U-A1b 的「取消运行 <id>」是不同意图:前者取消运行,后者放弃补槽)。
func taskCancelIntent(slots map[string]string) intentgate.MatchResult {
return intentgate.MatchResult{Branch: "matched", ID: "demo.task-cancel", Slots: slots,
Commands: []intentgate.Command{{Kind: intentgate.CommandCancel, Target: "demo.task-cancel"}}}
}
// liveRun 起一个仍在进行的运行(mock runner 的 #hitl 挂起等待审批),返回
// run id;它一直活到被取消或被审批——就是「我刚才的任务」。
func liveRun(t *testing.T, h http.Handler, convID string) string {
t.Helper()
runID := postMessage(t, h, convID, "#hitl")
waitActiveHITL(t, h, convID)
return runID
}
// waitActiveHITL 等到运行挂起在审批上(run 处于活跃/等待态)。
func waitActiveHITL(t *testing.T, h http.Handler, convID string) {
t.Helper()
deadline := time.Now().Add(testRunTerminalTimeout)
for time.Now().Before(deadline) {
if code, d := doJSON(t, h, "GET", "/api/conversations/"+convID, nil); code == 200 {
if cards, ok := d["activeHitl"].([]any); ok && len(cards) > 0 {
return
}
}
time.Sleep(20 * time.Millisecond)
}
t.Fatal("运行没有进入等待审批状态")
}
// waitRunStatus 等运行落到期望状态(取消是异步收尾的)。
func waitRunStatus(t *testing.T, h http.Handler, runID, want string) map[string]any {
t.Helper()
deadline := time.Now().Add(testRunTerminalTimeout)
for time.Now().Before(deadline) {
if code, d := doJSON(t, h, "GET", "/api/runs/"+runID, nil); code == 200 {
if r, ok := d["run"].(map[string]any); ok && r["status"] == want {
return r
}
}
time.Sleep(20 * time.Millisecond)
}
t.Fatalf("运行 %s 未到达状态 %s", runID, want)
return nil
}
// seedBoundRun 在会话上挂起一个已抽到运行号的预期(等价于「上一轮澄清里
// 抽到了这个运行」),用于驱动取消的会话来源。
func seedBoundRun(t *testing.T, st *store.Store, projectID, convID, runID string) {
t.Helper()
_, err := intentsession.New(st).Suspend(intentsession.Key{
Owner: store.OwnerDefault, Project: projectID, Conversation: convID},
intentsession.Expectation{IntentID: "demo.run-cancel",
Action: intentgate.ActionRef{Kind: intentgate.ActionCommand, ID: "demo.run-cancel"},
Missing: []intentgate.SlotQuestion{{Key: "reason", Question: "取消原因?"}},
Question: "取消原因?"},
map[string]string{"run_id": runID})
if err != nil {
t.Fatal(err)
}
}
// 会话里绑定了活跃运行 → 无澄清地派发既有取消动作(回执可见、运行真的停了)。
func TestIntentCancelDispatchesRunBoundInSession(t *testing.T) {
srv, st, gate, _ := cancelServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := liveRun(t, h, convID)
seedBoundRun(t, st, projectID, convID, runID)
gate.script("取消刚才的任务", taskCancelIntent(nil))
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["clarify"]; asked {
t.Fatalf("有唯一靶子时不得澄清:%v", out)
}
cancel, _ := out["cancel"].(map[string]any)
if cancel == nil || cancel["runId"] != runID || cancel["accepted"] != true || cancel["source"] != "session" {
t.Fatalf("取消回执:%v", out)
}
if _, ran := out["run"]; ran {
t.Fatalf("取消轮不得再起一个回合:%v", out)
}
note, _ := out["intent_session"].(map[string]any)
if note == nil || note["kind"] != "cancelled" || note["runId"] != runID || note["pending"] != false {
t.Fatalf("会话回话:%v", out)
}
// 派发真的生效:运行被取消。
waitRunStatus(t, h, runID, string(core.RunCancelled))
// 会话上留下短期「已取消」记录:挂起预期已清、记录指向被取消的运行。
sess := pendingIntent(t, st, projectID, convID)
if sess == nil || sess.Cancelled == nil || sess.Cancelled.RunID != runID {
t.Fatalf("取消记录:%+v", sess)
}
if len(sess.Expectation.Missing) != 0 || sess.Slots["run_id"] != "" {
t.Fatalf("取消后仍留着待补齐的槽位:%+v", sess)
}
}
// 原文点名的运行优先于会话绑定:被点的那个被取消,会话绑定的那个纹丝不动。
func TestIntentCancelExplicitRunIDWinsOverSessionBinding(t *testing.T) {
srv, st, gate, _ := cancelServer(t)
h := srv.Handler()
projectID, _ := setupConversation(t, h)
// 两个会话各起一个活跃运行(同一用户、同一项目)。
boundConv := mustCode(t, h, "POST", "/api/projects/"+projectID+"/conversations",
map[string]any{"title": "另一个会话"})["id"].(string)
boundRun := liveRun(t, h, boundConv)
_, convID := setupConversation(t, h)
target := liveRun(t, h, convID)
seedBoundRun(t, st, projectID, convID, boundRun)
gate.script("取消刚才的任务 编号:"+target, taskCancelIntent(map[string]string{"run_id": target}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消刚才的任务 编号:" + target})
if code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
cancel, _ := out["cancel"].(map[string]any)
if cancel == nil || cancel["runId"] != target || cancel["source"] != "explicit" || cancel["accepted"] != true {
t.Fatalf("显式运行号必须优先:%v", out)
}
waitRunStatus(t, h, target, string(core.RunCancelled))
if code, out := doJSON(t, h, "GET", "/api/runs/"+boundRun, nil); code == 200 {
if r, _ := out["run"].(map[string]any); r["status"] == string(core.RunCancelled) {
t.Fatalf("会话绑定的运行被误取消:%v", r)
}
}
}
// insertRun 写入一条合成运行行(未在本地执行的运行:候选与可见性用)。
func insertRun(t *testing.T, st *store.Store, id, convID, projectID, owner, status string, age time.Duration) {
t.Helper()
if _, err := st.DB().Exec(`INSERT INTO runs
(id,parent_run_id,conversation_id,project_id,owner,kind,profile_id,status,phase,agent_name,created_at)
VALUES (?,'',?,?,?,'conversation','',?,'','',?)`,
id, convID, projectID, owner, status, time.Now().UTC().Add(-age).Format(time.RFC3339Nano)); err != nil {
t.Fatal(err)
}
}
// 靶子不唯一:不猜。如实澄清,并把发起人自己最近的运行列出来(新到旧)。
func TestIntentCancelWithoutTargetClarifiesWithOwnRuns(t *testing.T) {
srv, st, gate, _ := cancelServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
// 会话里三条未终态运行:两条自己的、一条别人的(后者不得出现在候选里)。
insertRun(t, st, "run_old", convID, projectID, store.OwnerDefault, "running", 2*time.Minute)
insertRun(t, st, "run_new", convID, projectID, store.OwnerDefault, "waiting_hitl", 1*time.Minute)
insertRun(t, st, "run_foreign", convID, projectID, "mallory", "running", 30*time.Second)
insertRun(t, st, "run_done", convID, projectID, store.OwnerDefault, string(core.RunCompleted), 10*time.Second)
gate.script("取消刚才的任务", taskCancelIntent(nil))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消刚才的任务"})
if code != 200 {
t.Fatalf("澄清轮:%d %v", code, out)
}
if _, dispatched := out["cancel"]; dispatched {
t.Fatalf("没有唯一靶子时不得派发:%v", out)
}
c := clarifyOf(t, out)
if c["question"] != "要取消哪一个任务?" {
t.Fatalf("澄清提问:%v", c)
}
missing, _ := c["missing"].([]any)
got := []string{}
for _, m := range missing {
got = append(got, m.(map[string]any)["key"].(string))
}
if strings.Join(got, ",") != "run_new,run_old" {
t.Fatalf("候选必须是自己最近的未终态运行(新到旧):%v", got)
}
settle(t, h, out)
// 澄清不是取消:候选运行的状态一动不动(自己的可见,别人的依旧不可见)。
for _, id := range []string{"run_old", "run_new"} {
if _, err := st.GetRun(store.OwnerDefault, id); err != nil {
t.Fatalf("澄清轮改动了运行 %s:%v", id, err)
}
}
if _, err := st.GetRun("mallory", "run_foreign"); err != nil {
t.Fatalf("澄清轮改动了别人的运行:%v", err)
}
// 一个可取消的运行都没有时如实说明,而不是空问一句。
_, idle := setupConversation(t, h)
gate.script("取消任务", taskCancelIntent(nil))
code, out = doJSON(t, h, "POST", "/api/conversations/"+idle+"/messages",
map[string]any{"text": "取消任务"})
if code != 200 {
t.Fatalf("无候选轮:%d %v", code, out)
}
if q := clarifyOf(t, out)["question"]; q != "现在没有可取消的任务。" {
t.Fatalf("无候选时的提问:%v", q)
}
settle(t, h, out)
}
// 别人的运行既不可见也不可取消:不派发、不泄露存在性。
func TestIntentCancelNeverResolvesAnotherOwnerRun(t *testing.T) {
srv, st, gate, _ := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
foreign := "run_foreign"
insertRun(t, st, foreign, convID, "prj_foreign", "mallory", "running", time.Minute)
if _, err := st.GetRun("mallory", foreign); err != nil {
t.Fatalf("前置:别人的运行应存在:%v", err)
}
gate.script("取消别人的任务", taskCancelIntent(map[string]string{"run_id": foreign}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消别人的任务"})
if code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
if _, dispatched := out["cancel"]; dispatched {
t.Fatalf("别人的运行不得被取消:%v", out)
}
c := clarifyOf(t, out)
if reason, _ := c["reason"].(string); !strings.Contains(reason, "不可见") {
t.Fatalf("应如实说明不可见(与不存在同一出口):%v", c)
}
if missing, _ := c["missing"].([]any); len(missing) != 0 {
t.Fatalf("不得把别人的运行列为候选:%v", c)
}
settle(t, h, out)
if _, err := st.GetRun("mallory", foreign); err != nil {
t.Fatalf("别人的运行被改动了:%v", err)
}
}
// 确认语义不变:注册了高危取消命令时先出确认卡,确认前处理器不执行;
// 确认后才真的派发。会话不记录「已取消」(尚未生效)。
func TestIntentCancelSurfacesHighDangerConfirmCard(t *testing.T) {
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := liveRun(t, h, convID)
risky := agentcommand.CommandSpec{ID: intentCancelCommandID, Title: "取消运行(高危)",
Danger: agentcommand.DangerHigh,
ParamsSchema: json.RawMessage(`{"type":"object","required":["text","run_id"],
"properties":{"run_id":{"type":"string"},"text":{"type":"string"}}}`)}
if err := srv.Bus.RegisterBuiltin(risky, "", func(_ context.Context, req cmdbus.Request) (any, error) {
log.record(req)
return map[string]any{"ok": true}, nil
}); err != nil {
t.Fatal(err)
}
gate.script("取消刚才的任务", taskCancelIntent(map[string]string{"run_id": runID}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消刚才的任务"})
if code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
if out["command"] != intentCancelCommandID {
t.Fatalf("注册了取消命令时应走命令总线:%v", out)
}
receipt, _ := out["receipt"].(map[string]any)
if receipt == nil || receipt["state"] != string(agentcommand.StateAccepted) {
t.Fatalf("高危命令的受理回执:%v", out)
}
card, _ := out["card"].(map[string]any)
if card == nil {
t.Fatalf("高危取消必须出确认卡:%v", out)
}
if len(log.snapshot()) != 0 {
t.Fatalf("确认前不得执行处理器:%v", log.snapshot())
}
if _, noted := out["intent_session"]; noted {
t.Fatalf("确认前不得在会话上记「已取消」:%v", out)
}
if sess := pendingIntent(t, st, projectID, convID); sess != nil {
t.Fatalf("确认前会话不得被改写:%+v", sess)
}
// 卡上确认 → 与任何其他高后果命令同款:处理器执行,参数带运行号。
cardID, _ := card["id"].(string)
code, confirmed := doJSON(t, h, "POST", "/api/cards/"+cardID+"/actions/confirm", map[string]any{})
if code != 200 {
t.Fatalf("确认:%d %v", code, confirmed)
}
rec, _ := confirmed["receipt"].(map[string]any)
if rec == nil || rec["state"] != string(agentcommand.StateCommitted) {
t.Fatalf("确认回执:%v", confirmed)
}
calls := log.snapshot()
if len(calls) != 1 || calls[0]["run_id"] != runID || calls[0]["text"] != "取消刚才的任务" {
t.Fatalf("确认后的派发参数:%v", calls)
}
}
// 没有会话、也没有点名运行的历史行为逐字节不变:普通回合照常起,响应里既
// 没有取消字段,也没有会话语义字段。
func TestIntentCancelLegacyTurnUnchangedWithoutSession(t *testing.T) {
srv, _, gate, log := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
gate.script("今天天气如何", intentgate.MatchResult{Branch: "fallback"})
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "今天天气如何"})
if code != 200 {
t.Fatalf("普通轮:%d %v", code, out)
}
for _, key := range []string{"cancel", "clarify", "intent_session", "command"} {
if _, has := out[key]; has {
t.Fatalf("普通轮不应出现 %s:%v", key, out)
}
}
if _, ran := out["run"]; !ran {
t.Fatalf("普通轮照常起 run:%v", out)
}
settle(t, h, out)
if len(log.snapshot()) != 0 {
t.Fatalf("普通轮不得派发命令:%v", log.snapshot())
}
}
// 会话上的「已取消」记录不再是取消的靶子:重复取消如实回到澄清。
func TestIntentCancelRecordIsNotACancelTarget(t *testing.T) {
srv, st, gate, _ := cancelServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := liveRun(t, h, convID)
seedBoundRun(t, st, projectID, convID, runID)
gate.script("取消刚才的任务", taskCancelIntent(nil))
if code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消刚才的任务"}); code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
waitRunStatus(t, h, runID, string(core.RunCancelled))
gate.script("再取消一次", taskCancelIntent(nil))
// 第二轮是占槽位的澄清回合:等上一轮被取消的运行彻底收工再发车(忙 409 是
// 既有语义,见 postConversationTurn)。
code, out := postConversationTurn(t, srv, h, st, convID, "再取消一次")
if code != 200 {
t.Fatalf("第二轮取消:%d %v", code, out)
}
if _, dispatched := out["cancel"]; dispatched {
t.Fatalf("已取消的记录不得再当靶子:%v", out)
}
if _, kept := out["intent_session"]; kept {
t.Fatalf("澄清轮不得改写会话记录:%v", out)
}
settle(t, h, out)
}
// 命令值里的运行号(库层 cancel 命令自带 value)同样能解出靶子。
func TestIntentCancelResolvesRunIDFromCommandValue(t *testing.T) {
srv, _, gate, _ := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := liveRun(t, h, convID)
withValue := taskCancelIntent(nil)
withValue.Commands[0].Value = runID
gate.script("取消那个", withValue)
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消那个"})
if code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
cancel, _ := out["cancel"].(map[string]any)
if cancel == nil || cancel["runId"] != runID || cancel["accepted"] != true {
t.Fatalf("命令值里的运行号未解出:%v", out)
}
waitRunStatus(t, h, runID, string(core.RunCancelled))
}
// 消息原文进 params:取消命令与其它命令入口同款(恒带 text)。
func TestIntentCancelPassesOriginalText(t *testing.T) {
srv, _, gate, log := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
runID := liveRun(t, h, convID)
if err := srv.Bus.RegisterBuiltin(agentcommand.CommandSpec{ID: intentCancelCommandID,
Title: "取消运行", ParamsSchema: json.RawMessage(`{"type":"object","required":["text","run_id"],
"properties":{"run_id":{"type":"string"},"text":{"type":"string"}}}`)},
"", func(_ context.Context, req cmdbus.Request) (any, error) {
log.record(req)
raw, _ := json.Marshal(req)
if !strings.Contains(string(raw), `"Entrance":"chat"`) {
t.Errorf("Entrance 必须归一为 chat:%s", raw)
}
if req.Actor == "" || req.Owner == "" {
t.Errorf("Actor/Owner 必须是发起人:%+v", req)
}
return map[string]any{"ok": true}, nil
}); err != nil {
t.Fatal(err)
}
gate.script("取消刚才的任务", taskCancelIntent(map[string]string{"run_id": runID}))
code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
map[string]any{"text": "取消刚才的任务"})
if code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
calls := log.snapshot()
if len(calls) != 1 || calls[0]["run_id"] != runID || calls[0]["text"] != "取消刚才的任务" {
t.Fatalf("派发参数:%v", calls)
}
}
+328
View File
@@ -0,0 +1,328 @@
package server
// 意图会话语义(A1/U-A1b):chat 入口消费库层的会话命令(set_slot/
// change_slot/cancel/expectation),把「动作可执行但槽位未就绪」的那一轮
// 记成会话里的挂起预期,等用户后续轮次把槽位补齐。
//
// 纪律:
// - 补齐后走的是与一次完整识别完全相同的 route.Decide + admitIntentCommand
// + Bus.Dispatch 路径;会话只携带槽位与预期,绝不携带授权——高危动作
// 的确认仍由 cmdbus 收口。
// - 没有挂起预期时本段整体空转:响应逐字节与既有行为一致。
// - 取消分两件事,宿主如实区分,绝不混为一谈:
// · 「放弃本次补充」(会话语义):cancel 命令 + 有挂起预期 + 解不出要
// 取消的运行 → 清预期并如实确认,不派发任何业务动作;
// · 「取消一个运行」(业务动作,A1/U-A1c):解出了唯一靶子 → 走既有
// 取消路径派发,见 intent_cancel.go。
// 两者都不会绕过 cmdbus 的确认收口。
import (
"errors"
"fmt"
"strings"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/intentsession"
"agent-system-backend/internal/route"
core "agent-system-core"
)
// IntentSessions 是意图会话的宿主消费面(intentsession.Service 的端口)。
// 组合根/测试可注入替身;nil 时按存储自动装配。
type IntentSessions interface {
Pending(intentsession.Key) (*intentsession.Session, error)
Suspend(intentsession.Key, intentsession.Expectation, map[string]string) (*intentsession.Session, error)
ApplyAnswer(intentsession.Key, int, intentsession.Answer) (*intentsession.Outcome, error)
ApplyCommand(intentsession.Key, int, string, string, string) (*intentsession.Outcome, error)
Drop(intentsession.Key) error
Binding(intentsession.Key) (*intentsession.Binding, error)
RecordCancel(intentsession.Key, intentsession.Binding) (*intentsession.Session, error)
}
// SetIntentSessionService 注入意图会话服务(组合根与测试用;nil 回落按存储
// 自动装配)。
func (s *Server) SetIntentSessionService(svc IntentSessions) {
s.sessionsMu.Lock()
defer s.sessionsMu.Unlock()
s.sessions = svc
}
// intentSessionService 取会话服务:未注入时从存储惰性装配一次,并把槽位
// 校验接到命令总线的 DECLARED schema 上。
func (s *Server) intentSessionService() IntentSessions {
s.sessionsMu.Lock()
defer s.sessionsMu.Unlock()
if s.sessions == nil && s.st != nil {
svc := intentsession.New(s.st)
svc.SetSlotValidator(s.validateIntentSlot)
s.sessions = svc
}
return s.sessions
}
// intentSessionKey 是会话状态的键:owner + project + conversation(三元组
// 缺一不可,两个用户/项目在构造上互不可见)。
func intentSessionKey(owner string, conv core.Conversation) intentsession.Key {
return intentsession.Key{Owner: owner, Project: string(conv.ProjectID), Conversation: string(conv.ID)}
}
// validateIntentSlot 按动作声明的 schema 校验一个槽位值:声明不可解析
// (命令不存在/未启用)时不校验,交由既有准入面 fail-closed。
func (s *Server) validateIntentSlot(actionID, slot, value string) error {
if s.Bus == nil || actionID == "" {
return nil
}
spec, _, err := s.Bus.Registry().Resolve(actionID)
if err != nil {
return nil
}
props, _, ok := objectSchema(spec.ParamsSchema)
if !ok {
return nil
}
decl, declared := props[slot]
if !declared {
return fmt.Errorf("命令未声明参数:%s", slot)
}
if _, ok := castParam(decl, value); !ok {
return fmt.Errorf("参数类型与声明不符:%s", slot)
}
return nil
}
// intentSessionOutcome 是本轮会话语义的结果,供响应面如实回话。
type intentSessionOutcome struct {
// cancelled 表示用户取消了挂起的补充(不派发任何业务动作)。
cancelled bool
// clarify 是仍缺槽位时的澄清诉求(仍缺时才非空)。
clarify *clarifyPayload
// note 是取消这类「不需要派发」的如实回话。
note *intentSessionNote
}
// intentSessionNote 是会话语义的回话标记(既有响应上的附加字段,不新增 API 面)。
type intentSessionNote struct {
Kind string `json:"kind"`
Action string `json:"action,omitempty"`
Intent string `json:"intent,omitempty"`
RunID string `json:"runId,omitempty"`
Reply string `json:"reply"`
Pending bool `json:"pending"`
}
// fuseIntentSession 把本轮用户输入折叠进挂起预期:补齐即合成一条与完整
// 识别同形的 MatchResult(下游照常 route → admit → dispatch);仍缺则
// 如实重新澄清;取消则清理会话且不派发。
// 返回 nil 结果表示本轮与会话语义无关(无挂起预期、或属于别的任务),
// 调用方保持既有行为。
func (s *Server) fuseIntentSession(key intentsession.Key,
recognized intentgate.MatchResult) (intentgate.MatchResult, *intentSessionOutcome, error) {
svc := s.intentSessionService()
if svc == nil || !keyUsable(key) {
return recognized, nil, nil
}
pending, err := svc.Pending(key)
if err != nil {
return recognized, nil, err
}
// 「已取消」记录不是待补齐的动作:它不再吸收本轮的槽位补充(那一轮由
// 取消路由自己处理),只作为如实回话的依据。
if pending == nil || pending.Expectation == nil || pending.Cancelled != nil {
return recognized, nil, nil
}
answer := answerFromRecognition(pending, recognized)
if answer.Cancel {
return recognized, s.applyCancel(svc, key, pending, answer), nil
}
// 本轮对同一动作已是一次完整的新识别:它就是新的指令,按它派发;
// 旧累计槽位不混进新任务(会话就此作废)。
if recognized.Branch == "matched" && recognized.ID != "" &&
recognized.ID == pending.Expectation.IntentID && recognized.Complete() {
return recognized, nil, svc.Drop(key)
}
if len(answer.Slots) == 0 && len(answer.Invalid) == 0 {
// 本轮没有补槽信息(含「命中的是别的意图」):保持既有行为,挂起预期
// 原样留着,由续期或过期收尾——绝不拿别的任务的输入凑当前动作。
return recognized, nil, nil
}
out, err := svc.ApplyAnswer(key, pending.Revision, answer)
if errors.Is(err, intentsession.ErrStale) {
// 会话已被别的轮次推进或取消:陈旧回答作废,绝不据此执行动作。
return recognized, nil, nil
}
if err != nil || out == nil {
return recognized, nil, err
}
if out.Complete {
// 补齐:recognized 已是与完整识别同形的结果,直接走既有准入与
// 派发路径(无需任何附加回话字段)。
return completedRecognition(pending, out), &intentSessionOutcome{}, nil
}
return recognized, &intentSessionOutcome{clarify: sessionClarify(pending, out)}, nil
}
// applyCancel 消费库层 cancel:清空挂起预期并如实确认取消(不派发业务动作)。
func (s *Server) applyCancel(svc IntentSessions, key intentsession.Key,
pending *intentsession.Session, answer intentsession.Answer) *intentSessionOutcome {
out, err := svc.ApplyAnswer(key, pending.Revision, answer)
if errors.Is(err, intentsession.ErrStale) || err != nil || out == nil || !out.Cancelled {
return nil
}
action, intent := "", pending.Expectation.IntentID
if pending.Expectation.Action.ID != "" {
action = pending.Expectation.Action.ID
}
return &intentSessionOutcome{
cancelled: true,
note: &intentSessionNote{Kind: "cancelled", Action: action, Intent: intent, Pending: false,
Reply: "已取消本次槽位补充,未执行任何动作。"},
}
}
// answerFromRecognition 把一次识别投影成会话语义的答案:库层会话命令
// (cancel 优先)、挂起意图自身的槽位与非法原因。归属别的意图时一概不碰
// 本会话(别的任务的输入绝不混进当前动作)。
func answerFromRecognition(pending *intentsession.Session, recognized intentgate.MatchResult) intentsession.Answer {
answer := intentsession.Answer{}
sameIntent := recognized.ID != "" && recognized.ID == pending.Expectation.IntentID
otherIntent := recognized.ID != "" && !sameIntent
for _, cmd := range recognized.Commands {
switch cmd.Kind {
case intentgate.CommandCancel:
if !otherIntent { // 取消指向别的意图时不碰本会话
answer.Cancel = true
return answer
}
case intentgate.CommandSetSlot, intentgate.CommandChangeSlot:
if strings.TrimSpace(cmd.Name) == "" {
continue
}
if answer.Slots == nil {
answer.Slots = map[string]string{}
}
answer.Slots[cmd.Name] = cmd.Value // 显式落槽命令:不看意图归属
}
}
if !sameIntent {
return answer
}
if len(recognized.Slots) > 0 {
if answer.Slots == nil {
answer.Slots = make(map[string]string, len(recognized.Slots))
}
for k, v := range recognized.Slots {
answer.Slots[k] = v
}
}
for _, defect := range recognized.Invalid {
if answer.Invalid == nil {
answer.Invalid = map[string]string{}
}
answer.Invalid[defect.Key] = firstNonEmpty(defect.Reason, "value rejected")
}
return answer
}
// completedRecognition 合成一条与「一次完整识别」同形的 MatchResult:同一
// 意图 ID、累计槽位、同一动作引用的 Start 命令。风险与确认声明刻意不搬运
// ——会话不携带授权语义,确认始终由 cmdbus 按命令 danger 收口。
func completedRecognition(pending *intentsession.Session, out *intentsession.Outcome) intentgate.MatchResult {
action := pending.Expectation.Action
res := intentgate.MatchResult{
Branch: "matched",
ID: pending.Expectation.IntentID,
Slots: out.Slots,
Commands: []intentgate.Command{{Kind: intentgate.CommandStart, Action: action, Slots: out.Slots}},
}
if action.ID != "" {
ref := action
res.Action = &ref
}
return res
}
// sessionClarify 仍缺槽位时的澄清诉求:只报还差什么(缺失/非法各带原因)。
func sessionClarify(pending *intentsession.Session, out *intentsession.Outcome) *clarifyPayload {
action, intent := "", pending.Expectation.IntentID
if pending.Expectation.Action.ID != "" {
action = pending.Expectation.Action.ID
}
// 提问优先问「还差的那个」:缺失在前、非法在后,都没有才回落挂起时的
// 原始提问。
question := ""
if len(out.Remaining) > 0 {
question = strings.TrimSpace(out.Remaining[0].Question)
} else if len(out.Invalid) > 0 {
question = strings.TrimSpace(out.Invalid[0].Question)
}
if question == "" {
question = strings.TrimSpace(out.Question)
}
c := &clarifyPayload{Intent: intent, Command: action, Question: question,
Missing: append([]intentgate.SlotQuestion(nil), out.Remaining...),
Invalid: append([]intentgate.SlotDefect(nil), out.Invalid...)}
if c.Question == "" {
return nil
}
return c
}
// suspendIntentSession 把「动作可执行但槽位未就绪」的那一轮记成会话里的
// 挂起预期(U6 准入面:声明的必填参数无法由原文+槽位满足;路由面:缺失或
// 非法的必填槽位)。响应面不受影响——澄清话术仍由既有 clarify 字段给出。
func (s *Server) suspendIntentSession(key intentsession.Key, recognized intentgate.MatchResult,
dec route.Decision, adm *intentAdmission) error {
svc := s.intentSessionService()
if svc == nil || !keyUsable(key) {
return nil
}
commandID := dec.Clarification.Command
var missing []intentgate.SlotQuestion
question := strings.TrimSpace(dec.Clarification.Question)
switch {
case dec.Kind == route.KindCommand && adm != nil && adm.Blocked:
// 路由已认为动作可派发,但准入未过(声明的必填参数补不齐)。
commandID = dec.CommandID
for _, name := range adm.MissingParams {
missing = append(missing, intentgate.SlotQuestion{Key: name, Question: name})
}
case dec.Clarification.Needed && commandID != "":
// 路由层槽位缺口/非法:动作已解析到已存在的命令,只是槽位不齐。
missing = append(missing, dec.Clarification.Missing...)
for _, defect := range dec.Clarification.Invalid {
missing = append(missing, intentgate.SlotQuestion{Key: defect.Key, Question: defect.Question})
}
default:
return nil // 没有待补齐的可执行动作:不挂起
}
if commandID == "" {
return nil
}
if question == "" && len(missing) > 0 {
question = strings.TrimSpace(missing[0].Question)
}
_, err := svc.Suspend(key, intentsession.Expectation{
IntentID: recognized.ID,
Action: intentgate.ActionRef{Kind: intentgate.ActionCommand, ID: commandID},
Missing: missing, Question: question,
}, dec.Slots)
return err
}
// keyUsable 拒绝不完整的键(无归属即不落任何会话状态)。
func keyUsable(key intentsession.Key) bool { return key.Owner != "" && key.Conversation != "" }
// sessionRunTarget 把意图会话里绑定的运行上下文投影给路由器(取消靶子的
// 会话来源)。没有绑定、绑定已取消、或读取失败一律返回 false:宿主随即
// 如实澄清,绝不臆造一个靶子。
func (s *Server) sessionRunTarget(key intentsession.Key) (route.SessionRun, bool) {
svc := s.intentSessionService()
if svc == nil || !keyUsable(key) {
return route.SessionRun{}, false
}
b, err := svc.Binding(key)
if err != nil || b == nil || b.Cancelled || strings.TrimSpace(b.RunID) == "" {
return route.SessionRun{}, false
}
return route.SessionRun{RunID: b.RunID, Intent: b.Intent, Action: b.Action}, true
}
+676
View File
@@ -0,0 +1,676 @@
package server
import (
"context"
"encoding/json"
"fmt"
"net/http"
"sort"
"strings"
"sync"
"testing"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/cmdbus"
"agent-system-backend/internal/intentgate"
"agent-system-backend/internal/intentsession"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// A1/U-A1b 会话语义的 chat 端到端(合成数据 + 现有 fakes):缺槽位动作先
// 澄清挂起,下一轮补齐后按同一条准入/派发路径执行;取消清理会话;过期会话
// 不影响任何新轮次。高危确认仍由 cmdbus 收口(会话只带槽位)。
// scriptedGate 按输入文本返回脚本化的识别结果;未命中的输入走普通回合
// (fallback)。这是库层识别的替身,不改 Stage-1 语义。
type scriptedGate struct {
mu sync.Mutex
results map[string]intentgate.MatchResult
}
func (g *scriptedGate) script(input string, res intentgate.MatchResult) {
g.mu.Lock()
defer g.mu.Unlock()
if g.results == nil {
g.results = map[string]intentgate.MatchResult{}
}
g.results[input] = res
}
func (g *scriptedGate) Recognize(_ context.Context, input string) intentgate.MatchResult {
g.mu.Lock()
defer g.mu.Unlock()
res, ok := g.results[input]
if !ok {
return intentgate.MatchResult{Branch: "fallback"}
}
return res
}
func (g *scriptedGate) Match(_ context.Context, input string) (string, string) {
res := g.Recognize(context.Background(), input)
return res.Branch, res.ID
}
// cancelStart 是「取消运行」动作的一次完整/不完整识别(run-cancel 形态)。
func cancelStart(slots map[string]string, missing []intentgate.SlotQuestion) intentgate.MatchResult {
res := intentgate.MatchResult{
Branch: "matched", ID: "demo.run-cancel", Slots: slots, Missing: missing,
Commands: []intentgate.Command{{Kind: intentgate.CommandStart,
Action: intentgate.ActionRef{Kind: intentgate.ActionAuto, ID: "demo.run-cancel"}, Slots: slots}},
}
if len(missing) > 0 {
res.Commands = append(res.Commands, intentgate.Command{Kind: intentgate.CommandClarify,
Question: missing[0].Question})
}
return res
}
// cancelAnswer 是「对挂起预期的回答」(库层 expectation_hit 分支的同形结果)。
func cancelAnswer(slots map[string]string, missing []intentgate.SlotQuestion) intentgate.MatchResult {
res := intentgate.MatchResult{Branch: "expectation_hit", ID: "demo.run-cancel", Slots: slots, Missing: missing}
if len(missing) == 0 {
res.Commands = []intentgate.Command{{Kind: intentgate.CommandStart,
Action: intentgate.ActionRef{Kind: intentgate.ActionAuto, ID: "demo.run-cancel"}, Slots: slots}}
}
return res
}
var cancelSpec = agentcommand.CommandSpec{ID: "demo.run-cancel", Title: "取消运行",
ParamsSchema: json.RawMessage(`{"type":"object","required":["text","run_id","reason"],
"properties":{"run_id":{"type":"string"},"reason":{"type":"string"}}}`)}
// cancelServer 组装带脚本化意图闸门与 demo.run-cancel 命令的服务器。
func cancelServer(t *testing.T) (*Server, *store.Store, *scriptedGate, *dispatchLog) {
t.Helper()
return cancelServerWith(t, cancelSpec)
}
// cancelServerWith 同上但命令声明可替换(同 id 只注册一次)。
func cancelServerWith(t *testing.T, spec agentcommand.CommandSpec) (*Server, *store.Store, *scriptedGate, *dispatchLog) {
t.Helper()
srv, st := newTestServer(t)
log := &dispatchLog{}
if err := srv.Bus.RegisterBuiltin(spec, "", func(_ context.Context, req cmdbus.Request) (any, error) {
log.record(req)
return map[string]any{"ok": true}, nil
}); err != nil {
t.Fatal(err)
}
gate := &scriptedGate{}
srv.Gate = gate
return srv, st, gate, log
}
// settle 等本轮启动的回合落终态(会话槽位随即释放,便于下一轮)。
func settle(t *testing.T, h http.Handler, out map[string]any) map[string]any {
t.Helper()
if run, ok := out["run"].(map[string]any); ok {
if id, _ := run["id"].(string); id != "" {
waitRunTerminal(t, h, id, testRunTerminalTimeout)
}
}
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)
h := srv.Handler()
_, convID := setupConversation(t, h)
runIDQ := "取消哪个运行?"
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: runIDQ}, {Key: "reason", Question: "取消原因?"}}))
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
if _, executed := out["command"]; executed {
t.Fatalf("槽位未就绪不得派发业务动作:%v", out)
}
c := clarifyOf(t, out)
if c["command"] != "demo.run-cancel" || c["question"] != runIDQ {
t.Fatalf("澄清诉求:%v", c)
}
settle(t, h, out)
if len(log.snapshot()) != 0 {
t.Fatalf("澄清轮不得派发:%v", log.snapshot())
}
pending := pendingIntent(t, st, "", convID)
if pending == nil || pending.Expectation.Action.ID != "demo.run-cancel" {
t.Fatalf("未挂起预期:%+v", pending)
}
// 下一轮:仅补齐槽位(库层挂起预期命中分支的同形结果)。
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)
if code != 200 || out["command"] != "demo.run-cancel" {
t.Fatalf("补齐后应派发同一动作:%d %v", code, out)
}
if _, hasClarify := out["clarify"]; hasClarify {
t.Fatalf("补齐后不应再澄清:%v", out)
}
calls := log.snapshot()
if len(calls) != 1 {
t.Fatalf("处理器调用次数:%d", len(calls))
}
params := calls[0]
if params["run_id"] != "R-7" || params["reason"] != "误触发" || params["text"] != answerText {
t.Fatalf("累计槽位未进派发参数:%v", params)
}
if pendingIntent(t, st, "", convID) != nil {
t.Fatal("补齐后会话应被消费")
}
}
// 两轮部分填充:先 set_slot 落一个槽位(仍缺 → 重新澄清),再由回答补齐。
func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
settle(t, h, out)
// 第二轮:库层 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")
if code != 200 {
t.Fatalf("set_slot 轮:%d %v", code, out)
}
settle(t, h, out)
if _, executed := out["command"]; executed {
t.Fatalf("仍缺槽位不得派发:%v", out)
}
c := clarifyOf(t, out)
if c["question"] != "取消原因?" {
t.Fatalf("仍缺的提问:%v", c)
}
missing, _ := c["missing"].([]any)
if len(missing) != 1 || missing[0].(map[string]any)["key"] != "reason" {
t.Fatalf("仍缺清单:%v", c)
}
if len(log.snapshot()) != 0 {
t.Fatalf("部分填充不得派发:%v", log.snapshot())
}
// 第三轮:补上原因 → 派发,两个槽位都来自会话累计。
gate.script("原因:误触发", cancelAnswer(map[string]string{"reason": "误触发"}, nil))
code, out = postConversationTurn(t, srv, h, st, convID, "原因:误触发")
if code != 200 || out["command"] != "demo.run-cancel" {
t.Fatalf("补齐后应派发:%d %v", code, out)
}
calls := log.snapshot()
if len(calls) != 1 || calls[0]["run_id"] != "R-7" || calls[0]["reason"] != "误触发" {
t.Fatalf("累计槽位:%v", calls)
}
}
// 取消挂起的补充:如实确认取消,不派发任何业务动作,会话清空。
func TestIntentSessionCancelClearsPendingAction(t *testing.T) {
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
settle(t, h, out)
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, "算了不取消了")
if code != 200 {
t.Fatalf("取消轮:%d %v", code, out)
}
settle(t, h, out)
if _, executed := out["command"]; executed {
t.Fatalf("取消不得派发业务动作:%v", out)
}
note, ok := out["intent_session"].(map[string]any)
if !ok {
t.Fatalf("取消应如实回话:%v", out)
}
if note["kind"] != "cancelled" || note["action"] != "demo.run-cancel" || note["reply"] == "" || note["pending"] != false {
t.Fatalf("取消回话:%v", note)
}
if len(log.snapshot()) != 0 {
t.Fatalf("取消轮派发了动作:%v", log.snapshot())
}
if pendingIntent(t, st, "", convID) != nil {
t.Fatal("取消后仍挂起预期")
}
// 取消之后的回答不得补齐旧动作(这里识别仍不完整 → 只澄清,不派发)。
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")
if code != 200 {
t.Fatalf("取消后回答轮:%d %v", code, out)
}
settle(t, h, out)
if _, executed := out["command"]; executed {
t.Fatalf("取消后的回答不得补齐旧动作:%v", out)
}
if len(log.snapshot()) != 0 {
t.Fatalf("取消后的回答派发了动作:%v", log.snapshot())
}
}
// 过期会话既不补齐旧动作,也不影响新的动作:过期的累计槽位不得被新的一轮
// 借走,新的任务照常派发。
func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
projectID, convID := setupConversation(t, h)
// 第一轮:抽到 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")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
settle(t, h, out)
pending := pendingIntent(t, st, projectID, convID)
if pending == nil || pending.Slots["run_id"] != "R-7" {
t.Fatalf("未挂起预期/未累计槽位:%+v", pending)
}
// 把挂起预期推到过去(等价于 30 分钟未续期后仍在库里的行)。
expired := &store.IntentExpectation{IntentID: "demo.run-cancel", ActionKind: "command",
ActionID: "demo.run-cancel", Missing: []store.IntentSlotQuestion{{Key: "reason", Question: "取消原因?"}}}
if _, err := st.PutIntentSession(store.OwnerDefault, projectID, convID, expired,
map[string]string{"run_id": "R-7"}, -time.Minute); err != nil {
t.Fatal(err)
}
// 第二轮只补了原因:过期的 run_id 不得被借来凑齐动作。
answerText := "原因:误触发"
gate.script(answerText, cancelStart(map[string]string{"reason": "误触发"}, nil))
code, out = postConversationTurn(t, srv, h, st, convID, answerText)
if code != 200 {
t.Fatalf("过期后的回答轮:%d %v", code, out)
}
settle(t, h, out)
if _, executed := out["command"]; executed {
t.Fatalf("过期会话不得补齐旧动作:%v", out)
}
if len(log.snapshot()) != 0 {
t.Fatalf("过期会话派发了动作:%v", log.snapshot())
}
c := clarifyOf(t, out)
if c["command"] != "demo.run-cancel" {
t.Fatalf("澄清诉求:%v", c)
}
missing, _ := c["missing"].([]any)
foundRunID := false
for _, m := range missing {
if m.(map[string]any)["key"] == "run_id" {
foundRunID = true
}
}
if !foundRunID {
t.Fatalf("过期的累计槽位被借走了:%v", c)
}
if _, busy := out["intent_session"]; busy {
t.Fatalf("过期会话不得产生会话语义字段:%v", out)
}
if _, err := st.GetIntentSession(store.OwnerDefault, projectID, convID); err != store.ErrNotFound {
t.Fatalf("过期行未被惰性删除:%v", err)
}
// 新任务照常派发(过期会话从不干扰新轮次)。
if err := srv.Bus.RegisterBuiltin(agentcommand.CommandSpec{ID: "demo.purge", Title: "清理缓存",
ParamsSchema: json.RawMessage(`{"type":"object","required":["text"],"properties":{"text":{"type":"string"}}}`)},
"", func(_ context.Context, req cmdbus.Request) (any, error) {
log.record(req)
return map[string]any{"ok": true}, nil
}); err != nil {
t.Fatal(err)
}
gate.script("清理缓存", otherIntentStart("demo.purge"))
code, out = postConversationTurn(t, srv, h, st, convID, "清理缓存")
if code != 200 || out["command"] != "demo.purge" {
t.Fatalf("新任务应照常派发:%d %v", code, out)
}
if calls := log.snapshot(); len(calls) != 1 || calls[0]["text"] != "清理缓存" {
t.Fatalf("新任务派发参数:%v", calls)
}
}
// otherIntentStart 是另一个动作的完整识别(换任务)。
func otherIntentStart(id string) intentgate.MatchResult {
return intentgate.MatchResult{Branch: "matched", ID: id,
Commands: []intentgate.Command{{Kind: intentgate.CommandStart,
Action: intentgate.ActionRef{Kind: intentgate.ActionAuto, ID: id}}}}
}
// 别的会话/项目的挂起预期绝不互相影响(键含 project + conversation)。
func TestIntentSessionIsolatesConversations(t *testing.T) {
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
projectA, convA := setupConversation(t, h)
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, "取消运行")
if code != 200 {
t.Fatalf("A 会话第一轮:%d %v", code, out)
}
settle(t, h, out)
if pendingIntent(t, st, projectA, convA) == nil {
t.Fatal("A 会话未挂起预期")
}
if pendingIntent(t, st, projectB, convB) != nil {
t.Fatal("B 会话看到了 A 的挂起预期")
}
// 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)
if code != 200 {
t.Fatalf("B 会话回答轮:%d %v", code, out)
}
settle(t, h, out)
if len(log.snapshot()) != 0 {
t.Fatalf("无挂起预期的会话不得凭别人的槽位派发:%v", log.snapshot())
}
if pendingIntent(t, st, projectA, convA) == nil {
t.Fatal("B 会话的轮次改动了 A 的挂起预期")
}
}
// 高危动作仍走 cmdbus 确认:会话补齐不绕过确认卡(处理器不执行)。
func TestIntentSessionStillRequiresHighRiskConfirmation(t *testing.T) {
risky := cancelSpec
risky.Danger = agentcommand.DangerHigh
srv, st, gate, log := cancelServerWith(t, risky)
h := srv.Handler()
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
settle(t, h, out)
answer := "编号:R-7 原因:误触发"
gate.script(answer, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
code, out = postConversationTurn(t, srv, h, st, convID, answer)
if code != 200 || out["command"] != "demo.run-cancel" {
t.Fatalf("补齐后应派发到总线:%d %v", code, out)
}
if out["card"] == nil {
t.Fatalf("高危动作必须仍要确认卡:%v", out)
}
if len(log.snapshot()) != 0 {
t.Fatalf("确认前不得执行处理器:%v", log.snapshot())
}
}
// 非法槽位的补充照旧按非法追问:值不采纳、动作不派发。
func TestIntentSessionInvalidAnswerKeepsClarifying(t *testing.T) {
srv, st, gate, log := cancelServer(t)
h := srv.Handler()
_, convID := setupConversation(t, h)
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}}))
code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
if code != 200 {
t.Fatalf("第一轮:%d %v", code, out)
}
settle(t, h, out)
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")
if code != 200 {
t.Fatalf("非法补充轮:%d %v", code, out)
}
settle(t, h, out)
if _, executed := out["command"]; executed {
t.Fatalf("非法补充不得派发:%v", out)
}
c := clarifyOf(t, out)
invalid, _ := c["invalid"].([]any)
if len(invalid) != 1 || invalid[0].(map[string]any)["reason"] != "unknown run" {
t.Fatalf("非法原因未透传:%v", c)
}
if len(log.snapshot()) != 0 {
t.Fatalf("非法补充派发了动作:%v", log.snapshot())
}
pending := pendingIntent(t, st, "", convID)
if pending == nil || pending.Slots["run_id"] != "" {
t.Fatalf("非法值不得进入会话槽位表:%+v", pending)
}
}
// 无会话时行为逐字节不变:与「会话关闭」的服务器跑同一脚本,响应(去掉
// run/message 这类每次都不同的回执)逐字节一致,且不得出现会话语义字段。
func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
script := func(gate *scriptedGate) {
// 槽位齐全:照常派发(会话层不得介入)。
gate.script("取消运行 编号:R-3 原因:误触发", cancelStart(
map[string]string{"run_id": "R-3", "reason": "误触发"}, nil))
gate.script("今天天气如何", intentgate.MatchResult{Branch: "fallback"})
gate.script("算了", intentgate.MatchResult{Branch: "matched", ID: "demo.purge",
Commands: []intentgate.Command{{Kind: intentgate.CommandCancel, Target: "demo.purge"}}})
gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
{Key: "run_id", Question: "取消哪个运行?"}, {Key: "reason", Question: "取消原因?"}}))
}
// summarize 把响应投影成与运行无关的稳定形态:顶层键、命令、澄清诉求。
summarize := func(out map[string]any) string {
keys := make([]string, 0, len(out))
for k := range out {
keys = append(keys, k)
}
sort.Strings(keys)
var b strings.Builder
b.WriteString(strings.Join(keys, ","))
if cmd, ok := out["command"]; ok {
raw, _ := json.Marshal(cmd)
b.WriteString("|command=" + string(raw))
}
if c, ok := out["clarify"]; ok {
raw, _ := json.Marshal(c)
b.WriteString("|clarify=" + string(raw))
}
if n, ok := out["intent_session"]; ok {
raw, _ := json.Marshal(n)
b.WriteString("|session=" + string(raw))
}
return b.String()
}
run := func(t *testing.T, sessions IntentSessions) []string {
t.Helper()
srv, st, gate, log := cancelServer(t)
if sessions != nil {
srv.SetIntentSessionService(sessions)
}
script(gate)
h := srv.Handler()
_, convID := setupConversation(t, h)
var summaries []string
for _, text := range []string{"取消运行 编号:R-3 原因:误触发", "今天天气如何", "算了", "取消运行"} {
code, out := postConversationTurn(t, srv, h, st, convID, text)
if code != 200 {
t.Fatalf("%s:%d %v", text, code, out)
}
settle(t, h, out)
summaries = append(summaries, text+" -> "+summarize(out))
}
if len(log.snapshot()) != 1 {
t.Fatalf("脚本只应派发一次完整动作:%v", log.snapshot())
}
return summaries
}
withSessions := run(t, nil)
// 关闭会话语义(替身永不挂起/折叠)后同一脚本的响应必须逐字节一致。
without := run(t, inertSessions{})
if len(withSessions) != len(without) {
t.Fatalf("轮次数不一致:%v vs %v", withSessions, without)
}
for i := range withSessions {
if withSessions[i] != without[i] {
t.Fatalf("无会话时行为发生变化:\n 有:%s\n 无:%s", withSessions[i], without[i])
}
if strings.Contains(withSessions[i], "session=") {
t.Fatalf("无会话语义的轮次带上了会话语义字段:%s", withSessions[i])
}
}
if !strings.Contains(strings.Join(withSessions, "\n"), "|clarify=") {
t.Fatalf("脚本未触发澄清,测试失去意义:%v", withSessions)
}
}
// inertSessions 是「关闭会话语义」的替身:任何调用都如实报告没有会话。
type inertSessions struct{}
func (inertSessions) Pending(intentsession.Key) (*intentsession.Session, error) { return nil, nil }
func (inertSessions) Suspend(intentsession.Key, intentsession.Expectation, map[string]string) (*intentsession.Session, error) {
return nil, nil
}
func (inertSessions) ApplyAnswer(intentsession.Key, int, intentsession.Answer) (*intentsession.Outcome, error) {
return nil, nil
}
func (inertSessions) ApplyCommand(intentsession.Key, int, string, string, string) (*intentsession.Outcome, error) {
return nil, nil
}
func (inertSessions) Drop(intentsession.Key) error { return nil }
func (inertSessions) Binding(intentsession.Key) (*intentsession.Binding, error) {
return nil, nil
}
func (inertSessions) RecordCancel(intentsession.Key, intentsession.Binding) (*intentsession.Session, error) {
return nil, nil
}
// pendingIntent 读当前会话的挂起预期(测试只读断言面;过期即视为没有)。
func pendingIntent(t *testing.T, st *store.Store, projectID, convID string) *intentsession.Session {
t.Helper()
conv, err := st.GetConversation(store.OwnerDefault, convID)
if err != nil {
t.Fatal(err)
}
if projectID == "" {
projectID = string(conv.ProjectID)
}
sess, err := intentsession.New(st).Pending(intentsession.Key{
Owner: store.OwnerDefault, Project: projectID, Conversation: convID})
if err != nil {
t.Fatal(err)
}
return sess
}
+32
View File
@@ -0,0 +1,32 @@
package server
import (
"net/http"
"time"
"agent-system-backend/internal/rundiag"
)
// getRunDiagnostics 是单次 run 的只读诊断入口(feature unit A2):
//
// GET /api/runs/{id}/diagnostics
//
// 它回答值班/排障最常问的六个问题:这个 run 卡在哪(responsibility.state)、
// 谁该动(responsibility.role)、为什么在等(responsibility.wait_reason)、最后一个
// 稳定落盘点(last_checkpoint)、已知的外部副作用(side_effects)、下一步动作
// (next_step)。
//
// 授权口径与既有 run 读入口逐字同款:owner 只取自认证上下文,绝不接受请求参数;
// run 由 store.GetRun(owner, id) 做租户过滤,跨租户与不存在同为 404(不泄露存在性)。
// handler 路径上只有 SELECT——不给写操作、不改状态、不触发任何副作用;聚合逻辑在
// internal/rundiag,证据缺失一律降级为 unknown + notes,不编造检查点,也不把"没有
// 记录"读成"没有发生"。
func (s *Server) getRunDiagnostics(w http.ResponseWriter, r *http.Request) {
run, err := s.st.GetRun(s.owner(r), r.PathValue("id"))
if err != nil {
writeStoreErr(w, err)
return
}
evidence := rundiag.Gather(s.st, run, time.Now().UTC())
writeJSON(w, http.StatusOK, rundiag.Build(evidence))
}
+226
View File
@@ -0,0 +1,226 @@
package server
import (
"encoding/json"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
agentcommand "agent-command"
"agent-system-backend/internal/store"
core "agent-system-core"
)
// diagnosticsFixture 建一个 owner 名下的项目/会话/run(合成数据,无执行器)。
func diagnosticsFixture(t *testing.T, st *store.Store, owner string, status core.RunStatus) (core.Run, core.Conversation) {
t.Helper()
project, err := st.CreateProject(owner, "诊断项目")
if err != nil {
t.Fatal(err)
}
conversation, err := st.CreateConversation(owner, project.ID, "诊断会话", "")
if err != nil {
t.Fatal(err)
}
run := core.Run{
ConversationID: conversation.ID, ProjectID: project.ID, Owner: owner,
Kind: core.RunConversation, Status: status, Phase: core.PhaseActing, CreatedAt: time.Now().UTC(),
}
if err := st.CreateRun(&run); err != nil {
t.Fatal(err)
}
return run, conversation
}
func getDiagnostics(t *testing.T, h http.Handler, key, path string) *httptest.ResponseRecorder {
t.Helper()
req := httptest.NewRequest(http.MethodGet, path, nil)
if key != "" {
req.Header.Set("Authorization", "Bearer "+key)
}
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
return rec
}
func decodeReport(t *testing.T, rec *httptest.ResponseRecorder) map[string]any {
t.Helper()
if rec.Code != http.StatusOK {
t.Fatalf("diagnostics: %d %s", rec.Code, rec.Body.String())
}
var out map[string]any
if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
t.Fatalf("decode diagnostics: %v (%s)", err, rec.Body.String())
}
return out
}
// 授权口径与既有 run 读入口同款:未认证 401、跨 owner 404(不泄露存在性)、
// 不存在的 run 同样 404。
func TestRunDiagnosticsOwnerScopingMatchesRunReads(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
aliceKey, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
bobKey, err := createLegacyAPIKeyFixture(t, st, "bob", "e2e")
if err != nil {
t.Fatal(err)
}
run, _ := diagnosticsFixture(t, st, "alice", core.RunRunning)
path := "/api/runs/" + string(run.ID) + "/diagnostics"
if rec := getDiagnostics(t, h, "", path); rec.Code != http.StatusUnauthorized {
t.Fatalf("unauthenticated diagnostics must be rejected: %d", rec.Code)
}
if rec := getDiagnostics(t, h, bobKey, path); rec.Code != http.StatusNotFound {
t.Fatalf("cross-owner diagnostics must not leak the payload: %d %s", rec.Code, rec.Body.String())
}
if rec := getDiagnostics(t, h, aliceKey, "/api/runs/run_missing/diagnostics"); rec.Code != http.StatusNotFound {
t.Fatalf("missing run must be 404: %d", rec.Code)
}
rec := getDiagnostics(t, h, aliceKey, path)
report := decodeReport(t, rec)
runSection, _ := report["run"].(map[string]any)
if runSection["id"] != string(run.ID) || runSection["owner"] != "alice" || runSection["project"] != string(run.ProjectID) {
t.Fatalf("run identity wrong: %v", runSection)
}
responsibility, _ := report["responsibility"].(map[string]any)
if responsibility["state"] != "running" {
t.Fatalf("a running run with no blockers reads as running: %v", responsibility)
}
}
// 隔离中的 run:责任在副作用责任人,下一步是解除隔离;检查点与已记录的命令
// 回执一并给出。
func TestRunDiagnosticsReportsHoldCheckpointAndSideEffects(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
key, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
run, conversation := diagnosticsFixture(t, st, "alice", core.RunRunning)
if _, err := st.AppendEvents(string(conversation.ID), []core.StreamEvent{
{RunID: run.ID, Type: core.EvThinkingDelta, Payload: json.RawMessage(`{"index":0,"text":"核对差异"}`)},
{RunID: run.ID, Type: core.EvAgentProgress, Payload: json.RawMessage(`{"note":"核对中"}`)},
}); err != nil {
t.Fatal(err)
}
if err := st.AppendAudit("alice", agentcommand.AuditEnvelope{
Time: time.Now().UTC(), Actor: "alice", CommandID: "rdm.docs.instantiate",
OperationID: "op_diag_1", ConversationID: string(conversation.ID), State: agentcommand.StateUnknown,
}); err != nil {
t.Fatal(err)
}
if err := st.SetRecoveryHold(string(run.ID), "alice", string(conversation.ID), "外部副作用结果未知"); err != nil {
t.Fatal(err)
}
report := decodeReport(t, getDiagnostics(t, h, key, "/api/runs/"+string(run.ID)+"/diagnostics"))
responsibility, _ := report["responsibility"].(map[string]any)
if responsibility["state"] != "recovery_hold" || responsibility["role"] != "operation-owner" {
t.Fatalf("held run responsibility: %v", responsibility)
}
nextStep, _ := report["next_step"].(map[string]any)
if nextStep["action"] != "release_recovery_hold" {
t.Fatalf("held run next step: %v", nextStep)
}
checkpoint, _ := report["last_checkpoint"].(map[string]any)
if checkpoint == nil || checkpoint["type"] != string(core.EvAgentProgress) {
t.Fatalf("last persisted checkpoint must be surfaced: %v", report["last_checkpoint"])
}
effects, _ := report["side_effects"].([]any)
var foundOperation bool
for _, raw := range effects {
effect, _ := raw.(map[string]any)
if effect["kind"] == "operation" && effect["operation_id"] == "op_diag_1" {
foundOperation = true
if effect["command_id"] != "rdm.docs.instantiate" || effect["state"] != string(agentcommand.StateUnknown) {
t.Fatalf("operation projection: %v", effect)
}
}
}
if !foundOperation {
t.Fatalf("recorded operation missing from side effects: %v", effects)
}
}
// 只读契约:GET 不改任何状态,写方法不注册。
func TestRunDiagnosticsIsReadOnly(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
key, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
run, conversation := diagnosticsFixture(t, st, "alice", core.RunRunning)
if err := st.SetRecoveryHold(string(run.ID), "alice", string(conversation.ID), "unknown"); err != nil {
t.Fatal(err)
}
before, err := st.GetRun("alice", string(run.ID))
if err != nil {
t.Fatal(err)
}
path := "/api/runs/" + string(run.ID) + "/diagnostics"
if rec := getDiagnostics(t, h, key, path); rec.Code != http.StatusOK {
t.Fatalf("diagnostics: %d %s", rec.Code, rec.Body.String())
}
after, err := st.GetRun("alice", string(run.ID))
if err != nil {
t.Fatal(err)
}
if before.Status != after.Status || before.Phase != after.Phase || !before.CreatedAt.Equal(after.CreatedAt) {
t.Fatalf("read-only endpoint mutated the run: %+v -> %+v", before, after)
}
hold, err := st.GetRecoveryHold(string(run.ID))
if err != nil || hold.ReleasedAt != "" {
t.Fatalf("read-only endpoint changed the recovery hold: %+v err=%v", hold, err)
}
for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete, http.MethodPatch} {
req := httptest.NewRequest(method, path, nil)
req.Header.Set("Authorization", "Bearer "+key)
rec := httptest.NewRecorder()
h.ServeHTTP(rec, req)
if rec.Code == http.StatusOK || rec.Code == http.StatusAccepted {
t.Fatalf("%s must not be served on a read-only diagnostics path: %d", method, rec.Code)
}
}
}
// 读不到的证据降级为 unknown + note,不让整个诊断失败,也不给出安全重试。
func TestRunDiagnosticsUnknownStateDegradesWithNote(t *testing.T) {
srv, st := newAuthTestServer(t)
h := srv.Handler()
key, err := createLegacyAPIKeyFixture(t, st, "alice", "e2e")
if err != nil {
t.Fatal(err)
}
run, _ := diagnosticsFixture(t, st, "alice", core.RunCancelled)
report := decodeReport(t, getDiagnostics(t, h, key, "/api/runs/"+string(run.ID)+"/diagnostics"))
responsibility, _ := report["responsibility"].(map[string]any)
if responsibility["state"] != "unknown" {
t.Fatalf("unmapped status must degrade to unknown: %v", responsibility)
}
if report["last_checkpoint"] != nil {
t.Fatalf("no persisted event means no checkpoint reference: %v", report["last_checkpoint"])
}
notes, _ := report["notes"].([]any)
var sawStatusNote, sawEffectNote bool
for _, raw := range notes {
note, _ := raw.(string)
if strings.Contains(note, "no diagnostic mapping") {
sawStatusNote = true
}
if strings.Contains(note, "absence of records is not proof") {
sawEffectNote = true
}
}
if !sawStatusNote || !sawEffectNote {
t.Fatalf("missing degradation notes: %v", notes)
}
}
+63 -2
View File
@@ -19,6 +19,7 @@ import (
"sort"
"strconv"
"strings"
"sync"
"time"
agentcommand "agent-command"
@@ -71,6 +72,11 @@ type Server struct {
set *settings.Service
plugins *plugins.Host
static fs.FS
// 会话语义的端口(A1/U-A1b):chat 入口折叠「动作可执行但槽位未就绪」的
// 挂起预期。nil = 按存储惰性装配(intentsession.New);组合根/测试可
// 注入替身,见 SetIntentSessionService。
sessionsMu sync.Mutex
sessions IntentSessions
// Subscribe 注入 bus 订阅(事件在线转发;重放走 store 游标)。
Subscribe func(conversationID string) (<-chan core.StreamEvent, func())
// DisableUploadIndexing EXT-9:true 时上传端点不再自动写检索索引(原件与
@@ -386,6 +392,8 @@ func (s *Server) Handler() http.Handler {
mux.HandleFunc("DELETE /api/prompt-commands/{id}", s.deleteUserPromptCommand)
mux.HandleFunc("POST /api/runs/{id}/cancel", s.cancelRun)
mux.HandleFunc("GET /api/runs/{id}", s.getRun)
// 只读运行诊断(A2):卡在哪、谁负责、已知副作用、下一步;无写操作。
mux.HandleFunc("GET /api/runs/{id}/diagnostics", s.getRunDiagnostics)
// ---- HITL ----
mux.HandleFunc("POST /api/hitl/{id}/resolve", s.resolveHITL)
@@ -1231,8 +1239,21 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
var workflowDef core.WorkflowDef
var admission *intentAdmission
var clarify *clarifyPayload
var sessionNote *intentSessionNote
var cancelTarget *route.CancelTarget
var cancelRunID string
if s.Gate != nil && !strings.HasPrefix(text, "/") && req.TaskRef == nil {
recognized = s.recognize(r.Context(), text)
// 会话语义(A1/U-A1b):有挂起预期时先按它折叠本轮——补齐即按与
// 一次完整识别完全相同的准入与派发路径执行,仍缺则续挂并如实重新
// 澄清,取消则清理会话且不派发。无挂起预期时本段空转(行为不变)。
sessionKey := intentSessionKey(owner, conv)
fused, sessionOutcome, ferr := s.fuseIntentSession(sessionKey, recognized)
if ferr != nil {
writeStoreErr(w, ferr)
return
}
recognized = fused
dec := route.Decide(recognized, route.Env{
HasCommand: func(id string) bool { return s.Bus != nil && s.Bus.HasCommand(id) },
WorkflowFor: func(intentID string) (core.WorkflowDef, bool) {
@@ -1242,8 +1263,22 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
return s.Workflows.ByIntent(intentID)
},
// SquadReady 保持 false:per-run entry 覆盖未就位(见 route 包注释的已知边界)。
// SessionRun:取消靶子的会话来源(意图会话绑定的运行);没有绑定
// 时返回 false,路由层随即把「取消没有唯一靶子」交给宿主澄清。
SessionRun: func() (route.SessionRun, bool) { return s.sessionRunTarget(sessionKey) },
})
switch dec.Kind {
case route.KindCancel:
// 取消是既有动作(带明确对象):靶子唯一且对发起人可见才派发,
// 否则如实澄清(绝不猜一个)。本轮不启动新回合——取消与启动互斥。
// 会话层已经消费掉本轮 cancel(放弃补槽)时不再二次规划。
if sessionOutcome == nil || !sessionOutcome.cancelled {
runID, need := s.planIntentCancel(owner, conv, recognized, dec)
clarify = need
if need == nil {
routeKind, cancelTarget, cancelRunID = route.KindCancel, dec.Cancel, runID
}
}
case route.KindCommand:
// 接纳:按命令 DECLARED schema 把槽位映射成参数;准入未过
// (必填参数无法由原文+槽位满足)就不派发,降为澄清回合(F09)。
@@ -1263,6 +1298,22 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
default:
clarify = newClarifyPayload(recognized, dec, nil)
}
// 动作可执行但槽位未就绪:把这一轮记成会话里的挂起预期,下一轮
// 补齐即可按同一准入与派发路径执行(响应面不受影响)。
if err := s.suspendIntentSession(sessionKey, recognized, dec, admission); err != nil {
writeStoreErr(w, err)
return
}
if sessionOutcome != nil {
switch {
case sessionOutcome.clarify != nil:
// 仍缺槽位:如实按会话的累计状态重新澄清。
clarify = sessionOutcome.clarify
case sessionOutcome.cancelled:
// 取消挂起的补充:如实确认,不派发任何业务动作。
sessionNote = sessionOutcome.note
}
}
}
claimed := false
if routeKind == route.KindTurn || routeKind == route.KindWorkflow {
@@ -1294,6 +1345,12 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
writeStoreErr(w, err)
return
}
if routeKind == route.KindCancel {
// 取消意图落到既有取消路径(与 POST /api/runs/{id}/cancel 同款):
// 回执、确认卡与会话记录都在响应里如实给出;不启动新回合。
s.handleIntentCancel(w, r, conv, msg, cancelTarget, cancelRunID)
return
}
if routeKind == route.KindCommand {
// 参数按命令 DECLARED schema 映射(槽位→声明键/声明类型);原文恒供。
params := map[string]any{chatTextParam: text}
@@ -1344,6 +1401,11 @@ func (s *Server) postMessage(w http.ResponseWriter, r *http.Request) {
// 澄清诉求(业务动作未派发):既有端点上的附加字段,缺失/非法各带结构化原因。
out["clarify"] = clarify
}
if sessionNote != nil {
// 会话语义的如实回话(取消挂起的补充):同样不新增 API 面,且从不
// 声称已派发任何业务动作。
out["intent_session"] = sessionNote
}
writeJSON(w, http.StatusOK, out)
}
@@ -1516,8 +1578,7 @@ func (s *Server) cancelRun(w http.ResponseWriter, r *http.Request) {
// conversation_id 必须回填真实值:消费侧按执行权 JOIN 匹配,空串永远
// 无法送达(同批实测)。
if s.st.Dialect() == "postgres" && !s.mgr.RunActiveLocally(runID) {
if cerr := s.st.EnqueueCommandAccountChecked(s.owner(r), store.NewID("cmd"), "cancel", runID, 0,
runmgr.CancelCommandPayload{Owner: s.owner(r), RunID: runID}, s.Auth == nil || !s.Auth.Enabled); cerr != nil {
if cerr := s.enqueueCancelCommand(s.owner(r), runID); cerr != nil {
writeStoreErr(w, cerr)
return
}
+166
View File
@@ -0,0 +1,166 @@
package store
// 意图会话(A1/U-A1b):会话维度挂起一条「可执行动作预期」——动作引用、
// 仍缺的槽位、已收集的槽位、单调 revision 与到期时间。宿主
// (internal/intentsession)在澄清回合后写入它,下一轮把用户的补充折叠
// 回去;槽位齐全时宿主按同一条准入与派发路径执行(授权与高危确认始终
// 由 cmdbus 收口——本表不携带任何权限字段)。
//
// 键是 (owner, project_id, conversation_id):两个用户、两个项目的会话在
// 构造上互不可见,读写一律带 owner 过滤。行是运行态(默认 30 分钟后
// 过期、读侧惰性删除),不进 storage-migrate 的 copyOrder(同
// admission_leases:进程/会话瞬态,迁移没有意义)。
import (
"database/sql"
"errors"
"fmt"
"strings"
"time"
)
// IntentSlotQuestion 是一个仍缺的必填槽位及其建议提问。
type IntentSlotQuestion struct {
Key string `json:"key"`
Question string `json:"question,omitempty"`
}
// IntentExpectation 是挂起的可执行动作预期:等后续轮次补齐槽位。
// ActionKind/ActionID 是权威动作引用(command/workflow/blueprint/auto)。
type IntentExpectation struct {
IntentID string `json:"intent_id,omitempty"`
ActionKind string `json:"action_kind,omitempty"`
ActionID string `json:"action_id,omitempty"`
Missing []IntentSlotQuestion `json:"missing,omitempty"`
Question string `json:"question,omitempty"`
}
// IntentSession 是一个会话当前的意图会话状态。Expectation=nil 表示当前
// 没有挂起动作(例如用户已取消或动作已补齐),Slots 保留已收集值,
// Revision 单调递增用于挡住陈旧回答。
type IntentSession struct {
Owner string
ProjectID string
ConversationID string
Expectation *IntentExpectation
Slots map[string]string
Revision int
ExpiresAt time.Time
CreatedAt time.Time
UpdatedAt time.Time
}
// GetIntentSession 按 owner+project+conversation 读回意图会话。
// 没有行(别的用户/项目/会话、或尚未挂起)一律返回 ErrNotFound:
// 会话状态不跨用户可见,调用方不得据此猜测存在性。
func (s *Store) GetIntentSession(owner, projectID, conversationID string) (IntentSession, error) {
if owner == "" || conversationID == "" {
return IntentSession{}, ErrNotFound
}
row := s.db.QueryRow(s.q(`SELECT expectation,slots,revision,expires_at,created_at,updated_at
FROM intent_session WHERE owner=? AND project_id=? AND conversation_id=?`),
owner, projectID, conversationID)
return scanIntentSession(owner, projectID, conversationID, row)
}
// PutIntentSession 挂起一个动作预期并累计槽位(upsert,revision 每次 +1、
// 到期时间每次刷新)。expectation=nil 表示清空挂起动作(保留已收集槽位)。
// 返回写入后的会话(含新的 revision),调用方据此推进自己的视图。
func (s *Store) PutIntentSession(owner, projectID, conversationID string,
expectation *IntentExpectation, slots map[string]string, ttl time.Duration) (IntentSession, error) {
if owner == "" || conversationID == "" {
return IntentSession{}, errors.New("store: intent session key incomplete")
}
var rawExpectation any
if expectation != nil {
rawExpectation = marshalJSON(expectation)
}
if slots == nil {
slots = map[string]string{}
}
now := time.Now().UTC()
expiresAt := ""
if ttl > 0 {
expiresAt = now.Add(ttl).Format(time.RFC3339Nano)
}
if _, err := s.db.Exec(s.q(`INSERT INTO intent_session
(owner,project_id,conversation_id,expectation,slots,revision,expires_at,created_at,updated_at)
VALUES(?,?,?,?,?,1,?,?,?)
ON CONFLICT(owner,project_id,conversation_id) DO UPDATE SET
expectation=excluded.expectation, slots=excluded.slots,
revision=intent_session.revision+1, expires_at=excluded.expires_at,
updated_at=excluded.updated_at`),
owner, projectID, conversationID, rawExpectation, marshalJSON(slots),
expiresAt, now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano)); err != nil {
return IntentSession{}, fmt.Errorf("put intent session: %w", err)
}
return s.GetIntentSession(owner, projectID, conversationID)
}
// ClearIntentSession 删除整行会话状态(幂等;动作已派发或会话结束时用)。
func (s *Store) ClearIntentSession(owner, projectID, conversationID string) error {
if owner == "" || conversationID == "" {
return nil
}
if _, err := s.db.Exec(s.q(`DELETE FROM intent_session WHERE owner=? AND project_id=? AND conversation_id=?`),
owner, projectID, conversationID); err != nil {
return fmt.Errorf("clear intent session: %w", err)
}
return nil
}
// ClearIntentSessionExpectation 只清空挂起的动作预期(用户取消):行保留
// 已收集槽位,revision +1 让任何在途的陈旧回答都作废。无行时返回
// ErrNotFound。
func (s *Store) ClearIntentSessionExpectation(owner, projectID, conversationID string) (IntentSession, error) {
if owner == "" || conversationID == "" {
return IntentSession{}, ErrNotFound
}
res, err := s.db.Exec(s.q(`UPDATE intent_session SET expectation=NULL, revision=revision+1, updated_at=?
WHERE owner=? AND project_id=? AND conversation_id=?`), nowISO(), owner, projectID, conversationID)
if err != nil {
return IntentSession{}, fmt.Errorf("clear intent session expectation: %w", err)
}
if n, _ := res.RowsAffected(); n == 0 {
return IntentSession{}, ErrNotFound
}
return s.GetIntentSession(owner, projectID, conversationID)
}
// scanIntentSession 组装一行会话状态;缺行一律 ErrNotFound。
func scanIntentSession(owner, projectID, conversationID string, row *sql.Row) (IntentSession, error) {
var rawExpectation sql.NullString
var rawSlots string
var revision int
var expiresAt, createdAt, updatedAt string
if err := row.Scan(&rawExpectation, &rawSlots, &revision, &expiresAt, &createdAt, &updatedAt); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return IntentSession{}, ErrNotFound
}
return IntentSession{}, err
}
out := IntentSession{
Owner: owner, ProjectID: projectID, ConversationID: conversationID,
Slots: map[string]string{}, Revision: revision,
}
if rawExpectation.Valid && strings.TrimSpace(rawExpectation.String) != "" {
exp, err := unmarshalJSON[IntentExpectation](rawExpectation.String)
if err != nil {
return IntentSession{}, fmt.Errorf("intent session expectation: %w", err)
}
out.Expectation = &exp
}
if strings.TrimSpace(rawSlots) != "" {
slots, err := unmarshalJSON[map[string]string](rawSlots)
if err != nil {
return IntentSession{}, fmt.Errorf("intent session slots: %w", err)
}
if slots != nil {
out.Slots = slots
}
}
out.ExpiresAt, _ = parseTime(expiresAt)
out.CreatedAt, _ = parseTime(createdAt)
out.UpdatedAt, _ = parseTime(updatedAt)
return out, nil
}
+186
View File
@@ -0,0 +1,186 @@
package store
import (
"errors"
"testing"
"time"
)
// 意图会话表(v52 / A1-U-A1b):键含 owner+project+conversation、revision
// 单调递增、到期时间随更新刷新;旧库(v51)是纯增量升级。
func cancelExpectation() *IntentExpectation {
return &IntentExpectation{
IntentID: "demo.run-cancel", ActionKind: "command", ActionID: "demo.run-cancel",
Missing: []IntentSlotQuestion{{Key: "run_id", Question: "取消哪个运行?"}},
Question: "取消哪个运行?",
}
}
func TestIntentSessionUpsertIncrementsRevisionAndRefreshesExpiry(t *testing.T) {
st := openTestStore(t)
exp := cancelExpectation()
first, err := st.PutIntentSession("alice", "prj_1", "cnv_1", exp, map[string]string{"run_id": "R-1"}, time.Minute)
if err != nil {
t.Fatal(err)
}
if first.Revision != 1 || first.Expectation == nil || first.Expectation.ActionID != "demo.run-cancel" {
t.Fatalf("首次挂起:%+v", first)
}
if first.Slots["run_id"] != "R-1" || len(first.Expectation.Missing) != 1 {
t.Fatalf("槽位/缺失未落库:%+v", first)
}
if !first.ExpiresAt.After(time.Now()) {
t.Fatalf("到期时间未设置:%+v", first)
}
// 同一会话再挂一次:revision +1、槽位累计、到期时间刷新。
second, err := st.PutIntentSession("alice", "prj_1", "cnv_1", exp,
map[string]string{"run_id": "R-1", "reason": "误触发"}, 2*time.Minute)
if err != nil {
t.Fatal(err)
}
if second.Revision != 2 {
t.Fatalf("revision 未递增:%+v", second)
}
if second.Slots["reason"] != "误触发" || second.CreatedAt != first.CreatedAt {
t.Fatalf("槽位累计/创建时间被改写:%+v", second)
}
if !second.ExpiresAt.After(first.ExpiresAt) {
t.Fatalf("到期时间未刷新:%v -> %v", first.ExpiresAt, second.ExpiresAt)
}
got, err := st.GetIntentSession("alice", "prj_1", "cnv_1")
if err != nil || got.Revision != 2 || got.Expectation.Question != "取消哪个运行?" {
t.Fatalf("读回:%+v %v", got, err)
}
}
func TestIntentSessionClearKeepsNothingAndClearExpectationKeepsSlots(t *testing.T) {
st := openTestStore(t)
exp := cancelExpectation()
if _, err := st.PutIntentSession("alice", "prj_1", "cnv_1", exp, map[string]string{"run_id": "R-1"}, time.Minute); err != nil {
t.Fatal(err)
}
// 取消:清空预期、保留已收集槽位、revision +1(陈旧回答就此作废)。
cleared, err := st.ClearIntentSessionExpectation("alice", "prj_1", "cnv_1")
if err != nil {
t.Fatal(err)
}
if cleared.Expectation != nil || cleared.Revision != 2 || cleared.Slots["run_id"] != "R-1" {
t.Fatalf("取消后的状态:%+v", cleared)
}
// 动作补齐/会话结束:整行清掉(幂等)。
if err := st.ClearIntentSession("alice", "prj_1", "cnv_1"); err != nil {
t.Fatal(err)
}
if err := st.ClearIntentSession("alice", "prj_1", "cnv_1"); err != nil {
t.Fatalf("重复清理必须幂等:%v", err)
}
if _, err := st.GetIntentSession("alice", "prj_1", "cnv_1"); !errors.Is(err, ErrNotFound) {
t.Fatalf("清理后仍可读:%v", err)
}
if _, err := st.ClearIntentSessionExpectation("alice", "prj_1", "cnv_1"); !errors.Is(err, ErrNotFound) {
t.Fatalf("无行时清理预期必须报 not found:%v", err)
}
}
// 两个用户与两个项目在构造上互不可见:别人的挂起读不到、写不进。
func TestIntentSessionIsolatesOwnersAndProjects(t *testing.T) {
st := openTestStore(t)
exp := cancelExpectation()
if _, err := st.PutIntentSession("alice", "prj_1", "cnv_1", exp, map[string]string{"run_id": "R-alice"}, time.Minute); err != nil {
t.Fatal(err)
}
for _, probe := range []struct{ owner, project, conversation string }{
{"bob", "prj_1", "cnv_1"}, // 换用户
{"alice", "prj_2", "cnv_1"}, // 换项目
{"alice", "prj_1", "cnv_2"}, // 换会话
{"", "prj_1", "cnv_1"}, // 无归属
} {
if _, err := st.GetIntentSession(probe.owner, probe.project, probe.conversation); !errors.Is(err, ErrNotFound) {
t.Fatalf("%v 读到了别人的会话:%v", probe, err)
}
if err := st.ClearIntentSession(probe.owner, probe.project, probe.conversation); err != nil {
t.Fatalf("清理越权键:%v", err)
}
}
got, err := st.GetIntentSession("alice", "prj_1", "cnv_1")
if err != nil || got.Slots["run_id"] != "R-alice" {
t.Fatalf("越权清理影响了属主会话:%+v %v", got, err)
}
// 同 owner 换项目写入互不覆盖。
if _, err := st.PutIntentSession("alice", "prj_2", "cnv_1", exp, nil, time.Minute); err != nil {
t.Fatal(err)
}
if a, _ := st.GetIntentSession("alice", "prj_1", "cnv_1"); a.Slots["run_id"] != "R-alice" {
t.Fatalf("另一项目的写入串了会话:%+v", a)
}
}
// intentSessionTableExists 双档目录探测(SQLite sqlite_master / PG information_schema)。
func intentSessionTableExists(t *testing.T, st *Store) bool {
t.Helper()
query := `SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='intent_session'`
if st.Dialect() == "postgres" {
query = `SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=current_schema() AND table_name='intent_session'`
}
var n int
if err := st.db.QueryRow(query).Scan(&n); err != nil {
t.Fatal(err)
}
return n > 0
}
// 旧库(v51,无 intent_session 表)重开即升级:无数据迁移,既有行不受影响。
func TestIntentSessionUpgradesLegacyDatabase(t *testing.T) {
st := openTestStore(t)
if err := st.MigrateDown(51); err != nil {
t.Fatalf("回滚到 v51:%v", err)
}
if intentSessionTableExists(t, st) {
t.Fatal("v51 库不应有 intent_session 表")
}
if err := st.migrate(); err != nil {
t.Fatalf("升级 v51 -> v52:%v", err)
}
assertFullLedger(t, st)
if _, err := st.PutIntentSession("alice", "prj_1", "cnv_1", cancelExpectation(), map[string]string{"run_id": "R-1"}, time.Minute); err != nil {
t.Fatalf("升级后写入:%v", err)
}
got, err := st.GetIntentSession("alice", "prj_1", "cnv_1")
if err != nil || got.Slots["run_id"] != "R-1" || got.Revision != 1 {
t.Fatalf("升级后读回:%+v %v", got, err)
}
// 回滚可重复演练(down 语句齐备)。
if err := st.MigrateDown(51); err != nil {
t.Fatalf("回滚 v52:%v", err)
}
if err := st.migrate(); err != nil {
t.Fatalf("再次升级:%v", err)
}
assertFullLedger(t, st)
}
// 空值约定:没有挂起动作时 expectation 为 NULL,已收集槽位仍是合法 JSON。
func TestIntentSessionNullableExpectationAndEmptySlots(t *testing.T) {
st := openTestStore(t)
if _, err := st.PutIntentSession("alice", "", "cnv_1", nil, nil, 0); err != nil {
t.Fatal(err)
}
got, err := st.GetIntentSession("alice", "", "cnv_1")
if err != nil {
t.Fatal(err)
}
if got.Expectation != nil || got.Slots == nil || len(got.Slots) != 0 {
t.Fatalf("空预期/空槽位:%+v", got)
}
if !got.ExpiresAt.IsZero() {
t.Fatalf("ttl<=0 时不应有过期时间:%+v", got)
}
if _, err := st.PutIntentSession("", "prj_1", "cnv_1", nil, nil, time.Minute); err == nil {
t.Fatal("无 owner 的写入必须被拒")
}
}
+29
View File
@@ -1225,9 +1225,38 @@ func init() {
"postgres": {`DROP INDEX IF EXISTS idx_users_single_super_admin`, `ALTER TABLE users DROP COLUMN is_super_admin`},
},
},
migration{version: 52, name: "intent_session",
// 意图会话(A1/U-A1b):会话维度挂起一条可执行动作预期 + 已收集
// 槽位 + 单调 revision + 到期时间。纯新表,既有库是空操作(新表
// 增量升级,无需数据迁移);运行态行随到期惰性清理。
up: map[string][]string{"sqlite": intentSessionUp, "postgres": intentSessionUp},
down: map[string][]string{"sqlite": intentSessionDown, "postgres": intentSessionDown},
},
)
}
// v52 意图会话表:键含 owner+project+conversation,两个用户的会话在构造上
// 互不可见。expectation 可空(NULL = 没有挂起动作,只剩已收集槽位),
// slots/revision/expires_at 是折叠与陈旧回答防护的宿主状态。
var intentSessionUp = []string{
`CREATE TABLE IF NOT EXISTS intent_session (
owner TEXT NOT NULL,
project_id TEXT NOT NULL DEFAULT '',
conversation_id TEXT NOT NULL,
expectation TEXT,
slots TEXT NOT NULL DEFAULT '{}',
revision INTEGER NOT NULL DEFAULT 0,
expires_at TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY (owner, project_id, conversation_id)
)`,
}
var intentSessionDown = []string{
`DROP TABLE IF EXISTS intent_session`,
}
// v35 分层 Python 环境:扩展包申请与会话绑定。层目录本身在环境根,不入库。
var execEnvUp = []string{
`CREATE TABLE IF NOT EXISTS exec_env_requests (
+1
View File
@@ -58,6 +58,7 @@ var migrationBreaksOldReaders = map[int]bool{
49: false, // 角色权威仅新增独立表;消费者切换由公开提供者 readiness 单独围栏
50: false, // 角色导入 staging 为独立表;正式角色读取在激活前不变
51: true, // 超级管理员标记是账号权威边界;不认识它的旧账号写者必须排除
52: false, // intent_session 是新表;旧制品不读取会话内的意图状态
}
// SchemaTarget 是本二进制会迁移到的 schema 版本。
+1 -1
View File
@@ -2252,7 +2252,7 @@ func (s *Store) upsertRolePermissionsTx(tx *sql.Tx, actorID, roleID string, curr
for _, row := range writes {
if _, err := tx.Exec(s.q(`INSERT INTO role_permissions(role_id,action_id,allowed,cap_allowed,permission_version,updated_by,updated_at)
VALUES(?,?,?,?,?,?,?) ON CONFLICT(role_id,action_id) DO UPDATE SET allowed=excluded.allowed,cap_allowed=excluded.cap_allowed,
permission_version=permission_version+1,updated_by=excluded.updated_by,updated_at=excluded.updated_at`),
permission_version=role_permissions.permission_version+1,updated_by=excluded.updated_by,updated_at=excluded.updated_at`),
roleID, row.ActionID, row.Allowed, row.CapAllowed, 1, actorID, now); err != nil {
return err
}
+110
View File
@@ -0,0 +1,110 @@
package store
import (
"database/sql"
"errors"
"time"
)
// 运行诊断的只读投影(feature unit A2)。本文件只放 SELECT:不加表、不加索引、
// 不动迁移,也不引入新的写语义。诊断面回答的是"已经落库的证据是什么",因此
// 每个查询都遵守同一条口径:查不到就是"没有记录",调用方不得把它读成"没发生"
// (与 /api/commands/operations 的对账查询同源,见 cmdbus.OperationQuery)。
// RunCheckpoint 是 run 事件流上最后一个已持久化事件的稳定引用。
//
// 诚实边界:runs 行本身没有"检查点"列——事件流(events)才是 run 唯一的持久
// 进度痕迹(续跑、SSE 重放都以它为准)。所以这里给的是"最后一次稳定落盘"的
// 定位点(seq 可作游标),不是推断出来的中间状态;run 没有任何事件时不存在这条
// 记录,调用方必须显式处理"没有"而不是编一个。
type RunCheckpoint struct {
Seq int64 // 会话事件序(可作重放游标)
TS string // RFC3339(写入方原样落库)
Type string // 事件类型
}
// LastRunCheckpoint 返回该 run 在会话事件流里的最后一条已持久化事件。
// 没有事件时 ok=false,err 为 nil。
func (s *Store) LastRunCheckpoint(conversationID, runID string) (RunCheckpoint, bool, error) {
if conversationID == "" || runID == "" {
return RunCheckpoint{}, false, nil
}
var cp RunCheckpoint
err := s.db.QueryRow(s.q(`SELECT seq, ts, type FROM events
WHERE conversation_id=? AND run_id=? ORDER BY seq DESC LIMIT 1`), conversationID, runID).
Scan(&cp.Seq, &cp.TS, &cp.Type)
if errors.Is(err, sql.ErrNoRows) {
return RunCheckpoint{}, false, nil
}
if err != nil {
return RunCheckpoint{}, false, err
}
return cp, true, nil
}
// PendingHumanDecisionsByConversation 列出会话内未过期、尚未裁决的人工决策卡
// (owner + 会话双谓词,跨租户天然不可见;过期卡按 PendingHumanDecisions 的同一
// 口径过滤)。诊断面据此判断 run 是否在等人工裁决。
//
// 口径说明:决策卡按命令与会话归属,表里没有 run 列——把它归到某个 run 是调用方
// 的时间窗推断,不是存储事实(见 rundiag 的归属边界注释)。
func (s *Store) PendingHumanDecisionsByConversation(owner, conversationID string, now time.Time) ([]HumanDecision, error) {
if owner == "" || conversationID == "" {
return []HumanDecision{}, nil
}
rows, err := s.db.Query(s.q(`SELECT id,plugin_id,audience,command_id,human_actor,owner,project_id,proposal_id,
proposal_revision,payload_hash,question,status,decision,proof_id,conversation_id,issued_at,expires_at,responded_at
FROM human_decisions WHERE owner=? AND conversation_id=? AND status='pending' ORDER BY issued_at`),
owner, conversationID)
if err != nil {
return nil, err
}
defer rows.Close()
out := []HumanDecision{}
for rows.Next() {
rec, err := scanHumanDecision(rows)
if err != nil {
return nil, err
}
if now.Before(rec.ExpiresAt) {
out = append(out, rec)
}
}
return out, rows.Err()
}
// PendingConfirm 是危险命令确认挂起态(pending_confirms)的只读行。
// Payload 是 cmdbus 落下的分发请求 JSON(确认后据此二次分发);诊断面只解析出
// 归属与命令标识,不回显原始载荷。
type PendingConfirm struct {
CardID string
Owner string
Payload string
CreatedAt string
}
// ListOwnerPendingConfirms 列出该租户名下尚未确认的危险命令挂起行(时间正序)。
// owner 为空视为无作用域:不返回任何行(纵深防御,认证层恒注入非空 owner)。
func (s *Store) ListOwnerPendingConfirms(owner string, limit int) ([]PendingConfirm, error) {
if owner == "" {
return []PendingConfirm{}, nil
}
if limit <= 0 || limit > 200 {
limit = 50
}
rows, err := s.db.Query(s.q(`SELECT card_id, owner, payload, created_at FROM pending_confirms
WHERE owner=? ORDER BY created_at LIMIT ?`), owner, limit)
if err != nil {
return nil, err
}
defer rows.Close()
out := []PendingConfirm{}
for rows.Next() {
var p PendingConfirm
if err := rows.Scan(&p.CardID, &p.Owner, &p.Payload, &p.CreatedAt); err != nil {
return nil, err
}
out = append(out, p)
}
return out, rows.Err()
}
+120
View File
@@ -0,0 +1,120 @@
package store
import (
"encoding/json"
"testing"
"time"
core "agent-system-core"
)
// 运行诊断只读投影:检查点定位、HITL 决策卡的会话+租户作用域、确认挂起行的租户
// 作用域。口径与既有读入口一致:查不到就是"没有记录",不推断"没发生"。
func TestRunDiagnosticsReadHelpers(t *testing.T) {
st := openTestStore(t)
base := time.Now().UTC().Truncate(time.Second)
if err := st.EnsureUser("alice", ""); err != nil {
t.Fatal(err)
}
if err := st.EnsureUser("bob", ""); err != nil {
t.Fatal(err)
}
project, err := st.CreateProject("alice", "诊断项目")
if err != nil {
t.Fatal(err)
}
conversation, err := st.CreateConversation("alice", project.ID, "诊断会话", "")
if err != nil {
t.Fatal(err)
}
run := core.Run{ConversationID: conversation.ID, ProjectID: project.ID, Owner: "alice",
Kind: core.RunConversation, Status: core.RunRunning, CreatedAt: base}
if err := st.CreateRun(&run); err != nil {
t.Fatal(err)
}
other := core.Run{ConversationID: conversation.ID, ProjectID: project.ID, Owner: "alice",
Kind: core.RunSubagent, Status: core.RunRunning, CreatedAt: base}
if err := st.CreateRun(&other); err != nil {
t.Fatal(err)
}
// 检查点:只认本 run 的最后一条事件,且不跨 run 串味。
if cp, ok, err := st.LastRunCheckpoint(string(conversation.ID), string(run.ID)); err != nil || ok {
t.Fatalf("a run without events has no checkpoint: %+v ok=%v err=%v", cp, ok, err)
}
if _, err := st.AppendEvents(string(conversation.ID), []core.StreamEvent{
{RunID: run.ID, Type: core.EvThinkingDelta, TS: base.Format(time.RFC3339), Payload: json.RawMessage(`{}`)},
{RunID: other.ID, Type: core.EvToolOutput, TS: base.Format(time.RFC3339), Payload: json.RawMessage(`{}`)},
{RunID: run.ID, Type: core.EvAgentProgress, TS: base.Format(time.RFC3339), Payload: json.RawMessage(`{}`)},
}); err != nil {
t.Fatal(err)
}
cp, ok, err := st.LastRunCheckpoint(string(conversation.ID), string(run.ID))
if err != nil || !ok {
t.Fatalf("checkpoint read: ok=%v err=%v", ok, err)
}
if cp.Type != string(core.EvAgentProgress) || cp.TS == "" || cp.Seq <= 0 {
t.Fatalf("checkpoint must point at the newest persisted event of this run: %+v", cp)
}
if child, ok, err := st.LastRunCheckpoint(string(conversation.ID), string(other.ID)); err != nil || !ok || child.Type != string(core.EvToolOutput) {
t.Fatalf("child run checkpoint: %+v ok=%v err=%v", child, ok, err)
}
if cp, ok, err := st.LastRunCheckpoint("cnv_other", string(run.ID)); err != nil || ok {
t.Fatalf("another conversation must not expose this run's events: %+v ok=%v err=%v", cp, ok, err)
}
// HITL 决策卡:pending + 未过期才可见,且按 owner+会话双谓词隔离。
now := base.Add(time.Minute)
issue := func(id, owner, conversationID string, ttl time.Duration) {
t.Helper()
if _, err := st.IssueHumanDecision(HumanDecision{PluginID: "p", Audience: "a", CommandID: "rdm.confirm",
HumanActor: owner, Owner: owner, ProjectID: string(project.ID), ProposalID: id, ProposalRevision: 1,
PayloadHash: "h" + id, Question: "approve " + id + "?", ConversationID: conversationID},
now.Add(-time.Minute)); err != nil {
t.Fatal(err)
}
// 过期口径:把 expires_at 改到过去,模拟签发即过期的卡(走方言绑定)。
if ttl < 0 {
if _, err := st.DB().Exec(st.q(`UPDATE human_decisions SET expires_at=? WHERE proposal_id=?`),
now.Add(-time.Hour).Format(time.RFC3339Nano), id); err != nil {
t.Fatal(err)
}
}
}
issue("alive", "alice", string(conversation.ID), time.Hour)
issue("expired", "alice", string(conversation.ID), -time.Hour)
issue("otherowner", "bob", "cnv_bob", time.Hour)
pending, err := st.PendingHumanDecisionsByConversation("alice", string(conversation.ID), now)
if err != nil {
t.Fatal(err)
}
if len(pending) != 1 || pending[0].ProposalID != "alive" || pending[0].Status != "pending" {
t.Fatalf("only live pending cards in scope: %+v", pending)
}
if cross, err := st.PendingHumanDecisionsByConversation("bob", string(conversation.ID), now); err != nil || len(cross) != 0 {
t.Fatalf("cross-owner cards must stay invisible: %+v err=%v", cross, err)
}
if none, err := st.PendingHumanDecisionsByConversation("", string(conversation.ID), now); err != nil || len(none) != 0 {
t.Fatalf("an empty owner has no scope: %+v err=%v", none, err)
}
// 确认挂起行:按租户隔离,空 owner 无作用域。
if err := st.SavePendingConfirm("card_alice", "alice", `{"ID":"rdm.confirm"}`); err != nil {
t.Fatal(err)
}
if err := st.SavePendingConfirm("card_bob", "bob", `{"ID":"rdm.confirm"}`); err != nil {
t.Fatal(err)
}
confirms, err := st.ListOwnerPendingConfirms("alice", 0)
if err != nil {
t.Fatal(err)
}
if len(confirms) != 1 || confirms[0].CardID != "card_alice" || confirms[0].Owner != "alice" || confirms[0].Payload == "" || confirms[0].CreatedAt == "" {
t.Fatalf("owner-scoped pending confirms: %+v", confirms)
}
if none, err := st.ListOwnerPendingConfirms("", 0); err != nil || len(none) != 0 {
t.Fatalf("an empty owner has no scope: %+v err=%v", none, err)
}
}