diff --git a/internal/runtime/loop/fileclaim_leg_test.go b/internal/runtime/loop/fileclaim_leg_test.go new file mode 100644 index 0000000..994377c --- /dev/null +++ b/internal/runtime/loop/fileclaim_leg_test.go @@ -0,0 +1,191 @@ +package loop + +import ( + "context" + "errors" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "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,于是同一文件的第二条路拿到 +// tool.ErrFileBusy,并以**回灌给模型的工具结果文本**形式浮出来(显式、可消化)。 +// +// 单路口径同样钉住:ctx 上没有路标识(普通回合 / 子代理 / 后台思考)时,同文件 +// 的两条调用照跑 —— 完全不做互斥,行为逐字不变。 + +// legFileProbeTimeout 是「第二路必须立刻拿到判忙结果」的上界(只用来把阻塞式 +// 实现暴露成失败,不参与竞态)。 +const legFileProbeTimeout = 5 * time.Second + +// legFileTool 是实现了 tool.FileTargeter 的 mock 工具:声明本次要写的文件, +// 可阻塞执行,并记录它在 Execute 里读到的路标识(用来直接钉住桥的存在)。 +type legFileTool struct { + name string + file string + entered chan struct{} + hold chan struct{} + once sync.Once + + mu sync.Mutex + legs []string +} + +func (m *legFileTool) Name() string { return m.name } + +func (m *legFileTool) Description() string { return "leg file mock tool" } + +func (m *legFileTool) Parameters() (map[string]any, error) { + return map[string]any{ + "type": "object", + "properties": map[string]any{"path": map[string]any{"type": "string"}}, + "required": []any{"path"}, + }, nil +} + +// FileTargets 实现 tool.FileTargeter:本次调用要写 file。 +func (m *legFileTool) FileTargets(any) []tool.FileTarget { + return []tool.FileTarget{{Op: tool.FileOpWrite, Path: m.file}} +} + +func (m *legFileTool) 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 "ok", nil +} + +// seenLegs 返回 Execute 里读到的路标识快照。 +func (m *legFileTool) seenLegs() []string { + m.mu.Lock() + defer m.mu.Unlock() + return append([]string(nil), m.legs...) +} + +// awaitEntered 等工具进入 Execute(确定性同步,不靠 sleep 抢时序)。 +func awaitEntered(t *testing.T, ch <-chan struct{}, msg string) { + t.Helper() + select { + case <-ch: + case <-time.After(legFileProbeTimeout): + t.Fatal(msg) + } +} + +// TestInvokeOne_LegIDReachesTool 直接钉住桥:ctx 上挂了并发路标识 ⇒ 工具在 +// Execute 里读到同一个值;没挂(单路)⇒ 读到空串(⇒ 完全不做互斥)。 +func TestInvokeOne_LegIDReachesTool(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(WithParallelLeg(context.Background(), "turn-x"), d, call); out.Err != nil { + t.Fatalf("并行路调用: %v", out.Err) + } + if out := l.invokeOne(context.Background(), d, call); out.Err != nil { + 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) + } +} + +// TestInvokeOne_ParallelLegSameFileBusySurfacesToModel 钉住端到端的核心行为: +// 两路(不同路标识)同文件 ⇒ 第二路**不执行**、**不等待**,拿到 tool.ErrFileBusy, +// 且回灌给模型的文本里带中文文件状态(模型据此自己决定稍后重试)。 +func TestInvokeOne_ParallelLegSameFileBusySurfacesToModel(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{})} + peer := &legFileTool{name: "patch-b", file: file} + d := dispatcherWith(t, holder, peer) + l := New(Options{Tools: d}) + + callA := agent.ToolCall{Name: "patch-a", Args: map[string]any{"path": file}} + callB := agent.ToolCall{Name: "patch-b", Args: map[string]any{"path": file}} + + first := make(chan ToolOutcome, 1) + go func() { + first <- l.invokeOne(WithParallelLeg(context.Background(), "turn-a"), d, callA) + }() + awaitEntered(t, holder.entered, "第一路没有进入工具执行") + + done := make(chan ToolOutcome, 1) + go func() { + done <- l.invokeOne(WithParallelLeg(context.Background(), "turn-b"), d, callB) + }() + var outB ToolOutcome + select { + case outB = <-done: + case <-time.After(legFileProbeTimeout): + t.Fatal("第二路阻塞等待了 —— 口径是不排队死等,判忙即返回") + } + + if !errors.Is(outB.Err, tool.ErrFileBusy) { + t.Fatalf("第二路错误 = %v, want tool.ErrFileBusy", outB.Err) + } + if outB.ErrKind != tool.KindExecution { + t.Fatalf("判忙必须归类为执行面失败才能被回灌,得到 %q", outB.ErrKind) + } + text := toolResultText(callB, outB) + for _, want := range []string{"请稍后再试", "正在被另一路写入", "notes.txt"} { + if !strings.Contains(text, want) { + t.Fatalf("回灌给模型的文本缺少 %q: %s", want, text) + } + } + + close(holder.hold) + select { + case outA := <-first: + if outA.Err != nil { + t.Fatalf("持锁那一路应正常完成: %v", outA.Err) + } + case <-time.After(legFileProbeTimeout): + t.Fatal("第一路没有结束") + } +} + +// TestInvokeOne_SingleLegSameFileRunsConcurrently 钉住兼容硬要求③:**无路标识 +// ⇒ 完全不做互斥**(单路 / 既有路径逐字不变)—— 同文件的两条调用同时执行都成功。 +func TestInvokeOne_SingleLegSameFileRunsConcurrently(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}) + ctx := context.Background() + + first := make(chan ToolOutcome, 1) + go func() { + first <- l.invokeOne(ctx, d, agent.ToolCall{Name: "patch-a", Args: map[string]any{"path": file}}) + }() + 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) + } + close(blocker.hold) + if out := <-first; out.Err != nil { + t.Fatalf("无路标识的调用应正常完成: %v", out.Err) + } +} diff --git a/internal/runtime/loop/tools.go b/internal/runtime/loop/tools.go index d0472ac..5b7226b 100644 --- a/internal/runtime/loop/tools.go +++ b/internal/runtime/loop/tools.go @@ -299,7 +299,10 @@ 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) - result, err := d.Invoke(ctx, tc.Name, tc.Args) + // 路标识桥(design/15 §2.5.7、主人裁决 3):网关把「本回合属于哪条并发路」经 + // loop.WithParallelLeg 挂在 ctx 上,工具层按 tool.WithLegID 读它做文件级互斥 + // (工具侧唯一来源);未挂载 ⇒ 空串 ⇒ 完全不做互斥,行为逐字不变。 + result, err := d.Invoke(tool.WithLegID(ctx, ParallelLegFrom(ctx)), tc.Name, tc.Args) phase := model.ToolPhaseDone if err != nil { phase = model.ToolPhaseError diff --git a/internal/runtime/tool/fileclaim_dispatch_test.go b/internal/runtime/tool/fileclaim_dispatch_test.go new file mode 100644 index 0000000..d3238e7 --- /dev/null +++ b/internal/runtime/tool/fileclaim_dispatch_test.go @@ -0,0 +1,441 @@ +package tool + +import ( + "context" + "errors" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" +) + +// 本文件是**接线**的回归(design/15 §2.5.7、主人 2026-10-05 裁决 3):工具声明的 +// 目标文件在执行前登记、执行后(defer)释放,被另一条在途路占住时把带文件状态的 +// ErrFileBusy 直接回灌给模型。 +// +// 三条兼容硬要求在这里被钉死: +// - 无路标识(单路 / 既有路径)⇒ 完全不做互斥,行为逐字不变; +// - 工具没实现 FileTargeter(今天所有工具)⇒ 不参与互斥,行为逐字不变; +// - 声明面 panic 也不影响调用本身(声明是可选扩展点)。 + +// shortToolTimeout 用于把「超时路径必须释放占用」压成一个可断言的短窗口。 +const shortToolTimeout = 50 * time.Millisecond + +// declaringTool 是一个实现了 FileTargeter 的 sdk.Tool:能声明本次的目标文件, +// 也能阻塞在 Execute 里 —— 使「一路持有、另一路撞上」成为**确定性**事件,不靠 +// sleep 抢时序。 +type declaringTool struct { + name string + targets func(args any) []FileTarget + declareBoom bool // FileTargets 里 panic(声明面出错) + execBoom bool // Execute 里 panic(执行面出错) + execErr error // Execute 返回错误 + entered chan struct{} // 非 nil:进入 Execute 时关闭一次 + hold chan struct{} // 非 nil:进入后等它关闭(ctx 取消也返回) + once sync.Once +} + +func (d *declaringTool) Name() string { return d.name } + +func (d *declaringTool) Description() string { return "declaring mock tool" } + +func (d *declaringTool) Parameters() (map[string]any, error) { + return map[string]any{ + "type": "object", + "properties": map[string]any{"path": map[string]any{"type": "string"}}, + "required": []any{"path"}, + }, nil +} + +// FileTargets 实现 FileTargeter:声明本次调用要碰的文件。 +func (d *declaringTool) FileTargets(args any) []FileTarget { + if d.declareBoom { + panic("declare boom") + } + if d.targets == nil { + return nil + } + return d.targets(args) +} + +func (d *declaringTool) Execute(ctx context.Context, _ any) (any, error) { + if d.execBoom { + panic("exec boom") + } + if d.entered != nil { + d.once.Do(func() { close(d.entered) }) + } + if d.hold != nil { + select { + case <-d.hold: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if d.execErr != nil { + return nil, d.execErr + } + return "ok", nil +} + +// awaitClose 等一个信号通道关闭(不靠 sleep 抢时序)。 +func awaitClose(t *testing.T, ch <-chan struct{}, msg string) { + t.Helper() + select { + case <-ch: + case <-time.After(claimProbeTimeout): + t.Fatal(msg) + } +} + +// claimTableLen 返回占用表当前的键数(白盒):用于断言「无路标识 / 未声明目标 +// 不留痕迹」与「释放后有界」。 +func claimTableLen(t *testing.T, d *Dispatcher) int { + t.Helper() + c, ok := d.claimerRef().(*fileClaims) + if !ok { + t.Fatal("Dispatcher 未装配进程内占用表") + } + c.mu.Lock() + defer c.mu.Unlock() + return len(c.held) +} + +// invokeLeg 在指定并发路上调用工具。 +func invokeLeg(d *Dispatcher, leg, name string, args map[string]any) (any, error) { + return d.Invoke(WithLegID(context.Background(), leg), name, args) +} + +// declareWrite 造一个只声明「写某个文件」的声明函数。 +func declareWrite(file string) func(args any) []FileTarget { + return func(any) []FileTarget { return []FileTarget{{Op: FileOpWrite, Path: file}} } +} + +// TestInvoke_FileBusyIsExplicitAndNonBlocking 钉住主人裁决 3 的核心行为:第二路 +// **不执行**、**不等待**,拿到的是能被 errors.Is 识别的 ErrFileBusy,且文案带 +// 文件状态(含路径与「正在被另一路写入」),可直接回灌给模型。 +func TestInvoke_FileBusyIsExplicitAndNonBlocking(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + file := filepath.Join(t.TempDir(), "notes.txt") + holder := &declaringTool{name: "patch", targets: declareWrite(file), entered: make(chan struct{}), hold: make(chan struct{})} + if err := d.Register(holder); err != nil { + t.Fatalf("Register: %v", err) + } + + type result struct { + val any + err error + } + got := make(chan result, 1) + go func() { + val, err := invokeLeg(d, "turn-a", "patch", map[string]any{"path": file}) + got <- result{val: val, err: err} + }() + awaitClose(t, holder.entered, "第一路没有进入 Execute") + + done := make(chan result, 1) + go func() { + val, err := invokeLeg(d, "turn-b", "patch", map[string]any{"path": file}) + done <- result{val: val, err: err} + }() + var second result + select { + case second = <-done: + case <-time.After(claimProbeTimeout): + t.Fatal("第二路阻塞等待了 —— 主人口径是不排队死等,判忙即返回") + } + + if !errors.Is(second.err, ErrFileBusy) { + t.Fatalf("第二路错误 = %v,want ErrFileBusy", second.err) + } + if second.val != nil { + t.Fatalf("被判忙的调用绝不该执行(拿到结果 %v)", second.val) + } + msg := second.err.Error() + for _, want := range []string{"请稍后再试", "正在被另一路写入", "notes.txt", "本次写入无法进行"} { + if !strings.Contains(msg, want) { + t.Fatalf("回灌给模型的文案缺少 %q: %s", want, msg) + } + } + // 判忙发生在执行之前、且**不包 ToolError**:同一句话只该出现一次 + // (ToolError 的 message+cause 渲染会把它打印两遍,白烧模型的 token)。 + if got := strings.Count(msg, "请稍后再试"); got != 1 { + t.Fatalf("面向模型的文案重复渲染了 %d 次: %s", got, msg) + } + + close(holder.hold) + select { + case first := <-got: + if first.err != nil { + t.Fatalf("持锁那一路应正常完成: %v", first.err) + } + case <-time.After(claimProbeTimeout): + t.Fatal("第一路没有结束") + } +} + +// TestInvoke_DifferentFilesRunParallel 钉住「不是同一个文件 ⇒ 允许并行操作」。 +func TestInvoke_DifferentFilesRunParallel(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + dir := t.TempDir() + blocker := &declaringTool{name: "block", targets: declareWrite(filepath.Join(dir, "a.txt")), entered: make(chan struct{}), hold: make(chan struct{})} + other := &declaringTool{name: "other", targets: declareWrite(filepath.Join(dir, "b.txt"))} + for _, tk := range []*declaringTool{blocker, other} { + if err := d.Register(tk); err != nil { + t.Fatalf("Register %s: %v", tk.Name(), err) + } + } + + done := make(chan error, 1) + go func() { + _, err := invokeLeg(d, "turn-a", "block", map[string]any{"path": "a.txt"}) + done <- err + }() + awaitClose(t, blocker.entered, "第一路没有进入 Execute") + + if _, err := invokeLeg(d, "turn-b", "other", map[string]any{"path": "b.txt"}); err != nil { + t.Fatalf("不同文件必须允许并行操作,得到 %v", err) + } + close(blocker.hold) + if err := <-done; err != nil { + t.Fatalf("持锁那一路应正常完成: %v", err) + } +} + +// TestInvoke_TwoReadsRunParallel 钉住第 4 条的结论(**两个读允许并行**):同一 +// 文件、同一路标识不同,读与读互不阻挡;而写同一文件被判忙。 +func TestInvoke_TwoReadsRunParallel(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + file := filepath.Join(t.TempDir(), "notes.txt") + read := func(string) func(args any) []FileTarget { + return func(any) []FileTarget { return []FileTarget{{Op: FileOpRead, Path: file}} } + } + holder := &declaringTool{name: "read-holder", targets: read(file), entered: make(chan struct{}), hold: make(chan struct{})} + peer := &declaringTool{name: "read-peer", targets: read(file)} + if err := d.Register(holder); err != nil { + t.Fatalf("Register: %v", err) + } + if err := d.Register(peer); err != nil { + t.Fatalf("Register: %v", err) + } + + done := make(chan error, 1) + go func() { + _, err := invokeLeg(d, "turn-a", "read-holder", map[string]any{"path": file}) + done <- err + }() + awaitClose(t, holder.entered, "第一路没有进入 Execute") + + if _, err := invokeLeg(d, "turn-b", "read-peer", map[string]any{"path": file}); err != nil { + t.Fatalf("同一文件的两个读必须允许并行,得到 %v", err) + } + close(holder.hold) + if err := <-done; err != nil { + t.Fatalf("读那一路应正常完成: %v", err) + } +} + +// TestInvoke_NoLegIDNoMutualExclusion 钉住兼容硬要求③:**无路标识 ⇒ 完全不做 +// 互斥**。同一文件在同一时刻被两条无路标识的调用动,两边都成功,占用表里也 +// 不留任何痕迹(单路 / 既有路径行为逐字不变)。 +func TestInvoke_NoLegIDNoMutualExclusion(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + file := filepath.Join(t.TempDir(), "notes.txt") + blocker := &declaringTool{name: "block", targets: declareWrite(file), entered: make(chan struct{}), hold: make(chan struct{})} + peer := &declaringTool{name: "peer", targets: declareWrite(file)} + for _, tk := range []*declaringTool{blocker, peer} { + if err := d.Register(tk); err != nil { + t.Fatalf("Register %s: %v", tk.Name(), err) + } + } + + done := make(chan error, 1) + go func() { + _, err := d.Invoke(context.Background(), "block", map[string]any{"path": file}) + done <- err + }() + awaitClose(t, blocker.entered, "无路标识的调用没有进入 Execute") + + if _, err := d.Invoke(context.Background(), "peer", map[string]any{"path": file}); err != nil { + t.Fatalf("无路标识时完全不做互斥,得到 %v", err) + } + close(blocker.hold) + if err := <-done; err != nil { + t.Fatalf("无路标识的调用应正常完成: %v", err) + } + if got := claimTableLen(t, d); got != 0 { + t.Fatalf("无路标识不该登记任何占用,残留 %d 个键", got) + } +} + +// TestInvoke_NoDeclarationUnchanged 钉住兼容硬要求:没实现 FileTargeter 的工具 +// (今天所有工具)即使带路标识也完全不受影响。 +func TestInvoke_NoDeclarationUnchanged(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + plain := mockTool{name: "plain", params: map[string]any{"type": "object"}} + if err := d.Register(plain); err != nil { + t.Fatalf("Register: %v", err) + } + for _, leg := range []string{"turn-a", "turn-b"} { + if _, err := invokeLeg(d, leg, "plain", map[string]any{}); err != nil { + t.Fatalf("未声明目标的工具不该参与互斥(leg=%s): %v", leg, err) + } + } + if got := claimTableLen(t, d); got != 0 { + t.Fatalf("未声明目标的工具不该留下占用,残留 %d 个键", got) + } +} + +// TestInvoke_ReleaseOnEveryExitPath 钉住释放:成功 / 工具报错 / 工具 panic / 超时 +// 四条路径都必须释放占用 —— 否则一个中途报错的路会把文件永久锁住。 +func TestInvoke_ReleaseOnEveryExitPath(t *testing.T) { + t.Parallel() + cases := []struct { + name string + tk *declaringTool + }{ + {"success", &declaringTool{name: "tk"}}, + {"error", &declaringTool{name: "tk", execErr: errors.New("boom")}}, + {"panic", &declaringTool{name: "tk", execBoom: true}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + file := filepath.Join(t.TempDir(), "notes.txt") + tc.tk.targets = declareWrite(file) + if err := d.Register(tc.tk); err != nil { + t.Fatalf("Register: %v", err) + } + if _, err := invokeLeg(d, "turn-a", "tk", map[string]any{"path": file}); errors.Is(err, ErrFileBusy) { + t.Fatalf("第一次调用不该被判忙: %v", err) + } + // 另一路立刻接手 ⇒ 说明第一次调用已经把占用还回来了。 + if _, err := invokeLeg(d, "turn-b", "tk", map[string]any{"path": file}); errors.Is(err, ErrFileBusy) { + t.Fatalf("前一次调用退场后必须已释放占用,得到 %v", err) + } + if got := claimTableLen(t, d); got != 0 { + t.Fatalf("占用表残留 %d 个键", got) + } + }) + } +} + +// TestInvoke_ReleaseAfterTimeout 单独钉住超时路径(需要更短的每工具截止时间)。 +func TestInvoke_ReleaseAfterTimeout(t *testing.T) { + t.Parallel() + reg := registry.New() + d, err := New(reg, WithDefaultTimeout(shortToolTimeout)) + if err != nil { + t.Fatalf("New: %v", err) + } + file := filepath.Join(t.TempDir(), "notes.txt") + blocked := &declaringTool{name: "tk", targets: declareWrite(file), hold: make(chan struct{})} + if err := d.Register(blocked); err != nil { + t.Fatalf("Register: %v", err) + } + if _, err := invokeLeg(d, "turn-a", "tk", map[string]any{"path": file}); !errors.Is(err, ErrTimeout) { + t.Fatalf("阻塞调用应超时(ErrTimeout),得到 %v", err) + } + if _, err := invokeLeg(d, "turn-b", "tk", map[string]any{"path": file}); errors.Is(err, ErrFileBusy) { + t.Fatalf("超时退场后必须已释放占用,得到 %v", err) + } + if got := claimTableLen(t, d); got != 0 { + t.Fatalf("占用表残留 %d 个键", got) + } +} + +// TestInvoke_MultiTargetFailureRollsBack 钉住多目标调用的**回滚**:一个目标被判忙 +// 时,本次调用已登记的其余目标必须立刻还回去,绝不留下半登记状态。 +func TestInvoke_MultiTargetFailureRollsBack(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + dir := t.TempDir() + held := filepath.Join(dir, "held.txt") + free := filepath.Join(dir, "free.txt") + holder := &declaringTool{name: "holder", targets: declareWrite(held), entered: make(chan struct{}), hold: make(chan struct{})} + multi := &declaringTool{name: "multi", targets: func(any) []FileTarget { + // 先声明一个空闲文件,再声明被占住的那个 ⇒ 前者必须被回滚。 + return []FileTarget{{Op: FileOpWrite, Path: free}, {Op: FileOpWrite, Path: held}} + }} + for _, tk := range []*declaringTool{holder, multi} { + if err := d.Register(tk); err != nil { + t.Fatalf("Register %s: %v", tk.Name(), err) + } + } + + done := make(chan error, 1) + go func() { + _, err := invokeLeg(d, "turn-a", "holder", map[string]any{"path": held}) + done <- err + }() + awaitClose(t, holder.entered, "持锁那一路没有进入 Execute") + + if _, err := invokeLeg(d, "turn-b", "multi", map[string]any{"path": held}); !errors.Is(err, ErrFileBusy) { + t.Fatalf("多目标里有一个被占住就必须整调用判忙,得到 %v", err) + } + // free.txt 必须已被回滚:占用表里只剩被占住的 held.txt 一个键。 + if got := claimTableLen(t, d); got != 1 { + t.Fatalf("占用表应有且仅有 held.txt 一个键(free.txt 已回滚),实际 %d 个", got) + } + close(holder.hold) + if err := <-done; err != nil { + t.Fatalf("持锁那一路应正常完成: %v", err) + } +} + +// TestInvoke_DeclarationPanicDegradesSafely 钉住声明面的兜底:第三方工具在 +// FileTargets 里 panic,不该连带炸掉调用(降级为「不参与互斥」而不是「调用失败」)。 +func TestInvoke_DeclarationPanicDegradesSafely(t *testing.T) { + t.Parallel() + d, _ := newDispatcher(t) + boom := &declaringTool{name: "boom", declareBoom: true} + if err := d.Register(boom); err != nil { + t.Fatalf("Register: %v", err) + } + if _, err := invokeLeg(d, "turn-a", "boom", map[string]any{"path": "a.txt"}); err != nil { + t.Fatalf("声明面 panic 必须降级为「不参与互斥」,而不是让调用失败: %v", err) + } +} + +// TestInvoke_FileClaimerDisabled 钉住 WithFileClaimer(nil):显式关闭互斥时, +// 声明了目标的工具也完全不做互斥(装配点的逃生门)。 +func TestInvoke_FileClaimerDisabled(t *testing.T) { + t.Parallel() + reg := registry.New() + d, err := New(reg, WithFileClaimer(nil)) + if err != nil { + t.Fatalf("New: %v", err) + } + file := filepath.Join(t.TempDir(), "notes.txt") + blocker := &declaringTool{name: "block", targets: declareWrite(file), entered: make(chan struct{}), hold: make(chan struct{})} + peer := &declaringTool{name: "peer", targets: declareWrite(file)} + for _, tk := range []*declaringTool{blocker, peer} { + if err := d.Register(tk); err != nil { + t.Fatalf("Register %s: %v", tk.Name(), err) + } + } + + done := make(chan error, 1) + go func() { + _, err := invokeLeg(d, "turn-a", "block", map[string]any{"path": file}) + done <- err + }() + awaitClose(t, blocker.entered, "第一路没有进入 Execute") + + if _, err := invokeLeg(d, "turn-b", "peer", map[string]any{"path": file}); err != nil { + t.Fatalf("关闭占用表后不做任何互斥,得到 %v", err) + } + close(blocker.hold) + if err := <-done; err != nil { + t.Fatalf("第一路应正常完成: %v", err) + } +} diff --git a/internal/runtime/tool/fileclaims.go b/internal/runtime/tool/fileclaims.go new file mode 100644 index 0000000..7f6dcbf --- /dev/null +++ b/internal/runtime/tool/fileclaims.go @@ -0,0 +1,377 @@ +package tool + +import ( + "context" + "fmt" + "log/slog" + "path/filepath" + "runtime" + "strings" + "sync" + + "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" +) + +// 本文件把 legfile.go 预留的文件级互斥通道**实现并接线**(design/15 §2.5.7、 +// 主人 2026-10-05 裁决 3):写类工具执行前先在占用表里登记本次要动的文件,被 +// **另一条在途路**占住时立刻把「该文件正在被另一路操作」以面向模型的文案返回 +// 并提示稍后重试(不静默、不排队死等);不是同一个文件则照常并行。 +// +// 「本次要动哪个文件」由**工具自己**实现可选的 FileTargeter 接口声明,而不是 +// 由本包从 JSON Schema 标注或从参数正文里猜:同一个工具的文件参数会随调用而变 +// (内置 `file` 工具的 `path` 在 action=read 时是读、action=write 时是写;还有 +// 工具用 `name`、有的收多个路径),只有工具实现方知道本次调用的语义。静态标注 +// 要么表达不了「本次是读还是写」,要么逼工具把语义抄两遍;字符串猜文件名则会 +// 把正文里的任意文本当成路径(误挡 + 泄漏面)。声明不出来 ⇒ 该工具不参与互斥, +// 行为与加入本特性之前逐字相同。 +// +// 占用键 = **操作类别 + 规范化路径**(见 fileClaimKey):同一文件的两个**读** +// 允许并行(主人「优先考虑性能而不是预算」口径:检索类并行不该被互斥拖住), +// 读与写、写与写互斥。类别编码进键、而不是给 FileClaimer 多加一对方法,是为了 +// 不破坏 legfile.go 已钉死的两方法接口形状(既有用例零改动)。 + +// FileOp 是一次工具调用对目标文件的操作类别。粒度只到读/写两类:读不改变文件 +// 内容,读与读之间没有数据风险;写会改变内容,与任何路的读写都不能重叠。 +type FileOp string + +const ( + // FileOpRead 表示**只读**访问目标文件(可与别的读并行)。 + FileOpRead FileOp = "read" + // FileOpWrite 表示会**修改**目标文件(与其他路的任何读写都互斥)。 + FileOpWrite FileOp = "write" +) + +// FileTarget 是工具为**本次调用**声明的一个目标文件。Op 为空/未知一律按 +// FileOpWrite 处理:占用表的方向是宁可多挡一次(让模型稍后重试), +// 也不放过两路并发写。 +type FileTarget struct { + // Op 是本次调用的操作类别(读 / 写)。 + Op FileOp + // Path 是本次调用要碰的路径;以工具**自己执行时使用的那个路径**为准 + // (沙箱解析后的绝对路径最好)。 + Path string +} + +// FileTargeter 由需要参与文件级互斥的工具**可选**实现:返回本次调用会碰到的 +// 文件(含读/写类别)。未实现它的工具完全不参与互斥。 +// +// 约定: +// - Path 应是工具执行时真正用的路径;占用表只做一次**词法**规范化(清理相对 +// 段、Windows 折叠大小写),它不解析符号链接、不统一短名/UNC/映射盘 —— +// 别把沙箱解析责任推给占用表。 +// - 只声明真正会碰的文件:声明得越宽并行度越低(例如把每次写都声明成同一个 +// 目录名,就等于退化成全局串行)。 +// - 实现必须是纯函数(无副作用、不阻塞):它在每次工具调用前被调用一次, +// 而且它 panic 会被吞掉并降级为「不参与互斥」(见 safeFileTargets)。 +type FileTargeter interface { + FileTargets(args any) []FileTarget +} + +// fileClaimKeySep 分隔占用键里的操作类别与路径。用 NUL 是因为它在 Windows 与 +// POSIX 的文件名里都非法,真实路径不可能与它撞键(loop 的 toolFingerprint 用 +// 同一种分隔符表达同样的「不会撞」拼接)。 +const fileClaimKeySep = "\x00" + +// fileClaimKey 拼出占用键(调用方与占用表共用同一个拼法)。 +func fileClaimKey(op FileOp, path string) string { + return string(normalizeFileOp(op)) + fileClaimKeySep + path +} + +// normalizeFileOp 把空/未知类别收敛为「写」。 +func normalizeFileOp(op FileOp) FileOp { + if op == FileOpRead { + return FileOpRead + } + return FileOpWrite +} + +// parseFileClaimKey 把占用键拆回(类别,路径)。没有类别前缀的裸路径(外部 +// 实现方直接传路径 —— legfile_test 的 fakeFileClaimer 就是这种口径)按「写」 +// 处理,与 normalizeFileOp 同一个保守方向。 +func parseFileClaimKey(target string) (FileOp, string) { + op, path, ok := strings.Cut(target, fileClaimKeySep) + if !ok { + return FileOpWrite, target + } + if op == string(FileOpRead) { + return FileOpRead, path + } + return FileOpWrite, path +} + +// canonicalFilePath 把路径收敛成占用表的键。只做**词法**规范化(Abs + Clean + +// Windows 折叠大小写),绝不 touch 文件系统:占用判定在每次工具调用之前发生, +// 不能引入 I/O,也不该因为目标文件还不存在就放弃互斥。 +// +// 代价(诚实登记):符号链接、8.3 短名、UNC 与映射盘写法的**同一物理文件**不会 +// 被合并 —— 两个别名会被当成两个文件而放行并行。工具按自己执行时的真实落点 +// 声明路径即可避免绝大多数别名(见 FileTargeter 的约定)。 +func canonicalFilePath(p string) string { + p = strings.TrimSpace(p) + if p == "" { + return "" + } + if abs, err := filepath.Abs(p); err == nil { + p = abs + } + p = filepath.Clean(p) + if runtime.GOOS == "windows" { + // Windows 文件系统大小写不敏感:`A.TXT` 与 `a.txt` 是同一个文件,不折叠 + // 就会漏判成两个键,互斥形同虚设。 + p = strings.ToLower(p) + } + return p +} + +// fileClaims 是 FileClaimer 的进程内实现。单二进制、单进程 ⇒ 占用状态无需跨 +// 进程共享(项目红线:零跨进程 IPC)。 +// +// 语义是**读共享 / 写独占**(而不是「只登记写、读不登记」):一路在写时另一路 +// 读同一文件会读到撕开的内容(file 工具的读不是原子快照),与两路同时写属同一 +// 类数据风险,所以读也登记;但多个读互不阻挡,检索类并行不受影响。 +type fileClaims struct { + mu sync.Mutex + held map[string]*fileHolders +} + +// fileHolders 是某个文件当前的占用者集合(键是路标识):写至多一个,读可多个。 +// 两个集合都空时该键会被立刻删除,因此占用表随路收尾清空、绝不随会话长度增长。 +type fileHolders struct { + writers map[string]struct{} + readers map[string]struct{} +} + +// NewFileClaimer 构建一份进程内文件占用表。跨会话共享同一份即可:粒度的口径是 +// **文件**(主人裁决 3),不是会话 —— 两个会话同时改同一个文件同样是数据风险。 +func NewFileClaimer() FileClaimer { + return &fileClaims{held: make(map[string]*fileHolders)} +} + +// Claim 实现 FileClaimer:按占用键的类别做读共享/写独占判定,被别的路占住时 +// 返回带文件状态的 ErrFileBusy(直接回灌给模型)。它**绝不等待**:判忙即返回, +// 由模型稍后重试(主人裁决 3 的「不静默、不排队死等」)。 +// +// 空 legID 或空目标是空操作:单路 / 子代理 / 后台思考(没有并发路)不登记也不 +// 检查,行为与加入本特性之前逐字相同(兼容硬要求)。 +// +// 同一路重复 Claim 同一目标幂等;一路自己先读后写同一文件不会被自己挡住 +// (否则一条路自己的多步工具调用会互相锁死)。 +func (c *fileClaims) Claim(ctx context.Context, legID, target string) error { + if c == nil || legID == "" || target == "" { + return nil + } + op, path := parseFileClaimKey(target) + key := canonicalFilePath(path) + if key == "" { + return nil + } + c.mu.Lock() + defer c.mu.Unlock() + h := c.held[key] + if op == FileOpWrite { + if _, busy := h.otherWriter(legID); busy { + return FileBusyStatusError(key, fileBusyStatus(FileOpWrite, FileOpWrite)) + } + if _, busy := h.otherReader(legID); busy { + return FileBusyStatusError(key, fileBusyStatus(FileOpWrite, FileOpRead)) + } + } else if _, busy := h.otherWriter(legID); busy { + return FileBusyStatusError(key, fileBusyStatus(FileOpRead, FileOpWrite)) + } + if h == nil { + h = newFileHolders() + c.held[key] = h + } + if op == FileOpWrite { + h.writers[legID] = struct{}{} + } else { + h.readers[legID] = struct{}{} + } + return nil +} + +// Release 实现 FileClaimer:无条件摘掉该路在该目标上的占用(读、写两侧都摘), +// 最后一个占用者离开时删键。legID 不是占用者时是空操作。 +// +// 它必须挂在工具执行的生命周期上(defer):报错 / 超时 / panic 的路也绝不把 +// 文件永久锁住。 +func (c *fileClaims) Release(ctx context.Context, legID, target string) { + if c == nil || legID == "" || target == "" { + return + } + _, path := parseFileClaimKey(target) + key := canonicalFilePath(path) + if key == "" { + return + } + c.mu.Lock() + defer c.mu.Unlock() + h := c.held[key] + if h == nil { + return + } + delete(h.writers, legID) + delete(h.readers, legID) + if len(h.writers) == 0 && len(h.readers) == 0 { + delete(c.held, key) + } +} + +// fileBusyStatus 生成回灌给模型的**文件状态**短句:说清此刻是谁在动它, +// 模型才知道该等什么、为什么现在不能动(而不是只看到「忙」)。 +// 只含操作类别,不含另一路的任何身份或内容。 +func fileBusyStatus(want, holder FileOp) string { + verb := "写入" + if holder == FileOpRead { + verb = "读取" + } + return fmt.Sprintf("该文件正在被另一路%s,本次%s无法进行", verb, opLabel(want)) +} + +// opLabel 把操作类别渲染成中文,供面向模型的文案使用。 +func opLabel(op FileOp) string { + if op == FileOpRead { + return "读取" + } + return "写入" +} + +// newFileHolders 建一个空占用者集合(两个集合都必须非 nil,Claim 会直接写入)。 +func newFileHolders() *fileHolders { + return &fileHolders{writers: make(map[string]struct{}), readers: make(map[string]struct{})} +} + +// otherWriter 报告除 exclude 外是否还有路正在写该文件,并返回其中一条(供文案 +// 与调试使用)。nil 接收者表示「该文件当前无人占用」⇒ (空串, false),使 Claim +// 不必先建表再判定。 +// +// 判定用**布尔**而不是「返回空串表示没有」:路标识本身可以是空串(外部调用方 +// 的口径),用空串当「没有」的哨兵会把它们误判成无人占用。 +func (h *fileHolders) otherWriter(exclude string) (string, bool) { + if h == nil { + return "", false + } + for leg := range h.writers { + if leg != exclude { + return leg, true + } + } + return "", false +} + +// otherReader 报告除 exclude 外是否还有路正在读该文件,并返回其中一条。 +func (h *fileHolders) otherReader(exclude string) (string, bool) { + if h == nil { + return "", false + } + for leg := range h.readers { + if leg != exclude { + return leg, true + } + } + return "", false +} + +// fileTargetsOf 取工具为本次调用声明的目标文件。工具未实现 FileTargeter、 +// 声明为空、或某项路径为空时跳过 —— 声明不出来就不参与互斥(行为不变)。 +// 同一次调用里重复的目标只登记一次,避免同一把锁被重复占有/重复释放。 +func fileTargetsOf(t sdk.Tool, args any) []FileTarget { + targeter, ok := t.(FileTargeter) + if !ok { + return nil + } + targets := targeter.FileTargets(args) + if len(targets) == 0 { + return nil + } + out := make([]FileTarget, 0, len(targets)) + seen := make(map[string]struct{}, len(targets)) + for _, tg := range targets { + if strings.TrimSpace(tg.Path) == "" { + continue + } + key := fileClaimKey(tg.Op, tg.Path) + if _, dup := seen[key]; dup { + continue + } + seen[key] = struct{}{} + out = append(out, tg) + } + return out +} + +// safeFileTargets 是 fileTargetsOf 的抗 panic 包装:声明是**可选**扩展点, +// 一个第三方工具在它里面 panic 不该连带炸掉调用(callTool 只在 Execute 里 +// recover)。降级方向是「不参与互斥」,而不是「调用失败」。 +func safeFileTargets(log *slog.Logger, name string, t sdk.Tool, args any) (out []FileTarget) { + defer func() { + if r := recover(); r != nil { + out = nil + log.Warn("file target declaration panicked; skipping file claims", + "tool", name, "panic", fmt.Sprint(r)) + } + }() + return fileTargetsOf(t, args) +} + +// claimFiles 为一次工具调用登记它声明的目标文件,并返回释放函数(调用方必须 +// 在执行后 defer 调用它:成功 / 报错 / 超时 / panic 都要释放,否则一个中途报错 +// 的路会把文件永久锁住)。 +// +// 返回 nil 释放函数表示本次调用不参与互斥:未装占用表、未挂路标识(单路 / +// 子代理 / 未接线,兼容硬要求)、或工具没有声明目标文件。 +// +// 被另一条路占住时**不等待、不排队**,直接把带文件状态的 ErrFileBusy 交给模型 +// (主人裁决 3)。注意它**不包成 ToolError**:占用判定发生在执行之前,调用根本 +// 没跑,而且 ToolError 的渲染(message + cause)会把同一句话打印两遍 —— 回灌给 +// 模型的结果要一句话说清「谁在被谁占、稍后再试」(loop 的 errorKind 把它归为 +// execution,与执行面失败的分类一致)。 +// +// 多目标里任意一个失败时,本次调用**已登记的全部目标立刻回滚**,绝不留下半登记 +// 状态。 +func (d *Dispatcher) claimFiles(ctx context.Context, name string, t sdk.Tool, args map[string]any) (func(), error) { + claimer := d.claimerRef() + if claimer == nil { + return nil, nil + } + legID := LegIDFrom(ctx) + if legID == "" { + return nil, nil + } + targets := safeFileTargets(d.log, name, t, args) + if len(targets) == 0 { + return nil, nil + } + keys := make([]string, 0, len(targets)) + for _, tg := range targets { + key := fileClaimKey(tg.Op, tg.Path) + // 先记账再登记:失败的 Claim 也要进回滚清单(对第三方实现而言, + // 「失败即未登记」不是接口保证;多释放一次按契约是空操作)。 + keys = append(keys, key) + if err := claimer.Claim(ctx, legID, key); err != nil { + releaseFileClaims(ctx, claimer, legID, keys) + d.log.Info("file claim busy; asking the model to retry later", + "tool", name, "leg", legID, "target", key) + return nil, err + } + } + return func() { releaseFileClaims(ctx, claimer, legID, keys) }, nil +} + +// releaseFileClaims 释放一批占用(正常收尾与失败回滚共用一条路径)。 +func releaseFileClaims(ctx context.Context, claimer FileClaimer, legID string, keys []string) { + for _, key := range keys { + claimer.Release(ctx, legID, key) + } +} + +// claimerRef 返回已装配的占用表;未装配时返回 nil(⇒ 完全不做互斥)。 +// 占用表只在 New 的选项里装配一次,此后不再变更。 +func (d *Dispatcher) claimerRef() FileClaimer { + if d == nil { + return nil + } + d.mu.RLock() + defer d.mu.RUnlock() + return d.claimer +} diff --git a/internal/runtime/tool/fileclaims_test.go b/internal/runtime/tool/fileclaims_test.go new file mode 100644 index 0000000..584b553 --- /dev/null +++ b/internal/runtime/tool/fileclaims_test.go @@ -0,0 +1,294 @@ +package tool + +import ( + "context" + "errors" + "path/filepath" + "runtime" + "strings" + "testing" + "time" +) + +// 本文件钉住文件级互斥(design/15 §2.5.7、主人 2026-10-05 裁决 3)的**占用表 +// 语义**: +// - 同一文件被**另一路**占住 ⇒ 立刻判忙(文案带文件状态,可被 LLM 消化), +// **绝不阻塞等待**(不排队死等); +// - 不是同一个文件 ⇒ 并行放行; +// - 同一文件的两个**读** ⇒ 并行放行(结论见 TestFileClaimsTwoReadsRunParallel); +// - 同一路重复占用幂等、且不会被自己挡住(先读后写同一文件); +// - 释放后另一路立刻可接手;占用表随路收尾清空(有界); +// - **无路标识 ⇒ 不登记也不检查**(单路 / 既有路径逐字不变)。 + +// claimProbeTimeout 是「判忙必须立刻返回」的上界。它只把「阻塞等待」这种错误 +// 实现暴露成失败,不参与任何时序竞态(占用判定本身是同步的)。 +const claimProbeTimeout = 5 * time.Second + +// newTestFileClaims 返回占用表的具体实现,供白盒断言「表是否有界」。 +func newTestFileClaims() *fileClaims { + return NewFileClaimer().(*fileClaims) +} + +// writeTarget / readTarget 按占用键的正式拼法造目标(与 Dispatcher 侧同源)。 +func writeTarget(path string) string { return fileClaimKey(FileOpWrite, path) } +func readTarget(path string) string { return fileClaimKey(FileOpRead, path) } + +// TestFileClaimsSameFileOtherLegBusy 钉住主人口径的核心:同一文件正被另一路 +// 操作 ⇒ 明确判忙(errors.Is(ErrFileBusy))、**不阻塞等待**、文案里带出文件状态 +// 与路径,使模型能自己决定「等一会再操作」。 +func TestFileClaimsSameFileOtherLegBusy(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + file := filepath.Join(t.TempDir(), "notes.txt") + + if err := c.Claim(ctx, "turn-a", writeTarget(file)); err != nil { + t.Fatalf("第一路占用失败: %v", err) + } + + done := make(chan error, 1) + go func() { done <- c.Claim(ctx, "turn-b", writeTarget(file)) }() + select { + case err := <-done: + if !errors.Is(err, ErrFileBusy) { + t.Fatalf("同一文件被另一路占用必须判忙,得到 %v", err) + } + msg := err.Error() + for _, want := range []string{"请稍后再试", "正在被另一路写入", "notes.txt"} { + if !strings.Contains(msg, want) { + t.Fatalf("回灌给模型的文案缺少 %q: %s", want, msg) + } + } + case <-time.After(claimProbeTimeout): + t.Fatal("第二路在占用表上阻塞等待了 —— 主人口径是不排队死等,判忙即返回") + } + + // 判忙不改变占用归属:第三路同样判忙,第一路仍然持锁。 + if err := c.Claim(ctx, "turn-c", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("第三路也应判忙,得到 %v", err) + } +} + +// TestFileClaimsDifferentFilesRunParallel 钉住「不是同一个文件 ⇒ 允许并行」。 +func TestFileClaimsDifferentFilesRunParallel(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + dir := t.TempDir() + + if err := c.Claim(ctx, "turn-a", writeTarget(filepath.Join(dir, "a.txt"))); err != nil { + t.Fatalf("A 占用 a.txt: %v", err) + } + if err := c.Claim(ctx, "turn-b", writeTarget(filepath.Join(dir, "b.txt"))); err != nil { + t.Fatalf("不同文件必须允许并行操作,得到 %v", err) + } + // 同一目录下的另一个文件同样放行(互斥粒度是文件,不是目录)。 + if err := c.Claim(ctx, "turn-c", writeTarget(filepath.Join(dir, "sub", "c.txt"))); err != nil { + t.Fatalf("不同文件必须允许并行操作,得到 %v", err) + } +} + +// TestFileClaimsTwoReadsRunParallel 钉住第 4 条的结论:同一文件的**两个读允许 +// 并行**(读不改变内容,没有数据风险;主人「优先考虑性能」口径下检索类并行不该 +// 被互斥拖住)。而读与写之间仍然互斥 —— 读到「正在被写」的一半内容与两路同时 +// 写属同一类数据风险。 +func TestFileClaimsTwoReadsRunParallel(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + file := filepath.Join(t.TempDir(), "notes.txt") + + if err := c.Claim(ctx, "turn-a", readTarget(file)); err != nil { + t.Fatalf("第一路读: %v", err) + } + if err := c.Claim(ctx, "turn-b", readTarget(file)); err != nil { + t.Fatalf("同一文件的两个读必须允许并行,得到 %v", err) + } + if err := c.Claim(ctx, "turn-c", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("有读在途时写入必须判忙,得到 %v", err) + } + if !strings.Contains(c.Claim(ctx, "turn-c", writeTarget(file)).Error(), "正在被另一路读取") { + t.Fatal("写入被判忙时,文案必须说清是「另一路正在读取」") + } +} + +// TestFileClaimsWriteThenReadBusy 钉住相反方向:写在途时读同一文件判忙, +// 且文案说清是「另一路正在写入」。 +func TestFileClaimsWriteThenReadBusy(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + file := filepath.Join(t.TempDir(), "notes.txt") + + if err := c.Claim(ctx, "turn-a", writeTarget(file)); err != nil { + t.Fatalf("第一路写: %v", err) + } + err := c.Claim(ctx, "turn-b", readTarget(file)) + if !errors.Is(err, ErrFileBusy) { + t.Fatalf("写在途时读同一文件必须判忙,得到 %v", err) + } + if !strings.Contains(err.Error(), "正在被另一路写入") { + t.Fatalf("文案必须说清是「另一路正在写入」: %s", err) + } +} + +// TestFileClaimsSameLegIsIdempotent 钉住同一路的口径:重复占用同一目标幂等, +// 且**一路自己先读后写同一文件不会被自己挡住**(否则一条路自己的多步工具调用 +// 会互相锁死)。同时别的路仍被正确挡住。 +func TestFileClaimsSameLegIsIdempotent(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + file := filepath.Join(t.TempDir(), "notes.txt") + + if err := c.Claim(ctx, "turn-a", readTarget(file)); err != nil { + t.Fatalf("自己读: %v", err) + } + if err := c.Claim(ctx, "turn-a", readTarget(file)); err != nil { + t.Fatalf("同一路重复占用必须是幂等的: %v", err) + } + if err := c.Claim(ctx, "turn-a", writeTarget(file)); err != nil { + t.Fatalf("同一路先读后写同一文件不该被自己挡住: %v", err) + } + if err := c.Claim(ctx, "turn-a", readTarget(file)); err != nil { + t.Fatalf("同一路读写并存必须是幂等的: %v", err) + } + if err := c.Claim(ctx, "turn-b", readTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("别的路仍必须被挡住,得到 %v", err) + } + if err := c.Claim(ctx, "turn-b", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("别的路仍必须被挡住,得到 %v", err) + } +} + +// TestFileClaimsReleaseLetsOtherLegTakeOver 钉住释放:立刻可接手、非占用者释放 +// 是空操作、最后一个占用者离开后占用表清空(有界 —— 绝不随会话长度增长)。 +func TestFileClaimsReleaseLetsOtherLegTakeOver(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + file := filepath.Join(t.TempDir(), "notes.txt") + + if err := c.Claim(ctx, "turn-a", writeTarget(file)); err != nil { + t.Fatalf("A 占用: %v", err) + } + c.Release(ctx, "turn-a", writeTarget(file)) + if err := c.Claim(ctx, "turn-b", writeTarget(file)); err != nil { + t.Fatalf("释放后另一路应立即接手,得到 %v", err) + } + + // 非占用者释放是空操作:B 的占用不受影响。 + c.Release(ctx, "turn-z", writeTarget(file)) + if err := c.Claim(ctx, "turn-c", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("B 的占用被非占用者的释放动掉了,得到 %v", err) + } + + c.Release(ctx, "turn-b", writeTarget(file)) + if got := len(c.held); got != 0 { + t.Fatalf("占用表残留 %d 个键,必须随路收尾清空(有界)", got) + } + // 读一侧的释放同样要摘干净(读与写同属一个键)。 + if err := c.Claim(ctx, "turn-a", readTarget(file)); err != nil { + t.Fatalf("A 读: %v", err) + } + c.Release(ctx, "turn-a", writeTarget(file)) + if got := len(c.held); got != 0 { + t.Fatalf("释放必须同时摘掉读侧占用,残留 %d 个键", got) + } +} + +// TestFileClaimsEmptyLegIsNoOp 钉住兼容硬要求:**无路标识 ⇒ 不登记也不检查**。 +// 空路(单路 / 子代理 / 未接线)既不会被挡住,也不会在占用表里留下任何痕迹。 +func TestFileClaimsEmptyLegIsNoOp(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + file := filepath.Join(t.TempDir(), "notes.txt") + + if err := c.Claim(ctx, "", writeTarget(file)); err != nil { + t.Fatalf("空路标识不做互斥,得到 %v", err) + } + if err := c.Claim(ctx, "turn-a", writeTarget(file)); err != nil { + t.Fatalf("空路标识不该登记占用,真路必须能正常占用: %v", err) + } + if err := c.Claim(ctx, "turn-b", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("真路之间仍然要互斥,得到 %v", err) + } + // 真路占住时,空路标识照样放行(完全不做互斥)—— 这正是「单路逐字不变」。 + if err := c.Claim(ctx, "", writeTarget(file)); err != nil { + t.Fatalf("空路标识不做互斥,得到 %v", err) + } + // 空目标、nil ctx 也不 panic。 + if err := c.Claim(ctx, "turn-a", ""); err != nil { + t.Fatalf("空目标不做互斥,得到 %v", err) + } + if err := c.Claim(nil, "turn-x", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("nil ctx 不该改变判定,得到 %v", err) + } + c.Release(ctx, "", writeTarget(file)) + if err := c.Claim(ctx, "turn-c", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("空路标识的释放必须是空操作,得到 %v", err) + } +} + +// TestFileClaimsPathAliasesCollapse 钉住占用键的**词法规范化**:同一文件的 +// `sub/../notes.txt` 与 `./notes.txt` 写法必须撞成同一个键,否则互斥形同虚设。 +// Windows 上还要折叠大小写(`NOTES.TXT` 与 `notes.txt` 是同一个文件)。 +func TestFileClaimsPathAliasesCollapse(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + base := t.TempDir() + file := filepath.Join(base, "sub", "..", "notes.txt") + alias := filepath.Join(base, ".", "notes.txt") + + if err := c.Claim(ctx, "turn-a", writeTarget(file)); err != nil { + t.Fatalf("A 占用: %v", err) + } + if err := c.Claim(ctx, "turn-b", writeTarget(alias)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("同一文件的两种写法必须撞成同一个键,得到 %v", err) + } + if runtime.GOOS == "windows" { + upper := strings.ToUpper(alias) + if err := c.Claim(ctx, "turn-b", writeTarget(upper)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("Windows 大小写不敏感,必须撞成同一个键,得到 %v", err) + } + } +} + +// TestFileClaimKeyShape 钉住占用键的形状与「未知即写」的保守方向:裸路径 +// (外部实现方直接传路径的口径)按**写**处理,与读写都互斥。 +func TestFileClaimKeyShape(t *testing.T) { + t.Parallel() + if op, path := parseFileClaimKey(fileClaimKey(FileOpRead, "a.txt")); op != FileOpRead || path != "a.txt" { + t.Fatalf("读键往返 = (%q,%q),want (read,a.txt)", op, path) + } + if op, path := parseFileClaimKey(fileClaimKey(FileOpWrite, "a.txt")); op != FileOpWrite || path != "a.txt" { + t.Fatalf("写键往返 = (%q,%q),want (write,a.txt)", op, path) + } + // 空/未知类别与裸路径一律按写(宁可多挡一次,也不放过并发写)。 + for _, target := range []string{"a.txt", fileClaimKey("", "a.txt"), fileClaimKey("chmod", "a.txt")} { + if op, path := parseFileClaimKey(target); op != FileOpWrite || path != "a.txt" { + t.Fatalf("目标 %q 应保守按写处理,得到 (%q,%q)", target, op, path) + } + } +} + +// TestFileClaimsBareTargetBlocksReadAndWrite 把上一条的「未知即写」钉到行为上: +// 裸路径占用必须同时挡住读与写。 +func TestFileClaimsBareTargetBlocksReadAndWrite(t *testing.T) { + t.Parallel() + c := newTestFileClaims() + ctx := context.Background() + file := filepath.Join(t.TempDir(), "notes.txt") + + if err := c.Claim(ctx, "turn-a", file); err != nil { + t.Fatalf("裸路径占用: %v", err) + } + if err := c.Claim(ctx, "turn-b", readTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("裸路径按写处理 ⇒ 必须挡住读,得到 %v", err) + } + if err := c.Claim(ctx, "turn-b", writeTarget(file)); !errors.Is(err, ErrFileBusy) { + t.Fatalf("裸路径按写处理 ⇒ 必须挡住写,得到 %v", err) + } +} diff --git a/internal/runtime/tool/legfile.go b/internal/runtime/tool/legfile.go index 04ceabd..3f58599 100644 --- a/internal/runtime/tool/legfile.go +++ b/internal/runtime/tool/legfile.go @@ -6,10 +6,13 @@ import ( "fmt" ) -// 本文件是**接口预留**(design/15 §2.5.7 与主人 2026-10-05 裁决 3),本切片 -// **不实现也不接线**:忙时并行(design/15 §2.5)落地后,同一会话可以有多条在途 -// 回合,而写类工具(执行命令 / 修改文件 / 对外发消息 / 改配置,即 design/11 §4 -// 的敏感清单)在两路之间会真正争抢同一个副作用目标。 +// 本文件钉住**文件级**互斥(design/15 §2.5.7 与主人 2026-10-05 裁决 3)的对外 +// 形状:通道(WithLegID / LegIDFrom)、接口(FileClaimer)与错误(ErrFileBusy / +// FileBusyError / FileBusyStatusError)。**实现与接线在 fileclaims.go**(占用表、 +// 目标声明接口 FileTargeter、Dispatcher 的登记/释放)。忙时并行(design/15 §2.5) +// 落地后同一会话可以有多条在途回合,而写类工具(执行命令 / 修改文件 / 对外发 +// 消息 / 改配置,即 design/11 §4 的敏感清单)在两路之间会真正争抢同一个副作用 +// 目标。 // // 主人口径(裁决 3)是**文件级**互斥,而不是会话级串行: // - 工具先检测目标文件此刻是否正被**别的在途路**操作; @@ -17,11 +20,6 @@ import ( // LLM,并提示稍后再试(**不静默、不排队死等**); // - **不是同一个文件** ⇒ 允许并行操作。 // -// 为什么只留接口:写类工具的「目标文件」需要逐工具声明(命令、补丁、配置写入 -// 各自的目标语义不同),且占用登记要挂在**路**的生命周期上(路一收尾就必须释放, -// 否则一个中途报错的路会把那个文件永久锁住)。这两件都属于后续切片,本切片只把 -// 通道与判定口径钉死,使实现方不必再改工具接口。 -// // 身份与权限**不走这条通道**:它只回答「此刻有没有另一路正在动这个文件」,与 // actor 元数据 / 策略引擎无关(项目红线)。 @@ -39,8 +37,22 @@ func FileBusyError(target string) error { return fmt.Errorf("%w(目标:%s)", ErrFileBusy, target) } +// FileBusyStatusError 在 FileBusyError 之上补一句**文件状态**(该文件此刻被 +// 另一路读还是写、本次想做的操作因此做不了),使回灌给 LLM 的结果说清「为什么 +// 现在动不了它、该等什么」,而不是只丢一个「忙」。status 为空时退化为 +// FileBusyError(形状不变)。 +// +// 与 target 一样,status 只描述**状态**:绝不带另一条路的身份、正文或任何文件 +// 内容(红线:不泄露另一路的内容)。 +func FileBusyStatusError(target, status string) error { + if status == "" { + return FileBusyError(target) + } + return fmt.Errorf("%w(目标:%s;%s)", ErrFileBusy, target, status) +} + // FileClaimer 是按**文件级**粒度登记在途操作的接口(主人裁决 3)。实现方 -// (后续切片)负责: +// (进程内实现见 fileclaims.go 的 NewFileClaimer)负责: // - 以规范化后的目标为键,记录「哪一条路正在操作它」; // - Claim 在目标已被**另一条路**占用时返回 ErrFileBusy(同一路重复 Claim // 同一目标必须是幂等的,否则一条路自己的多步工具调用会互相挡); @@ -48,6 +60,9 @@ func FileBusyError(target string) error { // 占用者时是空操作。 // // 实现必须是**有界的**:占用表随路收尾清空,绝不随会话长度增长。 +// +// target 的拼法由调用方与本包的 fileClaimKey 统一给出(操作类别 + 规范化路径); +// 直接传裸路径的调用方按「写」处理(见 parseFileClaimKey)。 type FileClaimer interface { Claim(ctx context.Context, legID, target string) error Release(ctx context.Context, legID, target string) diff --git a/internal/runtime/tool/tool.go b/internal/runtime/tool/tool.go index 35218f9..e3a907a 100644 --- a/internal/runtime/tool/tool.go +++ b/internal/runtime/tool/tool.go @@ -110,9 +110,10 @@ type Dispatcher struct { reg *registry.Registry dft time.Duration log *slog.Logger - mu sync.RWMutex // guards timeouts and modes + mu sync.RWMutex // guards timeouts, modes and the file claimer timeouts map[string]time.Duration modes map[string]string // tool name -> ExecutionMode* (default parallel) + claimer FileClaimer // 文件级互斥的占用表(见 fileclaims.go) sinkMu sync.RWMutex // guards the soft-judge sink sink SoftJudgeSink } @@ -120,6 +121,11 @@ type Dispatcher struct { // New builds a Dispatcher backed by reg, applying opts in order. It returns an // error wrapping ErrNilRegistry when reg is nil. The default per-invoke timeout // is DefaultTimeout unless overridden with WithDefaultTimeout. +// +// 默认装配一份进程内文件占用表(fileclaims.go):它对**未声明目标文件**的工具、 +// 以及**未挂路标识**(单路 / 子代理 / 未接线)的调用完全无副作用,因此既有行为 +// 逐字不变;一旦工具实现 FileTargeter 且调用带路标识,写类工具的文件级互斥立即 +// 生效(无需上层再接线)。 func New(reg *registry.Registry, opts ...Option) (*Dispatcher, error) { if reg == nil { return nil, fmt.Errorf("tool: new: %w", ErrNilRegistry) @@ -130,6 +136,7 @@ func New(reg *registry.Registry, opts ...Option) (*Dispatcher, error) { log: sourcedLog(nil), timeouts: make(map[string]time.Duration), modes: make(map[string]string), + claimer: NewFileClaimer(), } for _, opt := range opts { if opt != nil { @@ -152,6 +159,14 @@ func WithLogger(l *slog.Logger) Option { return func(d *Dispatcher) { d.log = sourcedLog(l) } } +// WithFileClaimer 替换文件级互斥使用的占用表(设计/实现见 fileclaims.go)。 +// 传 nil 表示**关闭**文件级互斥(工具声明的目标一律不登记、不检查)—— 供测试 +// 与需要「完全不做互斥」的装配点使用;新构建的 Dispatcher 默认已带一份进程内 +// 占用表,通常无需调用它。 +func WithFileClaimer(c FileClaimer) Option { + return func(d *Dispatcher) { d.claimer = c } +} + // sourcedLog tags l with the module source "runtime/tool"; nil becomes a // discard logger, so the soft judge can always log safely. func sourcedLog(l *slog.Logger) *slog.Logger { @@ -236,6 +251,13 @@ func (d *Dispatcher) PolicyEngine(env restrict.ActorEnv) *restrict.Policy { // *ToolError-free soft result: the call still runs, and its value comes back as // SoftNotAllowed carrying the not_allowed notice for the model (design/14 §4, // 裁决 14.4.2). +// +// 参与文件级互斥的工具(实现 FileTargeter,见 fileclaims.go)在目标文件正被 +// **另一条在途路**操作时,其调用**不会执行**,而是直接返回 ErrFileBusy(带文件 +// 状态的中文文案)。它**不包 ToolError**:占用判定发生在执行之前,回灌给模型的 +// 结果就该是一句话(ToolError 的 message+cause 渲染会把同一句话打印两遍); +// 循环的 errorKind 仍把它归为 execution,与执行面失败的分类一致。 +// 无论如何都**不静默、不排队等待**。 func (d *Dispatcher) Invoke(ctx context.Context, name string, args map[string]any) (any, error) { if name == "" { return nil, &ToolError{Name: name, Message: "tool name is empty", cause: ErrUnknownTool} @@ -272,6 +294,18 @@ func (d *Dispatcher) Invoke(ctx context.Context, name string, args map[string]an // 不硬阻断(状态不是能力边界),但必须可观测、可回灌 —— 见 allowset.go。 notice := d.softJudge(ctx, name) + // 文件级互斥(design/15 §2.5.7、主人 2026-10-05 裁决 3):写类工具执行前先登记 + // 本次要动的文件;被**另一条在途路**占住时立刻把带文件状态的 ErrFileBusy 交给 + // 模型(不静默、不排队死等)。释放挂 defer:成功 / 报错 / 超时 / panic 都释放, + // 绝不把文件永久锁住。未挂路标识或工具未声明目标 ⇒ 完全不参与互斥,逐字不变。 + release, cerr := d.claimFiles(ctx, name, tool, args) + if cerr != nil { + return nil, cerr + } + if release != nil { + defer release() + } + invokeCtx, cancel := d.invokeCtx(ctx, name) defer cancel()