Cyrene_v2/internal/runtime/tool/fileclaim_dispatch_test.go

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)
	}
}