Harden intent-session tests against load-sensitive cancellation

Why: TestIntentSessionLegacyResponsesUnchanged raced the cancel
dispatch against the prior run settling and 409'd under CI load; the
tests now wait for settle and poll busy 409s with a deadline.
Evidence: repair receipt (a1ff-run.json); 3x race runs green
(verify-a1ff.json, exit=0).
Impact: test-only hardening.
Refs: feature/exec-recovery-diagnostics-a2

internal/server/intent_cancel_test.go

@@ -361,8 +361,9 @@ func TestIntentCancelRecordIsNotACancelTarget(t *testing.T) {
 	}
 	waitRunStatus(t, h, runID, string(core.RunCancelled))
 	gate.script("再取消一次", taskCancelIntent(nil))
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
-		map[string]any{"text": "再取消一次"})
+	// 第二轮是占槽位的澄清回合：等上一轮被取消的运行彻底收工再发车（忙 409 是
+	// 既有语义，见 postConversationTurn）。
+	code, out := postConversationTurn(t, srv, h, st, convID, "再取消一次")
 	if code != 200 {
 		t.Fatalf("第二轮取消：%d %v", code, out)
 	}

internal/server/intent_session_test.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"sort"
 	"strings"
@@ -15,6 +16,7 @@ import (
 	"agent-system-backend/internal/intentgate"
 	"agent-system-backend/internal/intentsession"
 	"agent-system-backend/internal/store"
+	core "agent-system-core"
 )
 
 // A1/U-A1b 会话语义的 chat 端到端（合成数据 + 现有 fakes）：缺槽位动作先
@@ -113,6 +115,105 @@ func settle(t *testing.T, h http.Handler, out map[string]any) map[string]any {
 	return out
 }
 
+// 会话忙（409）是既有语义，不改：占槽位的回合（普通回合、澄清回合）在忙时如
+// 实 409，意图命中的命令则不占槽位、会话忙时也能用。测试侧要解决的是「等发得
+// 出去」，而不是放宽断言：上一轮的 run 行在结算提交那一刻就终态了，槽位要到收
+// 尾（补锚点、发终态事件、摘执行句柄）之后才释放，机器繁忙时这段窗口会被放大，
+// 于是「run 一终态就立刻发下一轮」会偶发撞上忙 409。
+
+// postConversationTurn 发一轮消息：先等会话空闲（见 waitConversationIdle），
+// 发车时若仍撞上忙（409）则在「忙拒没有留下痕迹」的前提下按 deadline 重发。
+// 忙不写消息与运行，但抢槽位发生在意图会话折叠之后——被忙拒掉的轮次仍可能已经
+// 消费或挂起了会话，再发一次走的就是另一条路径了；所以每次重发前先比对意图会
+// 话状态：变过了就如实失败，没变过才重发。
+func postConversationTurn(t *testing.T, srv *Server, h http.Handler, st *store.Store,
+	convID, text string) (int, map[string]any) {
+	t.Helper()
+	waitConversationIdle(t, srv, h, convID)
+	deadline := time.Now().Add(testRunTerminalTimeout)
+	for {
+		before := intentSessionTrace(t, st, convID)
+		code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": text})
+		if code != http.StatusConflict {
+			return code, out
+		}
+		if after := intentSessionTrace(t, st, convID); after != before {
+			t.Fatalf("%s：忙（409）之前已改动意图会话，重发会走另一条路径：%s → %s", text, before, after)
+		}
+		if !time.Now().Before(deadline) {
+			t.Fatalf("%s：会话持续忙（409）超过 %v：%v", text, testRunTerminalTimeout, out)
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}
+
+// waitConversationIdle 等会话真正让出槽位：既没有未终态运行，也没有仍在本地
+// 执行的 run——执行句柄在槽位释放之后才摘除（runmgr 的 execute 收尾），所以
+// 「没有活跃句柄」就是槽位已空的可观测证据，不用赌收尾窗口有多宽。
+func waitConversationIdle(t *testing.T, srv *Server, h http.Handler, convID string) {
+	t.Helper()
+	deadline := time.Now().Add(testRunTerminalTimeout)
+	for time.Now().Before(deadline) {
+		if code, out := doJSON(t, h, "GET", "/api/conversations/"+convID, nil); code == 200 &&
+			!conversationHasOpenRun(out) && !conversationHasActiveRun(srv, out) {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("会话 %s 在 %v 内未空闲", convID, testRunTerminalTimeout)
+}
+
+// conversationHasOpenRun 看会话里是否还有未终态运行（结算尚未提交）。
+func conversationHasOpenRun(detail map[string]any) bool {
+	for _, run := range conversationRuns(detail) {
+		switch run["status"] {
+		case string(core.RunCompleted), string(core.RunFailed),
+			string(core.RunCancelled), string(core.RunInterrupted):
+		default:
+			return true
+		}
+	}
+	return false
+}
+
+// conversationHasActiveRun 看会话里是否还有本地在执行的 run（槽位未让出）。
+func conversationHasActiveRun(srv *Server, detail map[string]any) bool {
+	for _, run := range conversationRuns(detail) {
+		if id, _ := run["id"].(string); id != "" && srv.mgr.RunActiveLocally(id) {
+			return true
+		}
+	}
+	return false
+}
+
+func conversationRuns(detail map[string]any) []map[string]any {
+	raw, _ := detail["runs"].([]any)
+	runs := make([]map[string]any, 0, len(raw))
+	for _, item := range raw {
+		if run, ok := item.(map[string]any); ok {
+			runs = append(runs, run)
+		}
+	}
+	return runs
+}
+
+// intentSessionTrace 描述会话上当前的意图会话状态（忙拒前后各取一次比对）。
+func intentSessionTrace(t *testing.T, st *store.Store, convID string) string {
+	t.Helper()
+	conv, err := st.GetConversation(store.OwnerDefault, convID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	sess, err := st.GetIntentSession(store.OwnerDefault, string(conv.ProjectID), convID)
+	if err == store.ErrNotFound {
+		return "none"
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	return fmt.Sprintf("rev=%d exp=%v slots=%v", sess.Revision, sess.Expectation, sess.Slots)
+}
+
 // 缺必填槽位 → 澄清并挂起；下一轮补齐 → 按同一条准入与派发路径执行。
 func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
 	srv, st, gate, log := cancelServer(t)
@@ -122,7 +223,7 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
 	gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
 		{Key: "run_id", Question: runIDQ}, {Key: "reason", Question: "取消原因？"}}))
 
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
+	code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
 	if code != 200 {
 		t.Fatalf("第一轮：%d %v", code, out)
 	}
@@ -145,7 +246,7 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
 	// 下一轮：仅补齐槽位（库层挂起预期命中分支的同形结果）。
 	answerText := "编号：R-7 原因：误触发"
 	gate.script(answerText, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answerText})
+	code, out = postConversationTurn(t, srv, h, st, convID, answerText)
 	if code != 200 || out["command"] != "demo.run-cancel" {
 		t.Fatalf("补齐后应派发同一动作：%d %v", code, out)
 	}
@@ -167,12 +268,12 @@ func TestIntentSessionAnswerCompletesPendingAction(t *testing.T) {
 
 // 两轮部分填充：先 set_slot 落一个槽位（仍缺 → 重新澄清），再由回答补齐。
 func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
-	srv, _, gate, log := cancelServer(t)
+	srv, st, gate, log := cancelServer(t)
 	h := srv.Handler()
 	_, convID := setupConversation(t, h)
 	gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
 		{Key: "run_id", Question: "取消哪个运行？"}, {Key: "reason", Question: "取消原因？"}}))
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
+	code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
 	if code != 200 {
 		t.Fatalf("第一轮：%d %v", code, out)
 	}
@@ -181,7 +282,7 @@ func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
 	// 第二轮：库层 set_slot（只带一个槽位），仍缺原因 → 如实重新澄清。
 	gate.script("编号 R-7", intentgate.MatchResult{Branch: "expectation_hit", ID: "demo.run-cancel",
 		Commands: []intentgate.Command{{Kind: intentgate.CommandSetSlot, Name: "run_id", Value: "R-7"}}})
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号 R-7"})
+	code, out = postConversationTurn(t, srv, h, st, convID, "编号 R-7")
 	if code != 200 {
 		t.Fatalf("set_slot 轮：%d %v", code, out)
 	}
@@ -203,7 +304,7 @@ func TestIntentSessionSetSlotThenAnswerCompletes(t *testing.T) {
 
 	// 第三轮：补上原因 → 派发，两个槽位都来自会话累计。
 	gate.script("原因：误触发", cancelAnswer(map[string]string{"reason": "误触发"}, nil))
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "原因：误触发"})
+	code, out = postConversationTurn(t, srv, h, st, convID, "原因：误触发")
 	if code != 200 || out["command"] != "demo.run-cancel" {
 		t.Fatalf("补齐后应派发：%d %v", code, out)
 	}
@@ -220,7 +321,7 @@ func TestIntentSessionCancelClearsPendingAction(t *testing.T) {
 	_, convID := setupConversation(t, h)
 	gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
 		{Key: "run_id", Question: "取消哪个运行？"}, {Key: "reason", Question: "取消原因？"}}))
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
+	code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
 	if code != 200 {
 		t.Fatalf("第一轮：%d %v", code, out)
 	}
@@ -228,7 +329,7 @@ func TestIntentSessionCancelClearsPendingAction(t *testing.T) {
 
 	gate.script("算了不取消了", intentgate.MatchResult{Branch: "matched", ID: "demo.run-cancel",
 		Commands: []intentgate.Command{{Kind: intentgate.CommandCancel, Target: "demo.run-cancel"}}})
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "算了不取消了"})
+	code, out = postConversationTurn(t, srv, h, st, convID, "算了不取消了")
 	if code != 200 {
 		t.Fatalf("取消轮：%d %v", code, out)
 	}
@@ -253,7 +354,7 @@ func TestIntentSessionCancelClearsPendingAction(t *testing.T) {
 	// 取消之后的回答不得补齐旧动作（这里识别仍不完整 → 只澄清，不派发）。
 	gate.script("编号：R-7", cancelAnswer(map[string]string{"run_id": "R-7"},
 		[]intentgate.SlotQuestion{{Key: "reason", Question: "取消原因？"}}))
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号：R-7"})
+	code, out = postConversationTurn(t, srv, h, st, convID, "编号：R-7")
 	if code != 200 {
 		t.Fatalf("取消后回答轮：%d %v", code, out)
 	}
@@ -275,8 +376,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
 	// 第一轮：抽到 run_id，仍缺原因 → 挂起（累计槽位含 run_id）。
 	gate.script("取消运行 编号：R-7", cancelStart(map[string]string{"run_id": "R-7"},
 		[]intentgate.SlotQuestion{{Key: "reason", Question: "取消原因？"}}))
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages",
-		map[string]any{"text": "取消运行 编号：R-7"})
+	code, out := postConversationTurn(t, srv, h, st, convID, "取消运行 编号：R-7")
 	if code != 200 {
 		t.Fatalf("第一轮：%d %v", code, out)
 	}
@@ -297,7 +397,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
 	// 第二轮只补了原因：过期的 run_id 不得被借来凑齐动作。
 	answerText := "原因：误触发"
 	gate.script(answerText, cancelStart(map[string]string{"reason": "误触发"}, nil))
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answerText})
+	code, out = postConversationTurn(t, srv, h, st, convID, answerText)
 	if code != 200 {
 		t.Fatalf("过期后的回答轮：%d %v", code, out)
 	}
@@ -339,7 +439,7 @@ func TestIntentSessionExpiryDoesNotCompleteOldAction(t *testing.T) {
 		t.Fatal(err)
 	}
 	gate.script("清理缓存", otherIntentStart("demo.purge"))
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "清理缓存"})
+	code, out = postConversationTurn(t, srv, h, st, convID, "清理缓存")
 	if code != 200 || out["command"] != "demo.purge" {
 		t.Fatalf("新任务应照常派发：%d %v", code, out)
 	}
@@ -363,7 +463,7 @@ func TestIntentSessionIsolatesConversations(t *testing.T) {
 	projectB, convB := setupConversation(t, h)
 	gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
 		{Key: "run_id", Question: "取消哪个运行？"}, {Key: "reason", Question: "取消原因？"}}))
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convA+"/messages", map[string]any{"text": "取消运行"})
+	code, out := postConversationTurn(t, srv, h, st, convA, "取消运行")
 	if code != 200 {
 		t.Fatalf("A 会话第一轮：%d %v", code, out)
 	}
@@ -378,7 +478,7 @@ func TestIntentSessionIsolatesConversations(t *testing.T) {
 	// B 会话收到同样的回答：A 的槽位不得被借走（没有 A 的累计值）。
 	answer := "编号：R-7 原因：误触发"
 	gate.script(answer, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convB+"/messages", map[string]any{"text": answer})
+	code, out = postConversationTurn(t, srv, h, st, convB, answer)
 	if code != 200 {
 		t.Fatalf("B 会话回答轮：%d %v", code, out)
 	}
@@ -395,19 +495,19 @@ func TestIntentSessionIsolatesConversations(t *testing.T) {
 func TestIntentSessionStillRequiresHighRiskConfirmation(t *testing.T) {
 	risky := cancelSpec
 	risky.Danger = agentcommand.DangerHigh
-	srv, _, gate, log := cancelServerWith(t, risky)
+	srv, st, gate, log := cancelServerWith(t, risky)
 	h := srv.Handler()
 	_, convID := setupConversation(t, h)
 	gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
 		{Key: "run_id", Question: "取消哪个运行？"}, {Key: "reason", Question: "取消原因？"}}))
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
+	code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
 	if code != 200 {
 		t.Fatalf("第一轮：%d %v", code, out)
 	}
 	settle(t, h, out)
 	answer := "编号：R-7 原因：误触发"
 	gate.script(answer, cancelAnswer(map[string]string{"run_id": "R-7", "reason": "误触发"}, nil))
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": answer})
+	code, out = postConversationTurn(t, srv, h, st, convID, answer)
 	if code != 200 || out["command"] != "demo.run-cancel" {
 		t.Fatalf("补齐后应派发到总线：%d %v", code, out)
 	}
@@ -426,7 +526,7 @@ func TestIntentSessionInvalidAnswerKeepsClarifying(t *testing.T) {
 	_, convID := setupConversation(t, h)
 	gate.script("取消运行", cancelStart(nil, []intentgate.SlotQuestion{
 		{Key: "run_id", Question: "取消哪个运行？"}}))
-	code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "取消运行"})
+	code, out := postConversationTurn(t, srv, h, st, convID, "取消运行")
 	if code != 200 {
 		t.Fatalf("第一轮：%d %v", code, out)
 	}
@@ -435,7 +535,7 @@ func TestIntentSessionInvalidAnswerKeepsClarifying(t *testing.T) {
 	gate.script("编号：R-9", intentgate.MatchResult{Branch: "expectation_hit", ID: "demo.run-cancel",
 		Slots:   map[string]string{"run_id": "R-9"},
 		Invalid: []intentgate.SlotDefect{{Key: "run_id", Reason: "unknown run", Question: "取消哪个运行？", Invalid: true}}})
-	code, out = doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": "编号：R-9"})
+	code, out = postConversationTurn(t, srv, h, st, convID, "编号：R-9")
 	if code != 200 {
 		t.Fatalf("非法补充轮：%d %v", code, out)
 	}
@@ -495,7 +595,7 @@ func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
 	}
 	run := func(t *testing.T, sessions IntentSessions) []string {
 		t.Helper()
-		srv, _, gate, log := cancelServer(t)
+		srv, st, gate, log := cancelServer(t)
 		if sessions != nil {
 			srv.SetIntentSessionService(sessions)
 		}
@@ -504,7 +604,7 @@ func TestIntentSessionLegacyResponsesUnchanged(t *testing.T) {
 		_, convID := setupConversation(t, h)
 		var summaries []string
 		for _, text := range []string{"取消运行 编号：R-3 原因：误触发", "今天天气如何", "算了", "取消运行"} {
-			code, out := doJSON(t, h, "POST", "/api/conversations/"+convID+"/messages", map[string]any{"text": text})
+			code, out := postConversationTurn(t, srv, h, st, convID, text)
 			if code != 200 {
 				t.Fatalf("%s：%d %v", text, code, out)
 			}