Agent-Squad/squad/topology_conformance_test.go

package squad

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"
)

// topology_conformance_test.go — Q2 验收一的机器证据：每种拓扑一个
// 脚本化后端的 conformance（零模型、零网络）。

// ---- 脚本化夹具 ----

// collectSink 收集事件流供断言。
type collectSink struct {
	mu     sync.Mutex
	events []Event
}

func (c *collectSink) Emit(ev Event) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.events = append(c.events, ev)
}

func (c *collectSink) snapshot() []Event {
	c.mu.Lock()
	defer c.mu.Unlock()
	return append([]Event(nil), c.events...)
}

func (c *collectSink) countText(sub string) int {
	n := 0
	for _, ev := range c.snapshot() {
		if strings.Contains(ev.Text, sub) {
			n++
		}
	}
	return n
}

// iterBackend 按「步骤子目录 × 第几次执行」回放内容（循环/评估拓扑需要
// 逐轮不同的 fields.json）。
type iterBackend struct {
	mu      sync.Mutex
	calls   map[string]int
	content func(stepKey string, iter int, target string) []byte
	delay   time.Duration
}

func (b *iterBackend) Name() string { return "iter" }

func (b *iterBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) {
	key := filepath.Base(filepath.Clean(req.WorkDir))
	if b.delay > 0 {
		time.Sleep(b.delay)
	}
	b.mu.Lock()
	b.calls[key]++
	iter := b.calls[key]
	b.mu.Unlock()
	patches := map[string][]byte{}
	for _, t := range req.Targets {
		var data []byte
		if b.content != nil {
			data = b.content(key, iter, t)
		}
		if data == nil {
			data = []byte(goodReport)
		}
		out := filepath.Join(req.WorkDir, t)
		_ = os.MkdirAll(filepath.Dir(out), 0o755)
		if err := os.WriteFile(out, data, 0o644); err != nil {
			return nil, err
		}
		patches[t] = data
	}
	return &ExecResult{Success: true, Patches: patches}, nil
}

func (b *iterBackend) totalCalls() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	n := 0
	for _, v := range b.calls {
		n += v
	}
	return n
}

func newTopologyEngine(t *testing.T, planner Planner, be Executor, sink ProgressSink) *SquadEngine {
	t.Helper()
	dir := t.TempDir()
	reg := NewRoleRegistry()
	for _, spec := range []RoleSpec{
		{Role: "researcher", Name: "调研", Backend: "iter"},
		{Role: "writer", Name: "撰写", Backend: "iter"},
		{Role: "reviewer", Name: "审核", Backend: "iter"},
	} {
		if err := reg.Register(spec); err != nil {
			t.Fatal(err)
		}
	}
	backends := NewBackendResolver()
	backends.RegisterInstance("iter", be)
	if sink == nil {
		sink = NopSink{}
	}
	engine, err := NewEngine(Options{
		DataDir: dir, Registry: reg, Planner: planner, Backends: backends, Sink: sink,
	})
	if err != nil {
		t.Fatal(err)
	}
	return engine
}

func fieldsJSON(kv map[string]string) []byte {
	b, _ := json.Marshal(kv)
	return b
}

// ---- sequential / parallel：既有图语义 + 拓扑名受理 ----

func TestTopologyConformance_sequential(t *testing.T) {
	be := &iterBackend{calls: map[string]int{}}
	sink := &collectSink{}
	engine := newTopologyEngine(t, mockScriptPlanner(
		StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"notes.md"}},
		StepSpec{ID: "s2", Role: "writer", Title: "二", Targets: []string{"output.md"}},
	), be, sink)
	res, err := engine.Run(context.Background(), Goal{Text: "顺序任务"}, RunOptions{Topology: TopologySequential})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunCompleted {
		t.Fatalf("status = %s err = %v", res.Status, res.Error)
	}
	if be.totalCalls() != 2 {
		t.Fatalf("执行次数 = %d", be.totalCalls())
	}
	if sink.countText("拓扑 sequential") != 1 {
		t.Fatal("缺拓扑受理事件")
	}
}

func TestTopologyConformance_parallel(t *testing.T) {
	be := &iterBackend{calls: map[string]int{}}
	engine := newTopologyEngine(t, mockScriptPlanner(
		StepSpec{ID: "s1", Role: "researcher", Title: "并行甲", Parallel: true, Targets: []string{"a.md"}},
		StepSpec{ID: "s2", Role: "writer", Title: "并行乙", Parallel: true, Targets: []string{"b.md"}},
		StepSpec{ID: "s3", Role: "reviewer", Title: "汇合", Targets: []string{"output.md"}},
	), be, nil)
	res, err := engine.Run(context.Background(), Goal{Text: "并行任务"}, RunOptions{Topology: TopologyParallel})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunCompleted || be.totalCalls() != 3 {
		t.Fatalf("status = %s calls = %d err = %v", res.Status, be.totalCalls(), res.Error)
	}
}

// ---- loop：条件停机 + 预算触顶 ----

func TestTopologyConformance_loopUntilField(t *testing.T) {
	// 第 1-2 轮 done=no，第 3 轮 done=yes → 恰执行 3 次。
	be := &iterBackend{calls: map[string]int{}, content: func(_ string, iter int, target string) []byte {
		if target == "fields.json" {
			if iter >= 3 {
				return fieldsJSON(map[string]string{"done": "yes"})
			}
			return fieldsJSON(map[string]string{"done": "no"})
		}
		return nil
	}}
	sink := &collectSink{}
	engine := newTopologyEngine(t, mockScriptPlanner(
		StepSpec{ID: "s1", Role: "researcher", Title: "循环体", Targets: []string{"output.md", "fields.json"},
			Loop: &LoopSpec{MaxIterations: 5, UntilField: "done", UntilEquals: "yes"}},
	), be, sink)
	res, err := engine.Run(context.Background(), Goal{Text: "循环任务"}, RunOptions{Topology: TopologyLoop})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunCompleted {
		t.Fatalf("status = %s err = %v", res.Status, res.Error)
	}
	if be.totalCalls() != 3 {
		t.Fatalf("应在条件满足时停机（3 次），实际 %d", be.totalCalls())
	}
	if sink.countText("再武装") != 2 {
		t.Fatalf("再武装事件应 2 次：%d", sink.countText("再武装"))
	}
	// 计划里的 LoopRuns 持久化为 2。
	if len(res.Plan.Steps) != 1 || res.Plan.Steps[0].LoopRuns != 2 {
		t.Fatalf("LoopRuns = %+v", res.Plan.Steps)
	}
}

func TestTopologyConformance_loopBudgetExhausted(t *testing.T) {
	// 条件永不满足：恰执行 MaxIterations 次后触顶（非失败）。
	be := &iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte {
		if target == "fields.json" {
			return fieldsJSON(map[string]string{"done": "no"})
		}
		return nil
	}}
	sink := &collectSink{}
	engine := newTopologyEngine(t, mockScriptPlanner(
		StepSpec{ID: "s1", Role: "researcher", Title: "循环体", Targets: []string{"output.md", "fields.json"},
			Loop: &LoopSpec{MaxIterations: 3, UntilField: "done", UntilEquals: "yes"}},
	), be, sink)
	res, err := engine.Run(context.Background(), Goal{Text: "循环任务"}, RunOptions{Topology: TopologyLoop})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunCompleted {
		t.Fatalf("触顶应是完成而非失败：%s %v", res.Status, res.Error)
	}
	if be.totalCalls() != 3 {
		t.Fatalf("恰 3 次，实际 %d", be.totalCalls())
	}
	if sink.countText("loop_exhausted") != 1 {
		t.Fatal("缺 loop_exhausted 事件")
	}
}

// ---- router：分支选中/落选跳过/无命中 fail-closed ----

func routerSteps() []StepSpec {
	return []StepSpec{
		{ID: "r1", Role: "researcher", Title: "路由", Targets: []string{"output.md", "fields.json"},
			Branches: []Branch{
				{When: BranchCond{Field: "route", Equals: "fast"}, Then: []string{"fast"}},
				{When: BranchCond{Field: "route", NotEquals: "fast"}, Then: []string{"slow"}},
			}},
		{ID: "fast", Role: "writer", Title: "快线", Needs: []string{"r1"}, Targets: []string{"output.md"}},
		{ID: "slow", Role: "writer", Title: "慢线", Needs: []string{"r1"}, Targets: []string{"output.md"}},
	}
}

func TestTopologyConformance_routerSelectsBranch(t *testing.T) {
	be := &iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte {
		if target == "fields.json" {
			return fieldsJSON(map[string]string{"route": "fast"})
		}
		return nil
	}}
	engine := newTopologyEngine(t, mockScriptPlanner(routerSteps()...), be, nil)
	res, err := engine.Run(context.Background(), Goal{Text: "路由任务"}, RunOptions{Topology: TopologyRouter})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunCompleted {
		t.Fatalf("分支落选是预期路径，不得降级/失败：%s %v", res.Status, res.Error)
	}
	byID := map[string]StepSpec{}
	for _, s := range res.Plan.Steps {
		byID[s.ID] = s
	}
	if byID["fast"].Status != "completed" {
		t.Fatalf("fast = %s", byID["fast"].Status)
	}
	if byID["slow"].Status != "skipped" || byID["slow"].Error != routerSkipReason {
		t.Fatalf("slow = %s %q", byID["slow"].Status, byID["slow"].Error)
	}
}

func TestTopologyConformance_routerNoMatchFailsClosed(t *testing.T) {
	be := &iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte {
		if target == "fields.json" {
			return fieldsJSON(map[string]string{"route": ""}) // 两条分支都不命中
		}
		return nil
	}}
	steps := routerSteps()
	// 去掉 NotEquals 兜底分支，制造无命中。
	steps[0].Branches = []Branch{{When: BranchCond{Field: "route", Equals: "fast"}, Then: []string{"fast"}}}
	engine := newTopologyEngine(t, mockScriptPlanner(steps...), be, nil)
	res, err := engine.Run(context.Background(), Goal{Text: "路由任务"}, RunOptions{Topology: TopologyRouter})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunFailed || !strings.Contains(res.Error, "无分支命中") {
		t.Fatalf("无命中必须 fail-closed：%s %v", res.Status, res.Error)
	}
}

// ---- evaluator-optimizer：draft→eval 循环复合，收敛即停 ----

func TestTopologyConformance_evaluatorOptimizer(t *testing.T) {
	// eval 第 1-2 轮 verdict=fail，第 3 轮 pass；draft 随 body 再武装。
	be := &iterBackend{calls: map[string]int{}, content: func(key string, iter int, target string) []byte {
		if target == "fields.json" {
			if iter >= 3 {
				return fieldsJSON(map[string]string{"verdict": "pass"})
			}
			return fieldsJSON(map[string]string{"verdict": "fail"})
		}
		if target == "verdict.md" {
			return []byte(fmt.Sprintf("评审第 %d 轮\n\n%s", iter, goodReport))
		}
		return nil
	}}
	engine := newTopologyEngine(t, mockScriptPlanner(
		StepSpec{ID: "draft", Role: "writer", Title: "起草", Targets: []string{"output.md"}},
		StepSpec{ID: "eval", Role: "reviewer", Title: "评审", Targets: []string{"verdict.md", "fields.json"},
			Loop: &LoopSpec{MaxIterations: 4, UntilField: "verdict", UntilEquals: "pass", Body: []string{"draft"}}},
	), be, nil)
	res, err := engine.Run(context.Background(), Goal{Text: "起草并评审"}, RunOptions{Topology: TopologyEvaluatorOptimizer})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunCompleted {
		t.Fatalf("status = %s err = %v", res.Status, res.Error)
	}
	byID := map[string]StepSpec{}
	for _, s := range res.Plan.Steps {
		byID[s.ID] = s
	}
	if byID["eval"].LoopRuns != 2 {
		t.Fatalf("eval 应再武装 2 轮：%+v", byID["eval"])
	}
	// draft 3 次 + eval 3 次。
	if be.totalCalls() != 6 {
		t.Fatalf("总执行次数 = %d，期望 6", be.totalCalls())
	}
}

// ---- long-running：截止时刻 → 未完步骤 skipped、降级收口 ----

func TestTopologyConformance_longRunningDeadline(t *testing.T) {
	// 时序鲁棒设计：每步 300ms、截止 +500ms——装配耗时不影响断言核心：
	// s1 必完成（首个 superstep 在截止前开跑），s3 必被跳过（第三个
	// superstep 边界必超期），收口必为 degraded。s2 完成与否取决于装配
	// 耗时，不做断言。
	be := &iterBackend{calls: map[string]int{}, delay: 300 * time.Millisecond}
	sink := &collectSink{}
	engine := newTopologyEngine(t, mockScriptPlanner(
		StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"notes.md"}},
		StepSpec{ID: "s2", Role: "writer", Title: "二", Targets: []string{"output.md"}},
		StepSpec{ID: "s3", Role: "reviewer", Title: "三", Targets: []string{"output.md"}},
	), be, sink)
	res, err := engine.Run(context.Background(), Goal{Text: "长任务"}, RunOptions{
		Topology: TopologyLongRunning,
		Deadline: time.Now().Add(500 * time.Millisecond),
	})
	if err != nil {
		t.Fatal(err)
	}
	if res.Status != RunDegraded {
		t.Fatalf("截止后应降级收口：%s %v", res.Status, res.Error)
	}
	byID := map[string]StepSpec{}
	for _, s := range res.Plan.Steps {
		byID[s.ID] = s
	}
	if byID["s1"].Status != "completed" {
		t.Fatalf("s1 = %s", byID["s1"].Status)
	}
	if byID["s3"].Status != "skipped" || byID["s3"].Error != "deadline exceeded" {
		t.Fatalf("s3 = %s %q", byID["s3"].Status, byID["s3"].Error)
	}
	if sink.countText("截止时刻") != 1 {
		t.Fatal("缺截止事件")
	}
}

// ---- 拓扑校验 fail-closed ----

func TestTopologyValidationFailsClosed(t *testing.T) {
	be := &iterBackend{calls: map[string]int{}}
	cases := []struct {
		name    string
		topo    string
		wantErr string
	}{
		{"未注册名", "bogus_topology", "未知拓扑"},
		{"primitive 单跑", TopologyMailbox, "跨拓扑原语"},
		{"primitive 单跑 handoff", TopologyHandoff, "跨拓扑原语"},
		// review/group_chat（WP3）与 task_board（WP4）驱动均已点亮；
		// 12/12 受理面由 TestTopologyConformance_allTwelve 汇总钉住。
	}
	for _, tc := range cases {
		engine := newTopologyEngine(t, mockScriptPlanner(
			StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}},
		), be, nil)
		res, err := engine.Run(context.Background(), Goal{Text: "x"}, RunOptions{Topology: tc.topo})
		if err != nil {
			t.Fatalf("%s：err = %v", tc.name, err)
		}
		if res.Status != RunFailed || !strings.Contains(res.Error, tc.wantErr) {
			t.Fatalf("%s：%s %q（期望含 %q）", tc.name, res.Status, res.Error, tc.wantErr)
		}
	}
}

// leader 拓扑名映射到既有 Mode 委派循环（本测只钉「受理且映射」，
// leader 行为 conformance 在 leader_test.go / WP2 增强）。
func TestTopologyLeaderMapsToMode(t *testing.T) {
	spec, ok := LookupTopology(TopologyLeader)
	if !ok || spec.Kind != TopologyMode {
		t.Fatalf("leader 注册形态：%+v %v", spec, ok)
	}
	if _, err := checkTopology(TopologyLeader); err != nil {
		t.Fatalf("leader 拓扑应可受理：%v", err)
	}
}

// 12 种内置拓扑全部注册且 Kind 合法（清单完整性）。
func TestBuiltinTopologiesComplete(t *testing.T) {
	want := map[string]TopologyKind{
		TopologySequential: TopologyGraph, TopologyParallel: TopologyGraph,
		TopologyLoop: TopologyGraph, TopologyRouter: TopologyGraph,
		TopologyLeader: TopologyMode, TopologyHandoff: TopologyPrimitive,
		TopologyGroupChat: TopologyDriver, TopologyReview: TopologyDriver,
		TopologyEvaluatorOptimizer: TopologyGraph, TopologyLongRunning: TopologyGraph,
		TopologyTaskBoard: TopologyDriver, TopologyMailbox: TopologyPrimitive,
	}
	got := BuiltinTopologies()
	if len(got) != len(want) {
		t.Fatalf("注册数 = %d，期望 %d", len(got), len(want))
	}
	for _, spec := range got {
		if k, ok := want[spec.Name]; !ok || k != spec.Kind {
			t.Fatalf("拓扑 %s kind = %s（期望 %s）", spec.Name, spec.Kind, k)
		}
	}
}