fix(gateway,loop,tool): 解耦「工具互斥身份」与「checkpoint 归属」(修普通回合不受保护)
缺口:「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 三包全绿;诚实边界含一个既有竞态(同一路并发登记共享,建议后续小切片)。
This commit is contained in:
1 parent
d64b26d499
commit
ef3dcb699b
9 files changed
+673
-53
No files matched your search
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 只命中自己的那份 —— 任何一路都绝不会把对方的崩溃快照
|
||||
// 当成自己的续接。
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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**:占用判定发生在执行之前,调用根本
|
||||
|
||||
@@ -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 ""
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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))
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user