Compare commits

...
Author SHA1 Message Date
weichen 7776399fe7 Reapply the A1 review m1-m4 fixes lost in a stale apply
Why: the m1-m4 changes (intent_session comment sync, cancel reason,
expiry fail-closed, slot ownership) were applied to a backend state
that later got reset, losing them from the branch; the reviewer
correctly flagged their absence.
Evidence: repair receipt (a1m-run.json) reusing the preserved
transport diff; per-package tests below.
Impact: the four review-minor behaviors restored on the current tip.
Refs: feature/audit-review-minors-cleanup
2026-10-23 12:28:01 +00:00
agent-exec-transport 67102901d6 agent-exec transport 7a27ae40d354fd19ec3b760cf1a7ce9a/d2827ccb55ff 2026-10-23 11:51:03 +00:00
weichen bd5c0f39e6 Pin nested intent, native and squad to the mis-execution fix tips
Why: composition consistency - Dev pins the three containers at their
negation-aware tips; the backend container's nested copies must match
(check-pins).
Evidence: pin only.
Impact: nested gitlink pointer changes.
Refs: feature/recall-misexec-fixes
2026-10-23 04:05:44 +00:00
weichen 8dc91c1783 Pin nested intent, native and squad to the A3 tips
Why: composition consistency - Dev pins the three containers at their
A3 evaluation/repair tips; the backend container's nested copies must
match (check-pins).
Evidence: pin only.
Impact: nested gitlink pointer changes.
Refs: feature/catalog-recall-eval-a3
2026-10-23 01:39:59 +00:00
14 changed files with 481 additions and 23 deletions
+46 -6
View File
@@ -218,6 +218,12 @@ func (s *Service) ApplyAnswer(key Key, revision int, answer Answer) (*Outcome, e
if err != nil || cur == nil {
return nil, err
}
return s.applyAnswer(key, cur, answer)
}
// applyAnswer 把一轮回答折叠进已读出的会话(读一次判一次:同一条命令里
// 不重复读库,判定口径与读到的状态一致)。
func (s *Service) applyAnswer(key Key, cur *Session, answer Answer) (*Outcome, error) {
if cur.Expectation == nil || cur.Cancelled != nil {
return nil, nil
}
@@ -247,16 +253,33 @@ func (s *Service) ApplyAnswer(key Key, revision int, answer Answer) (*Outcome, e
}
// ApplyCommand 消费库层会话命令:
// - set_slot / change_slot 写槽位(动作声明可解析时按声明校验,非法值不采纳);
// - 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) == "" {
name = strings.TrimSpace(name)
if name == "" {
return nil, nil
}
return s.ApplyAnswer(key, revision, Answer{Slots: map[string]string{name: value}})
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 !cur.ownsSlot(name) {
// 显式落槽命令只许落在当前预期认得的槽位键上:别的动作的键绝不
// 注入当前会话(如给「取消运行」的会话塞「清理对象」的键)。如
// 实忽略(返回 nil=本轮没有任何会话语义),响应面继续按仍缺的
// 槽位澄清,绝不假装消费过这条命令。
return nil, nil
}
return s.applyAnswer(key, cur, Answer{Slots: map[string]string{name: value}})
case string(intentgate.CommandCancel):
cur, err := s.current(key, revision)
if err != nil || cur == nil {
@@ -362,13 +385,14 @@ func (s *Service) load(key Key) (*Session, error) {
}
// live 报告会话是否仍在有效期内。没有挂起预期的行(已取消/已补齐)对宿主
// 等于「没有会话」。
// 等于「没有会话」;到期时间缺失或不可解析(行被写坏)同样 fail-closed:
// 当作已过期,绝不把损坏的到期时间当成「永不过期」。
func (s *Service) live(cur store.IntentSession) bool {
if cur.Expectation == nil {
return false
}
if cur.ExpiresAt.IsZero() {
return true
if !cur.ExpiresValid || cur.ExpiresAt.IsZero() {
return false
}
return s.now().Before(cur.ExpiresAt)
}
@@ -446,6 +470,22 @@ func fromStore(key Key, cur store.IntentSession) *Session {
return out
}
// ownsSlot 报告这个槽位键是否属于当前预期认得的槽位:仍缺的(Missing)
// 与已收集的(Slots)都是当前动作的键;别的动作的键一律不认(显式落槽
// 命令不得把它们注入当前会话)。
func (s *Session) ownsSlot(key string) bool {
if s == nil || s.Expectation == nil {
return false
}
for _, q := range s.Expectation.Missing {
if q.Key == key {
return true
}
}
_, collected := s.Slots[key]
return collected
}
// runIDOf 从槽位表里取运行号(固定键序,结果与 map 迭代序无关;无则空)。
func runIDOf(slots map[string]string) string {
for _, key := range runSlotKeys {
@@ -142,6 +142,60 @@ func TestSetSlotThenAnswerCompletes(t *testing.T) {
}
}
// 显式 set_slot/change_slot 只许落在当前预期认得的槽位上:别的动作的键
// 一律忽略(不落槽、不推进 revision);认得的键照旧生效。
func TestSetSlotOnlyAcceptsKeysOfCurrentExpectation(t *testing.T) {
svc := newService(t)
key := cancelKey()
// 当前预期是「取消运行」:它认得的键只有 run_id 与 reason。
suspended, err := svc.Suspend(key, cancelExpectation(), nil)
if err != nil {
t.Fatal(err)
}
// 别的动作(demo.purge)的键:忽略——绝不注入当前会话。
out, err := svc.ApplyCommand(key, suspended.Revision, string(intentgate.CommandSetSlot), "target", "T-1")
if err != nil || out != nil {
t.Fatalf("不属于当前预期的键必须被忽略:%+v %v", out, err)
}
cur, err := svc.Pending(key)
if err != nil || cur == nil || cur.Revision != suspended.Revision || cur.Slots["target"] != "" {
t.Fatalf("被忽略的命令改动了会话:%+v %v", cur, err)
}
// 认得的键(仍缺的):照旧落槽。
ok, err := svc.ApplyCommand(key, suspended.Revision, string(intentgate.CommandSetSlot), "run_id", "R-7")
if err != nil || ok == nil || ok.Session == nil {
t.Fatalf("认得的键必须照旧落槽:%+v %v", ok, err)
}
if ok.Session.Slots["run_id"] != "R-7" || ok.Session.Revision != suspended.Revision+1 {
t.Fatalf("落槽结果:%+v", ok.Session)
}
// 已收集的键也认(否则「改一下编号」永远改不动已抽到的槽位)。
changed, err := svc.ApplyCommand(key, ok.Session.Revision, string(intentgate.CommandChangeSlot), "run_id", "R-8")
if err != nil || changed == nil || changed.Session == nil || changed.Session.Slots["run_id"] != "R-8" {
t.Fatalf("change_slot 未改写已收集的键:%+v %v", changed, err)
}
// 补上最后一个缺键:照旧补齐(消费会话)。
filled, err := svc.ApplyCommand(key, changed.Session.Revision, string(intentgate.CommandSetSlot), "reason", "误触发")
if err != nil || filled == nil || !filled.Complete {
t.Fatalf("补齐最后一个键应完成:%+v %v", filled, err)
}
// 反过来:另一个动作的预期认不得上一个动作的键。
purge, 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 out, err := svc.ApplyCommand(key, purge.Revision, string(intentgate.CommandSetSlot), "run_id", "R-9"); err != nil || out != nil {
t.Fatalf("上个动作的键必须被忽略:%+v %v", out, err)
}
after, err := svc.Pending(key)
if err != nil || after == nil || after.Revision != purge.Revision || after.Slots["run_id"] != "" {
t.Fatalf("被忽略的命令改动了会话:%+v %v", after, err)
}
}
// 取消清空挂起预期:确认取消、不假装成功,陈旧回答也作废。
func TestCancelClearsPendingExpectation(t *testing.T) {
svc := newService(t)
@@ -221,6 +275,35 @@ func TestExpiredSessionNeverCompletesOldAction(t *testing.T) {
}
}
// 库里的到期时间被写坏(不可解析):fail-closed 当作已过期——会话对宿主
// 不存在,行被惰性删除;绝不把损坏的到期时间当成「永不过期」。
func TestCorruptedExpiryIsTreatedAsExpired(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)
}
// 直接改坏库里的 expires_at(模拟损坏的行)。
if _, err := svc.st.DB().Exec(`UPDATE intent_session SET expires_at = ? WHERE owner = ? AND conversation_id = ?`,
"not-a-timestamp", key.Owner, key.Conversation); err != nil {
t.Fatal(err)
}
// 损坏的到期时间=已过期:会话对宿主不存在。
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{"reason": "误触发"}}); err != nil || out != nil {
t.Fatalf("过期回答不得补齐旧动作:%+v %v", out, err)
}
// 行已被惰性删除。
if _, err := svc.st.GetIntentSession(key.Owner, key.Project, key.Conversation); !errors.Is(err, store.ErrNotFound) {
t.Fatalf("过期行未被惰性删除:%v", err)
}
}
// 两个用户、两个项目在构造上互不可见。
func TestSessionsAreIsolatedByOwnerAndProject(t *testing.T) {
svc := newService(t)
+67 -7
View File
@@ -7,6 +7,8 @@
// 读不到隔离记录时更按 fail-closed 处理(不声称"可以安全重试")。
// - 「没有记录」≠「没有发生」:side_effects 为空必须带 note,说明这是"未记录"
// 而不是"未发生"。
// - 有界读取也要说出来:命中读取上限(扫描/聚合上限)时报告显式带截断 note,
// 不让"读了前 N 条"冒充"全部证据"。
// - 归属边界:HITL 决策卡、危险命令确认卡、命令审计信封都按 owner + 会话归属,
// 其中决策卡与确认卡表里根本没有 run 列,审计信封虽有 runId 字段但宿主总线
// 当前不写。因此"归到这个 run"是调用方按签发时间窗做的有界推断,报告以
@@ -17,6 +19,7 @@ import (
"encoding/json"
"errors"
"fmt"
"strings"
"time"
agentcommand "agent-command"
@@ -190,6 +193,10 @@ func Gather(src Source, run core.Run, now time.Time) Evidence {
ev = ev.withNote("pending confirmations unreadable: " + err.Error())
} else {
ev.Confirms = confirms
if len(confirms) >= maxConfirms {
// 有界读到的行数正好等于上限 = 列表可能被截断;不说出来就会被读成"就这些"。
ev = ev.withNote(fmt.Sprintf("pending confirmation read reached the %d-row cap; the list may be truncated", maxConfirms))
}
}
cp, ok, err := src.LastRunCheckpoint(conversation, string(run.ID))
@@ -205,6 +212,11 @@ func Gather(src Source, run core.Run, now time.Time) Evidence {
ev = ev.withNote("command audit unreadable: " + err.Error())
} else {
ev.Audits = audits
if len(audits) >= maxAuditScan {
// 审计信封按时间倒序取最近 maxAuditScan 条;命中上限说明更早的回执没被读,
// 副作用证据可能不完整。
ev = ev.withNote(fmt.Sprintf("command audit scan reached the %d-envelope cap (newest first); older receipts were not read, so side effect evidence may be incomplete", maxAuditScan))
}
}
return ev
}
@@ -215,13 +227,32 @@ func Build(ev Evidence) Report {
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)
effects, effectsTruncated := sideEffects(ev)
rep.SideEffects = effects
rep.Notes = append([]string(nil), ev.Notes...)
if effectsTruncated {
rep.Notes = append(rep.Notes, fmt.Sprintf(
"side_effects truncated at the %d-entry cap; recorded effects beyond the cap are not represented in this report",
maxSideEffects))
}
card := waitingCard(ev)
decision := waitingDecision(ev)
confirm := waitingConfirm(ev)
if isTerminalRun(ev.Run.Status) {
// 终态 run 上仍未关闭的决策卡 / 人工决策 / 危险命令确认行是残留记录,不是
// 「现在在等人」。终态优先于证据驱动的等待映射:按终态报结论,并把残留记录
// 写进 note——否则值班会拿着一个已经结束、永远不会回跑的 run 去确认一张卡。
// run.status 原样保留(不隐藏),这里只改责任判定与措辞。
if stale := staleWaitRecords(card, decision, confirm); len(stale) > 0 {
rep.Notes = append(rep.Notes, "run reached terminal status "+string(ev.Run.Status)+
" while "+strings.Join(stale, ", ")+
" stayed open; the run is not waiting on them, treat the leftover record(s) as stale instead of confirming")
card, decision, confirm = nil, nil, nil
}
}
switch {
case ev.HoldUnknown:
// 隔离记录读不到:不猜没有 hold,直接 unknown,且不给"安全重试"。
@@ -338,8 +369,9 @@ func runSummary(ev Evidence) RunSummary {
}
// sideEffects 汇总已记录的外部副作用证据:命令回执对账记录(U7b 审计信封)加上
// 恢复隔离行(U3b 的"结果未知"证据)。
func sideEffects(ev Evidence) []SideEffect {
// 恢复隔离行(U3b 的"结果未知"证据)。第二个返回值是截断标志:命中 maxSideEffects
// 上限时为 true,调用方据此给报告加显式截断 note(有界读不冒充全量)。
func sideEffects(ev Evidence) ([]SideEffect, bool) {
out := []SideEffect{}
if ev.Hold != nil {
state := "held"
@@ -354,15 +386,16 @@ func sideEffects(ev Evidence) []SideEffect {
if !belongsToRun(ev, env) {
continue
}
if len(out) >= maxSideEffects {
// 上限已满且还有归属本次 run 的回执:后面的没进报告,如实标出截断。
return out, true
}
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
return out, false
}
// belongsToRun 是审计信封的归属边界:信封自带 runId 时以它为准;否则只在"同一
@@ -385,6 +418,33 @@ func openHold(ev Evidence) bool {
return ev.Hold != nil && ev.Hold.ReleasedAt == ""
}
// isTerminalRun 判定 run 是否已到终态(failed / completed / cancelled:runner 不会再
// 推进它)。终态优先于证据驱动的等待映射——终态 run 上残留的卡不再构成"在等人"。
func isTerminalRun(status core.RunStatus) bool {
switch status {
case core.RunFailed, core.RunCompleted, core.RunCancelled:
return true
default:
return false
}
}
// staleWaitRecords 列出终态 run 上仍未关闭的等待类记录,供 note 如实点名(不点名就等于
// 把残留记录藏起来)。它不改变任何结论,只决定报告里怎么称呼残留证据。
func staleWaitRecords(card *core.HITLCard, decision *store.HumanDecision, confirm *scopedConfirm) []string {
out := []string{}
if card != nil {
out = append(out, "hitl card "+string(card.ID))
}
if decision != nil {
out = append(out, "pending human decision "+decision.ID)
}
if confirm != nil {
out = append(out, "pending command confirmation "+confirm.row.CardID)
}
return out
}
// unknownOperations 统计结果未知的已记录 operation(四态语义里的 unknown:
// 执行过但结果不确定)。它是"重试会不会复制副作用"的直接证据。
func unknownOperations(effects []SideEffect) int {
+142
View File
@@ -3,6 +3,7 @@ package rundiag
import (
"encoding/json"
"errors"
"fmt"
"strings"
"testing"
"time"
@@ -107,6 +108,38 @@ func TestBuildResponsibilityStatesAndNextStep(t *testing.T) {
state: StateReconciling, role: RoleOperationOwner, action: ActionNeedsManualReconciliation,
waitHint: "still interrupted", since: afterRun.Add(time.Minute).Format(time.RFC3339),
},
{
name: "released hold reconciles even while a HITL card is still open",
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)},
Cards: []core.HITLCard{{ID: "card_hitl", CreatedAt: afterRun.Format(time.RFC3339), Question: "approve?"}},
},
state: StateReconciling, role: RoleOperationOwner, action: ActionNeedsManualReconciliation,
waitHint: "still interrupted", since: afterRun.Add(time.Minute).Format(time.RFC3339),
},
{
name: "a pending decision outranks a pending dangerous command confirmation",
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)}},
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: StateWaitingHITL, role: RoleHuman, action: ActionConfirmCard, cardID: "hd_1",
waitHint: "approve delivery?", since: afterRun.Format(time.RFC3339),
},
{
name: "a terminal failed run with a leftover card is failed, not a live wait",
ev: Evidence{
Run: ended(run(core.RunFailed), afterRun),
Cards: []core.HITLCard{{ID: "card_hitl", CreatedAt: afterRun.Format(time.RFC3339), Question: "approve?"}},
},
state: StateFailed, role: RoleRunner, action: ActionSafeRetry,
waitHint: "run failed", since: afterRun.Format(time.RFC3339),
},
{
name: "waiting on a HITL card",
ev: Evidence{
@@ -421,6 +454,115 @@ func TestBuildStatusDrivenWaitHasNoCardAndSaysSo(t *testing.T) {
}
}
// MI1:终态 run 上的残留卡不再驱动「等人确认」。终态映射优先,run.status 原样可见,
// 残留记录逐一点名——而不是把一个永远不会回跑的 run 的卡交给值班去确认。
func TestBuildTerminalRunSuppressesStaleWaitEvidence(t *testing.T) {
leftovers := func() Evidence {
return Evidence{
Run: ended(run(core.RunFailed), afterRun),
Cards: []core.HITLCard{{ID: "card_hitl", CreatedAt: afterRun.Format(time.RFC3339), Question: "approve?"}},
Decisions: []store.HumanDecision{{ID: "hd_1", CommandID: "rdm.confirm", Status: "pending",
Question: "approve delivery?", IssuedAt: afterRun, ExpiresAt: afterRun.Add(time.Hour)}},
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"})}},
}
}
rep := Build(leftovers())
if rep.Responsibility.State != StateFailed {
t.Fatalf("a terminal run must not read as waiting on a card: %+v", rep.Responsibility)
}
if rep.NextStep.Action == ActionConfirmCard || rep.NextStep.CardID != "" {
t.Fatalf("a terminal run must not advise confirming a leftover card: %+v", rep.NextStep)
}
if rep.Run.Status != string(core.RunFailed) {
t.Fatalf("run.status must stay visible: %q", rep.Run.Status)
}
for _, want := range []string{"stale", "hitl card card_hitl", "pending human decision hd_1", "pending command confirmation card_cmd"} {
if !containsNote(rep.Notes, want) {
t.Fatalf("terminal run with leftovers must note %q: %v", want, rep.Notes)
}
}
completed := leftovers()
completed.Run = ended(run(core.RunCompleted), afterRun)
rep = Build(completed)
if rep.Responsibility.State != StateCompleted || rep.NextStep.Action != ActionNone || rep.NextStep.CardID != "" {
t.Fatalf("completed run with leftovers: %+v", rep)
}
if !containsNote(rep.Notes, "stale") {
t.Fatalf("completed run with leftovers must name them: %v", rep.Notes)
}
cancelled := leftovers()
cancelled.Run = ended(run(core.RunCancelled), afterRun)
rep = Build(cancelled)
if rep.Responsibility.State != StateUnknown || rep.NextStep.Action == ActionConfirmCard || rep.NextStep.CardID != "" {
t.Fatalf("cancelled run with leftovers: %+v", rep)
}
if !containsNote(rep.Notes, "stale") {
t.Fatalf("cancelled run with leftovers must name them: %v", rep.Notes)
}
// 未终态的 run 不受影响:卡仍然驱动等待映射,且不得被说成残留。
live := leftovers()
live.Run = run(core.RunRunning)
rep = Build(live)
if rep.Responsibility.State != StateWaitingHITL || rep.NextStep.Action != ActionConfirmCard || rep.NextStep.CardID != "card_hitl" {
t.Fatalf("a live run must keep its wait mapping: %+v", rep)
}
if containsNote(rep.Notes, "stale") {
t.Fatalf("a live run must not be reported as stale: %v", rep.Notes)
}
}
// MI2:有界读取命中上限时必须显式说截断,不能让「读了前 N 条」冒充「全部证据」。
func TestBuildReportsReadCapsInsteadOfSilentTruncation(t *testing.T) {
// 审计信封按时间倒序取最近 maxAuditScan 条;命中上限即说明更早的回执没被读。
rep := Build(Gather(fakeSource{audits: envelopes(maxAuditScan + 5)}, run(core.RunRunning), afterRun))
if !containsNote(rep.Notes, "audit scan reached the") {
t.Fatalf("a capped audit scan must be reported: %v", rep.Notes)
}
// 副作用聚合上限:超限的回执不进报告,必须标出截断。
rep = Build(Evidence{Run: ended(run(core.RunFailed), afterRun), Audits: envelopes(maxSideEffects + 10)})
if len(rep.SideEffects) != maxSideEffects {
t.Fatalf("side effects must stop at the read cap: %d", len(rep.SideEffects))
}
if !containsNote(rep.Notes, "side_effects truncated at the") {
t.Fatalf("capped side effects must be reported as truncated: %v", rep.Notes)
}
// 确认行上限同样要标出来。
confirms := make([]store.PendingConfirm, maxConfirms)
for i := range confirms {
confirms[i] = store.PendingConfirm{CardID: fmt.Sprintf("card_%d", i), Owner: "alice",
CreatedAt: afterRun.Format(time.RFC3339), Payload: "{}"}
}
rep = Build(Gather(fakeSource{confirms: confirms}, run(core.RunRunning), afterRun))
if !containsNote(rep.Notes, "confirmation read reached the") {
t.Fatalf("a capped confirmation read must be reported: %v", rep.Notes)
}
// 未命中上限时不加噪音 note。
clean := Build(Evidence{Run: run(core.RunRunning), Audits: envelopes(3)})
if containsNote(clean.Notes, "truncat") {
t.Fatalf("an uncapped read must not claim truncation: %v", clean.Notes)
}
}
// envelopes 造 n 条归属本次 run 的回执信封(RunID 显式带上,绕过时间窗推断)。
func envelopes(n int) []agentcommand.AuditEnvelope {
out := make([]agentcommand.AuditEnvelope, 0, n)
for i := 0; i < n; i++ {
out = append(out, agentcommand.AuditEnvelope{
RunID: "run_1", CommandID: "rdm.docs.instantiate", OperationID: fmt.Sprintf("op_%d", i),
State: agentcommand.StateCommitted, Time: afterRun.Add(time.Duration(i) * time.Second),
})
}
return out
}
func mustRequest(t *testing.T, req cmdbus.Request) string {
t.Helper()
raw, err := json.Marshal(req)
+14 -1
View File
@@ -54,7 +54,7 @@ type directRun struct {
acting bool
queued []agentllm.ToolCall
inflight map[string]*toolSlot // 并行执行下按内核事件 CallID 配对
toolMu sync.Mutex // 并行 onEvent 下 inflight/queued/finalParts 互斥
toolMu sync.Mutex // 事件回调与工具 goroutine 之间互斥(inflight/finalParts/queued)
turnBody string
evidence []toolEvidence
@@ -763,6 +763,15 @@ func (st *directRun) onEvent(ev nruntime.Event) {
_ = st.rc.Emit(core.EvRunPhase, core.RunPhasePayload{Phase: core.PhaseActing})
st.acting = true
}
// queued 的判空与 takeQueued 不持 toolMu:它们与 message_end 里对 queued 的
// 写入都在同一个事件回调(onEvent)里,而内核把宿主回调串行化——Run 入口的
// serializeHostCallbacks 包装 Config.OnEvent(agent-native 循环),并行
// executeParallel 的每个 goroutine 各发一个 tool_execution_start,但它们逐个
// 进入 onEvent,不会并发;于是取用与写入天然互斥(配对结果不因此改变,只是
// 不再依赖“恰好没人同时来”)。对比:inflight 槽位与 finalParts 还被工具
// goroutine(directRun.writeSlot / appendFinalPart)写,那条路径不在回调串行化
// 范围内,必须持 toolMu——takeInflight / writeSlot 的“调用方须持有 toolMu”
// 靠那把锁兑现。
if len(st.queued) == 0 {
return
}
@@ -1440,6 +1449,10 @@ func toolExecMode(eff core.EffectiveSettings) string {
// takeQueued 按内核事件 CallID 取出队列中对应调用(并行下 start 事件乱序到达);
// CallID 为空(旧内核/未知来源)时按原先进先出。调用方须持有 toolMu。
//
// queued 的并发前提:内核串行化宿主事件回调,takeQueued 与写 queued 的 message_end
// 不会并发(见 onEvent 的 tool_execution_start 分支说明),因此调用点可以不持锁;
// 本函数的其余调用方仍应持 toolMu(读改写非原子,脱离回调串行化即数据竞争)。
func (st *directRun) takeQueued(callID string) (agentllm.ToolCall, bool) {
if callID != "" {
for i, c := range st.queued {
+9
View File
@@ -59,6 +59,13 @@ type toolFailure struct {
// 因此流式文本、工具部件、用量与 pending 配对表全部在同一把锁下更新:
// 既不丢条目,也不会把 A 调用的 start 与 B 调用的 end 交错写坏。
// 事件发射与部件追加在同一临界区内,保持到达顺序与最终部件顺序一致。
//
// sink 契约(rc.Emit):同步、快、不得阻塞。onEvent 全程持 l.mu,发射在临界区内完成,
// 所以任何慢 sink 都会把所有过程帧串成一条队:并行工具的 goroutine 一起堵在锁上,
// 帧的到达顺序与最终部件顺序的不变式随之失去意义(且帧可能被重排)。既有 sink 只做
// 鉴权、投影与投递,不在回调里做重活;新增 sink 必须守住这条契约。真要做重活就放到
// 锁外(自建 goroutine 或交给 hub 排队),但那会放弃"发射与部件同序"这条不变式,
// 需要一并权衡,而不是顺手挪出来。
type squadEventLog struct {
rc RunCtx
@@ -76,6 +83,8 @@ func newSquadEventLog(rc RunCtx) *squadEventLog {
}
// onEvent 是小队过程帧回调(并发安全)。
//
// 锁内调 rc.Emit:sink 契约见 squadEventLog 的说明(同步、快、不得阻塞)。
func (l *squadEventLog) onEvent(ev squad.ProcessEvent) {
l.mu.Lock()
defer l.mu.Unlock()
+8
View File
@@ -140,6 +140,14 @@ func (s *Server) handleIntentCancel(w http.ResponseWriter, r *http.Request, conv
AnchorMessageID: msg.ID, HostProjectID: string(conv.ProjectID),
})
result.Via, result.Accepted = "bus", resp.Receipt.State != agentcommand.StateRejected
if !result.Accepted {
// 拒绝原因如实透出(与本地投递路径对称):不受理就要说清为什么,
// 绝不只报一个「没成功」的空理由。
result.Reason = strings.TrimSpace(resp.Receipt.Reason)
if result.Reason == "" {
result.Reason = "命令被拒绝"
}
}
awaitingConfirm = resp.Card != nil
out["command"], out["receipt"] = intentCancelCommandID, resp.Receipt
if resp.Card != nil {
+55
View File
@@ -10,6 +10,7 @@ package server
import (
"context"
"encoding/json"
"errors"
"net/http"
"strings"
"testing"
@@ -398,6 +399,60 @@ func TestIntentCancelResolvesRunIDFromCommandValue(t *testing.T) {
waitRunStatus(t, h, runID, string(core.RunCancelled))
}
// 命令总线拒绝派发(处理器声明「确定未执行」):不受理,并如实给出拒绝
// 原因(与本地投递路径对称:绝不只报一个空理由)。
func TestIntentCancelRejectedDispatchReportsReason(t *testing.T) {
srv, st := newTestServer(t)
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) {
return nil, cmdbus.NotExecuted(errors.New("取消被策略拒绝"))
}); err != nil {
t.Fatal(err)
}
gate := &scriptedGate{}
srv.Gate = gate
h := srv.Handler()
projectID, convID := setupConversation(t, h)
runID := liveRun(t, h, convID)
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)
}
cancel, _ := out["cancel"].(map[string]any)
if cancel == nil {
t.Fatalf("取消回执缺失:%v", out)
}
if cancel["accepted"] != false || cancel["via"] != "bus" || cancel["runId"] != runID {
t.Fatalf("被拒绝的派发不得写成已受理:%v", cancel)
}
reason, _ := cancel["reason"].(string)
if !strings.Contains(reason, "取消被策略拒绝") {
t.Fatalf("拒绝原因必须如实透出:%v", cancel)
}
receipt, _ := out["receipt"].(map[string]any)
if receipt == nil || receipt["state"] != string(agentcommand.StateRejected) {
t.Fatalf("拒绝回执:%v", out)
}
// 没受理就不写会话记录,也不谎报已取消。
if _, noted := out["intent_session"]; noted {
t.Fatalf("被拒绝的取消不得写会话记录:%v", out)
}
if sess := pendingIntent(t, st, projectID, convID); sess != nil {
t.Fatalf("被拒绝的取消改动了会话:%+v", sess)
}
// 运行没被取消(拒绝即无副作用)。
if code, run := doJSON(t, h, "GET", "/api/runs/"+runID, nil); code == 200 {
if r, _ := run["run"].(map[string]any); r["status"] == string(core.RunCancelled) {
t.Fatalf("被拒绝的取消却取消了运行:%v", r)
}
}
}
// 消息原文进 params:取消命令与其它命令入口同款(恒带 text)。
func TestIntentCancelPassesOriginalText(t *testing.T) {
srv, _, gate, log := cancelServer(t)
+19 -5
View File
@@ -8,8 +8,9 @@ package store
//
// 键是 (owner, project_id, conversation_id):两个用户、两个项目的会话在
// 构造上互不可见,读写一律带 owner 过滤。行是运行态(默认 30 分钟后
// 过期、读侧惰性删除),不进 storage-migrate 的 copyOrder(同
// admission_leases:进程/会话瞬态,迁移没有意义)。
// 过期、读侧惰性删除);但它随库迁移:v52 起 intent_session 已登记在
// storage-migrate 的 copyOrder 里(夹具即文档:新表漏登记会被那条用例
// 挡住),源库没有这张表时由 sqliteHasTable 跳过(旧库无此表)。
import (
"database/sql"
@@ -46,8 +47,12 @@ type IntentSession struct {
Slots map[string]string
Revision int
ExpiresAt time.Time
CreatedAt time.Time
UpdatedAt time.Time
// ExpiresValid 报告到期时间是否可用:库里的 expires_at 为空或不可解析
// 时为假(此时 ExpiresAt 不可信)。宿主据此 fail-closed 把该行当成
// 已过期,绝不把损坏的到期时间当成「永不过期」。
ExpiresValid bool
CreatedAt time.Time
UpdatedAt time.Time
}
// GetIntentSession 按 owner+project+conversation 读回意图会话。
@@ -159,7 +164,16 @@ func scanIntentSession(owner, projectID, conversationID string, row *sql.Row) (I
out.Slots = slots
}
}
out.ExpiresAt, _ = parseTime(expiresAt)
// 到期时间:可解析才算有效。空值(没有 TTL)或解析失败(行被写坏)都
// 如实标成不可用——宿主据此 fail-closed 当作已过期,绝不 fail-open 地
// 当成「永不过期」。
expires, err := parseTime(strings.TrimSpace(expiresAt))
if err != nil {
out.ExpiresValid = false
} else {
out.ExpiresValid = true
}
out.ExpiresAt = expires
out.CreatedAt, _ = parseTime(createdAt)
out.UpdatedAt, _ = parseTime(updatedAt)
return out, nil
+34
View File
@@ -120,6 +120,40 @@ func TestIntentSessionIsolatesOwnersAndProjects(t *testing.T) {
}
}
// 到期时间:可解析才算有效。为空(没有 TTL)或被写坏都如实标成「不可用」,
// 宿主据此 fail-closed 当作已过期——绝不当成「永不过期」。
func TestIntentSessionExpiryIsReportedUnusableWhenCorrupted(t *testing.T) {
st := openTestStore(t)
if _, err := st.PutIntentSession("alice", "prj_1", "cnv_1", cancelExpectation(),
map[string]string{"run_id": "R-1"}, time.Minute); err != nil {
t.Fatal(err)
}
got, err := st.GetIntentSession("alice", "prj_1", "cnv_1")
if err != nil || !got.ExpiresValid || got.ExpiresAt.IsZero() {
t.Fatalf("正常写入的到期时间应可用:%+v %v", got, err)
}
// 直接改坏库里的 expires_at(模拟损坏的行)。
if _, err := st.DB().Exec(st.q(`UPDATE intent_session SET expires_at = ? WHERE owner = ? AND conversation_id = ?`),
"not-a-timestamp", "alice", "cnv_1"); err != nil {
t.Fatal(err)
}
broken, err := st.GetIntentSession("alice", "prj_1", "cnv_1")
if err != nil {
t.Fatal(err)
}
if broken.ExpiresValid || !broken.ExpiresAt.IsZero() {
t.Fatalf("损坏的到期时间必须如实标成不可用:%+v", broken)
}
// 没有 TTL(ttl<=0)的写入同样没有可用到期时间。
if _, err := st.PutIntentSession("alice", "prj_1", "cnv_2", cancelExpectation(), nil, 0); err != nil {
t.Fatal(err)
}
none, err := st.GetIntentSession("alice", "prj_1", "cnv_2")
if err != nil || none.ExpiresValid {
t.Fatalf("没有到期时间的行不得被当成有效:%+v %v", none, err)
}
}
// intentSessionTableExists 双档目录探测(SQLite sqlite_master / PG information_schema)。
func intentSessionTableExists(t *testing.T, st *Store) bool {
t.Helper()