Compare commits

...
Author SHA1 Message Date
weichen c88100967d Add review driver, handoff primitive, and group-chat driver (Q2 WP3)
ci / vet-and-test (pull_request) Successful in 34s
ci / docker-checkout-hostgateway (pull_request) Successful in 2s
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
2026-10-08 01:11:51 +00:00
weichen 01d5b96cda Merge pull request 'Add leader progress ledger, replan cap, and mailbox with actor attribution (Q2 WP2)' (#43) from feature/q2-leader-mailbox-wp2 into main
ci / vet-and-test (push) Successful in 25s
ci / docker-checkout-hostgateway (push) Successful in 0s
Merge pull request 'Add leader progress ledger, replan cap, and mailbox with actor attribution (Q2 WP2)' (#43) from feature/q2-leader-mailbox-wp2 into main
2026-10-08 00:36:38 +00:00
8 changed files with 1366 additions and 3 deletions
+66
View File
@@ -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 == "" {
+355
View File
@@ -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] + "…"
}
+91
View File
@@ -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)})
}
+366
View File
@@ -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}
}
+6
View File
@@ -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
}
+1 -2
View File
@@ -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 {
+29 -1
View File
@@ -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)
+452
View File
@@ -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
}