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