From c88100967d73e4572aa7eaa32c1e6bcd1a33e050 Mon Sep 17 00:00:00 2001 From: weichen Date: Thu, 8 Oct 2026 01:11:51 +0000 Subject: [PATCH] Add review driver, handoff primitive, and group-chat driver (Q2 WP3) Why: Q2 acceptance 2 requires the squad review topology to be equivalent to today's backend runReview flow (parallel finders, per-finding verification, drop unverified); handoff and group_chat complete 11 of 12 builtin topologies. Evidence: wp3_conformance_test.go mirrors backend review_test.go assertions (true finder concurrency via release gate, verify calls = 5 unique after dedupe, only PASS kept, dropped-count line, severity order); handoff records ownership transfer, target receives filtered history, finalize takes chain-tail output, unknown target role fails closed; group-chat rotate / close-marker / moderator conformance with budget stop and fail-closed unknown nominee. Full suite green with -race. Impact: review/group_chat drivers lit in the topology registry (task_board remains fail-closed until WP4); RunState gains Handoffs (persisted); RunOptions gains GroupChat*/Review* knobs; runOneStep validates handoff targets and injects handoff context; finalize prefers chain-tail deliverable. Parse/dedupe/verdict/severity/report formats ported verbatim from backend internal/runtime/review.go for byte-level report equivalence. Refs: Q2 plan WP3 (2026-10-07-agent-stack-q2q3-plans.md); task card .task/feature-agent-stack-q2.md --- squad/engine.go | 66 +++++ squad/groupchat.go | 355 ++++++++++++++++++++++ squad/handoff.go | 91 ++++++ squad/review.go | 366 +++++++++++++++++++++++ squad/runstate.go | 6 + squad/topology_conformance_test.go | 3 +- squad/workflow.go | 30 +- squad/wp3_conformance_test.go | 452 +++++++++++++++++++++++++++++ 8 files changed, 1366 insertions(+), 3 deletions(-) create mode 100644 squad/groupchat.go create mode 100644 squad/handoff.go create mode 100644 squad/review.go create mode 100644 squad/wp3_conformance_test.go diff --git a/squad/engine.go b/squad/engine.go index d050f14..64847cb 100644 --- a/squad/engine.go +++ b/squad/engine.go @@ -114,6 +114,16 @@ type RunOptions struct { // MaxReplans leader 模式的重规划上限（Q2；与 MaxLeaderTurns 独立）。 // 0 = 默认 DefaultMaxReplans。 MaxReplans int `json:"max_replans,omitempty"` + // GroupChatPolicy group_chat 拓扑发言人策略（Q2）：rotate（默认）| moderator。 + GroupChatPolicy string `json:"group_chat_policy,omitempty"` + // GroupChatModerator 主持人角色 key（moderator 策略必填，须在名册内）。 + GroupChatModerator string `json:"group_chat_moderator,omitempty"` + // GroupChatMaxTurns 群聊轮次预算（0 = DefaultMaxGroupChatTurns）。 + GroupChatMaxTurns int `json:"group_chat_max_turns,omitempty"` + // ReviewVerifyRole review 拓扑核实角色 key（默认 "verifier"）。 + ReviewVerifyRole string `json:"review_verify_role,omitempty"` + // ReviewTarget 被评审材料（默认取 context/goal.md）。 + ReviewTarget string `json:"review_target,omitempty"` } // SquadEngine 小队引擎门面：Run / Respond / Resume / Inspect / ListRuns。 @@ -258,6 +268,21 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (* if runOpts.MaxReplans > 0 { meta.Metadata["max_replans"] = fmt.Sprintf("%d", runOpts.MaxReplans) } + if runOpts.GroupChatPolicy != "" { + meta.Metadata["group_chat_policy"] = runOpts.GroupChatPolicy + } + if runOpts.GroupChatModerator != "" { + meta.Metadata["group_chat_moderator"] = runOpts.GroupChatModerator + } + if runOpts.GroupChatMaxTurns > 0 { + meta.Metadata["group_chat_max_turns"] = fmt.Sprintf("%d", runOpts.GroupChatMaxTurns) + } + if runOpts.ReviewVerifyRole != "" { + meta.Metadata["review_verify_role"] = runOpts.ReviewVerifyRole + } + if runOpts.ReviewTarget != "" { + meta.Metadata["review_target"] = runOpts.ReviewTarget + } e.deps.Sink.Emit(Event{Type: EvRunStarted, RunID: runID, Text: goal.Text, At: time.Now()}) // 意图闸门（可选，入口一次裁决；显式 TaskKey 优先，不重复过闸）。 @@ -434,6 +459,15 @@ func (e *SquadEngine) planAndExecute(ctx context.Context, st *RunState, runOpts } return out, nil } + // driver 类拓扑（Q2 WP3+）：按名分发专用驱动（review/group_chat； + // task_board 属 WP4，未点亮前 checkTopology 已 fail-closed）。 + if out, handled, derr := e.dispatchTopologyDriver(ctx, st); handled { + if derr != nil { + return nil, derr + } + e.finalizeEvent(out) + return out, nil + } out, _, err := drv.advance(ctx, st) if err != nil { return nil, err @@ -494,6 +528,15 @@ func (e *SquadEngine) resumeAfterHold(ctx context.Context, st *RunState) (*RunRe } return out, nil } + // driver 类拓扑续跑：驱动自身幂等（review 已完成 finder 不重跑； + // group_chat 按已落盘轮次续转）。 + if out, handled, err := e.dispatchTopologyDriver(ctx, st); handled { + if err != nil { + return nil, err + } + e.finalizeEvent(out) + return out, nil + } drv := &flowDriver{opts: e.deps} out, _, err := drv.advance(ctx, st) if err != nil { @@ -503,6 +546,29 @@ func (e *SquadEngine) resumeAfterHold(ctx context.Context, st *RunState) (*RunRe return out, nil } +// dispatchTopologyDriver 按运行拓扑分发 driver 类拓扑。 +// handled=false 表示非 driver 拓扑（走既有图/模式路径）。 +func (e *SquadEngine) dispatchTopologyDriver(ctx context.Context, st *RunState) (*RunResult, bool, error) { + topo := st.Meta.Metadata["topology"] + if topo == "" { + return nil, false, nil + } + spec, ok := LookupTopology(topo) + if !ok || spec.Kind != TopologyDriver { + return nil, false, nil + } + switch topo { + case TopologyReview: + out, err := e.runReviewDriver(ctx, st) + return out, true, err + case TopologyGroupChat: + out, err := e.runGroupChatDriver(ctx, st) + return out, true, err + default: + return e.driverFail(st, fmt.Sprintf("topology: %q 驱动未实现（fail-closed）", topo)), true, nil + } +} + // materializeWorkspace 落盘工作区与子工作区（已存在则只更新计划并重建待执行子目录）。 func (e *SquadEngine) materializeWorkspace(meta *RunMetadata, st *RunState, plan *SquadPlan) error { if st.Meta.WorkDir == "" { diff --git a/squad/groupchat.go b/squad/groupchat.go new file mode 100644 index 0000000..8501999 --- /dev/null +++ b/squad/groupchat.go @@ -0,0 +1,355 @@ +package squad + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// groupchat.go — group_chat 拓扑驱动（Q2 WP3）。 +// 形态（计划 §三 WP3 与边界 §五）：共享转录（BB 字段 transcript + +// group_chat/transcript.md 全量落盘）；发言人策略只做**轮转**与**主持人** +// 两种（脚本可判定，LLM 选人策略不进库）；预算停机（max_turns，缺省 8）。 +// 蓝图步骤 = 参与者名册（按序，不直接执行，driver 标记 skipped 并派生轮次 +// 步骤）；任一发言含 [close] 标记即收口；主持人模式下主持人输出 JSON 决策 +// （{"next":"role"} / {"action":"close"}），未知角色/非法 JSON fail-closed。 +// 驱动幂等可恢复：轮次由已落盘的 gc 步骤数重建，挂起（权限/HITL）后 +// Respond 重入续转。 + +func init() { + _ = RegisterTopologyDriver(TopologyGroupChat) +} + +// DefaultMaxGroupChatTurns 群聊轮次预算缺省值（预算停机）。 +const DefaultMaxGroupChatTurns = 8 + +// gcRosterSkipReason 名册步骤的 skipped 原因（非执行步骤，不算降级）。 +const gcRosterSkipReason = "group_chat: 名册步骤（参与者由驱动接管）" + +// gcCloseMarker 任一发言包含该标记（大小写不敏感）即收口。 +const gcCloseMarker = "[close]" + +// gcTranscriptCapBB 是 BB 转录字段的容量上限（TASK.md 快照只带近端）。 +const gcTranscriptCapBB = 4000 + +// runGroupChatDriver group_chat 拓扑入口。 +func (e *SquadEngine) runGroupChatDriver(ctx context.Context, st *RunState) (*RunResult, error) { + drv := &flowDriver{opts: e.deps} + + // 名册：首次进入从蓝图步骤取参与者序列并持久化到元数据（恢复时直读）。 + roster, err := e.gcRoster(st) + if err != nil { + return e.driverFail(st, err.Error()), nil + } + policy := orDefault(st.Meta.Metadata["group_chat_policy"], "rotate") + moderator := st.Meta.Metadata["group_chat_moderator"] + if policy != "rotate" && policy != "moderator" { + return e.driverFail(st, fmt.Sprintf("group_chat: 未知发言人策略 %q（只支持 rotate/moderator）", policy)), nil + } + if policy == "moderator" { + if moderator == "" || !containsString(roster, moderator) { + return e.driverFail(st, fmt.Sprintf("group_chat: 主持人 %q 不在名册内（fail-closed）", moderator)), nil + } + } + maxTurns := DefaultMaxGroupChatTurns + if v := st.Meta.Metadata["group_chat_max_turns"]; v != "" { + if n := parseIntDefault(v, DefaultMaxGroupChatTurns); n > 0 { + maxTurns = n + } + } + + // 名册步骤标记 skipped（首次进入时；恢复重入无 pending 名册步骤）。 + st.mu.Lock() + for i := range st.Plan.Workflow.Steps { + s := &st.Plan.Workflow.Steps[i] + if !strings.HasPrefix(s.ID, "gc") && (s.Status == "pending" || s.Status == "running") { + s.Status = "skipped" + s.Error = gcRosterSkipReason + } + } + _ = st.save() + st.mu.Unlock() + + for { + // 恢复重入：上次挂起留下的未完 gc 轮次步骤优先续跑（不重新选人）。 + if idx := e.gcHeldStepIdx(st); idx >= 0 { + out, gate, err := e.gcRunStep(ctx, drv, st, idx) + if err != nil { + return nil, err + } + if gate != nil { + if out == nil { + out = heldResult(st, gate) + } + return out, nil + } + if out != nil { + return out, nil + } + if e.gcMessageClosed(st, idx) { + drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("group_chat：%s 发出收口标记，停机", st.Plan.Workflow.Steps[idx].Role)}) + break + } + continue + } + turns := e.gcTurnCount(st) + if turns >= maxTurns { + drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("group_chat：轮次预算 %d 用尽，停机收口", maxTurns)}) + break + } + speaker, closed, derr := e.gcNextSpeaker(st, roster, policy, moderator, turns) + if derr != nil { + return e.driverFail(st, derr.Error()), nil + } + if closed { + drv.emit(st, Event{Type: EvPhase, Text: "group_chat：主持人收口"}) + break + } + out, gate, err := e.gcRunTurn(ctx, drv, st, turns, speaker) + if err != nil { + return nil, err + } + if gate != nil { + if out == nil { + out = heldResult(st, gate) + } + return out, nil + } + if out != nil { // 轮次步骤失败（关键）→ fail-closed 传播 + return out, nil + } + if e.gcMessageClosed(st, len(st.Plan.Workflow.Steps)-1) { + drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("group_chat：%s 发出收口标记，停机", speaker)}) + break + } + } + return e.gcFinish(st, drv), nil +} + +// gcRoster 参与者名册（首次从蓝图步骤派生并写入元数据；恢复直读）。 +func (e *SquadEngine) gcRoster(st *RunState) ([]string, error) { + if v := st.Meta.Metadata["group_chat_roster"]; v != "" { + return strings.Split(v, ","), nil + } + var roster []string + for _, s := range st.Plan.Workflow.Steps { + if s.Role != "" && !containsString(roster, s.Role) { + roster = append(roster, s.Role) + } + } + if len(roster) == 0 { + return nil, fmt.Errorf("group_chat: 名册为空（蓝图步骤即参与者）") + } + st.mu.Lock() + if st.Meta.Metadata == nil { + st.Meta.Metadata = map[string]string{} + } + st.Meta.Metadata["group_chat_roster"] = strings.Join(roster, ",") + _ = st.save() + st.mu.Unlock() + return roster, nil +} + +// gcTurnCount 已完成的轮次数（预算与轮转序号的基准；恢复重入自然续算）。 +func (e *SquadEngine) gcTurnCount(st *RunState) int { + n := 0 + for _, s := range st.Plan.Workflow.Steps { + if strings.HasPrefix(s.ID, "gc") && s.Status == "completed" { + n++ + } + } + return n +} + +// gcHeldStepIdx 返回上次挂起留下的未完 gc 轮次步骤下标（无则 -1）。 +func (e *SquadEngine) gcHeldStepIdx(st *RunState) int { + for i := range st.Plan.Workflow.Steps { + s := &st.Plan.Workflow.Steps[i] + if strings.HasPrefix(s.ID, "gc") && (s.Status == "pending" || s.Status == "running") { + return i + } + } + return -1 +} + +// gcNextSpeaker 决定下一发言人。rotate = 轮转；moderator = 主持人裁决 +// （上一步是参与者发言 → 主持人接手；主持人决策 close → 收口，next → 指名， +// 非法 JSON / 未知角色 fail-closed）。 +func (e *SquadEngine) gcNextSpeaker(st *RunState, roster []string, policy, moderator string, turns int) (string, bool, error) { + if policy == "rotate" { + return roster[turns%len(roster)], false, nil + } + last := e.gcLastStep(st) + if last == nil || last.Role == moderator { + // 首轮或上一步就是主持人（不该连续，防御性让主持人接手）。 + if last != nil && last.Role == moderator { + // 解析主持人决策。 + dec, err := e.gcParseDecision(st, last) + if err != nil { + return "", false, err + } + if dec.Action == "close" { + return "", true, nil + } + if !containsString(roster, dec.Next) { + return "", false, fmt.Errorf("group_chat: 主持人指名 %q 不在名册内（fail-closed）", dec.Next) + } + return dec.Next, false, nil + } + return moderator, false, nil + } + // 上一步是参与者发言 → 主持人接手裁决。 + return moderator, false, nil +} + +type gcDecision struct { + Action string `json:"action"` + Next string `json:"next"` +} + +// gcParseDecision 宽松解析主持人 message.md 里的 JSON 决策（首 `{` 末 `}`）。 +func (e *SquadEngine) gcParseDecision(st *RunState, step *StepSpec) (*gcDecision, error) { + msg := e.gcMessageOf(st, step) + start := strings.Index(msg, "{") + end := strings.LastIndex(msg, "}") + if start < 0 || end <= start { + return nil, fmt.Errorf("group_chat: 主持人输出无 JSON 决策（fail-closed）：%q", truncateStr(msg, 80)) + } + var dec gcDecision + if err := json.Unmarshal([]byte(msg[start:end+1]), &dec); err != nil { + return nil, fmt.Errorf("group_chat: 主持人决策解析失败（fail-closed）: %v", err) + } + if dec.Action == "close" { + return &dec, nil + } + if strings.TrimSpace(dec.Next) == "" { + return nil, fmt.Errorf("group_chat: 主持人决策缺 next/action（fail-closed）") + } + return &dec, nil +} + +// gcRunTurn 派生一轮发言步骤并执行。 +func (e *SquadEngine) gcRunTurn(ctx context.Context, drv *flowDriver, st *RunState, turns int, speaker string) (*RunResult, *HitlGate, error) { + step := StepSpec{ + ID: fmt.Sprintf("gc%02d", turns+1), + Role: speaker, + Title: fmt.Sprintf("群聊第 %d 轮：%s 发言（把发言写入 message.md；同意收口就写 [close]）", turns+1, speaker), + Targets: []string{"message.md"}, + SubWorkspace: fmt.Sprintf("group_chat/turn_%02d_%s", turns+1, speaker), + Status: "pending", + } + st.mu.Lock() + st.Plan.Workflow.Steps = append(st.Plan.Workflow.Steps, step) + idx := len(st.Plan.Workflow.Steps) - 1 + _ = st.save() + st.mu.Unlock() + return e.gcRunStep(ctx, drv, st, idx) +} + +// gcRunStep 执行一个轮次步骤：过门 → runOneStep → 转录并入共享状态。 +// 挂起（gate）时步骤保持未完，恢复重入经 gcHeldStepIdx 续跑同一步骤。 +func (e *SquadEngine) gcRunStep(ctx context.Context, drv *flowDriver, st *RunState, idx int) (*RunResult, *HitlGate, error) { + out, gate, err := drv.gateBeforeRun(ctx, st, idx) + if err != nil || gate != nil || out != nil { + return out, gate, err + } + step := st.Plan.Workflow.Steps[idx] + drv.emit(st, Event{Type: EvPhase, StepID: step.ID, Role: step.Role, + Text: fmt.Sprintf("group_chat：%s 发言（第 %d 轮）", step.Role, e.gcTurnCount(st)+1)}) + out, gate, err = drv.runOneStep(ctx, st, idx) + if err != nil || gate != nil || out != nil { + return out, gate, err + } + // 转录并入共享状态：全量落盘 transcript.md，近端进 BB 字段（下一位 + // 发言人的 TASK.md 黑板快照可见）。 + msg := e.gcMessageOf(st, &st.Plan.Workflow.Steps[idx]) + e.gcAppendTranscript(st, e.gcTurnCount(st), step.Role, msg) + return nil, nil, nil +} + +// gcAppendTranscript 追加转录（全量文件 + BB 近端快照）。 +func (e *SquadEngine) gcAppendTranscript(st *RunState, turn int, speaker, msg string) { + dir := filepath.Join(st.Meta.WorkDir, "group_chat") + _ = os.MkdirAll(dir, 0o755) + line := fmt.Sprintf("%d. %s：%s\n", turn, speaker, strings.TrimSpace(msg)) + f, _ := os.OpenFile(filepath.Join(dir, "transcript.md"), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644) + if f != nil { + _, _ = f.WriteString(line) + _ = f.Close() + } + st.mu.Lock() + if st.BB.Fields == nil { + st.BB.Fields = map[string]string{} + } + cur := st.BB.Fields["transcript"] + line + if len(cur) > gcTranscriptCapBB { + cur = cur[len(cur)-gcTranscriptCapBB:] + } + st.BB.Fields["transcript"] = cur + _ = st.save() + st.mu.Unlock() +} + +// gcLastStep 最后一个已完成的派生轮次步骤（主持人决策解析的锚点）。 +func (e *SquadEngine) gcLastStep(st *RunState) *StepSpec { + for i := len(st.Plan.Workflow.Steps) - 1; i >= 0; i-- { + s := &st.Plan.Workflow.Steps[i] + if strings.HasPrefix(s.ID, "gc") && s.Status == "completed" { + return s + } + } + return nil +} + +// gcMessageOf 读取某轮的 message.md（缺文件 = 空串）。 +func (e *SquadEngine) gcMessageOf(st *RunState, step *StepSpec) string { + data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, filepath.FromSlash(step.SubWorkspace), "message.md")) + if err != nil { + return "" + } + return string(data) +} + +// gcMessageClosed 第 idx 步的发言是否带收口标记。 +func (e *SquadEngine) gcMessageClosed(st *RunState, idx int) bool { + if idx < 0 || idx >= len(st.Plan.Workflow.Steps) { + return false + } + msg := e.gcMessageOf(st, &st.Plan.Workflow.Steps[idx]) + return strings.Contains(strings.ToLower(msg), gcCloseMarker) +} + +// gcFinish 收口：交付物 = 全量转录；写 output.md；状态落盘；构造结果。 +func (e *SquadEngine) gcFinish(st *RunState, drv *flowDriver) *RunResult { + transcript := "" + if data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "group_chat", "transcript.md")); err == nil { + transcript = string(data) + } + report := "# 群聊转录\n\n" + transcript + _ = os.WriteFile(filepath.Join(st.Meta.WorkDir, "output.md"), []byte(report), 0o644) + turns := e.gcTurnCount(st) + st.mu.Lock() + wf := st.Workflow() + wf.Status = string(RunCompleted) + st.setWorkflow(wf) + st.Meta.Status = string(RunCompleted) + now := time.Now() + st.Meta.CompletedAt = &now + _ = st.save() + st.mu.Unlock() + summary := fmt.Sprintf("group_chat：%d 轮发言，转录已收口", turns) + drv.emit(st, Event{Type: EvPhase, Text: summary}) + return &RunResult{RunID: st.Meta.ID, Status: RunCompleted, Plan: planPtr(st), + Deliverable: "output.md", Summary: summary, WorkDir: st.Meta.WorkDir} +} + +func truncateStr(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/squad/handoff.go b/squad/handoff.go new file mode 100644 index 0000000..b58be68 --- /dev/null +++ b/squad/handoff.go @@ -0,0 +1,91 @@ +package squad + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// handoff.go — handoff 跨拓扑原语（Q2 WP3）。 +// 语义（计划 §三 WP3）：StepSpec.Handoff 非空的步骤完成后，最终答案所有权 +// 移交给目标角色——① 记录所有权转移（Handoffs，随 RunState 持久化）； +// ② 目标角色的下一个步骤收到「过滤历史」：只带移交步骤的主产出，不带全量 +// 转录（一次性并入任务书，与 mailbox 同款机制）；③ finalize 汇总交付物时 +// 取链尾（最后一次移交的目标角色）的最后一个完成步骤产出，而非全图最后 +// 完成步骤。目标角色不在小队内 = fail-closed（步骤启动前拒绝）。 + +// HandoffRecord 是一次所有权转移记录。 +type HandoffRecord struct { + FromRole string `json:"from_role"` + FromStep string `json:"from_step"` + ToRole string `json:"to_role"` + Output string `json:"output,omitempty"` // 移交步骤主产出（截断上限内） + At time.Time `json:"at"` + Delivered bool `json:"delivered,omitempty"` +} + +// handoffOutputCap 移交上下文的产出截断上限（过滤历史：只带主产出，且限量）。 +const handoffOutputCap = 64 * 1024 + +// takeHandoffFor 取出移交给某角色的未投递记录，标记已投递（一次性）。 +// 调用方负责随后 st.save()（或在临界区内）。 +func takeHandoffFor(st *RunState, role string) *HandoffRecord { + for i := range st.Handoffs { + h := &st.Handoffs[i] + if h.Delivered || h.ToRole != role { + continue + } + h.Delivered = true + out := *h + return &out + } + return nil +} + +// formatHandoffSection 渲染任务书里的移交上下文段（无则空串）。 +func formatHandoffSection(h *HandoffRecord) string { + if h == nil { + return "" + } + body := h.Output + if len(body) > handoffOutputCap { + body = body[:handoffOutputCap] + "\n…（已截断）" + } + var b strings.Builder + fmt.Fprintf(&b, "\n\n## 移交上下文（所有权来自 %s，步骤 %s）\n", h.FromRole, h.FromStep) + b.WriteString("以下是移交方的主产出（过滤历史：只含该产出，不含其过程转录）：\n\n") + b.WriteString(body) + b.WriteString("\n") + return b.String() +} + +// handoffChainTail 返回移交链尾角色（无移交记录返回空串）。 +func handoffChainTail(st *RunState) string { + if n := len(st.Handoffs); n > 0 { + return st.Handoffs[n-1].ToRole + } + return "" +} + +// recordHandoff 步骤完成时的所有权转移记录：主产出快照（截断上限内）+ +// 持久化 + 事件。目标角色合法性已在步骤启动前 fail-closed 校验。 +func (d *flowDriver) recordHandoff(st *RunState, step StepSpec, subDir string, targets []string) { + output := "" + if data, err := os.ReadFile(filepath.Join(subDir, firstTarget(targets))); err == nil { + output = string(data) + } + if len(output) > handoffOutputCap { + output = output[:handoffOutputCap] + "\n…（已截断）" + } + st.mu.Lock() + st.Handoffs = append(st.Handoffs, HandoffRecord{ + FromRole: step.Role, FromStep: step.ID, ToRole: step.Handoff, + Output: output, At: time.Now(), + }) + _ = st.save() + st.mu.Unlock() + d.emit(st, Event{Type: EvPhase, StepID: step.ID, Role: step.Role, + Text: fmt.Sprintf("handoff：%s → %s（最终答案所有权移交）", step.Role, step.Handoff)}) +} diff --git a/squad/review.go b/squad/review.go new file mode 100644 index 0000000..edc934e --- /dev/null +++ b/squad/review.go @@ -0,0 +1,366 @@ +package squad + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + "time" +) + +// review.go — review 拓扑驱动（Q2 WP3，验收二）。 +// 与 backend internal/runtime/review.go 的 runReview 等价： +// 并行 N finder（semaphore 3）→ 宽松 JSON 解析（首 `[` 末 `]`，丢空 issue） +// → 按 (location,issue) 去重、上限 10 → 每条独立 verifier（尾行 +// VERDICT: PASS/FAIL/PARTIAL）→ 只留 PASS（fail-closed）→ severity 稳定排序 +// → 报告恒带「已丢弃 N 条未验证」。解析/去重/判定/排序/文案与 backend 同一 +// 格式串（计划 §一.5：以报告字节形态为等价判据）。 +// +// squad 侧接线：finder = 计划步骤（各自的 findings.json 为目标产物，经 +// runOneStep 走完整工作区/事件/HITL 机制）；verifier = 驱动内按 verify 角色 +// 直接 Execute 的核实调用（对应 backend 的 spawn(verifyProfileID,…)）， +// 每次核实落盘 reviews/verify_NN/ 留痕。 + +func init() { + _ = RegisterTopologyDriver(TopologyReview) +} + +// reviewFinding 与 backend reviewFinding 同构（JSON 键一致）。 +type reviewFinding struct { + Location string `json:"location"` + Severity string `json:"severity"` + Issue string `json:"issue"` + Evidence string `json:"evidence"` +} + +// reviewVerbatimTask 是核实调用的任务文案（与 backend runReview 同串）。 +const reviewVerbatimTask = "独立核实这条问题是否成立。材料如下：\n%s\n\n问题：%s\n只在结尾给出一行 VERDICT: PASS、FAIL 或 PARTIAL。" + +// parseFindings 宽松解析：首 `[` 末 `]`，丢空 issue（backend 同款）。 +func parseFindings(text string) []reviewFinding { + start := strings.Index(text, "[") + end := strings.LastIndex(text, "]") + if start < 0 || end <= start { + return nil + } + var out []reviewFinding + if err := json.Unmarshal([]byte(text[start:end+1]), &out); err != nil { + return nil + } + var kept []reviewFinding + for _, f := range out { + if strings.TrimSpace(f.Issue) == "" { + continue + } + kept = append(kept, f) + } + return kept +} + +// dedupeFindings 按 (location,issue) 去重，上限 limit（backend 同款）。 +func dedupeFindings(groups [][]reviewFinding, limit int) []reviewFinding { + seen := map[string]bool{} + var out []reviewFinding + for _, group := range groups { + for _, f := range group { + key := strings.TrimSpace(f.Location) + "\n" + strings.TrimSpace(f.Issue) + if seen[key] { + continue + } + seen[key] = true + out = append(out, f) + if len(out) >= limit { + return out + } + } + } + return out +} + +// reviewVerdict 尾行 VERDICT 判定（backend 同款：PASS/FAIL/PARTIAL， +// 解析不出 = 空串 = 不通过）。 +func reviewVerdict(text string) string { + lines := strings.Split(text, "\n") + for i := len(lines) - 1; i >= 0; i-- { + line := strings.TrimSpace(lines[i]) + if line == "" || !strings.HasPrefix(strings.ToUpper(line), "VERDICT:") { + continue + } + v := strings.TrimSpace(line[len("VERDICT:"):]) + v = strings.ToUpper(v) + switch { + case strings.HasPrefix(v, "PASS"): + return "PASS" + case strings.HasPrefix(v, "FAIL"): + return "FAIL" + case strings.HasPrefix(v, "PARTIAL"): + return "PARTIAL" + default: + return "" + } + } + return "" +} + +// severityRank 稳定排序键（backend 同款）。 +func severityRank(s string) int { + switch strings.ToLower(strings.TrimSpace(s)) { + case "critical": + return 0 + case "high": + return 1 + case "medium": + return 2 + case "low": + return 3 + default: + return 4 + } +} + +// reviewParallel semaphore-3 并行（backend parallel 同款）。 +func reviewParallel(n int, fn func(i int)) { + if n <= 0 { + return + } + sem := make(chan struct{}, 3) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + fn(i) + }(i) + } + wg.Wait() +} + +// runReviewDriver review 拓扑入口。finder 阶段复用计划步骤（可挂起可恢复： +// 重入时已完成步骤不重跑，findings.json 从各自子工作区读取）。 +func (e *SquadEngine) runReviewDriver(ctx context.Context, st *RunState) (*RunResult, error) { + drv := &flowDriver{opts: e.deps} + steps := st.Plan.Workflow.Steps + + // verify 角色先行解析（fail-fast：缺角色不浪费 finder 轮次）。 + verifyRoleName := orDefault(st.Meta.Metadata["review_verify_role"], "verifier") + verifyRole, ok := findAgent(st.Plan.Squad.Agents, verifyRoleName) + if !ok { + if r, exists := e.deps.Registry.Get(verifyRoleName); exists { + verifyRole, ok = r, true + } + } + if !ok { + return e.driverFail(st, fmt.Sprintf("review: verify 角色 %q 不在小队内（fail-closed）", verifyRoleName)), nil + } + verifyBackend, _, rerr := e.deps.Backends.Resolve(verifyRole, st.Meta.Backend) + if rerr != nil { + return e.driverFail(st, "review: verify 后端解析失败: "+rerr.Error()), nil + } + + // 1) finder 阶段：先全组过门（任一要挂起则整组不执行，与 advance 并行组 + // 同款语义），再 semaphore-3 并行执行。 + var finderIdx []int + for i := range steps { + if steps[i].Status == "pending" || steps[i].Status == "running" { + finderIdx = append(finderIdx, i) + } + } + for _, idx := range finderIdx { + out, gate, err := drv.gateBeforeRun(ctx, st, idx) + if err != nil { + return nil, err + } + if gate != nil { + if out == nil { + out = heldResult(st, gate) + } + return out, nil + } + if out != nil { + return out, nil + } + } + drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("review：%d 个 finder 并行找问题", len(finderIdx))}) + var holdGate *HitlGate + var holdOut *RunResult + var fatal *RunResult + finderFailed := 0 + var fmu sync.Mutex + reviewParallel(len(finderIdx), func(k int) { + idx := finderIdx[k] + out, gate, err := drv.runOneStep(ctx, st, idx) + fmu.Lock() + defer fmu.Unlock() + if err != nil { + if fatal == nil { + fatal = &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), + Error: err.Error(), WorkDir: st.Meta.WorkDir} + } + return + } + if gate != nil && holdGate == nil { + holdGate = gate + holdOut = out + return + } + if out != nil && out.Status == RunFailed && fatal == nil { + fatal = out // 关键 finder 失败：fail-closed 传播 + return + } + if st.Plan.Workflow.Steps[idx].Status == "failed" { + finderFailed++ // 非关键 finder 失败：宽容（backend spawn err 同款），收口降级 + } + }) + if fatal != nil { + return fatal, nil + } + if holdGate != nil { + if holdOut == nil { + holdOut = heldResult(st, holdGate) + } + return holdOut, nil + } + + // 2) 收集各 finder 的 findings.json（宽松解析；缺文件/坏 JSON = 该 finder + // 零发现，与 backend 同款）。 + steps = st.Plan.Workflow.Steps + groups := make([][]reviewFinding, 0, len(steps)) + for i := range steps { + if steps[i].Status != "completed" { + continue + } + subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(steps[i].SubWorkspace)) + data, err := os.ReadFile(filepath.Join(subDir, "findings.json")) + if err != nil { + continue + } + groups = append(groups, parseFindings(string(data))) + } + unique := dedupeFindings(groups, 10) + drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("review：去重后 %d 条，逐条独立核实", len(unique))}) + + // 3) 逐条 verify（semaphore-3 并行；fail-closed：核实调用失败/无判定 = 丢弃）。 + target := reviewTargetMaterial(st) + kept := make([]reviewFinding, len(unique)) + pass := make([]bool, len(unique)) + reviewParallel(len(unique), func(i int) { + f := unique[i] + body, _ := json.Marshal(f) + task := fmt.Sprintf(reviewVerbatimTask, target, body) + vDir := filepath.Join(st.Meta.WorkDir, "reviews", fmt.Sprintf("verify_%02d", i+1)) + if err := os.MkdirAll(vDir, 0o755); err != nil { + return + } + instructions := "" + if e.deps.Workspaces != nil { + instructions = e.deps.Workspaces.ResolveInstructions(verifyRole) + } + res, err := verifyBackend.Execute(ctx, ExecRequest{ + WorkDir: vDir, + Prompt: task, + Instructions: instructions, + Targets: []string{"verdict.md"}, + Timeout: e.deps.StepTimeout, + SessionScope: st.Meta.ID + "/verify_" + fmt.Sprintf("%02d", i+1), + }) + if err != nil || res == nil { + return + } + text := string(res.Patches["verdict.md"]) + "\n" + res.Logs + if data, rerr2 := os.ReadFile(filepath.Join(vDir, "verdict.md")); rerr2 == nil { + text = string(data) + "\n" + text + } + if reviewVerdict(text) == "PASS" { + kept[i] = f + pass[i] = true + } + }) + var verified []reviewFinding + for i, ok := range pass { + if ok { + verified = append(verified, kept[i]) + } + } + + // 4) severity 稳定排序 + 等价报告（backend 同格式串）。 + sort.SliceStable(verified, func(i, j int) bool { + return severityRank(verified[i].Severity) < severityRank(verified[j].Severity) + }) + dropped := len(unique) - len(verified) + var b strings.Builder + if len(verified) == 0 { + b.WriteString("没有验证通过的问题。\n") + } + for _, f := range verified { + fmt.Fprintf(&b, "- [%s] %s：%s\n 证据：%s\n", + orDefault(f.Severity, "medium"), orDefault(f.Location, "未定位"), f.Issue, f.Evidence) + } + fmt.Fprintf(&b, "已丢弃 %d 条未验证\n", dropped) + report := b.String() + + // 5) 收口：报告落盘（reviews/report.md + output.md 交付物约定）， + // finder 有失败则降级（诚实信号），否则完成。 + reviewsDir := filepath.Join(st.Meta.WorkDir, "reviews") + _ = os.MkdirAll(reviewsDir, 0o755) + _ = os.WriteFile(filepath.Join(reviewsDir, "report.md"), []byte(report), 0o644) + _ = os.WriteFile(filepath.Join(st.Meta.WorkDir, "output.md"), []byte(report), 0o644) + st.mu.Lock() + if st.BB.Fields == nil { + st.BB.Fields = map[string]string{} + } + st.BB.Fields["review_verified"] = fmt.Sprintf("%d", len(verified)) + st.BB.Fields["review_dropped"] = fmt.Sprintf("%d", dropped) + wf := st.Workflow() + status := RunCompleted + if finderFailed > 0 { + status = RunDegraded + } + wf.Status = string(status) + st.setWorkflow(wf) + st.Meta.Status = string(status) + now := time.Now() + st.Meta.CompletedAt = &now + _ = st.save() + st.mu.Unlock() + summary := fmt.Sprintf("review：保留 %d 条 / 丢弃 %d 条未验证", len(verified), dropped) + if finderFailed > 0 { + summary += fmt.Sprintf("（%d 个 finder 失败，降级）", finderFailed) + } + drv.emit(st, Event{Type: EvPhase, Text: summary}) + return &RunResult{RunID: st.Meta.ID, Status: status, Plan: planPtr(st), + Deliverable: "output.md", Summary: summary, WorkDir: st.Meta.WorkDir}, nil +} + +// reviewTargetMaterial 取被评审材料：元数据 review_target 优先， +// 其次运行目标（context/goal.md）。 +func reviewTargetMaterial(st *RunState) string { + if v := strings.TrimSpace(st.Meta.Metadata["review_target"]); v != "" { + return v + } + if data, err := os.ReadFile(filepath.Join(st.Meta.WorkDir, "context", "goal.md")); err == nil { + return string(data) + } + return "" +} + +// driverFail 驱动级 fail-closed 收口（与 leaderFail 同款语义）。 +func (e *SquadEngine) driverFail(st *RunState, reason string) *RunResult { + now := time.Now() + st.mu.Lock() + st.Meta.Status = string(RunFailed) + st.Meta.CompletedAt = &now + wf := st.Workflow() + wf.Status = "failed" + st.setWorkflow(wf) + _ = st.save() + st.mu.Unlock() + (&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, Text: reason}) + return &RunResult{RunID: st.Meta.ID, Status: RunFailed, Plan: planPtr(st), + Error: reason, WorkDir: st.Meta.WorkDir} +} diff --git a/squad/runstate.go b/squad/runstate.go index 23509ed..99aedaf 100644 --- a/squad/runstate.go +++ b/squad/runstate.go @@ -30,6 +30,9 @@ type RunState struct { // Mailbox 运行内点对点消息（Q2）：随 RunState 持久化；只在步骤启动时 // 并入任务书，永不触碰 HITL 门。 Mailbox []MailMessage `json:"mailbox,omitempty"` + // Handoffs 所有权转移记录（Q2 handoff 原语）：随 RunState 持久化； + // 目标角色的下一步骤一次性接收移交上下文，finalize 取链尾产出。 + Handoffs []HandoffRecord `json:"handoffs,omitempty"` // Usage 按角色累计的 token 用量。 Usage map[string]*Usage `json:"usage,omitempty"` // Leader leader-delegation 模式的循环状态（workspace 内 leader.json 持久化）。 @@ -96,6 +99,7 @@ func (st *RunState) persist() error { RepairBudgetGranted: st.RepairBudgetGranted, Clarifications: st.Clarifications, Mailbox: st.Mailbox, + Handoffs: st.Handoffs, }) } @@ -105,6 +109,7 @@ type diskRunState struct { RepairBudgetGranted int `json:"repair_budget_granted,omitempty"` Clarifications []string `json:"clarifications,omitempty"` Mailbox []MailMessage `json:"mailbox,omitempty"` + Handoffs []HandoffRecord `json:"handoffs,omitempty"` } // Workflow 返回计划的编排清单（简洁访问器）。 @@ -160,6 +165,7 @@ func loadRunState(wsDir string) (*RunState, error) { st.RepairBudgetGranted = disk.RepairBudgetGranted st.Clarifications = disk.Clarifications st.Mailbox = disk.Mailbox + st.Handoffs = disk.Handoffs } else if !os.IsNotExist(err) { return nil, err } diff --git a/squad/topology_conformance_test.go b/squad/topology_conformance_test.go index 5f263f7..d936f3b 100644 --- a/squad/topology_conformance_test.go +++ b/squad/topology_conformance_test.go @@ -384,8 +384,7 @@ func TestTopologyValidationFailsClosed(t *testing.T) { {"未注册名", "bogus_topology", "未知拓扑"}, {"primitive 单跑", TopologyMailbox, "跨拓扑原语"}, {"primitive 单跑 handoff", TopologyHandoff, "跨拓扑原语"}, - {"driver 未点亮", TopologyReview, "驱动未注册"}, - {"driver 未点亮 group_chat", TopologyGroupChat, "驱动未注册"}, + // review/group_chat 驱动已于 WP3 点亮；task_board 属 WP4，仍未点亮。 {"driver 未点亮 task_board", TopologyTaskBoard, "驱动未注册"}, } for _, tc := range cases { diff --git a/squad/workflow.go b/squad/workflow.go index 26dcd1e..2da4082 100644 --- a/squad/workflow.go +++ b/squad/workflow.go @@ -279,6 +279,16 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru return d.failStep(ctx, st, idx) } } + // handoff 原语 fail-closed（Q2）：目标角色必须可解析，启动前拒绝。 + if step.Handoff != "" { + if _, ok := findAgent(st.Plan.Squad.Agents, step.Handoff); !ok { + if _, exists := d.opts.Registry.Get(step.Handoff); !exists { + step.Status, step.Error = "failed", "handoff: 未知目标角色 "+step.Handoff + st.setStep(idx, step) + return d.failStep(ctx, st, idx) + } + } + } subDir := filepath.Join(st.Meta.WorkDir, filepath.FromSlash(orDefault(step.SubWorkspace, fmt.Sprintf("tasks/%02d_%s", idx+1, step.Role)))) if err := os.MkdirAll(subDir, 0o755); err != nil { return nil, nil, err @@ -313,11 +323,14 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru // 任务书；只走 prompt 面，永不触碰 HITL 门（验收三构造性保证）。 st.mu.Lock() mail := takeMailFor(st, step.Role) - if len(mail) > 0 { + handoff := takeHandoffFor(st, step.Role) + if len(mail) > 0 || handoff != nil { _ = st.save() } st.mu.Unlock() prompt += formatMailSection(mail) + // handoff 原语（Q2）：目标角色的步骤接收过滤历史（只带移交方主产出）。 + prompt += formatHandoffSection(handoff) instructions := readFileOr(filepath.Join(subDir, "AGENTS.md"), d.opts.Workspaces.ResolveInstructions(role)) var backend Executor @@ -464,6 +477,12 @@ func (d *flowDriver) runOneStep(ctx context.Context, st *RunState, idx int) (*Ru d.emit(st, Event{Type: EvStepCompleted, StepID: step.ID, Role: step.Role, Text: step.Title, Data: marshalJSON(map[string]any{"duration_ms": step.DurationMs, "targets": targets})}) + // handoff 原语（Q2）：完成即移交最终答案所有权——记录 + 主产出快照， + // 供目标角色下一步骤（过滤历史）与 finalize（链尾产出）消费。 + if step.Handoff != "" { + d.recordHandoff(st, step, subDir, targets) + } + // 循环步再武装（Q2 loop）：停机条件不满足且预算未尽 → 本步与 loop.body // 一起重置为 pending；触顶发 loop_exhausted 事件后按完成继续。 if step.Loop != nil { @@ -775,12 +794,21 @@ func (d *flowDriver) finalize(ctx context.Context, st *RunState) (*RunResult, *H } } // 汇总交付物：最后一个 completed 步骤的主产出 → output.md。 + // handoff 链尾优先（Q2）：有所有权移交时，交付物取链尾角色的最后一个 + // completed 步骤产出（最终答案所有权在链尾），无则该角色无完成步骤时回落。 var srcStep *StepSpec for i := range wf.Steps { if wf.Steps[i].Status == "completed" { srcStep = &wf.Steps[i] } } + if tail := handoffChainTail(st); tail != "" { + for i := range wf.Steps { + if wf.Steps[i].Status == "completed" && wf.Steps[i].Role == tail { + srcStep = &wf.Steps[i] + } + } + } if srcStep == nil { wf.Status = "failed" st.setWorkflow(wf) diff --git a/squad/wp3_conformance_test.go b/squad/wp3_conformance_test.go new file mode 100644 index 0000000..0d50e15 --- /dev/null +++ b/squad/wp3_conformance_test.go @@ -0,0 +1,452 @@ +package squad + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" +) + +// wp3_conformance_test.go — Q2 WP3 验收：review 拓扑与 backend runReview 等价 +// （验收二：并行找问题、逐条验证、丢掉未验证）、handoff 原语（所有权转移 + +// 过滤历史 + finalize 链尾）、group_chat 驱动（轮转/主持人 + 预算停机）。 + +// ---- review：与 backend review_test.go 同款断言 ---- + +// reviewScriptBackend finder 目录回放 findings.json（第 3 个进入者放行，钉住 +// 真并发）；verify_* 目录按 prompt 内容回放判定（与 backend 夹具同文案）。 +type reviewScriptBackend struct { + mu sync.Mutex + once sync.Once + entered int + verifyCalls int + release chan struct{} +} + +func (b *reviewScriptBackend) Name() string { return "review-script" } + +func (b *reviewScriptBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) { + key := filepath.Base(filepath.Clean(req.WorkDir)) + if strings.HasPrefix(key, "verify_") { + b.mu.Lock() + b.verifyCalls++ + b.mu.Unlock() + verdict := "看不出对错" + switch { + case strings.Contains(req.Prompt, "无来源"): + verdict = "核对原文确实没有来源。\nVERDICT: PASS" + case strings.Contains(req.Prompt, "错字"): + verdict = "核对后成立。\nVERDICT: PASS" + case strings.Contains(req.Prompt, "过期数据"): + verdict = "数据仍有效。\nVERDICT: FAIL" + } + return writeOne(req.WorkDir, "verdict.md", verdict), nil + } + // finder：三个必须同时在跑（backend review_test.go:16-33 同款门）。 + b.mu.Lock() + b.entered++ + n := b.entered + b.mu.Unlock() + if n == 3 { + b.once.Do(func() { close(b.release) }) + } + select { + case <-b.release: + case <-time.After(time.Second): + return nil, fmt.Errorf("finders did not run together") + } + var findings string + switch key { + case "01_researcher": + findings = `[{"location":"第2段","severity":"high","issue":"无来源","evidence":"显然"},{"location":"第1段","severity":"low","issue":"错字","evidence":"的地"},{"location":"表1","severity":"medium","issue":"过期数据","evidence":"2020"}]` + case "02_writer": + findings = `[{"location":"第2段","severity":"high","issue":"无来源","evidence":"显然"},{"location":"附录","severity":"medium","issue":"数字不一致","evidence":"3 与 4"}]` + default: + findings = `[{"location":"结尾","severity":"low","issue":"没有结论","evidence":"最后一句"}]` + } + return writeOne(req.WorkDir, "findings.json", findings), nil +} + +func writeOne(dir, name, content string) *ExecResult { + out := filepath.Join(dir, name) + _ = os.MkdirAll(filepath.Dir(out), 0o755) + _ = os.WriteFile(out, []byte(content), 0o644) + return &ExecResult{Success: true, Patches: map[string][]byte{name: []byte(content)}} +} + +func TestTopologyConformance_review(t *testing.T) { + be := &reviewScriptBackend{release: make(chan struct{})} + sink := &collectSink{} + planner := mockScriptPlanner( + StepSpec{ID: "f1", Role: "researcher", Title: "评审一", Targets: []string{"findings.json"}}, + StepSpec{ID: "f2", Role: "writer", Title: "评审二", Targets: []string{"findings.json"}}, + StepSpec{ID: "f3", Role: "reviewer", Title: "评审三", Targets: []string{"findings.json"}}, + ) + planner.Agents = []RoleSpec{ + {Role: "researcher", Name: "评审一", Backend: "review-script"}, + {Role: "writer", Name: "评审二", Backend: "review-script"}, + {Role: "reviewer", Name: "评审三", Backend: "review-script"}, + {Role: "verifier", Name: "核实", Backend: "review-script"}, + } + engine := newWP3Engine(t, planner, be, sink, []RoleSpec{ + {Role: "researcher", Name: "评审一", Backend: "review-script"}, + {Role: "writer", Name: "评审二", Backend: "review-script"}, + {Role: "reviewer", Name: "评审三", Backend: "review-script"}, + {Role: "verifier", Name: "核实", Backend: "review-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "评审材料正文"}, + RunOptions{Topology: TopologyReview, ReviewTarget: "材料正文"}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + data, err := os.ReadFile(filepath.Join(res.WorkDir, "output.md")) + if err != nil { + t.Fatal(err) + } + out := string(data) + // 只留 PASS（fail-closed）：无来源/错字保留；FAIL 与无判定丢弃。 + if !strings.Contains(out, "无来源") || !strings.Contains(out, "错字") || + strings.Contains(out, "数字不一致") || strings.Contains(out, "没有结论") || + strings.Contains(out, "过期数据") { + t.Fatalf("filter = %s", out) + } + // 报告恒带丢弃计数（backend 同格式串）。 + if !strings.Contains(out, "已丢弃 3 条未验证") { + t.Fatalf("dropped = %s", out) + } + // 去重后逐条验证：verify 调用数 = unique 数 = 5。 + if be.verifyCalls != 5 { + t.Fatalf("verify calls = %d, want 5 after dedupe", be.verifyCalls) + } + // severity 稳定排序：high(无来源) 在 low(错字) 前。 + if strings.Index(out, "无来源") > strings.Index(out, "错字") { + t.Fatalf("severity order: %s", out) + } + if sink.countText("review：3 个 finder 并行找问题") != 1 || + sink.countText("review：去重后 5 条") != 1 { + t.Fatal("缺 review 阶段事件") + } +} + +// verify 角色缺失 → fail-closed。 +func TestReviewVerifyRoleMissingFailsClosed(t *testing.T) { + be := &reviewScriptBackend{release: make(chan struct{})} + planner := mockScriptPlanner( + StepSpec{ID: "f1", Role: "researcher", Title: "评审", Targets: []string{"findings.json"}}, + ) + engine := newWP3Engine(t, planner, be, NopSink{}, []RoleSpec{ + {Role: "researcher", Name: "评审", Backend: "review-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "评审"}, + RunOptions{Topology: TopologyReview}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunFailed || !strings.Contains(res.Error, "verify 角色") { + t.Fatalf("应 fail-closed：%s %q", res.Status, res.Error) + } +} + +// ---- handoff 原语 ---- + +// handoffScriptBackend 记录 prompt 并按目录回放带标记的产出（≥50 字过最小质检）。 +type handoffScriptBackend struct { + mu sync.Mutex + prompts map[string][]string +} + +func (b *handoffScriptBackend) Name() string { return "handoff-script" } + +func (b *handoffScriptBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) { + key := filepath.Base(filepath.Clean(req.WorkDir)) + b.mu.Lock() + b.prompts[key] = append(b.prompts[key], req.Prompt) + b.mu.Unlock() + pad := "\n\n这是为了满足最小质检长度要求而补充的正文内容，超过五十个字符。" + var content string + switch key { + case "01_researcher": + content = "MARK_RESEARCHER 调研笔记。" + pad + case "02_writer": + content = "MARK_WRITER 终稿。" + pad + default: + content = "MARK_REVIEWER 评审意见。" + pad + } + name := "output.md" + if len(req.Targets) > 0 { + name = req.Targets[0] + } + return writeOne(req.WorkDir, name, content), nil +} + +func TestTopologyConformance_handoff(t *testing.T) { + be := &handoffScriptBackend{prompts: map[string][]string{}} + sink := &collectSink{} + engine := newWP3Engine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "调研", Targets: []string{"notes.md"}, Handoff: "writer"}, + StepSpec{ID: "s2", Role: "writer", Title: "成文", Targets: []string{"output.md"}, Needs: []string{"s1"}}, + StepSpec{ID: "s3", Role: "reviewer", Title: "终审", Targets: []string{"output.md"}, Needs: []string{"s2"}}, + ), be, sink, []RoleSpec{ + {Role: "researcher", Name: "调研", Backend: "handoff-script"}, + {Role: "writer", Name: "撰写", Backend: "handoff-script"}, + {Role: "reviewer", Name: "审核", Backend: "handoff-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "带移交的三步"}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + // ① 所有权转移记录持久化。 + st, err := engine.loadRun(engine.runDir(res.RunID)) + if err != nil { + t.Fatal(err) + } + if len(st.Handoffs) != 1 || st.Handoffs[0].FromRole != "researcher" || st.Handoffs[0].ToRole != "writer" { + t.Fatalf("Handoffs = %+v", st.Handoffs) + } + if !strings.Contains(st.Handoffs[0].Output, "MARK_RESEARCHER") { + t.Fatalf("移交产出快照缺失：%q", st.Handoffs[0].Output) + } + if sink.countText("handoff：researcher → writer") != 1 { + t.Fatal("缺所有权移交事件") + } + // ② 目标收过滤历史：writer 的任务书带移交上下文，reviewer 不带。 + var writerPrompt, reviewerPrompt string + for _, p := range be.prompts["02_writer"] { + writerPrompt += p + } + for _, p := range be.prompts["03_reviewer"] { + reviewerPrompt += p + } + if !strings.Contains(writerPrompt, "移交上下文") || !strings.Contains(writerPrompt, "MARK_RESEARCHER") { + t.Fatalf("writer 未收到过滤历史：%q", writerPrompt) + } + if strings.Contains(reviewerPrompt, "移交上下文") { + t.Fatal("非目标角色不应收到移交上下文") + } + // ③ finalize 取链尾输出：交付物 = writer（链尾）产出，而非最后完成的 reviewer。 + data, err := os.ReadFile(filepath.Join(res.WorkDir, "output.md")) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(data), "MARK_WRITER") || strings.Contains(string(data), "MARK_REVIEWER") { + t.Fatalf("交付物应为链尾 writer 产出：%s", data) + } +} + +// handoff 目标角色未知 → 步骤启动前 fail-closed。 +func TestHandoffUnknownRoleFailsClosed(t *testing.T) { + be := &handoffScriptBackend{prompts: map[string][]string{}} + engine := newWP3Engine(t, mockScriptPlanner( + StepSpec{ID: "s1", Role: "researcher", Title: "调研", Handoff: "ghost"}, + ), be, NopSink{}, []RoleSpec{ + {Role: "researcher", Name: "调研", Backend: "handoff-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "x"}, RunOptions{}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunFailed || !strings.Contains(res.Error, "handoff: 未知目标角色 ghost") { + t.Fatalf("应 fail-closed：%s %q", res.Status, res.Error) + } +} + +// ---- group_chat 驱动 ---- + +// gcScriptBackend 按目录回放 message.md。 +type gcScriptBackend struct { + messages map[string]string +} + +func (b *gcScriptBackend) Name() string { return "gc-script" } + +func (b *gcScriptBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) { + key := filepath.Base(filepath.Clean(req.WorkDir)) + msg := b.messages[key] + if msg == "" { + msg = "发言：" + key + } + return writeOne(req.WorkDir, "message.md", msg), nil +} + +func TestTopologyConformance_groupChatRotateBudget(t *testing.T) { + be := &gcScriptBackend{messages: map[string]string{ + "turn_01_researcher": "第一句", + "turn_02_writer": "第二句", + "turn_03_researcher": "第三句", + }} + sink := &collectSink{} + engine := newWP3Engine(t, mockScriptPlanner( + StepSpec{ID: "p1", Role: "researcher", Title: "参与者甲"}, + StepSpec{ID: "p2", Role: "writer", Title: "参与者乙"}, + ), be, sink, []RoleSpec{ + {Role: "researcher", Name: "甲", Backend: "gc-script"}, + {Role: "writer", Name: "乙", Backend: "gc-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "群聊"}, + RunOptions{Topology: TopologyGroupChat, GroupChatMaxTurns: 3}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + // 预算停机：恰好 3 轮，轮转序 researcher→writer→researcher。 + if sink.countText("轮次预算 3 用尽") != 1 { + t.Fatal("缺预算停机事件") + } + data, err := os.ReadFile(filepath.Join(res.WorkDir, "output.md")) + if err != nil { + t.Fatal(err) + } + out := string(data) + for _, want := range []string{"1. researcher：第一句", "2. writer：第二句", "3. researcher：第三句"} { + if !strings.Contains(out, want) { + t.Fatalf("转录缺 %q：%s", want, out) + } + } + if strings.Contains(out, "4. ") { + t.Fatalf("超出预算仍在发言：%s", out) + } + // 名册步骤不执行，标记 skipped（非降级原因）。 + st, _ := engine.loadRun(engine.runDir(res.RunID)) + for _, s := range st.Plan.Workflow.Steps { + if s.ID == "p1" || s.ID == "p2" { + if s.Status != "skipped" || s.Error != gcRosterSkipReason { + t.Fatalf("名册步骤 %s = %s/%q", s.ID, s.Status, s.Error) + } + } + } +} + +func TestTopologyConformance_groupChatCloseMarker(t *testing.T) { + be := &gcScriptBackend{messages: map[string]string{ + "turn_01_researcher": "我先说", + "turn_02_writer": "同意收口 [close]", + }} + sink := &collectSink{} + engine := newWP3Engine(t, mockScriptPlanner( + StepSpec{ID: "p1", Role: "researcher", Title: "甲"}, + StepSpec{ID: "p2", Role: "writer", Title: "乙"}, + ), be, sink, []RoleSpec{ + {Role: "researcher", Name: "甲", Backend: "gc-script"}, + {Role: "writer", Name: "乙", Backend: "gc-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "群聊"}, + RunOptions{Topology: TopologyGroupChat, GroupChatMaxTurns: 8}) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s", res.Status) + } + if sink.countText("发出收口标记") != 1 { + t.Fatal("缺收口标记事件") + } + st, _ := engine.loadRun(engine.runDir(res.RunID)) + if n := countGcSteps(st); n != 2 { + t.Fatalf("收口标记后应停在 2 轮，实际 %d", n) + } +} + +func TestTopologyConformance_groupChatModerator(t *testing.T) { + be := &gcScriptBackend{messages: map[string]string{ + "turn_01_researcher": `{"next":"writer","reason":"请乙补充"}`, + "turn_02_writer": "补一段分析", + "turn_03_researcher": `{"action":"close","reason":"讨论充分"}`, + }} + sink := &collectSink{} + engine := newWP3Engine(t, mockScriptPlanner( + StepSpec{ID: "p1", Role: "researcher", Title: "主持人"}, + StepSpec{ID: "p2", Role: "writer", Title: "乙"}, + ), be, sink, []RoleSpec{ + {Role: "researcher", Name: "主持", Backend: "gc-script"}, + {Role: "writer", Name: "乙", Backend: "gc-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "主持人群聊"}, RunOptions{ + Topology: TopologyGroupChat, GroupChatPolicy: "moderator", + GroupChatModerator: "researcher", GroupChatMaxTurns: 8, + }) + if err != nil { + t.Fatal(err) + } + if res.Status != RunCompleted { + t.Fatalf("status = %s err = %v", res.Status, res.Error) + } + if sink.countText("主持人收口") != 1 { + t.Fatal("缺主持人收口事件") + } + data, _ := os.ReadFile(filepath.Join(res.WorkDir, "output.md")) + out := string(data) + for _, want := range []string{"1. researcher", "2. writer：补一段分析", "3. researcher"} { + if !strings.Contains(out, want) { + t.Fatalf("转录缺 %q：%s", want, out) + } + } + st, _ := engine.loadRun(engine.runDir(res.RunID)) + if n := countGcSteps(st); n != 3 { + t.Fatalf("主持人模式应 3 轮收口，实际 %d", n) + } +} + +// 主持人指名不在名册 → fail-closed。 +func TestGroupChatModeratorUnknownNextFailsClosed(t *testing.T) { + be := &gcScriptBackend{messages: map[string]string{ + "turn_01_researcher": `{"next":"ghost"}`, + }} + engine := newWP3Engine(t, mockScriptPlanner( + StepSpec{ID: "p1", Role: "researcher", Title: "主持人"}, + StepSpec{ID: "p2", Role: "writer", Title: "乙"}, + ), be, NopSink{}, []RoleSpec{ + {Role: "researcher", Name: "主持", Backend: "gc-script"}, + {Role: "writer", Name: "乙", Backend: "gc-script"}, + }) + res, err := engine.Run(context.Background(), Goal{Text: "x"}, RunOptions{ + Topology: TopologyGroupChat, GroupChatPolicy: "moderator", GroupChatModerator: "researcher", + }) + if err != nil { + t.Fatal(err) + } + if res.Status != RunFailed || !strings.Contains(res.Error, "不在名册内") { + t.Fatalf("应 fail-closed：%s %q", res.Status, res.Error) + } +} + +// ---- 夹具 ---- + +func countGcSteps(st *RunState) int { + n := 0 + for _, s := range st.Plan.Workflow.Steps { + if strings.HasPrefix(s.ID, "gc") && s.Status == "completed" { + n++ + } + } + return n +} + +func newWP3Engine(t *testing.T, planner Planner, be Executor, sink ProgressSink, roles []RoleSpec) *SquadEngine { + t.Helper() + dir := t.TempDir() + reg := NewRoleRegistry() + for _, r := range roles { + if err := reg.Register(r); err != nil { + t.Fatal(err) + } + } + backends := NewBackendResolver() + backends.RegisterInstance(be.Name(), be) + engine, err := NewEngine(Options{DataDir: dir, Registry: reg, Planner: planner, Backends: backends, Sink: sink}) + if err != nil { + t.Fatal(err) + } + return engine +} -- 2.54.0