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

internal/intentsession/intentsession.go

@@ -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 {

internal/intentsession/intentsession_test.go

@@ -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)

internal/rundiag/rundiag.go

@@ -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 {

internal/rundiag/rundiag_test.go

@@ -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)

internal/runtime/llm_direct.go

@@ -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 {

internal/runtime/squad_events.go

@@ -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()

internal/server/intent_cancel.go

@@ -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 {

internal/server/intent_cancel_test.go

@@ -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)

internal/store/intent_session.go

@@ -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

internal/store/intent_session_test.go

@@ -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()