From ef3dcb699b01603bba42ecdc266cbd338e6e1c91 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 09:53:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway,loop,tool):=20=E8=A7=A3=E8=80=A6?= =?UTF-8?q?=E3=80=8C=E5=B7=A5=E5=85=B7=E4=BA=92=E6=96=A5=E8=BA=AB=E4=BB=BD?= =?UTF-8?q?=E3=80=8D=E4=B8=8E=E3=80=8Ccheckpoint=20=E5=BD=92=E5=B1=9E?= =?UTF-8?q?=E3=80=8D=EF=BC=88=E4=BF=AE=E6=99=AE=E9=80=9A=E5=9B=9E=E5=90=88?= =?UTF-8?q?=E4=B8=8D=E5=8F=97=E4=BF=9D=E6=8A=A4=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 缺口：「1 普通回合 + 1 并行路」时普通回合既不登记也不被检查 —— 因为互斥身份取自 loop.ParallelLegFrom，而网关只给 parallel=true 那一路盖它。直接给普通回合盖 WithParallelLeg 会把它的 checkpoint 键从 sess 变成 sess#turn-x（改崩溃恢复语义），故改为两条独立通道： - 工具互斥身份：loop.WithToolLegID / ToolLegIDFrom（新增），网关在 beginLegLocked 内【无条件】盖， 取值 = 本路 turnID ⇒ 普通回合 / 队首续跑路 / 并行路 三条路都覆盖 - checkpoint 归属：WithParallelLeg 一行未改，仍只有真并行路 ⇒ 恢复语义零变化 守门断言 TestCheckpointOwnerIgnoresToolLegID：普通回合 ctx 下 AgentID 必须逐字仍是 sess。 变异 3 条：互斥身份回退只认 ParallelLegFrom（判忙失效）/ 顺手把 checkpoint 改读新通道 （守门断言红，证明确实解耦）/ 普通回合不盖身份（判忙失效）。 另修 race：-race -count=2 三包全绿；诚实边界含一个既有竞态（同一路并发登记共享，建议后续小切片）。 --- internal/runtime/loop/fileclaim_leg_test.go | 158 +++++-- internal/runtime/loop/parallel_leg_test.go | 42 +- internal/runtime/loop/session_ctx.go | 49 ++- internal/runtime/loop/tools.go | 14 +- internal/runtime/tool/fileclaims.go | 8 +- internal/runtime/tool/legfile.go | 25 +- .../services/gateway/session_legmutex_test.go | 390 ++++++++++++++++++ internal/services/gateway/session_parallel.go | 31 +- .../services/gateway/session_parallel_test.go | 9 + 9 files changed, 673 insertions(+), 53 deletions(-) create mode 100644 internal/services/gateway/session_legmutex_test.go diff --git a/internal/runtime/loop/fileclaim_leg_test.go b/internal/runtime/loop/fileclaim_leg_test.go index 994377c..43c8476 100644 --- a/internal/runtime/loop/fileclaim_leg_test.go +++ b/internal/runtime/loop/fileclaim_leg_test.go @@ -13,20 +13,26 @@ import ( "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" ) -// 本文件是「路标识 → 工具层」这座桥的回归（design/15 §2.5.7、主人 2026-10-05 -// 裁决 3）：网关按 loop.WithParallelLeg 把并发路标识挂在 ctx 上，工具执行点 -// （tools.go 的 invokeOne）把它转成 tool.WithLegID，于是同一文件的第二条路拿到 +// 本文件是「互斥身份 → 工具层」这座桥的回归（design/15 §2.5.7、主人 2026-10-05 +// 裁决 3）：**每一条在途路**（普通回合 / 并行路 / 队首续跑路）都带互斥身份 —— +// 网关按 loop.WithToolLegID 把它挂在 ctx 上（取值 = 该路的 turnID），工具执行点 +// （tools.go 的 invokeOne）把它桥接成 tool.WithLegID，于是同一文件的第二条路拿到 // tool.ErrFileBusy，并以**回灌给模型的工具结果文本**形式浮出来（显式、可消化）。 // -// 单路口径同样钉住：ctx 上没有路标识（普通回合 / 子代理 / 后台思考）时，同文件 -// 的两条调用照跑 —— 完全不做互斥，行为逐字不变。 +// 它与 checkpoint 归属（loop.WithParallelLeg）是**两条独立通道**，因此这里刻意 +// 钉住两个方向：① 只有并发路标识（没有互斥身份）的 ctx **不**带来互斥；② 只有 +// 互斥身份的 ctx（普通回合的实际形状）**照样**带来互斥 —— 后者正是「1 普通回合 +// + 1 并行路」同文件写的修复目标。 +// +// 单路口径同样钉住：ctx 上没有互斥身份（子代理 / 后台思考 / 未接编排器）时，同一 +// 文件的两条调用照跑 —— 完全不做互斥，行为逐字不变。 // legFileProbeTimeout 是「第二路必须立刻拿到判忙结果」的上界（只用来把阻塞式 // 实现暴露成失败，不参与竞态）。 const legFileProbeTimeout = 5 * time.Second // legFileTool 是实现了 tool.FileTargeter 的 mock 工具：声明本次要写的文件， -// 可阻塞执行，并记录它在 Execute 里读到的路标识（用来直接钉住桥的存在）。 +// 可阻塞执行，并记录它在 Execute 里读到的互斥身份（用来直接钉住桥的存在）。 type legFileTool struct { name string file string @@ -89,30 +95,40 @@ func awaitEntered(t *testing.T, ch <-chan struct{}, msg string) { } } -// TestInvokeOne_LegIDReachesTool 直接钉住桥：ctx 上挂了并发路标识 ⇒ 工具在 -// Execute 里读到同一个值；没挂（单路）⇒ 读到空串（⇒ 完全不做互斥）。 -func TestInvokeOne_LegIDReachesTool(t *testing.T) { +// TestInvokeOne_ToolLegIDReachesTool 直接钉住桥：ctx 上挂了**互斥身份** +// ⇒ 工具在 Execute 里读到同一个值；没挂（子代理 / 后台思考 / 未接编排器） +// ⇒ 读到空串（⇒ 完全不做互斥）。它同时钉住**被解耦的那个方向**：只挂 checkpoint +// 归属键（loop.WithParallelLeg）**不**构成互斥身份 —— 两条通道各管各的。 +func TestInvokeOne_ToolLegIDReachesTool(t *testing.T) { t.Parallel() holder := &legFileTool{name: "patch", file: filepath.Join(t.TempDir(), "notes.txt")} d := dispatcherWith(t, holder) l := New(Options{Tools: d}) call := agent.ToolCall{Name: "patch", Args: map[string]any{"path": "notes.txt"}} + if out := l.invokeOne(WithToolLegID(context.Background(), "turn-x"), d, call); out.Err != nil { + t.Fatalf("带互斥身份的调用: %v", out.Err) + } if out := l.invokeOne(WithParallelLeg(context.Background(), "turn-x"), d, call); out.Err != nil { - t.Fatalf("并行路调用: %v", out.Err) + t.Fatalf("只挂 checkpoint 归属键的调用: %v", out.Err) } if out := l.invokeOne(context.Background(), d, call); out.Err != nil { - t.Fatalf("单路调用: %v", out.Err) + t.Fatalf("无身份的调用: %v", out.Err) } - if legs := holder.seenLegs(); len(legs) != 2 || legs[0] != "turn-x" || legs[1] != "" { - t.Fatalf("工具读到的路标识 = %v, want [turn-x ]（空串 = 单路不参与互斥）", legs) + if legs := holder.seenLegs(); len(legs) != 3 || legs[0] != "turn-x" || legs[1] != "" || legs[2] != "" { + t.Fatalf("工具读到的互斥身份 = %v, want [turn-x ]（空串 = 不参与互斥；checkpoint 通道不构成身份）", legs) } } -// TestInvokeOne_ParallelLegSameFileBusySurfacesToModel 钉住端到端的核心行为： -// 两路（不同路标识）同文件 ⇒ 第二路**不执行**、**不等待**，拿到 tool.ErrFileBusy， -// 且回灌给模型的文本里带中文文件状态（模型据此自己决定稍后重试）。 -func TestInvokeOne_ParallelLegSameFileBusySurfacesToModel(t *testing.T) { +// TestInvokeOne_NormalTurnVsParallelLegSameFileBusy 钉住**本次修复的目标**： +// 「1 普通回合 + 1 并行路」的会话里，持锁的是那个**普通回合** —— 它的 ctx 上只有 +// 互斥身份（普通回合不挂 checkpoint 归属键），照样必须登记并对同文件的并行路判忙： +// 第二条路**不执行**、**不等待**，拿到 tool.ErrFileBusy，且回灌给模型的文本里带中文 +// 文件状态（模型据此自己决定稍后重试）。 +// +// 两条 ctx 刻意按**生产形状**构造：普通回合 = 只有 WithToolLegID；并行路 = 两条 +// 通道都有（WithToolLegID + WithParallelLeg，取值同为该路 turnID）。 +func TestInvokeOne_NormalTurnVsParallelLegSameFileBusy(t *testing.T) { t.Parallel() file := filepath.Join(t.TempDir(), "notes.txt") holder := &legFileTool{name: "patch-a", file: file, entered: make(chan struct{}), hold: make(chan struct{})} @@ -123,25 +139,28 @@ func TestInvokeOne_ParallelLegSameFileBusySurfacesToModel(t *testing.T) { callA := agent.ToolCall{Name: "patch-a", Args: map[string]any{"path": file}} callB := agent.ToolCall{Name: "patch-b", Args: map[string]any{"path": file}} + normalCtx := WithToolLegID(context.Background(), "turn-normal") + parallelCtx := WithParallelLeg(WithToolLegID(context.Background(), "turn-par"), "turn-par") + first := make(chan ToolOutcome, 1) go func() { - first <- l.invokeOne(WithParallelLeg(context.Background(), "turn-a"), d, callA) + first <- l.invokeOne(normalCtx, d, callA) }() - awaitEntered(t, holder.entered, "第一路没有进入工具执行") + awaitEntered(t, holder.entered, "普通回合没有进入工具执行") done := make(chan ToolOutcome, 1) go func() { - done <- l.invokeOne(WithParallelLeg(context.Background(), "turn-b"), d, callB) + done <- l.invokeOne(parallelCtx, d, callB) }() var outB ToolOutcome select { case outB = <-done: case <-time.After(legFileProbeTimeout): - t.Fatal("第二路阻塞等待了 —— 口径是不排队死等，判忙即返回") + t.Fatal("并行路阻塞等待了 —— 口径是不排队死等，判忙即返回") } if !errors.Is(outB.Err, tool.ErrFileBusy) { - t.Fatalf("第二路错误 = %v, want tool.ErrFileBusy", outB.Err) + t.Fatalf("普通回合持有同一文件时，并行路必须判忙，得到 %v（nil 说明普通回合压根没登记）", outB.Err) } if outB.ErrKind != tool.KindExecution { t.Fatalf("判忙必须归类为执行面失败才能被回灌，得到 %q", outB.ErrKind) @@ -152,20 +171,67 @@ func TestInvokeOne_ParallelLegSameFileBusySurfacesToModel(t *testing.T) { t.Fatalf("回灌给模型的文本缺少 %q: %s", want, text) } } + // 被判忙的调用绝不执行：只有普通回合那一次进了工具。 + if legs := peer.seenLegs(); len(legs) != 0 { + t.Fatalf("被判忙的并行路绝不该执行工具，实际执行了 %d 次（身份 %v）", len(legs), legs) + } close(holder.hold) select { case outA := <-first: if outA.Err != nil { - t.Fatalf("持锁那一路应正常完成: %v", outA.Err) + t.Fatalf("持锁的普通回合应正常完成: %v", outA.Err) } case <-time.After(legFileProbeTimeout): - t.Fatal("第一路没有结束") + t.Fatal("普通回合没有结束") } } -// TestInvokeOne_SingleLegSameFileRunsConcurrently 钉住兼容硬要求③：**无路标识 -// ⇒ 完全不做互斥**（单路 / 既有路径逐字不变）—— 同文件的两条调用同时执行都成功。 +// TestInvokeOne_DifferentFilesRunParallel 钉住「不是同一个文件 ⇒ 允许并行操作」 +// 在**两条在途路**（普通回合 + 并行路）下同样成立：互斥按文件粒度，绝不能退化成 +// 会话级串行。 +func TestInvokeOne_DifferentFilesRunParallel(t *testing.T) { + t.Parallel() + dir := t.TempDir() + holder := &legFileTool{name: "patch-a", file: filepath.Join(dir, "a.txt"), entered: make(chan struct{}), hold: make(chan struct{})} + peer := &legFileTool{name: "patch-b", file: filepath.Join(dir, "b.txt")} + d := dispatcherWith(t, holder, peer) + l := New(Options{Tools: d}) + + first := make(chan ToolOutcome, 1) + go func() { + first <- l.invokeOne(WithToolLegID(context.Background(), "turn-normal"), d, + agent.ToolCall{Name: "patch-a", Args: map[string]any{"path": "a.txt"}}) + }() + awaitEntered(t, holder.entered, "普通回合没有进入工具执行") + + out := l.invokeOne(WithParallelLeg(WithToolLegID(context.Background(), "turn-par"), "turn-par"), d, + agent.ToolCall{Name: "patch-b", Args: map[string]any{"path": "b.txt"}}) + if out.Err != nil { + t.Fatalf("不同文件必须允许并行操作，得到 %v", out.Err) + } + if legs := peer.seenLegs(); len(legs) != 1 || legs[0] != "turn-par" { + t.Fatalf("不同文件的调用必须真的执行，且带着自己的身份，得到 %v", legs) + } + + close(holder.hold) + select { + case outA := <-first: + if outA.Err != nil { + t.Fatalf("持锁的普通回合应正常完成: %v", outA.Err) + } + case <-time.After(legFileProbeTimeout): + t.Fatal("普通回合没有结束") + } +} + +// TestInvokeOne_SingleLegSameFileRunsConcurrently 钉住兼容硬要求③：**无互斥身份 +// ⇒ 完全不做互斥**（ctx 上没有 WithToolLegID 的调用方逐字不变）—— 同文件的两条 +// 调用同时执行都成功。 +// +// 「单路」这件事本身不再等于「无身份」：接上编排器之后，单条普通回合也有自己的 +// 身份（见 TestInvokeOne_NormalTurnVsParallelLegSameFileBusy），因此它**受保护**； +// 这里钉的是**没有身份**的那一类调用方（子代理 / 后台思考 / 未接编排器）。 func TestInvokeOne_SingleLegSameFileRunsConcurrently(t *testing.T) { t.Parallel() file := filepath.Join(t.TempDir(), "notes.txt") @@ -179,13 +245,47 @@ func TestInvokeOne_SingleLegSameFileRunsConcurrently(t *testing.T) { go func() { first <- l.invokeOne(ctx, d, agent.ToolCall{Name: "patch-a", Args: map[string]any{"path": file}}) }() - awaitEntered(t, blocker.entered, "无路标识的调用没有进入工具执行") + awaitEntered(t, blocker.entered, "无互斥身份的调用没有进入工具执行") if out := l.invokeOne(ctx, d, agent.ToolCall{Name: "patch-b", Args: map[string]any{"path": file}}); out.Err != nil { - t.Fatalf("无路标识时完全不做互斥（同文件也应放行），得到 %v", out.Err) + t.Fatalf("无互斥身份时完全不做互斥（同文件也应放行），得到 %v", out.Err) } close(blocker.hold) if out := <-first; out.Err != nil { - t.Fatalf("无路标识的调用应正常完成: %v", out.Err) + t.Fatalf("无互斥身份的调用应正常完成: %v", out.Err) + } +} + +// TestInvokeOne_ParallelLegOnlyOwnsCheckpointNotMutex 钉住解耦的另一个方向：只挂 +// **checkpoint 归属键**（WithParallelLeg，没有互斥身份）的 ctx 完全不参与文件级 +// 互斥 —— 两条这样的调用同文件同时执行都成功。它挡住「把两个用途又焊回去」的 +// 反向回归：checkpoint 归属键绝不是互斥身份。 +func TestInvokeOne_ParallelLegOnlyOwnsCheckpointNotMutex(t *testing.T) { + t.Parallel() + file := filepath.Join(t.TempDir(), "notes.txt") + blocker := &legFileTool{name: "patch-a", file: file, entered: make(chan struct{}), hold: make(chan struct{})} + peer := &legFileTool{name: "patch-b", file: file} + d := dispatcherWith(t, blocker, peer) + l := New(Options{Tools: d}) + + first := make(chan ToolOutcome, 1) + go func() { + first <- l.invokeOne(WithParallelLeg(context.Background(), "turn-a"), d, + agent.ToolCall{Name: "patch-a", Args: map[string]any{"path": file}}) + }() + awaitEntered(t, blocker.entered, "只挂归属键的调用没有进入工具执行") + + out := l.invokeOne(WithParallelLeg(context.Background(), "turn-b"), d, + agent.ToolCall{Name: "patch-b", Args: map[string]any{"path": file}}) + if out.Err != nil { + t.Fatalf("只有归属键（无互斥身份）时必须放行，得到 %v", out.Err) + } + if legs := peer.seenLegs(); len(legs) != 1 || legs[0] != "" { + t.Fatalf("只挂归属键的调用不该带互斥身份，得到 %v", legs) + } + + close(blocker.hold) + if out := <-first; out.Err != nil { + t.Fatalf("只挂归属键的调用应正常完成: %v", out.Err) } } diff --git a/internal/runtime/loop/parallel_leg_test.go b/internal/runtime/loop/parallel_leg_test.go index fef3dc7..437abdc 100644 --- a/internal/runtime/loop/parallel_leg_test.go +++ b/internal/runtime/loop/parallel_leg_test.go @@ -14,7 +14,9 @@ import ( // 真并行时两路各写各的归属键，LoadLatest 因此天然按路隔离 —— 否则两路合写同一个 // key，后写的那份会把前一路的窗口整段顶掉（LangGraph `Send` 扇出的同族教训）。 // -// 单路口径是**逐字不变**的：ctx 上不挂路标识时归属键仍等于 sessionID。 +// 单路口径是**逐字不变**的：ctx 上不挂路标识时归属键仍等于 sessionID —— 包括那种 +// 「挂了工具互斥身份、但没挂并发路标识」的 ctx（普通回合的实际形状，见 +// TestCheckpointOwnerIgnoresToolLegID：两条通道必须保持解耦）。 // keyedCheckpointStore 是按 AgentID 分桶的内存 checkpoint 存储，用来证明归属键 // 真的按路隔离（判据是**键**，而不是「循环怎么调用 store」）。 @@ -60,6 +62,44 @@ func TestCheckpointOwnerSingleLegUnchanged(t *testing.T) { } } +// TestCheckpointOwnerIgnoresToolLegID 是**守门断言**（design/15 §2.5.7 与 §2.5.9 +// 解耦的守卫）：checkpoint 归属键只认 WithParallelLeg，**绝不**认工具互斥身份 +// （WithToolLegID）。 +// +// 它的存在是为了让「两个用途又被焊回去」变成一条会红的用例：网关给**每一条**在途路 +// 都盖互斥身份（普通回合也有），一旦 checkpointAgentID 改读那条通道，单条普通回合的 +// 归属键就会从 `sess` 升成 `sess#turn-1` —— 崩溃恢复语义随之改变（旧快照读不到、 +// 凭空多出按路隔离）。 +func TestCheckpointOwnerIgnoresToolLegID(t *testing.T) { + t.Parallel() + store := newKeyedCheckpointStore() + l := New(Options{Checkpoint: store}) + // 普通回合的实际形状：有互斥身份（本路 turnID），**没有**并发路标识。 + ctx := WithToolLegID(context.Background(), "turn-1") + window := []model.Message{model.TextMessage(model.RoleUser, "hi")} + l.saveCheckpoint(ctx, "sess", 3, window, agent.StopReasonToolCalls) + + if len(store.saved) != 1 { + t.Fatalf("saved = %d, want 1", len(store.saved)) + } + if got := store.saved[0].AgentID; got != "sess" { + t.Fatalf("互斥身份绝不能改变归属键：AgentID = %q, want %q（逐字不变）", got, "sess") + } + if got := store.saved[0].ID; got != "sess:3" { + t.Fatalf("互斥身份绝不能改变归属键：checkpoint id = %q, want %q", got, "sess:3") + } + + // 续接侧同样只看归属键：LoadLatest 请求的键仍是 sessionID，快照照样能接上。 + reader := &recordingLoadStore{inner: store} + l.checkpoint = reader + if _, step, resumed := l.loadCheckpoint(ctx, "sess", window); !resumed || step != 4 { + t.Fatalf("普通回合 resumed=%v step=%d, want true/4（有互斥身份也必须照旧续接）", resumed, step) + } + if len(reader.calls) != 1 || reader.calls[0] != "sess" { + t.Fatalf("LoadLatest 键 = %v, want [sess]", reader.calls) + } +} + // TestCheckpointOwnerIsolatedPerLeg 钉住归属键按路隔离：同一会话的两条路各写各的 // 键，且各自 LoadLatest 只命中自己的那份 —— 任何一路都绝不会把对方的崩溃快照 // 当成自己的续接。 diff --git a/internal/runtime/loop/session_ctx.go b/internal/runtime/loop/session_ctx.go index 7c93e65..e14c14e 100644 --- a/internal/runtime/loop/session_ctx.go +++ b/internal/runtime/loop/session_ctx.go @@ -57,8 +57,11 @@ func loopTurn(ctx context.Context) string { } // parallelLegKey 是「本 Run 驱动的是某条并发路（leg）」的 context key。 -// 它只影响 checkpoint 归属（design/15 §2.5.9），不参与身份判定，也不改变 -// 循环的任何行为。 +// 它**只**影响 checkpoint 归属（design/15 §2.5.9），不参与身份判定，也不改变 +// 循环的任何行为。工具层的文件级互斥走**另一条独立通道**（toolLegIDKey，见 +// 本文件下半部分）—— 两条通道必须分开：这条只有真并行路才有，而互斥身份 +// 每条在途路都有。焊在一起会把普通回合的 checkpoint 键从 `sess` 改成 +// `sess#turn-x`，连带改变崩溃恢复语义（checkpoint.go 的 checkpointAgentID）。 type parallelLegKey struct{} // WithParallelLeg 把本回合所属**并发路**的标识挂到 ctx 上（design/15 §2.5.9）： @@ -77,6 +80,9 @@ func WithParallelLeg(ctx context.Context, legID string) context.Context { // ParallelLegFrom 返回 ctx 上挂的并发路标识；未挂载时返回 ""。它是 // WithParallelLeg 的读侧，供 checkpoint 归属键与网关侧断言使用。 +// +// ⚠️ 它**只**服务 checkpoint 归属，绝不是工具互斥身份 —— 工具侧要读 +// ToolLegIDFrom（那条通道每条在途路都有）。 func ParallelLegFrom(ctx context.Context) string { if ctx == nil { return "" @@ -84,3 +90,42 @@ func ParallelLegFrom(ctx context.Context) string { s, _ := ctx.Value(parallelLegKey{}).(string) return s } + +// toolLegIDKey 是「本 Run 属于哪一条**在途路**」的 context key，**只**服务 +// 工具层的文件级互斥（design/15 §2.5.7、主人 2026-10-05 裁决 3）。 +// +// 它与 parallelLegKey 刻意分成两条通道，因为二者的**受众与取舍完全不同**： +// - 这条：每条在途路都挂（普通回合 / 并行路 / 队首续跑路），回答「此刻有没有 +// **另一条路**正在动这个文件」；它不能改变 checkpoint 归属，否则「1 普通回合 +// + 1 并行路」的会话里普通回合的快照键会被升成 `sess#turn-x`，崩溃恢复语义 +// 随之改变（硬要求：恢复语义零变化）； +// - 那条：只有真并行路才有，回答「这条路的 checkpoint 该写哪个键」。 +// +// 把两者焊成一个（复用 ParallelLegFrom）是曾经的真实缺口：普通回合既不登记也不 +// 被检查，它与并行路同文件写仍可能并发。 +type toolLegIDKey struct{} + +// WithToolLegID 把本路的互斥身份挂到 ctx 上（design/15 §2.5.7）。身份由**网关/ +// 编排器**在**每一条在途路**上挂载，取值就是该路的 turnID（编排器生成、唯一， +// 普通回合也有）。它只回答「这条工具调用属于哪条在途路」，与 actor 元数据 / +// 策略引擎无关（项目红线：身份识别绝不走这条通道），也不影响 checkpoint 归属。 +// +// 空 legID 是空操作：子代理、后台思考、未接编排器的调用方不挂该键 ⇒ 工具层读到 +// 空串 ⇒ 完全不做互斥，行为与加入本特性之前逐字相同（兼容硬要求）。 +func WithToolLegID(ctx context.Context, legID string) context.Context { + if ctx == nil || legID == "" { + return ctx + } + return context.WithValue(ctx, toolLegIDKey{}, legID) +} + +// ToolLegIDFrom 返回 ctx 上挂的互斥身份；未挂载时返回 ""（⇒ 不登记也不检查）。 +// 它是 WithToolLegID 的读侧，只被循环的工具执行点读一次（tools.go 的 invokeOne +// 把它桥接成 tool.WithLegID —— 工具侧凭 tool.LegIDFrom 取到）。 +func ToolLegIDFrom(ctx context.Context) string { + if ctx == nil { + return "" + } + s, _ := ctx.Value(toolLegIDKey{}).(string) + return s +} diff --git a/internal/runtime/loop/tools.go b/internal/runtime/loop/tools.go index 5b7226b..cbcd269 100644 --- a/internal/runtime/loop/tools.go +++ b/internal/runtime/loop/tools.go @@ -299,10 +299,16 @@ func (l *Loop) invokeOne(ctx context.Context, d *tool.Dispatcher, tc agent.ToolC } _, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool) l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionStart, model.ToolPhaseRunning) - // 路标识桥（design/15 §2.5.7、主人裁决 3）：网关把「本回合属于哪条并发路」经 - // loop.WithParallelLeg 挂在 ctx 上，工具层按 tool.WithLegID 读它做文件级互斥 - // （工具侧唯一来源）；未挂载 ⇒ 空串 ⇒ 完全不做互斥，行为逐字不变。 - result, err := d.Invoke(tool.WithLegID(ctx, ParallelLegFrom(ctx)), tc.Name, tc.Args) + // 路标识桥（design/15 §2.5.7、主人裁决 3）：网关在**每一条在途路**上把互斥 + // 身份经 loop.WithToolLegID 挂在 ctx 上（普通回合 / 并行路 / 队首续跑路都挂， + // 取值 = 该路的 turnID），这里把它桥接成工具层的 tool.WithLegID —— 工具侧 + // 凭 tool.LegIDFrom 读到「我是哪一路」做文件级互斥；未挂载 ⇒ 空串 ⇒ 完全 + // 不做互斥，行为逐字不变（兼容硬要求）。 + // + // 刻意**不**读 ParallelLegFrom：那条通道只有真并行路才有、且只决定 checkpoint + // 归属（checkpoint.go 的 checkpointAgentID），拿它当互斥身份会让普通回合在 + // 「1 普通回合 + 1 并行路」时完全不受保护。 + result, err := d.Invoke(tool.WithLegID(ctx, ToolLegIDFrom(ctx)), tc.Name, tc.Args) phase := model.ToolPhaseDone if err != nil { phase = model.ToolPhaseError diff --git a/internal/runtime/tool/fileclaims.go b/internal/runtime/tool/fileclaims.go index 7f6dcbf..8011d71 100644 --- a/internal/runtime/tool/fileclaims.go +++ b/internal/runtime/tool/fileclaims.go @@ -151,8 +151,8 @@ func NewFileClaimer() FileClaimer { // 返回带文件状态的 ErrFileBusy（直接回灌给模型）。它**绝不等待**：判忙即返回， // 由模型稍后重试（主人裁决 3 的「不静默、不排队死等」）。 // -// 空 legID 或空目标是空操作：单路 / 子代理 / 后台思考（没有并发路）不登记也不 -// 检查，行为与加入本特性之前逐字相同（兼容硬要求）。 +// 空 legID 或空目标是空操作：子代理 / 后台思考 / 未接编排器的调用方（ctx 上没有 +// 互斥身份）不登记也不检查，行为与加入本特性之前逐字相同（兼容硬要求）。 // // 同一路重复 Claim 同一目标幂等；一路自己先读后写同一文件不会被自己挡住 // （否则一条路自己的多步工具调用会互相锁死）。 @@ -318,8 +318,8 @@ func safeFileTargets(log *slog.Logger, name string, t sdk.Tool, args any) (out [ // 在执行后 defer 调用它：成功 / 报错 / 超时 / panic 都要释放，否则一个中途报错 // 的路会把文件永久锁住）。 // -// 返回 nil 释放函数表示本次调用不参与互斥：未装占用表、未挂路标识（单路 / -// 子代理 / 未接线，兼容硬要求）、或工具没有声明目标文件。 +// 返回 nil 释放函数表示本次调用不参与互斥：未装占用表、ctx 上没有互斥身份 +// （子代理 / 后台思考 / 未接编排器，兼容硬要求）、或工具没有声明目标文件。 // // 被另一条路占住时**不等待、不排队**，直接把带文件状态的 ErrFileBusy 交给模型 // （主人裁决 3）。注意它**不包成 ToolError**：占用判定发生在执行之前，调用根本 diff --git a/internal/runtime/tool/legfile.go b/internal/runtime/tool/legfile.go index 3f58599..014d6ea 100644 --- a/internal/runtime/tool/legfile.go +++ b/internal/runtime/tool/legfile.go @@ -68,15 +68,20 @@ type FileClaimer interface { Release(ctx context.Context, legID, target string) } -// legIDKey 是「本回合属于哪条并发路」的 context key。 +// legIDKey 是「本回合属于哪条在途路（并发路）」的 context key。 type legIDKey struct{} -// WithLegID 把本回合所属的并发路标识挂到 ctx 上。循环在 Run 顶部按 ctx 上挂的 -// 路标识播种一次（单路 / 子代理 / 后台思考为空串 ⇒ 不挂载），工具经 -// Dispatcher.Invoke 的 ctx 读到它，从而知道「我是哪一路」——这是文件级占用 -// 登记的唯一来源，绝不从模型输出推断。 +// WithLegID 把本路（在途回合）的互斥身份挂到 ctx 上。身份由**网关/编排器**在 +// **每一条在途路**上挂载（普通回合 / 并行路 / 队首续跑路，取值 = 该路的 turnID）， +// 循环的工具执行点（loop/tools.go 的 invokeOne）按 loop.ToolLegIDFrom 读到它并 +// 桥接到本键；工具经 Dispatcher.Invoke 的 ctx 读到它，从而知道「我是哪一路」—— +// 这是文件级占用登记的唯一来源，绝不从模型输出推断。 // -// 空 legID 是空操作：单路时未挂载，读取侧返回空串，与加入本特性之前逐字相同。 +// 它与 checkpoint 归属（loop.WithParallelLeg，只有真并行路才有）是**两条独立 +// 通道**：那条只回答「快照该写哪个键」，这条只回答「有没有另一路正在动这个文件」。 +// +// 空 legID 是空操作：子代理 / 后台思考 / 未接编排器的调用方未挂载，读取侧返回 +// 空串，与加入本特性之前逐字相同。 func WithLegID(ctx context.Context, legID string) context.Context { if ctx == nil || legID == "" { return ctx @@ -84,8 +89,12 @@ func WithLegID(ctx context.Context, legID string) context.Context { return context.WithValue(ctx, legIDKey{}, legID) } -// LegIDFrom 返回 ctx 上挂的并发路标识；未挂载（单路、子代理、未接线）时返回 -// 空串。空串表示「本回合不与其他路并行」，此时文件级互斥**无需生效**。 +// LegIDFrom 返回 ctx 上挂的互斥身份；未挂载（子代理、后台思考、未接编排器的 +// 调用方）时返回空串。空串 ⇒ 不登记也不检查，行为逐字不变（兼容硬要求）。 +// +// ⚠️ 空串**不**等于「单路」：接上编排器之后，单条普通回合也有自己的身份（它的 +// turnID）—— 同文件的两条**不同**在途路正是靠它才互相判忙，而同一路自己的多次 +// 调用凭同一个值互相放行（Claim 对同一 legID 幂等）。 func LegIDFrom(ctx context.Context) string { if ctx == nil { return "" diff --git a/internal/services/gateway/session_legmutex_test.go b/internal/services/gateway/session_legmutex_test.go new file mode 100644 index 0000000..2ed22b3 --- /dev/null +++ b/internal/services/gateway/session_legmutex_test.go @@ -0,0 +1,390 @@ +package gateway + +import ( + "context" + "fmt" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" + "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" +) + +// 本文件是「工具互斥身份由网关在**每一条在途路**上盖」的端到端回归（design/15 +// §2.5.7、主人 2026-10-05 裁决 3）。 +// +// 被修掉的缺口：「1 普通回合 + 1 并行路」的会话里，普通回合既不登记也不被检查 +// —— 它与并行路同文件写仍会并发。根因是两条**用途不同的** ctx 通道被焊在一起： +// checkpoint 归属键（loop.WithParallelLeg）只有真并行路才有，而它同时被当成了 +// 工具互斥身份。现在互斥身份走独立通道（loop.WithToolLegID，每一路都盖）。 +// +// 断言走**真路径**：真编排器（真路登记、真 pending、真取消链）→ **真 loop** +// （真模型步、真工具批次）→ 真工具分发器（真占用表）—— 只有模型适配器是脚本化的。 + +// legMutexProbeTimeout 是「另一条路必须已经拿到判忙结果」的上界（只把错误实现暴露 +// 成失败，不参与竞态）。 +const legMutexProbeTimeout = 5 * time.Second + +// legMutexTool 是实现了 tool.FileTargeter 的 mock 写类工具：声明自己操作的文件、 +// 可阻塞执行，并记录它在 Execute 里读到的工具互斥身份（用来钉住身份真的到了工具）。 +type legMutexTool struct { + name string + file string + entered chan struct{} // 非 nil：进入 Execute 时关闭一次 + hold chan struct{} // 非 nil：进入后等它关闭（ctx 取消也返回） + once sync.Once + + mu sync.Mutex + legs []string +} + +func (m *legMutexTool) Name() string { return m.name } + +func (m *legMutexTool) Description() string { return "leg mutex mock tool" } + +func (m *legMutexTool) Parameters() (map[string]any, error) { + return map[string]any{ + "type": "object", + "properties": map[string]any{"path": map[string]any{"type": "string"}}, + }, nil +} + +// FileTargets 实现 tool.FileTargeter：本次调用要**写**自己那个文件。 +func (m *legMutexTool) FileTargets(any) []tool.FileTarget { + return []tool.FileTarget{{Op: tool.FileOpWrite, Path: m.file}} +} + +func (m *legMutexTool) Execute(ctx context.Context, _ any) (any, error) { + m.mu.Lock() + m.legs = append(m.legs, tool.LegIDFrom(ctx)) + m.mu.Unlock() + if m.entered != nil { + m.once.Do(func() { close(m.entered) }) + } + if m.hold != nil { + select { + case <-m.hold: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + return "written", nil +} + +// executedLegIDs 返回 Execute 里读到的互斥身份快照（每个元素 = 一次真的执行）。 +func (m *legMutexTool) executedLegIDs() []string { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.legs...) +} + +// legMutexAdapter 是脚本化的模型适配器：按**全局**步序返回一批工具调用 +// （script[i] = 第 i+1 步要请求的工具名），脚本越界之后的步返回收尾文本。 +// +// 两路共享同一个适配器，因此步序必须由测试钉成确定的：先用「第一路阻塞在工具里」 +// 把它的下一个模型步卡住，再投递第二路，于是 step1 必属于第一路、step2 必属于 +// 第二路。参数一律 `{"path":"notes.txt"}`：目标文件由工具**自己**声明 +// （tool.FileTargeter），所以此处不需要真实路径，也避开了 Windows 路径转义。 +type legMutexAdapter struct { + mu sync.Mutex + steps int + reqs []model.Request + script [][]string +} + +func (a *legMutexAdapter) Name() string { return "leg-mutex-script" } + +func (a *legMutexAdapter) Chat(_ context.Context, req model.Request, _ model.Options) (model.Response, error) { + a.mu.Lock() + a.steps++ + step := a.steps + a.reqs = append(a.reqs, req) + a.mu.Unlock() + if step <= len(a.script) { + return model.Response{Content: toolCallBatchJSON(a.script[step-1])}, nil + } + return model.Response{Content: fmt.Sprintf("第 %d 步收尾", step)}, nil +} + +func (a *legMutexAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return a.Chat(ctx, req, model.Options{}) +} + +// sawText 报告任何一次模型调用收到的窗口里是否出现过 want（工具结果文本也在这里， +// 因此它能证明「判忙/放行」真的回灌到了模型）。 +func (a *legMutexAdapter) sawText(want string) bool { + a.mu.Lock() + defer a.mu.Unlock() + for _, req := range a.reqs { + for _, m := range req.Messages { + if strings.Contains(messageText(m), want) { + return true + } + } + } + return false +} + +// toolCallBatchJSON 把一批工具名渲染成模型请求工具调用的 JSON。 +func toolCallBatchJSON(names []string) string { + calls := make([]string, 0, len(names)) + for _, n := range names { + calls = append(calls, fmt.Sprintf(`{"name":%q,"args":{"path":"notes.txt"}}`, n)) + } + return `{"tool_calls":[` + strings.Join(calls, ",") + `]}` +} + +// legChannelRecord 是一路的 ctx 上三条通道的取值（按 Run 进入顺序记录）。 +type legChannelRecord struct { + turnID string // loop.WithTurnID：本路的权威回合标识 + toolLeg string // loop.WithToolLegID：工具互斥身份（每一路都应有） + parallel string // loop.WithParallelLeg：checkpoint 归属键（只有并行路应有） +} + +// legMutexFixture 装配「真 loop + 真工具分发器 + 真编排器」三条腿，只把模型适配器 +// 换成脚本。 +type legMutexFixture struct { + o *Orchestrator + ad *legMutexAdapter + mu sync.Mutex + leg []legChannelRecord +} + +// newLegMutexFixture 构建夹具。Run 刻意与生产同形（server.go 的 newOrchestrator → +// hub.runTurn）：先把 env 挂到 ctx（loop.WithSessionEnv），再把编排器给的 ctx 原样 +// 交给 loop.Run —— 互斥身份能不能到工具，就取决于这条 ctx 链。 +func newLegMutexFixture(t *testing.T, script [][]string, tools ...sdk.Tool) *legMutexFixture { + t.Helper() + ad := &legMutexAdapter{script: script} + dispatcher := newSteerDispatcher(t, tools...) + ag, err := agent.New("你是测试助手。", dispatcher, nil, agent.WithModel(ad, nil)) + if err != nil { + t.Fatalf("agent.New: %v", err) + } + realLoop := loop.New(loop.Options{Agent: ag, Tools: dispatcher, MaxRounds: 8, Log: gwTestLogger()}) + f := &legMutexFixture{ad: ad} + f.o = NewOrchestrator(OrchestratorOptions{ + Run: func(ctx context.Context, sessionID string, messages []model.Message, env loop.SessionEnv) (*loop.Result, error) { + f.recordLeg(ctx) + return realLoop.Run(loop.WithSessionEnv(ctx, env), sessionID, messages) + }, + Emit: func(string, string, map[string]any) {}, + EmitTurn: func(string, string, string, map[string]any) {}, + Log: gwTestLogger(), + MaxPending: 5, + MaxTail: 20, + History: &fakeHistory{}, + MaxParallelTurns: 2, + }) + return f +} + +// recordLeg 记下这一路 ctx 上三条通道的取值。 +func (f *legMutexFixture) recordLeg(ctx context.Context) { + f.mu.Lock() + defer f.mu.Unlock() + f.leg = append(f.leg, legChannelRecord{ + turnID: loop.TurnIDFrom(ctx), + toolLeg: loop.ToolLegIDFrom(ctx), + parallel: loop.ParallelLegFrom(ctx), + }) +} + +// legRecords 返回已记录的路（副本）。 +func (f *legMutexFixture) legRecords() []legChannelRecord { + f.mu.Lock() + defer f.mu.Unlock() + return append([]legChannelRecord(nil), f.leg...) +} + +// waitLegMutex 轮询直到条件成立（超时即失败），使断言不靠睡眠。 +func waitLegMutex(t *testing.T, what string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(legMutexProbeTimeout) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("超时等待：%s", what) +} + +// awaitLegMutexEntered 等工具进入 Execute（确定性同步，不靠 sleep 抢时序）。 +func awaitLegMutexEntered(t *testing.T, ch <-chan struct{}, msg string) { + t.Helper() + select { + case <-ch: + case <-time.After(legMutexProbeTimeout): + t.Fatal(msg) + } +} + +// assertLegChannels 断言两条 ctx 通道**各司其职**：每一路都有互斥身份（取值 = 该路 +// turnID），而 checkpoint 归属键只有并行路才有 —— 普通回合的归属键仍是 sessionID +// （崩溃恢复语义零变化）。 +func assertLegChannels(t *testing.T, normal, parallel legChannelRecord, normalTurn, parallelTurn string) { + t.Helper() + if normal.turnID != normalTurn { + t.Fatalf("普通回合 ctx turnId = %q, want %q", normal.turnID, normalTurn) + } + if normal.toolLeg != normalTurn { + t.Fatalf("普通回合的互斥身份 = %q, want %q（普通回合也必须受文件级互斥保护）", normal.toolLeg, normalTurn) + } + if normal.parallel != "" { + t.Fatalf("普通回合不该有 checkpoint 归属键，得到 %q（归属键 ≠ 互斥身份）", normal.parallel) + } + + if parallel.turnID != parallelTurn { + t.Fatalf("并行路 ctx turnId = %q, want %q", parallel.turnID, parallelTurn) + } + if parallel.toolLeg != parallelTurn { + t.Fatalf("并行路的互斥身份 = %q, want %q", parallel.toolLeg, parallelTurn) + } + if parallel.parallel != parallelTurn { + t.Fatalf("并行路的 checkpoint 归属键 = %q, want %q", parallel.parallel, parallelTurn) + } +} + +// --------------------------------------------------------------------------- +// ① 普通回合 + 并行路同文件 ⇒ 判忙（本次修复的目标） +// --------------------------------------------------------------------------- + +// TestLegMutexNormalTurnBlocksParallelLegOnSameFile 钉住缺口被修掉：普通回合持有 +// 某文件时，同会话的并行路写**同一个**文件必须判忙 —— 它不执行、不排队死等，而是 +// 把中文文件状态作为工具结果回灌给模型（模型据此稍后重试）。 +// +// 修复前这条会红：普通回合没盖互斥身份（既不登记也不被检查），并行路照写不误。 +func TestLegMutexNormalTurnBlocksParallelLegOnSameFile(t *testing.T) { + t.Parallel() + file := filepath.Join(t.TempDir(), "notes.txt") + holder := &legMutexTool{name: "patch-a", file: file, entered: make(chan struct{}), hold: make(chan struct{})} + peer := &legMutexTool{name: "patch-b", file: file} + f := newLegMutexFixture(t, [][]string{{"patch-a"}, {"patch-b"}}, holder, peer) + + first := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv()) + if !first.Immediate || first.TurnID == "" { + t.Fatalf("第一路 = %+v, want Immediate 且带 turnId", first) + } + awaitLegMutexEntered(t, holder.entered, "普通回合没有进入工具执行") + + second := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv()) + if !second.Immediate || !second.Parallel { + t.Fatalf("显式并行提交 = %+v, want Immediate 且 Parallel", second) + } + + // 并行路撞上普通回合持有的同一文件 ⇒ 判忙文案回灌给模型。 + waitLegMutex(t, "并行路应拿到同文件判忙结果", func() bool { return f.ad.sawText("正在被另一路写入") }) + if got := peer.executedLegIDs(); len(got) != 0 { + t.Fatalf("被判忙的并行路绝不该执行写工具，实际执行 %d 次（身份 %v）", len(got), got) + } + if !f.ad.sawText("请稍后再试") { + t.Fatal("回灌给模型的文案必须提示稍后重试（判忙是显式的，不是静默失败）") + } + + recs := f.legRecords() + if len(recs) < 2 { + t.Fatalf("在途路数 = %d, want ≥2", len(recs)) + } + assertLegChannels(t, recs[0], recs[1], first.TurnID, second.TurnID) + + close(holder.hold) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } + // 放行之后普通回合自己正常收尾（判忙只挡另一路，绝不连坐持锁方）。 + if got := holder.executedLegIDs(); len(got) != 1 || got[0] != first.TurnID { + t.Fatalf("执行过的普通回合身份 = %v, want [%s]", got, first.TurnID) + } +} + +// --------------------------------------------------------------------------- +// ② 不同文件 ⇒ 放行（互斥按文件粒度，绝不退化成会话级串行） +// --------------------------------------------------------------------------- + +// TestLegMutexDifferentFilesKeepBothLegsRunning 钉住另一条口径：两条在途路动**不同** +// 文件时照常并行 —— 并行路自己的写工具必须真的执行（而不是被判忙）。 +func TestLegMutexDifferentFilesKeepBothLegsRunning(t *testing.T) { + t.Parallel() + dir := t.TempDir() + holder := &legMutexTool{name: "patch-a", file: filepath.Join(dir, "a.txt"), entered: make(chan struct{}), hold: make(chan struct{})} + peer := &legMutexTool{name: "patch-b", file: filepath.Join(dir, "b.txt")} + f := newLegMutexFixture(t, [][]string{{"patch-a"}, {"patch-b"}}, holder, peer) + + first := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv()) + awaitLegMutexEntered(t, holder.entered, "普通回合没有进入工具执行") + second := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv()) + if !second.Parallel { + t.Fatalf("显式并行提交 = %+v, want Parallel", second) + } + + waitLegMutex(t, "不同文件的并行路应真的执行写工具", func() bool { return len(peer.executedLegIDs()) == 1 }) + if got := peer.executedLegIDs()[0]; got != second.TurnID { + t.Fatalf("并行路执行时读到的互斥身份 = %q, want %q", got, second.TurnID) + } + + close(holder.hold) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } + if f.ad.sawText("正在被另一路写入") { + t.Fatal("不同文件绝不该被判忙（互斥是文件粒度，不是会话级串行）") + } + if got := holder.executedLegIDs(); len(got) != 1 || got[0] != first.TurnID { + t.Fatalf("普通回合执行身份 = %v, want [%s]", got, first.TurnID) + } +} + +// --------------------------------------------------------------------------- +// ④ 单路（只有一个回合）时工具互斥行为不变：同一路自己的多个调用互不判忙 +// --------------------------------------------------------------------------- + +// TestLegMutexSingleLegOwnToolCallsDoNotBlockEachOther 钉住「给普通回合盖上身份」 +// 没有引入自锁：单条路在同一批里请求两个写工具、写**同一个**文件，两次都必须真的 +// 执行（同一路幂等，绝不自己挡自己）—— 互斥只回答「有没有**另一条**路在动它」。 +// +// 它同时钉住单路的 ctx 形状：有互斥身份（取值 = 本路 turnID），没有 checkpoint +// 归属键（归属键仍是 sessionID）。 +func TestLegMutexSingleLegOwnToolCallsDoNotBlockEachOther(t *testing.T) { + t.Parallel() + file := filepath.Join(t.TempDir(), "notes.txt") + a := &legMutexTool{name: "patch-a", file: file} + b := &legMutexTool{name: "patch-b", file: file} + f := newLegMutexFixture(t, [][]string{{"patch-a", "patch-b"}}, a, b) + + sr := f.o.Submit(context.Background(), "web", []model.Message{userMsg("单路")}, loop.OwnerEnv()) + if !sr.Immediate || sr.TurnID == "" { + t.Fatalf("提交 = %+v, want Immediate 且带 turnId", sr) + } + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } + + // 同一批里的两次调用都真的执行了（另一次调用不会因为「文件已被占用」被拦下）。 + if got := a.executedLegIDs(); len(got) != 1 || got[0] != sr.TurnID { + t.Fatalf("第一路自己的第一次写 = %v, want [%s]", got, sr.TurnID) + } + if got := b.executedLegIDs(); len(got) != 1 || got[0] != sr.TurnID { + t.Fatalf("第一路自己的第二次写 = %v, want [%s]（同一路绝不自己挡自己）", got, sr.TurnID) + } + if f.ad.sawText("正在被另一路写入") { + t.Fatal("同一路的多次调用不该互相判忙") + } + + recs := f.legRecords() + if len(recs) != 1 { + t.Fatalf("在途路数 = %d, want 1", len(recs)) + } + if recs[0].toolLeg != sr.TurnID { + t.Fatalf("单路的互斥身份 = %q, want %q", recs[0].toolLeg, sr.TurnID) + } + if recs[0].parallel != "" { + t.Fatalf("单路不该有 checkpoint 归属键，得到 %q（归属键逐字不变 = sessionID）", recs[0].parallel) + } +} diff --git a/internal/services/gateway/session_parallel.go b/internal/services/gateway/session_parallel.go index e16f31f..7ca9a06 100644 --- a/internal/services/gateway/session_parallel.go +++ b/internal/services/gateway/session_parallel.go @@ -22,8 +22,10 @@ import ( // kind:"turn"、并行路 kind:"parallel"，两条路径都发。`steer_ack` 的帧壳 // turnId 由 session_steer.go 的转向游标带出，与这里**取值同源**（leg.turnID）。 // -// 它**不做**（留给后续切片）：写类工具的文件级互斥、第二路的进行态 system 摘要。 -// 预留点见 turnLeg 的字段。 +// 它**不做**（留给后续切片）：第二路的进行态 system 摘要。写类工具的文件级互斥 +// 在**两个地方**落地：互斥**身份**由 beginLegLocked 对每一条在途路盖 +// （loop.WithToolLegID，取值 = 本路 turnID），登记 / 判忙 / 释放由工具侧 +// （internal/runtime/tool/fileclaims.go）在执行前后来做。 // 并发额度的取值口径（design/15 §2.5.7）。 const ( @@ -68,7 +70,9 @@ type turnLeg struct { // 展示口径）。除了发帧，日志也用它。 leg int // parallel 报告本路是否为**显式声明的并行路**：它决定 checkpoint 归属键 - // （单路逐字等于 sessionID，并行路为 {sessionID}#{turnID}）。 + // （单路逐字等于 sessionID，并行路为 {sessionID}#{turnID}）。它**不**决定 + // 工具互斥身份 —— 那条身份每一路都有（loop.WithToolLegID），否则普通回合 + // 与并行路同文件写会并发。 parallel bool // cancel 派生自提交时的 ctx，是**本路独占**的取消链：任一路的注入 / 取消 // 都不影响另一路已发起的工具（I3′）。 @@ -178,20 +182,32 @@ func (o *Orchestrator) startParallelLocked(ctx context.Context, st *sessionState } // beginLegLocked 在状态锁下登记一条新路（design/15 §2.5.9）。它分配该路自己的 -// 世代号、派生**独立的**取消链（I3′），把 turnID 与（并行时的）路标识盖进 runCtx， -// 并把 runCtx 交回调用方 —— 调用方解锁后再起 goroutine。 +// 世代号、派生**独立的**取消链（I3′），把 turnID 与（并行时的）**checkpoint 归属 +// 键**、以及（每一路的）**工具互斥身份**盖进 runCtx，并把 runCtx 交回调用方 —— +// 调用方解锁后再起 goroutine。 // // turnID 由**编排器**生成并由 loop 复用（design/15 §2.5.9 的 ⚠️ 必改项）： // `turn_begin` 帧由编排器在调用 Run 之前发出，若 Run 内部再自己生成一个， // 客户端按 turnId 两级分桶会立即失配。单路时该 id 仍是每回合全新的随机值， // 因此对客户端**无可见差异**。 // +// 两条 ctx 通道在这里分道（见 loop/session_ctx.go）：checkpoint 归属键 +// （loop.WithParallelLeg）**只给并行路**，工具互斥身份（loop.WithToolLegID） +// **每一路都盖**。它们刻意不共用 —— 共用会把普通回合的 checkpoint 键从 +// sessionID 升成 `{sessionID}#{turnID}`，连带改变崩溃恢复语义。 +// // 它同时把会话标成忙（ModeThinking），使 mode 在任一路在途时恒非空闲。 func (o *Orchestrator) beginLegLocked(ctx context.Context, st *sessionState, turnID string, parallel bool, fence turnFence) (*turnLeg, context.Context, context.CancelFunc) { st.gen++ st.legSeq++ runCtx, cancel := context.WithCancel(ctx) runCtx = loop.WithTurnID(runCtx, turnID) + // 工具互斥身份：**每一条在途路都盖**（普通回合 / 并行路 / 队首续跑路）。 + // 漏掉普通回合的代价是真实的：「1 普通回合 + 1 并行路」时普通回合既不登记 + // 也不被检查，它与并行路同文件写仍会并发（design/15 §2.5.7 的文件级互斥 + // 因此形同虚设）。取值就是本路的 turnID：路内唯一且稳定，于是同一条路自己 + // 的多次工具调用不会互相判忙，而两条不同的路一定取到不同的值。 + runCtx = loop.WithToolLegID(runCtx, turnID) leg := &turnLeg{turnID: turnID, gen: st.gen, leg: st.legSeq, parallel: parallel, cancel: cancel, fence: fence} if parallel { runCtx = loop.WithParallelLeg(runCtx, turnID) @@ -305,6 +321,10 @@ func userMessageIDs(msgs []history.Message) []string { // **服务端权威的**消息 id（客户端靠它把这些气泡置为「已受理·回答中」并按 turnId // 分桶），因此它只能在落库之后发。 // +// ctx 上会先挂上本路的回合标识（design/15 §2.5.9 的两段式第一段）：这一刻落库的 +// 用户消息随**写入**带上 turn_id，于是它与该路的助手回复同值 —— 客户端重连后靠 +// 这一列精确配对「回复 ↔ 提问」，而流帧早就丢了。 +// // 这里落库的只有**此刻到达**的那些消息（已播种的转录早就持久了，重追加会重复整个 // 对话），且只写**引用** —— 抽取结论要等回合跑起来才有，由回合结算补进同一列 // （两段式见 session_document_status.go）。 @@ -312,6 +332,7 @@ func userMessageIDs(msgs []history.Message) []string { // 单路与并行路共用这一段，于是两条启动路径的帧序完全同形 —— 客户端只需要实现一 // 遍「turn_begin → 灰态」。 func (o *Orchestrator) publishLegStart(ctx context.Context, sessionID string, leg *turnLeg, messages []model.Message, deviceID, mode, personaID string, activeTurns int) { + ctx = withTurnID(ctx, leg.turnID) persisted := o.persistUserMessages(ctx, sessionID, messages, deviceID, mode, personaID) o.rememberTurnDocuments(sessionID, leg.gen, persisted) o.rememberVoiceTurn(sessionID, leg.gen, voiceReplySinkFrom(ctx)) diff --git a/internal/services/gateway/session_parallel_test.go b/internal/services/gateway/session_parallel_test.go index dbca4dd..80993aa 100644 --- a/internal/services/gateway/session_parallel_test.go +++ b/internal/services/gateway/session_parallel_test.go @@ -266,6 +266,15 @@ func TestParallelOmittedKeepsQueueingBehaviour(t *testing.T) { if loop.TurnIDFrom(calls[1].ctx) == loop.TurnIDFrom(calls[0].ctx) { t.Fatal("续跑路必须是一条**新路**，turnId 不能与上一路相同") } + // 工具互斥身份是**另一条通道**（design/15 §2.5.7）：续跑路同样必须带上它， + // 取值就是它自己的 turnId —— 否则串行续跑路在写类工具上不受保护。它与 + // checkpoint 归属键无关，所以上面那条「空串」断言不该被它推翻。 + if got := loop.ToolLegIDFrom(calls[1].ctx); got != loop.TurnIDFrom(calls[1].ctx) { + t.Fatalf("续跑路的互斥身份 = %q, want %q（每一路都要盖）", got, loop.TurnIDFrom(calls[1].ctx)) + } + if loop.ToolLegIDFrom(calls[1].ctx) == loop.ToolLegIDFrom(calls[0].ctx) { + t.Fatal("两条路的互斥身份必须不同，否则同文件写会被当成「同一路自己的调用」而放行") + } drains := f.sink.ofType(EventPendingDrain) if len(drains) != 1 || stateInt(drains[0], "activeTurns") != 1 { t.Fatalf("pending_drain = %v, want 一帧且 activeTurns=1", drains)