weichen/Agent-System-Backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden intent-session tests against load-sensitive cancellation

Browse Source 
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
This commit is contained in:
agent-exec-transport
2026-10-22 14:53:10 +00:00
committed by weichen
parent 9ee752c380
commit bd051ec895
2 changed files with 125 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/server/intent_cancel_test.go
+3 -2
View File
@@ -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
+122 -22
View File
@@ -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)
}