Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
df1f877a4f | ||
|
|
663e89ec58 | ||
|
|
424bfc0e29 | ||
|
|
633e853bfc | ||
|
|
d6669cd9c8 | ||
|
|
bcc8d080b0 | ||
|
|
acbe744142 | ||
|
|
aec2a4aa0f | ||
|
|
c88100967d | ||
|
|
01d5b96cda |
@@ -0,0 +1,179 @@
|
||||
// Package a2a 把小队接入 A2A(Agent2Agent)1.0 互通面(Q3 WP4):
|
||||
// - 发布侧:BuildCard 从角色注册表与蓝图注册表渲染 Agent Card(描述本
|
||||
// 团队能力),CardHandler 以 JSON 服务(挂载在
|
||||
// /.well-known/agent-card.json);
|
||||
// - 消费侧:FetchCard 发现远端 Agent Card(fail-closed 校验必备字段),
|
||||
// Executor 实现 squad.Executor——message/send 发起任务,tasks/get 轮询
|
||||
// A2A 任务状态机(submitted/working/input-required → completed/failed/
|
||||
// canceled),终态映射 ExecResult:completed 的 artifacts(TextPart/
|
||||
// FilePart)落成 patches/目标文件;failed/canceled/input-required 一律
|
||||
// Success=false(跨进程挂起不在本适配面内,fail-closed 交引擎重试/降级
|
||||
// 语义处理)。
|
||||
//
|
||||
// 传输为 JSON-RPC 2.0 over HTTP(A2A 1.0 形态);不引入协议之外的依赖。
|
||||
package a2a
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"agent-squad/squad"
|
||||
)
|
||||
|
||||
// ProtocolVersion 对齐的 A2A 协议版本。
|
||||
const ProtocolVersion = "1.0"
|
||||
|
||||
// WellKnownPath 是 Agent Card 的发现路径(A2A 1.0)。
|
||||
const WellKnownPath = "/.well-known/agent-card.json"
|
||||
|
||||
// AgentCard 是 A2A 1.0 Agent Card(发布与消费共用形态)。
|
||||
type AgentCard struct {
|
||||
ProtocolVersion string `json:"protocolVersion"`
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description,omitempty"`
|
||||
// URL 是 JSON-RPC 服务端点(消费侧据此发 message/send)。
|
||||
URL string `json:"url"`
|
||||
Version string `json:"version"`
|
||||
Capabilities CardCapabilities `json:"capabilities"`
|
||||
DefaultInputModes []string `json:"defaultInputModes,omitempty"`
|
||||
DefaultOutputModes []string `json:"defaultOutputModes,omitempty"`
|
||||
Skills []AgentSkill `json:"skills,omitempty"`
|
||||
}
|
||||
|
||||
// CardCapabilities 能力声明(本适配面:非流式、无推送、无状态历史)。
|
||||
type CardCapabilities struct {
|
||||
Streaming bool `json:"streaming"`
|
||||
PushNotifications bool `json:"pushNotifications"`
|
||||
StateTransitionHistory bool `json:"stateTransitionHistory"`
|
||||
}
|
||||
|
||||
// AgentSkill 是一项可被远端调用的能力。
|
||||
type AgentSkill struct {
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Tags []string `json:"tags,omitempty"`
|
||||
Examples []string `json:"examples,omitempty"`
|
||||
}
|
||||
|
||||
// CardOptions 覆盖 BuildCard 的身份字段(空则用缺省)。
|
||||
type CardOptions struct {
|
||||
Name string
|
||||
Description string
|
||||
// URL JSON-RPC 端点(必填:远端没有端点就无法调用本团队)。
|
||||
URL string
|
||||
// Version 卡片版本(空 = "1.0.0")。
|
||||
Version string
|
||||
// Skills 显式能力清单;空 = 从注册表派生。
|
||||
Skills []AgentSkill
|
||||
}
|
||||
|
||||
// BuildCard 从角色注册表与蓝图注册表渲染 Agent Card(描述本团队能力)。
|
||||
// 角色 → 技能(关键词作 tags);蓝图 → 技能(match 词作 examples)。
|
||||
func BuildCard(reg *squad.RoleRegistry, bps *squad.BlueprintRegistry, opts CardOptions) (AgentCard, error) {
|
||||
if strings.TrimSpace(opts.URL) == "" {
|
||||
return AgentCard{}, fmt.Errorf("a2a: Agent Card 需要 JSON-RPC 端点 URL")
|
||||
}
|
||||
card := AgentCard{
|
||||
ProtocolVersion: ProtocolVersion,
|
||||
Name: orDefault(opts.Name, "agent-squad"),
|
||||
Description: opts.Description,
|
||||
URL: opts.URL,
|
||||
Version: orDefault(opts.Version, "1.0.0"),
|
||||
DefaultInputModes: []string{"text/plain"},
|
||||
DefaultOutputModes: []string{"text/plain"},
|
||||
Skills: append([]AgentSkill(nil), opts.Skills...),
|
||||
}
|
||||
if len(card.Skills) == 0 {
|
||||
if reg != nil {
|
||||
for _, role := range reg.List() {
|
||||
desc := firstLine(role.Instructions)
|
||||
if desc == "" {
|
||||
desc = role.Name
|
||||
}
|
||||
card.Skills = append(card.Skills, AgentSkill{
|
||||
ID: role.Role, Name: orDefault(role.Name, role.Role),
|
||||
Description: desc, Tags: append([]string(nil), role.Keywords...),
|
||||
})
|
||||
}
|
||||
}
|
||||
if bps != nil {
|
||||
for _, bp := range bps.List() {
|
||||
card.Skills = append(card.Skills, AgentSkill{
|
||||
ID: bp.Key, Name: orDefault(bp.Title, bp.Key),
|
||||
Description: bp.Description, Examples: append([]string(nil), bp.Match...),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
return card, nil
|
||||
}
|
||||
|
||||
// CardHandler 以 JSON 服务 Agent Card(调用方挂载到 WellKnownPath)。
|
||||
func CardHandler(card AgentCard) http.Handler {
|
||||
body, _ := json.Marshal(card)
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write(body)
|
||||
})
|
||||
}
|
||||
|
||||
// FetchCard 发现并校验远端 Agent Card(fail-closed:必备字段缺失即拒)。
|
||||
func FetchCard(client *http.Client, wellKnownURL string) (AgentCard, error) {
|
||||
if client == nil {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
resp, err := client.Get(wellKnownURL)
|
||||
if err != nil {
|
||||
return AgentCard{}, fmt.Errorf("a2a: 获取 Agent Card: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return AgentCard{}, fmt.Errorf("a2a: Agent Card 端点返回 %d", resp.StatusCode)
|
||||
}
|
||||
var card AgentCard
|
||||
if err := json.NewDecoder(resp.Body).Decode(&card); err != nil {
|
||||
return AgentCard{}, fmt.Errorf("a2a: Agent Card 解析失败: %w", err)
|
||||
}
|
||||
if err := validateCard(card); err != nil {
|
||||
return AgentCard{}, err
|
||||
}
|
||||
return card, nil
|
||||
}
|
||||
|
||||
// validateCard 消费侧最小校验(A2A 1.0 必备字段)。
|
||||
func validateCard(card AgentCard) error {
|
||||
if card.ProtocolVersion == "" {
|
||||
return fmt.Errorf("a2a: Agent Card 缺 protocolVersion")
|
||||
}
|
||||
if strings.TrimSpace(card.Name) == "" {
|
||||
return fmt.Errorf("a2a: Agent Card 缺 name")
|
||||
}
|
||||
if strings.TrimSpace(card.URL) == "" {
|
||||
return fmt.Errorf("a2a: Agent Card 缺 url")
|
||||
}
|
||||
if strings.TrimSpace(card.Version) == "" {
|
||||
return fmt.Errorf("a2a: Agent Card 缺 version")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func orDefault(v, def string) string {
|
||||
if strings.TrimSpace(v) != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func firstLine(s string) string {
|
||||
s = strings.TrimSpace(s)
|
||||
if i := strings.IndexByte(s, '\n'); i >= 0 {
|
||||
s = s[:i]
|
||||
}
|
||||
if len(s) > 120 {
|
||||
s = s[:120] + "…"
|
||||
}
|
||||
return s
|
||||
}
|
||||
@@ -0,0 +1,345 @@
|
||||
package a2a
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"agent-squad/squad"
|
||||
)
|
||||
|
||||
// conformance_test.go — Q3 WP4 验收:双向 conformance(本地假 A2A 服务端)。
|
||||
// 发布侧:BuildCard + CardHandler 在 well-known 路径服务合法 Agent Card;
|
||||
// 消费侧:FetchCard 发现校验 + Executor 走完任务状态机
|
||||
// (submitted→working→completed / failed / input-required),终态对齐
|
||||
// squad.ExecResult。
|
||||
|
||||
// fakeA2AServer 最小 A2A 1.0 服务端:well-known 卡片 + JSON-RPC
|
||||
// (message/send 建任务;tasks/get 按脚本推进状态)。
|
||||
type fakeA2AServer struct {
|
||||
mu sync.Mutex
|
||||
card AgentCard
|
||||
tasks map[string]*Task
|
||||
plan []string // 每次 tasks/get 推进到的状态序列(走完停在末态)
|
||||
calls map[string]int
|
||||
artifacts []Artifact
|
||||
statusMsg string
|
||||
}
|
||||
|
||||
func newFakeServer(t *testing.T, card AgentCard, plan []string, artifacts []Artifact) *fakeA2AServer {
|
||||
t.Helper()
|
||||
s := &fakeA2AServer{card: card, tasks: map[string]*Task{}, plan: plan,
|
||||
calls: map[string]int{}, artifacts: artifacts}
|
||||
return s
|
||||
}
|
||||
|
||||
func (s *fakeA2AServer) handler() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc(WellKnownPath, func(w http.ResponseWriter, r *http.Request) {
|
||||
s.mu.Lock()
|
||||
body, _ := json.Marshal(s.card)
|
||||
s.mu.Unlock()
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write(body)
|
||||
})
|
||||
mux.HandleFunc("/rpc", func(w http.ResponseWriter, r *http.Request) {
|
||||
var req rpcRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
writeRPCError(w, 0, -32700, "parse error")
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.calls[req.Method]++
|
||||
switch req.Method {
|
||||
case "message/send":
|
||||
id := fmt.Sprintf("task_%d", len(s.tasks)+1)
|
||||
state := StateSubmitted
|
||||
if len(s.plan) > 0 {
|
||||
state = s.plan[0]
|
||||
}
|
||||
task := &Task{Kind: "task", ID: id, ContextID: "ctx_1",
|
||||
Status: TaskStatus{State: state, Message: s.statusMessage()}}
|
||||
s.tasks[id] = task
|
||||
writeRPCResult(w, req.ID, task)
|
||||
case "tasks/get":
|
||||
var params struct {
|
||||
ID string `json:"id"`
|
||||
}
|
||||
raw, _ := json.Marshal(req.Params)
|
||||
_ = json.Unmarshal(raw, ¶ms)
|
||||
task, ok := s.tasks[params.ID]
|
||||
if !ok {
|
||||
writeRPCError(w, req.ID, -32001, "task not found")
|
||||
return
|
||||
}
|
||||
// 按脚本推进状态机(停留在终态)。
|
||||
if !isTerminal(task.Status.State) && len(s.plan) > 0 {
|
||||
n := s.calls["tasks/get#"+params.ID]
|
||||
s.calls["tasks/get#"+params.ID] = n + 1
|
||||
idx := n + 1
|
||||
if idx >= len(s.plan) {
|
||||
idx = len(s.plan) - 1
|
||||
}
|
||||
task.Status.State = s.plan[idx]
|
||||
task.Status.Message = s.statusMessage()
|
||||
if task.Status.State == StateCompleted {
|
||||
task.Artifacts = s.artifacts
|
||||
}
|
||||
}
|
||||
writeRPCResult(w, req.ID, task)
|
||||
default:
|
||||
writeRPCError(w, req.ID, -32601, "method not found")
|
||||
}
|
||||
})
|
||||
return mux
|
||||
}
|
||||
|
||||
func (s *fakeA2AServer) statusMessage() *Message {
|
||||
if s.statusMsg == "" {
|
||||
return nil
|
||||
}
|
||||
return &Message{Kind: "message", Role: "agent",
|
||||
Parts: []Part{{Kind: "text", Text: s.statusMsg}}}
|
||||
}
|
||||
|
||||
func writeRPCResult(w http.ResponseWriter, id int64, result any) {
|
||||
body, _ := json.Marshal(result)
|
||||
env, _ := json.Marshal(rpcResponse{JSONRPC: "2.0", ID: id, Result: body})
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write(env)
|
||||
}
|
||||
|
||||
func writeRPCError(w http.ResponseWriter, id int64, code int, msg string) {
|
||||
env, _ := json.Marshal(rpcResponse{JSONRPC: "2.0", ID: id, Error: &rpcError{Code: code, Message: msg}})
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write(env)
|
||||
}
|
||||
|
||||
// ---- 发布侧 ----
|
||||
|
||||
func TestBuildCardFromRegistries(t *testing.T) {
|
||||
reg := squad.NewRoleRegistry()
|
||||
for _, r := range []squad.RoleSpec{
|
||||
{Role: "researcher", Name: "调研员", Keywords: []string{"调研", "检索"},
|
||||
Instructions: "围绕目标调研关键问题。\n第二行不进摘要。"},
|
||||
{Role: "writer", Name: "撰稿员", Keywords: []string{"撰写"}},
|
||||
} {
|
||||
if err := reg.Register(r); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
bps := squad.NewBlueprintRegistry()
|
||||
if err := bps.Register(squad.Blueprint{Key: "brief", Title: "简报",
|
||||
Description: "生成简报", Match: []string{"简报", "brief"},
|
||||
Steps: []squad.StepSpec{{ID: "s1", Role: "writer", Title: "成文"}}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
card, err := BuildCard(reg, bps, CardOptions{
|
||||
Name: "示例小队", Description: "演示团队能力", URL: "https://team.example/rpc",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if card.ProtocolVersion != ProtocolVersion || card.Version != "1.0.0" {
|
||||
t.Fatalf("card = %+v", card)
|
||||
}
|
||||
byID := map[string]AgentSkill{}
|
||||
for _, s := range card.Skills {
|
||||
byID[s.ID] = s
|
||||
}
|
||||
if byID["researcher"].Name != "调研员" || byID["researcher"].Description != "围绕目标调研关键问题。" {
|
||||
t.Fatalf("role skill = %+v", byID["researcher"])
|
||||
}
|
||||
if len(byID["researcher"].Tags) != 2 {
|
||||
t.Fatalf("tags = %v", byID["researcher"].Tags)
|
||||
}
|
||||
if byID["brief"].Name != "简报" || strings.Join(byID["brief"].Examples, ",") != "简报,brief" {
|
||||
t.Fatalf("blueprint skill = %+v", byID["brief"])
|
||||
}
|
||||
if _, err := BuildCard(reg, bps, CardOptions{}); err == nil {
|
||||
t.Fatal("缺 URL 必须 fail-closed")
|
||||
}
|
||||
}
|
||||
|
||||
// CardHandler 以 JSON 服务卡片(发布端点形态)。
|
||||
func TestCardHandlerServesJSON(t *testing.T) {
|
||||
card := AgentCard{ProtocolVersion: ProtocolVersion, Name: "小队", URL: "https://x/rpc", Version: "1.0.0"}
|
||||
ts := httptest.NewServer(CardHandler(card))
|
||||
defer ts.Close()
|
||||
resp, err := http.Get(ts.URL + WellKnownPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if ct := resp.Header.Get("Content-Type"); ct != "application/json" {
|
||||
t.Fatalf("content-type = %s", ct)
|
||||
}
|
||||
var got AgentCard
|
||||
if err := json.NewDecoder(resp.Body).Decode(&got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Name != "小队" || got.ProtocolVersion != ProtocolVersion {
|
||||
t.Fatalf("card = %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 消费侧:发现 + 状态机 ----
|
||||
|
||||
func TestFetchCardAndExecuteCompleted(t *testing.T) {
|
||||
fileBytes := base64.StdEncoding.EncodeToString([]byte("notes 内容"))
|
||||
artifacts := []Artifact{{
|
||||
ArtifactID: "art1", Name: "调研结果",
|
||||
Parts: []Part{
|
||||
{Kind: "text", Text: "# 调研结论\n\n结论正文,长度足够作为交付物内容。"},
|
||||
{Kind: "file", File: &FilePart{Name: "notes.md", Bytes: fileBytes, MimeType: "text/markdown"}},
|
||||
},
|
||||
}}
|
||||
card := AgentCard{ProtocolVersion: ProtocolVersion, Name: "远端小队",
|
||||
URL: "PLACEHOLDER", Version: "1.0.0"}
|
||||
srv := newFakeServer(t, card, []string{StateWorking, StateWorking, StateCompleted}, artifacts)
|
||||
ts := httptest.NewServer(srv.handler())
|
||||
defer ts.Close()
|
||||
srv.mu.Lock()
|
||||
srv.card.URL = ts.URL + "/rpc"
|
||||
srv.mu.Unlock()
|
||||
|
||||
// 发现:well-known 路径取卡并校验。
|
||||
got, err := FetchCard(ts.Client(), ts.URL+WellKnownPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.Name != "远端小队" || got.URL != ts.URL+"/rpc" {
|
||||
t.Fatalf("card = %+v", got)
|
||||
}
|
||||
|
||||
// 执行:submitted/working → completed,工件映射 patches。
|
||||
var states []string
|
||||
ex := NewExecutor(got)
|
||||
ex.Client = ts.Client()
|
||||
ex.PollInterval = 5 * time.Millisecond
|
||||
ex.OnState = func(s string) { states = append(states, s) }
|
||||
res, err := ex.Execute(context.Background(), squad.ExecRequest{
|
||||
Prompt: "调研主题 X", Targets: []string{"output.md"},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !res.Success {
|
||||
t.Fatalf("result = %+v", res)
|
||||
}
|
||||
if string(res.Patches["notes.md"]) != "notes 内容" {
|
||||
t.Fatalf("file part 未落成 patch:%q", res.Patches["notes.md"])
|
||||
}
|
||||
if !strings.Contains(string(res.Patches["output.md"]), "调研结论") {
|
||||
t.Fatalf("文本部件未写入目标:%q", res.Patches["output.md"])
|
||||
}
|
||||
joined := strings.Join(states, ",")
|
||||
if !strings.Contains(joined, StateWorking) || !strings.HasSuffix(joined, StateCompleted) {
|
||||
t.Fatalf("状态机迁移 = %v", states)
|
||||
}
|
||||
srv.mu.Lock()
|
||||
sends := srv.calls["message/send"]
|
||||
srv.mu.Unlock()
|
||||
if sends != 1 {
|
||||
t.Fatalf("message/send = %d", sends)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteMapsFailedAndInputRequired(t *testing.T) {
|
||||
card := AgentCard{ProtocolVersion: ProtocolVersion, Name: "远端", URL: "PLACEHOLDER", Version: "1"}
|
||||
srv := newFakeServer(t, card, []string{StateWorking, StateFailed}, nil)
|
||||
srv.statusMsg = "远端执行崩溃"
|
||||
ts := httptest.NewServer(srv.handler())
|
||||
defer ts.Close()
|
||||
srv.mu.Lock()
|
||||
srv.card.URL = ts.URL + "/rpc"
|
||||
srv.mu.Unlock()
|
||||
|
||||
ex := NewExecutor(card)
|
||||
ex.BaseURL = ts.URL + "/rpc"
|
||||
ex.Client = ts.Client()
|
||||
ex.PollInterval = 5 * time.Millisecond
|
||||
res, err := ex.Execute(context.Background(), squad.ExecRequest{Prompt: "x", Targets: []string{"output.md"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Success || !strings.Contains(res.ErrorReason, "failed") || !strings.Contains(res.ErrorReason, "远端执行崩溃") {
|
||||
t.Fatalf("failed 映射 = %+v", res)
|
||||
}
|
||||
|
||||
// input-required:fail-closed(跨进程人工输入不在适配面内)。
|
||||
srv2 := newFakeServer(t, card, []string{StateInputRequired}, nil)
|
||||
srv2.statusMsg = "请补充授权"
|
||||
ts2 := httptest.NewServer(srv2.handler())
|
||||
defer ts2.Close()
|
||||
ex2 := &Executor{BaseURL: ts2.URL + "/rpc", Client: ts2.Client(), PollInterval: 5 * time.Millisecond}
|
||||
res2, err := ex2.Execute(context.Background(), squad.ExecRequest{Prompt: "x", Targets: []string{"output.md"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res2.Success || !strings.Contains(res2.ErrorReason, "input-required") {
|
||||
t.Fatalf("input-required 映射 = %+v", res2)
|
||||
}
|
||||
}
|
||||
|
||||
// 直接回 Message(无任务态)→ 文本落目标。
|
||||
func TestExecuteDirectMessageResult(t *testing.T) {
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
var req rpcRequest
|
||||
_ = json.NewDecoder(r.Body).Decode(&req)
|
||||
writeRPCResult(w, req.ID, Message{Kind: "message", Role: "agent",
|
||||
Parts: []Part{{Kind: "text", Text: "直接回答正文"}}})
|
||||
}))
|
||||
defer ts.Close()
|
||||
ex := &Executor{BaseURL: ts.URL, Client: ts.Client()}
|
||||
res, err := ex.Execute(context.Background(), squad.ExecRequest{Prompt: "x", Targets: []string{"output.md"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !res.Success || !strings.Contains(string(res.Patches["output.md"]), "直接回答正文") {
|
||||
t.Fatalf("res = %+v", res)
|
||||
}
|
||||
}
|
||||
|
||||
// 传输/协议错误 → error(引擎按瞬态评估),不吞成成功。
|
||||
func TestExecuteProtocolErrors(t *testing.T) {
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
writeRPCError(w, 1, -32601, "method not found")
|
||||
}))
|
||||
defer ts.Close()
|
||||
ex := &Executor{BaseURL: ts.URL, Client: ts.Client()}
|
||||
if _, err := ex.Execute(context.Background(), squad.ExecRequest{Prompt: "x"}); err == nil ||
|
||||
!strings.Contains(err.Error(), "协议错误") {
|
||||
t.Fatalf("err = %v", err)
|
||||
}
|
||||
// ctx 取消传播。
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
slow := newFakeServer(t, AgentCard{ProtocolVersion: "1.0", Name: "x", URL: ts.URL, Version: "1"},
|
||||
[]string{StateWorking}, nil)
|
||||
ts2 := httptest.NewServer(slow.handler())
|
||||
defer ts2.Close()
|
||||
ex2 := &Executor{BaseURL: ts2.URL + "/rpc", Client: ts2.Client(), PollInterval: time.Second}
|
||||
if _, err := ex2.Execute(ctx, squad.ExecRequest{Prompt: "x"}); err == nil {
|
||||
t.Fatal("取消必须传播")
|
||||
}
|
||||
}
|
||||
|
||||
// FetchCard fail-closed:缺必备字段拒绝。
|
||||
func TestFetchCardValidates(t *testing.T) {
|
||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte(`{"protocolVersion":"1.0","name":"","url":"","version":""}`))
|
||||
}))
|
||||
defer ts.Close()
|
||||
if _, err := FetchCard(ts.Client(), ts.URL); err == nil || !strings.Contains(err.Error(), "name") {
|
||||
t.Fatalf("err = %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,331 @@
|
||||
package a2a
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"agent-squad/squad"
|
||||
)
|
||||
|
||||
// executor.go — 远端智能体 Executor 客户端(A2A 任务状态机对齐 ExecResult)。
|
||||
|
||||
// 任务状态(A2A 1.0 TaskState)。
|
||||
const (
|
||||
StateSubmitted = "submitted"
|
||||
StateWorking = "working"
|
||||
StateInputRequired = "input-required"
|
||||
StateCompleted = "completed"
|
||||
StateCanceled = "canceled"
|
||||
StateFailed = "failed"
|
||||
StateRejected = "rejected"
|
||||
)
|
||||
|
||||
// DefaultPollInterval 是 tasks/get 轮询缺省间隔。
|
||||
const DefaultPollInterval = 500 * time.Millisecond
|
||||
|
||||
// Executor 把远端 A2A 智能体接成 squad.Executor:一次步骤执行 = 一个
|
||||
// A2A 任务(message/send 发起,tasks/get 轮询到终态)。
|
||||
type Executor struct {
|
||||
// BaseURL 远端 JSON-RPC 端点(Agent Card 的 url)。
|
||||
BaseURL string
|
||||
// Client HTTP 客户端(nil = http.DefaultClient)。
|
||||
Client *http.Client
|
||||
// PollInterval 轮询间隔(0 = DefaultPollInterval)。
|
||||
PollInterval time.Duration
|
||||
// OnState 可选状态回调(观测任务状态机迁移)。
|
||||
OnState func(state string)
|
||||
|
||||
callID atomic.Int64
|
||||
}
|
||||
|
||||
// NewExecutor 按 Agent Card 构造远端执行器。
|
||||
func NewExecutor(card AgentCard) *Executor {
|
||||
return &Executor{BaseURL: card.URL}
|
||||
}
|
||||
|
||||
// Name 实现 squad.Executor。
|
||||
func (e *Executor) Name() string { return "a2a" }
|
||||
|
||||
// ---- A2A 1.0 报文形态(JSON-RPC 2.0) ----
|
||||
|
||||
type rpcRequest struct {
|
||||
JSONRPC string `json:"jsonrpc"`
|
||||
ID int64 `json:"id"`
|
||||
Method string `json:"method"`
|
||||
Params any `json:"params"`
|
||||
}
|
||||
|
||||
type rpcError struct {
|
||||
Code int `json:"code"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
type rpcResponse struct {
|
||||
JSONRPC string `json:"jsonrpc"`
|
||||
ID int64 `json:"id"`
|
||||
Result json.RawMessage `json:"result,omitempty"`
|
||||
Error *rpcError `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// Part 是消息/工件的部件(text | file | data)。
|
||||
type Part struct {
|
||||
Kind string `json:"kind"`
|
||||
Text string `json:"text,omitempty"`
|
||||
File *FilePart `json:"file,omitempty"`
|
||||
Data *json.RawMessage `json:"data,omitempty"`
|
||||
}
|
||||
|
||||
// FilePart 文件部件:bytes(base64)或 uri 二选一。
|
||||
type FilePart struct {
|
||||
Name string `json:"name"`
|
||||
Bytes string `json:"bytes,omitempty"`
|
||||
URI string `json:"uri,omitempty"`
|
||||
MimeType string `json:"mimeType,omitempty"`
|
||||
}
|
||||
|
||||
// Message 是一条 A2A 消息。
|
||||
type Message struct {
|
||||
Kind string `json:"kind"`
|
||||
Role string `json:"role"`
|
||||
Parts []Part `json:"parts"`
|
||||
MessageID string `json:"messageId"`
|
||||
}
|
||||
|
||||
// TaskStatus 任务状态。
|
||||
type TaskStatus struct {
|
||||
State string `json:"state"`
|
||||
Message *Message `json:"message,omitempty"`
|
||||
Timestamp string `json:"timestamp,omitempty"`
|
||||
}
|
||||
|
||||
// Artifact 任务工件。
|
||||
type Artifact struct {
|
||||
ArtifactID string `json:"artifactId,omitempty"`
|
||||
Name string `json:"name,omitempty"`
|
||||
Parts []Part `json:"parts"`
|
||||
}
|
||||
|
||||
// Task 是 A2A 任务(状态机载体)。
|
||||
type Task struct {
|
||||
Kind string `json:"kind"`
|
||||
ID string `json:"id"`
|
||||
ContextID string `json:"contextId,omitempty"`
|
||||
Status TaskStatus `json:"status"`
|
||||
Artifacts []Artifact `json:"artifacts,omitempty"`
|
||||
}
|
||||
|
||||
// Execute 实现 squad.Executor:发起任务并轮询到终态,映射 ExecResult。
|
||||
// 传输/协议错误返回 error(引擎按瞬态失败评估);任务级失败
|
||||
// (failed/canceled/rejected/input-required)返回 Success=false 的结果
|
||||
// (fail-closed:跨进程人工输入不在本适配面内)。
|
||||
func (e *Executor) Execute(ctx context.Context, req squad.ExecRequest) (*squad.ExecResult, error) {
|
||||
if strings.TrimSpace(e.BaseURL) == "" {
|
||||
return nil, fmt.Errorf("a2a: executor 缺 BaseURL")
|
||||
}
|
||||
prompt := strings.TrimSpace(req.Prompt)
|
||||
if prompt == "" {
|
||||
prompt = strings.TrimSpace(req.Instructions)
|
||||
}
|
||||
msg := Message{
|
||||
Kind: "message", Role: "user",
|
||||
Parts: []Part{{Kind: "text", Text: prompt}},
|
||||
MessageID: fmt.Sprintf("msg-%d-%d", time.Now().UnixNano(), e.callID.Add(1)),
|
||||
}
|
||||
raw, err := e.call(ctx, "message/send", map[string]any{"message": msg})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 服务端可直接回 Message(无任务态)或 Task。
|
||||
var probe struct {
|
||||
Kind string `json:"kind"`
|
||||
}
|
||||
_ = json.Unmarshal(raw, &probe)
|
||||
if probe.Kind == "message" {
|
||||
var m Message
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
return nil, fmt.Errorf("a2a: message 结果解析失败: %w", err)
|
||||
}
|
||||
return e.resultFromParts(m.Parts, req.Targets), nil
|
||||
}
|
||||
var task Task
|
||||
if err := json.Unmarshal(raw, &task); err != nil {
|
||||
return nil, fmt.Errorf("a2a: task 结果解析失败: %w", err)
|
||||
}
|
||||
e.observe(task.Status.State)
|
||||
if shouldStop(task.Status.State) {
|
||||
return e.resultFromTask(task, req.Targets), nil
|
||||
}
|
||||
// 轮询任务状态机直至终态(ctx 取消传播)。
|
||||
interval := e.PollInterval
|
||||
if interval <= 0 {
|
||||
interval = DefaultPollInterval
|
||||
}
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-ticker.C:
|
||||
}
|
||||
raw, err := e.call(ctx, "tasks/get", map[string]any{"id": task.ID})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var next Task
|
||||
if err := json.Unmarshal(raw, &next); err != nil {
|
||||
return nil, fmt.Errorf("a2a: tasks/get 解析失败: %w", err)
|
||||
}
|
||||
if next.Status.State != task.Status.State {
|
||||
task = next
|
||||
e.observe(task.Status.State)
|
||||
} else {
|
||||
task = next
|
||||
}
|
||||
if shouldStop(task.Status.State) {
|
||||
return e.resultFromTask(task, req.Targets), nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// call 发一次 JSON-RPC 调用并取 result。
|
||||
func (e *Executor) call(ctx context.Context, method string, params any) (json.RawMessage, error) {
|
||||
client := e.Client
|
||||
if client == nil {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
body, _ := json.Marshal(rpcRequest{
|
||||
JSONRPC: "2.0", ID: e.callID.Add(1), Method: method, Params: params,
|
||||
})
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.BaseURL, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("a2a: %s 传输失败: %w", method, err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
data, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("a2a: %s 读取响应失败: %w", method, err)
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("a2a: %s 返回 HTTP %d", method, resp.StatusCode)
|
||||
}
|
||||
var env rpcResponse
|
||||
if err := json.Unmarshal(data, &env); err != nil {
|
||||
return nil, fmt.Errorf("a2a: %s 响应不是合法 JSON-RPC: %w", method, err)
|
||||
}
|
||||
if env.Error != nil {
|
||||
return nil, fmt.Errorf("a2a: %s 协议错误 %d: %s", method, env.Error.Code, env.Error.Message)
|
||||
}
|
||||
return env.Result, nil
|
||||
}
|
||||
|
||||
func (e *Executor) observe(state string) {
|
||||
if e.OnState != nil {
|
||||
e.OnState(state)
|
||||
}
|
||||
}
|
||||
|
||||
func isTerminal(state string) bool {
|
||||
switch state {
|
||||
case StateCompleted, StateFailed, StateCanceled, StateRejected:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// shouldStop 轮询停止条件:真终态,或 input-required(跨进程人工输入不在
|
||||
// 本适配面内,fail-closed 交引擎语义处理,不死等)。
|
||||
func shouldStop(state string) bool {
|
||||
return isTerminal(state) || state == StateInputRequired
|
||||
}
|
||||
|
||||
// resultFromTask 终态任务 → ExecResult。
|
||||
func (e *Executor) resultFromTask(task Task, targets []string) *squad.ExecResult {
|
||||
switch task.Status.State {
|
||||
case StateCompleted:
|
||||
var parts []Part
|
||||
for _, art := range task.Artifacts {
|
||||
parts = append(parts, art.Parts...)
|
||||
}
|
||||
return e.resultFromParts(parts, targets)
|
||||
case StateInputRequired:
|
||||
return &squad.ExecResult{Success: false,
|
||||
ErrorReason: "a2a: 远端要求人工输入(input-required):" + statusText(task.Status)}
|
||||
case StateCanceled:
|
||||
return &squad.ExecResult{Success: false,
|
||||
ErrorReason: "a2a: 远端任务被取消(canceled):" + statusText(task.Status)}
|
||||
case StateRejected:
|
||||
return &squad.ExecResult{Success: false,
|
||||
ErrorReason: "a2a: 远端拒绝任务(rejected):" + statusText(task.Status)}
|
||||
default: // failed 及未知终态一律 fail-closed
|
||||
return &squad.ExecResult{Success: false,
|
||||
ErrorReason: "a2a: 远端任务失败(" + task.Status.State + "):" + statusText(task.Status)}
|
||||
}
|
||||
}
|
||||
|
||||
// resultFromParts 部件流 → patches:FilePart(bytes) 按名落 patch;
|
||||
// 文本部件拼接后写入全部声明目标(无文件工件时的交付物兜底,
|
||||
// 与其他后端「正文写目标文件」契约一致)。
|
||||
func (e *Executor) resultFromParts(parts []Part, targets []string) *squad.ExecResult {
|
||||
patches := map[string][]byte{}
|
||||
var text strings.Builder
|
||||
for _, p := range parts {
|
||||
switch p.Kind {
|
||||
case "text":
|
||||
if text.Len() > 0 {
|
||||
text.WriteString("\n")
|
||||
}
|
||||
text.WriteString(p.Text)
|
||||
case "file":
|
||||
if p.File == nil || p.File.Name == "" {
|
||||
continue
|
||||
}
|
||||
data, err := base64.StdEncoding.DecodeString(p.File.Bytes)
|
||||
if err != nil {
|
||||
continue // uri 形态与坏 base64 不落 patch(消费侧最小面)
|
||||
}
|
||||
patches[p.File.Name] = data
|
||||
}
|
||||
}
|
||||
body := strings.TrimSpace(text.String())
|
||||
if len(targets) == 0 {
|
||||
targets = []string{"output.md"}
|
||||
}
|
||||
if body != "" {
|
||||
for _, t := range targets {
|
||||
if _, ok := patches[t]; !ok {
|
||||
patches[t] = []byte(body + "\n")
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(patches) == 0 {
|
||||
return &squad.ExecResult{Success: false, ErrorReason: "a2a: 完成任务无任何工件"}
|
||||
}
|
||||
return &squad.ExecResult{Success: true, Patches: patches, Logs: body}
|
||||
}
|
||||
|
||||
func statusText(st TaskStatus) string {
|
||||
if st.Message == nil {
|
||||
return ""
|
||||
}
|
||||
var b strings.Builder
|
||||
for _, p := range st.Message.Parts {
|
||||
if p.Kind == "text" {
|
||||
b.WriteString(p.Text)
|
||||
}
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
module agent-squad/adapters/a2a
|
||||
|
||||
go 1.27
|
||||
|
||||
require agent-squad v0.0.0
|
||||
|
||||
require error-code-system v0.0.0 // indirect
|
||||
|
||||
replace agent-squad => ../..
|
||||
|
||||
replace error-code-system => ../../deps/error-code-system
|
||||
Vendored
+1
-1
Submodule deps/agent-native updated: 1ac21c68d4...37c5e6fbd6
@@ -0,0 +1,41 @@
|
||||
package squad
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// brief_test.go — Q3 WP1a:StepSpec.Brief 渲染进 TASK.md「任务指令」节。
|
||||
|
||||
func TestStepBriefRendersIntoTaskFile(t *testing.T) {
|
||||
rec := &promptRecordingBackend{prompts: map[string][]string{}}
|
||||
engine := newRecordingEngine(t, rec, []StepSpec{
|
||||
{ID: "s1", Role: "researcher", Title: "起草", Targets: []string{"output.md"},
|
||||
Brief: "写一份关于「太空电梯」的简报,篇幅:短。\n必须包含单独一行「核对标记:P4」。"},
|
||||
})
|
||||
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)
|
||||
}
|
||||
var prompt string
|
||||
for _, p := range rec.prompts["01_researcher"] {
|
||||
prompt += p
|
||||
}
|
||||
if !strings.Contains(prompt, "## 任务指令") {
|
||||
t.Fatalf("TASK.md 缺任务指令节:%q", prompt)
|
||||
}
|
||||
if !strings.Contains(prompt, "核对标记:P4") {
|
||||
t.Fatalf("任务指令正文缺失:%q", prompt)
|
||||
}
|
||||
// 指令节在期望产出之后、上下文表之前。
|
||||
iOut := strings.Index(prompt, "期望产出")
|
||||
iBrief := strings.Index(prompt, "## 任务指令")
|
||||
iCtx := strings.Index(prompt, "## 任务上下文资源")
|
||||
if !(iOut < iBrief && iBrief < iCtx) {
|
||||
t.Fatalf("任务指令节位置错误:%d/%d/%d", iOut, iBrief, iCtx)
|
||||
}
|
||||
}
|
||||
@@ -114,6 +114,19 @@ 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"`
|
||||
// TaskBoardLeaseSeconds task_board 任务租约时限(秒;0 = 默认 600)。
|
||||
// 孤儿租约(claimed 且过期)在驱动重入时再武装回 open。
|
||||
TaskBoardLeaseSeconds int `json:"task_board_lease_seconds,omitempty"`
|
||||
}
|
||||
|
||||
// SquadEngine 小队引擎门面:Run / Respond / Resume / Inspect / ListRuns。
|
||||
@@ -250,6 +263,11 @@ func (e *SquadEngine) Run(ctx context.Context, goal Goal, runOpts RunOptions) (*
|
||||
}
|
||||
if runOpts.Topology != "" {
|
||||
meta.Metadata["topology"] = runOpts.Topology
|
||||
// leader 拓扑映射到委派循环须发生在计划确认之前(确认门按
|
||||
// mode=leader 豁免);后置映射会让 leader 运行误触计划确认。
|
||||
if runOpts.Topology == TopologyLeader && meta.Metadata["mode"] == "" {
|
||||
meta.Metadata["mode"] = "leader"
|
||||
}
|
||||
}
|
||||
if !runOpts.Deadline.IsZero() {
|
||||
// RFC3339Nano:秒级 RFC3339 会截断亚秒精度,短截止会被立刻判超期。
|
||||
@@ -258,6 +276,24 @@ 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
|
||||
}
|
||||
if runOpts.TaskBoardLeaseSeconds > 0 {
|
||||
meta.Metadata["task_board_lease_seconds"] = fmt.Sprintf("%d", runOpts.TaskBoardLeaseSeconds)
|
||||
}
|
||||
e.deps.Sink.Emit(Event{Type: EvRunStarted, RunID: runID, Text: goal.Text, At: time.Now()})
|
||||
|
||||
// 意图闸门(可选,入口一次裁决;显式 TaskKey 优先,不重复过闸)。
|
||||
@@ -434,6 +470,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 +539,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 +557,32 @@ 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
|
||||
case TopologyTaskBoard:
|
||||
out, err := e.runTaskBoardDriver(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 == "" {
|
||||
|
||||
@@ -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] + "…"
|
||||
}
|
||||
@@ -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)})
|
||||
}
|
||||
@@ -128,6 +128,10 @@ type StepSpec struct {
|
||||
Branches []Branch `json:"branches,omitempty"`
|
||||
// Handoff 非空 = 完成即把最终答案所有权移交给该角色(Q2 handoff 原语)。
|
||||
Handoff string `json:"handoff,omitempty"`
|
||||
// Brief 编译期任务指令(Q3 编译器):非空时渲染进 TASK.md 的
|
||||
// 「任务指令」节,位于标题/产出之后、上下文表之前。图步的富文本任务书
|
||||
// 载体(此前只有 leader 委派有 briefing)。
|
||||
Brief string `json:"brief,omitempty"`
|
||||
}
|
||||
|
||||
// LoopSpec 是循环步的预算与停机条件(Q2)。
|
||||
|
||||
+366
@@ -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}
|
||||
}
|
||||
@@ -30,6 +30,12 @@ type RunState struct {
|
||||
// Mailbox 运行内点对点消息(Q2):随 RunState 持久化;只在步骤启动时
|
||||
// 并入任务书,永不触碰 HITL 门。
|
||||
Mailbox []MailMessage `json:"mailbox,omitempty"`
|
||||
// Handoffs 所有权转移记录(Q2 handoff 原语):随 RunState 持久化;
|
||||
// 目标角色的下一步骤一次性接收移交上下文,finalize 取链尾产出。
|
||||
Handoffs []HandoffRecord `json:"handoffs,omitempty"`
|
||||
// Board task_board 拓扑的队列状态(Q2):claim/lease/complete 三原语,
|
||||
// 随 RunState 持久化;孤儿租约在驱动重入时再武装。
|
||||
Board []BoardTask `json:"board,omitempty"`
|
||||
// Usage 按角色累计的 token 用量。
|
||||
Usage map[string]*Usage `json:"usage,omitempty"`
|
||||
// Leader leader-delegation 模式的循环状态(workspace 内 leader.json 持久化)。
|
||||
@@ -96,6 +102,8 @@ func (st *RunState) persist() error {
|
||||
RepairBudgetGranted: st.RepairBudgetGranted,
|
||||
Clarifications: st.Clarifications,
|
||||
Mailbox: st.Mailbox,
|
||||
Handoffs: st.Handoffs,
|
||||
Board: st.Board,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -105,6 +113,8 @@ 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"`
|
||||
Board []BoardTask `json:"board,omitempty"`
|
||||
}
|
||||
|
||||
// Workflow 返回计划的编排清单(简洁访问器)。
|
||||
@@ -160,6 +170,8 @@ func loadRunState(wsDir string) (*RunState, error) {
|
||||
st.RepairBudgetGranted = disk.RepairBudgetGranted
|
||||
st.Clarifications = disk.Clarifications
|
||||
st.Mailbox = disk.Mailbox
|
||||
st.Handoffs = disk.Handoffs
|
||||
st.Board = disk.Board
|
||||
} else if !os.IsNotExist(err) {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,308 @@
|
||||
package squad
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// taskboard.go — task_board 拓扑驱动(Q2 WP4)。
|
||||
// 形态:计划步骤 = 板上任务(Needs = 依赖门);执行语义是队列化的
|
||||
// claim/lease/complete 三原语——工作者认领任务(带租约时限)、执行、完成;
|
||||
// 进程死亡留下的孤儿租约(claimed 且租约过期)在驱动重入时再武装回 open
|
||||
// (Attempts 保留,超限由步骤级重试护栏兜底)。全部任务收口后走标准
|
||||
// finalize(交付物/质检/降级判定与图拓扑同一收口面)。
|
||||
// 挂起(HITL)安全:claimed 且步骤未完的任务在重入时续跑,不重复认领。
|
||||
|
||||
func init() {
|
||||
_ = RegisterTopologyDriver(TopologyTaskBoard)
|
||||
}
|
||||
|
||||
// DefaultTaskLeaseSeconds 任务租约缺省时限(秒)。
|
||||
const DefaultTaskLeaseSeconds = 600
|
||||
|
||||
// boardSkipReason 依赖门无法满足(上游 failed/skipped)时的跳过原因。
|
||||
const boardSkipReason = "task_board: 依赖未满足(上游失败或跳过)"
|
||||
|
||||
// BoardTask 是任务板上一条任务的队列状态。
|
||||
type BoardTask struct {
|
||||
ID string `json:"id"`
|
||||
Status string `json:"status"` // open / claimed / done / failed / skipped
|
||||
ClaimedBy string `json:"claimed_by,omitempty"`
|
||||
// LeaseExpires 认领租约到期时刻;过期未 complete = 孤儿租约,重入再武装。
|
||||
LeaseExpires time.Time `json:"lease_expires,omitempty"`
|
||||
// Attempts 认领次数(孤儿再武装累加;执行重试由步骤级护栏负责)。
|
||||
Attempts int `json:"attempts,omitempty"`
|
||||
}
|
||||
|
||||
// runTaskBoardDriver task_board 拓扑入口(幂等可恢复)。
|
||||
func (e *SquadEngine) runTaskBoardDriver(ctx context.Context, st *RunState) (*RunResult, error) {
|
||||
drv := &flowDriver{opts: e.deps}
|
||||
e.initBoard(st)
|
||||
e.rearmOrphanLeases(st)
|
||||
leaseTTL := DefaultTaskLeaseSeconds
|
||||
if v := st.Meta.Metadata["task_board_lease_seconds"]; v != "" {
|
||||
if n := parseIntDefault(v, DefaultTaskLeaseSeconds); n > 0 {
|
||||
leaseTTL = n
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
idx, held := e.boardHeldTask(st)
|
||||
if !held {
|
||||
var ok bool
|
||||
idx, ok = e.boardClaimNext(st, leaseTTL)
|
||||
if !ok {
|
||||
// 无可认领:把依赖已死(上游 failed/skipped)的 open 任务级联
|
||||
// 置 skipped;再无变化则收口。
|
||||
if !e.boardSkipDeadlocked(st) {
|
||||
break
|
||||
}
|
||||
continue
|
||||
}
|
||||
}
|
||||
out, gate, err := e.boardRunTask(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 { // 关键任务失败:fail-closed 传播(failStep 已置运行失败)
|
||||
return out, nil
|
||||
}
|
||||
e.boardSettleTask(st, idx)
|
||||
}
|
||||
return e.boardFinish(ctx, drv, st)
|
||||
}
|
||||
|
||||
// initBoard 首次进入时从计划步骤派生任务板(恢复重入直读持久化状态)。
|
||||
func (e *SquadEngine) initBoard(st *RunState) {
|
||||
st.mu.Lock()
|
||||
defer st.mu.Unlock()
|
||||
if len(st.Board) > 0 {
|
||||
return
|
||||
}
|
||||
for _, s := range st.Plan.Workflow.Steps {
|
||||
status := "open"
|
||||
switch s.Status {
|
||||
case "completed":
|
||||
status = "done"
|
||||
case "failed":
|
||||
status = "failed"
|
||||
case "skipped":
|
||||
status = "skipped"
|
||||
}
|
||||
st.Board = append(st.Board, BoardTask{ID: s.ID, Status: status})
|
||||
}
|
||||
_ = st.save()
|
||||
}
|
||||
|
||||
// rearmOrphanLeases 孤儿租约再武装:claimed 且租约过期 → open(进程死亡
|
||||
// 恢复路径;Attempts 保留作观测)。
|
||||
func (e *SquadEngine) rearmOrphanLeases(st *RunState) {
|
||||
now := time.Now()
|
||||
st.mu.Lock()
|
||||
var rearmed []string
|
||||
for i := range st.Board {
|
||||
t := &st.Board[i]
|
||||
if t.Status == "claimed" && !t.LeaseExpires.IsZero() && t.LeaseExpires.Before(now) {
|
||||
t.Status = "open"
|
||||
t.ClaimedBy = ""
|
||||
rearmed = append(rearmed, t.ID)
|
||||
}
|
||||
}
|
||||
changed := len(rearmed) > 0
|
||||
if changed {
|
||||
_ = st.save()
|
||||
}
|
||||
st.mu.Unlock()
|
||||
if changed {
|
||||
(&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase,
|
||||
Text: fmt.Sprintf("task_board:孤儿租约再武装 %s", strings.Join(rearmed, ","))})
|
||||
}
|
||||
}
|
||||
|
||||
// boardHeldTask 返回 claimed 且步骤未完的任务(HITL 挂起/中断重入的续跑点)。
|
||||
func (e *SquadEngine) boardHeldTask(st *RunState) (int, bool) {
|
||||
for ti := range st.Board {
|
||||
t := &st.Board[ti]
|
||||
if t.Status != "claimed" {
|
||||
continue
|
||||
}
|
||||
if idx, s := e.boardStep(st, t.ID); idx >= 0 && (s.Status == "pending" || s.Status == "running") {
|
||||
return idx, true
|
||||
}
|
||||
}
|
||||
return -1, false
|
||||
}
|
||||
|
||||
// boardClaimNext 认领下一个可执行任务(open 且依赖门全 done),返回步骤下标。
|
||||
func (e *SquadEngine) boardClaimNext(st *RunState, leaseTTL int) (int, bool) {
|
||||
st.mu.Lock()
|
||||
defer st.mu.Unlock()
|
||||
for ti := range st.Board {
|
||||
t := &st.Board[ti]
|
||||
if t.Status != "open" {
|
||||
continue
|
||||
}
|
||||
if !e.boardDepsDoneLocked(st, t.ID) {
|
||||
continue
|
||||
}
|
||||
idx, s := e.boardStepLocked(st, t.ID)
|
||||
if idx < 0 {
|
||||
continue
|
||||
}
|
||||
t.Status = "claimed"
|
||||
t.ClaimedBy = s.Role
|
||||
t.Attempts++
|
||||
t.LeaseExpires = time.Now().Add(time.Duration(leaseTTL) * time.Second)
|
||||
_ = st.save()
|
||||
(&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, StepID: t.ID, Role: s.Role,
|
||||
Text: fmt.Sprintf("task_board:%s 认领 %s(租约 %ds,第 %d 次)", s.Role, t.ID, leaseTTL, t.Attempts)})
|
||||
return idx, true
|
||||
}
|
||||
return -1, false
|
||||
}
|
||||
|
||||
// boardSkipDeadlocked 把依赖已死(上游 failed/skipped)的 open 任务级联置
|
||||
// skipped;返回是否有变化(有则继续循环级联)。
|
||||
func (e *SquadEngine) boardSkipDeadlocked(st *RunState) bool {
|
||||
st.mu.Lock()
|
||||
changed := false
|
||||
for ti := range st.Board {
|
||||
t := &st.Board[ti]
|
||||
if t.Status != "open" {
|
||||
continue
|
||||
}
|
||||
if e.boardDepsDeadLocked(st, t.ID) {
|
||||
t.Status = "skipped"
|
||||
if idx, s := e.boardStepLocked(st, t.ID); idx >= 0 {
|
||||
s.Status = "skipped"
|
||||
s.Error = boardSkipReason
|
||||
st.Plan.Workflow.Steps[idx] = *s
|
||||
}
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if changed {
|
||||
_ = st.save()
|
||||
}
|
||||
st.mu.Unlock()
|
||||
return changed
|
||||
}
|
||||
|
||||
// boardRunTask 执行已认领任务:过门 → runOneStep。
|
||||
func (e *SquadEngine) boardRunTask(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
|
||||
}
|
||||
return drv.runOneStep(ctx, st, idx)
|
||||
}
|
||||
|
||||
// boardSettleTask 任务结算:步骤 completed → done;failed(非关键降级)→ failed。
|
||||
func (e *SquadEngine) boardSettleTask(st *RunState, idx int) {
|
||||
step := st.Plan.Workflow.Steps[idx]
|
||||
status := "done"
|
||||
if step.Status == "failed" {
|
||||
status = "failed"
|
||||
}
|
||||
st.mu.Lock()
|
||||
for i := range st.Board {
|
||||
if st.Board[i].ID == step.ID {
|
||||
st.Board[i].Status = status
|
||||
st.Board[i].LeaseExpires = time.Time{}
|
||||
}
|
||||
}
|
||||
_ = st.save()
|
||||
st.mu.Unlock()
|
||||
(&flowDriver{opts: e.deps}).emit(st, Event{Type: EvPhase, StepID: step.ID, Role: step.Role,
|
||||
Text: fmt.Sprintf("task_board:%s → %s", step.ID, status)})
|
||||
}
|
||||
|
||||
// boardFinish 全部任务收口后走标准 finalize(交付物/质检/降级与图拓扑同面)。
|
||||
// 仍有 open/claimed 任务 = 依赖不可满足,fail-closed 不静默收口。
|
||||
func (e *SquadEngine) boardFinish(ctx context.Context, drv *flowDriver, st *RunState) (*RunResult, error) {
|
||||
done, total := 0, len(st.Board)
|
||||
for _, t := range st.Board {
|
||||
switch t.Status {
|
||||
case "done":
|
||||
done++
|
||||
case "open", "claimed":
|
||||
return e.driverFail(st, fmt.Sprintf("task_board: 任务 %s 无法收口(依赖不可满足)", t.ID)), nil
|
||||
}
|
||||
}
|
||||
drv.emit(st, Event{Type: EvPhase, Text: fmt.Sprintf("task_board:收口(done %d/%d)", done, total)})
|
||||
out, gate, err := drv.finalize(ctx, st)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if gate != nil {
|
||||
return heldResult(st, gate), nil
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// boardStep 按任务 ID 找步骤(只读副本)。
|
||||
func (e *SquadEngine) boardStep(st *RunState, id string) (int, *StepSpec) {
|
||||
st.mu.Lock()
|
||||
defer st.mu.Unlock()
|
||||
return e.boardStepLocked(st, id)
|
||||
}
|
||||
|
||||
// boardStepLocked 按任务 ID 找步骤(调用方持锁)。返回下标与副本指针。
|
||||
func (e *SquadEngine) boardStepLocked(st *RunState, id string) (int, *StepSpec) {
|
||||
for i := range st.Plan.Workflow.Steps {
|
||||
if st.Plan.Workflow.Steps[i].ID == id {
|
||||
s := st.Plan.Workflow.Steps[i]
|
||||
return i, &s
|
||||
}
|
||||
}
|
||||
return -1, nil
|
||||
}
|
||||
|
||||
// boardDepsDoneLocked 依赖门:Needs 全部 done(调用方持锁)。
|
||||
func (e *SquadEngine) boardDepsDoneLocked(st *RunState, id string) bool {
|
||||
needs := e.boardNeedsLocked(st, id)
|
||||
for _, dep := range needs {
|
||||
found := false
|
||||
for i := range st.Board {
|
||||
if st.Board[i].ID == dep {
|
||||
found = true
|
||||
if st.Board[i].Status != "done" {
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
return false // 依赖不在板上 = fail-closed 不认领
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// boardDepsDeadLocked 依赖已死:任一 Need 为 failed/skipped(级联跳过依据)。
|
||||
func (e *SquadEngine) boardDepsDeadLocked(st *RunState, id string) bool {
|
||||
for _, dep := range e.boardNeedsLocked(st, id) {
|
||||
for i := range st.Board {
|
||||
if st.Board[i].ID == dep && (st.Board[i].Status == "failed" || st.Board[i].Status == "skipped") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (e *SquadEngine) boardNeedsLocked(st *RunState, id string) []string {
|
||||
for i := range st.Plan.Workflow.Steps {
|
||||
if st.Plan.Workflow.Steps[i].ID == id {
|
||||
return st.Plan.Workflow.Steps[i].Needs
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -384,9 +384,8 @@ func TestTopologyValidationFailsClosed(t *testing.T) {
|
||||
{"未注册名", "bogus_topology", "未知拓扑"},
|
||||
{"primitive 单跑", TopologyMailbox, "跨拓扑原语"},
|
||||
{"primitive 单跑 handoff", TopologyHandoff, "跨拓扑原语"},
|
||||
{"driver 未点亮", TopologyReview, "驱动未注册"},
|
||||
{"driver 未点亮 group_chat", TopologyGroupChat, "驱动未注册"},
|
||||
{"driver 未点亮 task_board", TopologyTaskBoard, "驱动未注册"},
|
||||
// review/group_chat(WP3)与 task_board(WP4)驱动均已点亮;
|
||||
// 12/12 受理面由 TestTopologyConformance_allTwelve 汇总钉住。
|
||||
}
|
||||
for _, tc := range cases {
|
||||
engine := newTopologyEngine(t, mockScriptPlanner(
|
||||
|
||||
+29
-1
@@ -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)
|
||||
|
||||
@@ -189,6 +189,13 @@ func (e *WorkspaceEngine) writeTaskFile(subDir string, step *StepSpec, role Role
|
||||
fmt.Fprintf(&b, "- 角色:%s(%s)\n", role.Name, role.Role)
|
||||
fmt.Fprintf(&b, "- 期望产出:%v\n\n", step.Targets)
|
||||
|
||||
// 编译期任务指令(Q3):编译器渲染好的富文本任务书,权威仅次于目标。
|
||||
if strings.TrimSpace(step.Brief) != "" {
|
||||
b.WriteString("## 任务指令\n\n")
|
||||
b.WriteString(strings.TrimSpace(step.Brief))
|
||||
b.WriteString("\n\n")
|
||||
}
|
||||
|
||||
b.WriteString("## 任务上下文资源与权威等级(已快照到本目录,只引用本地路径)\n\n")
|
||||
b.WriteString("| 资源路径 | 权威等级 | 用途说明 | 使用与引用指导 |\n|---|---|---|---|\n")
|
||||
b.WriteString("| `context/goal.md` | **authoritative** | 运行目标 | 不可推翻 |\n")
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -0,0 +1,355 @@
|
||||
package squad
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// wp4_conformance_test.go — Q2 WP4 验收:task_board 拓扑(claim/lease/complete
|
||||
// 原语 + 依赖门 + 孤儿租约再武装)与全 12 拓扑 conformance 汇总跑。
|
||||
|
||||
// ---- task_board ----
|
||||
|
||||
// 菱形依赖 a→(b,c)→d:认领序尊重依赖门,全部 done,标准 finalize 收口。
|
||||
func TestTopologyConformance_taskBoard(t *testing.T) {
|
||||
be := &iterBackend{calls: map[string]int{}}
|
||||
sink := &collectSink{}
|
||||
engine := newTopologyEngine(t, mockScriptPlanner(
|
||||
StepSpec{ID: "a", Role: "researcher", Title: "任务a", Targets: []string{"a.md"}},
|
||||
StepSpec{ID: "b", Role: "writer", Title: "任务b", Needs: []string{"a"}, Targets: []string{"b.md"}},
|
||||
StepSpec{ID: "c", Role: "reviewer", Title: "任务c", Needs: []string{"a"}, Targets: []string{"c.md"}},
|
||||
StepSpec{ID: "d", Role: "writer", Title: "任务d", Needs: []string{"b", "c"}, Targets: []string{"output.md"}},
|
||||
), be, sink)
|
||||
res, err := engine.Run(context.Background(), Goal{Text: "任务板"}, RunOptions{Topology: TopologyTaskBoard})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Status != RunCompleted {
|
||||
t.Fatalf("status = %s err = %v", res.Status, res.Error)
|
||||
}
|
||||
// 认领序尊重依赖门:a 先于 b/c,d 最后(板上按步骤序认领,确定性)。
|
||||
var order []string
|
||||
for _, ev := range sink.snapshot() {
|
||||
if strings.Contains(ev.Text, "认领") {
|
||||
order = append(order, ev.StepID)
|
||||
}
|
||||
}
|
||||
want := []string{"a", "b", "c", "d"}
|
||||
if strings.Join(order, ",") != strings.Join(want, ",") {
|
||||
t.Fatalf("认领序 = %v,期望 %v", order, want)
|
||||
}
|
||||
st, _ := engine.loadRun(engine.runDir(res.RunID))
|
||||
for _, task := range st.Board {
|
||||
if task.Status != "done" {
|
||||
t.Fatalf("任务 %s = %s,期望 done", task.ID, task.Status)
|
||||
}
|
||||
}
|
||||
if sink.countText("task_board:收口(done 4/4)") != 1 {
|
||||
t.Fatal("缺收口事件")
|
||||
}
|
||||
}
|
||||
|
||||
// 孤儿租约再武装:崩溃(ctx 取消)留下 claimed+过期租约 → Resume 重入
|
||||
// 再武装回 open、重新认领(Attempts=2)并跑完。
|
||||
func TestTaskBoardOrphanLeaseRearms(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
crashed := false
|
||||
be := &iterBackend{calls: map[string]int{}, content: func(key string, _ int, _ string) []byte {
|
||||
if key == "02_writer" && !crashed {
|
||||
crashed = true
|
||||
cancel() // 模拟进程死亡:runOneStep 在 ctx 检查处退出,步骤回到 pending
|
||||
}
|
||||
return nil
|
||||
}}
|
||||
sink := &collectSink{}
|
||||
engine := newTopologyEngine(t, mockScriptPlanner(
|
||||
StepSpec{ID: "t1", Role: "researcher", Title: "一", Targets: []string{"a.md"}},
|
||||
StepSpec{ID: "t2", Role: "writer", Title: "二", Needs: []string{"t1"}, Targets: []string{"output.md"}},
|
||||
), be, sink)
|
||||
res0, err := engine.Run(ctx, Goal{Text: "任务板"}, RunOptions{Topology: TopologyTaskBoard})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res0.Status != RunCancelled {
|
||||
t.Fatalf("取消后应为 RunCancelled:%s", res0.Status)
|
||||
}
|
||||
// 崩溃现场:t2 claimed、租约未过期;步骤在盘上是 running(setStep 只改
|
||||
// 内存,取消路径的 pending 未落盘)——进程死亡即此形态。测试把租约回拨
|
||||
// 成过期(孤儿)并把步骤置回 pending(模拟重启前的一致化),再走 Resume。
|
||||
st, err := engine.loadRun(engine.runDir(res0.RunID))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
found := false
|
||||
for i := range st.Board {
|
||||
if st.Board[i].ID == "t2" {
|
||||
found = true
|
||||
if st.Board[i].Status != "claimed" || st.Board[i].Attempts != 1 {
|
||||
t.Fatalf("崩溃现场 t2 = %+v", st.Board[i])
|
||||
}
|
||||
st.Board[i].LeaseExpires = time.Now().Add(-time.Hour) // 租约过期 = 孤儿
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatal("板上无 t2")
|
||||
}
|
||||
for i := range st.Plan.Workflow.Steps {
|
||||
if st.Plan.Workflow.Steps[i].ID == "t2" && st.Plan.Workflow.Steps[i].Status == "running" {
|
||||
st.Plan.Workflow.Steps[i].Status = "pending"
|
||||
}
|
||||
}
|
||||
if err := st.save(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// 重入:孤儿租约再武装 → 重新认领 → 跑完收口。
|
||||
res, err := engine.Resume(context.Background(), st.Meta.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Status != RunCompleted {
|
||||
t.Fatalf("恢复后应完成:%s %v", res.Status, res.Error)
|
||||
}
|
||||
if sink.countText("孤儿租约再武装 t2") != 1 {
|
||||
t.Fatal("缺孤儿租约再武装事件")
|
||||
}
|
||||
st2, _ := engine.loadRun(engine.runDir(res.RunID))
|
||||
for _, task := range st2.Board {
|
||||
if task.ID == "t2" && task.Attempts != 2 {
|
||||
t.Fatalf("再武装后 Attempts = %d,期望 2", task.Attempts)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 依赖门级联:非关键任务失败 → 下游 open 任务级联 skipped → 降级收口。
|
||||
func TestTaskBoardDependencyGateCascade(t *testing.T) {
|
||||
be := &iterBackend{calls: map[string]int{}, content: func(key string, _ int, _ string) []byte {
|
||||
return nil
|
||||
}}
|
||||
failBe := &failOnceBackend{inner: be, failKey: "02_writer"}
|
||||
crit := false
|
||||
sink := &collectSink{}
|
||||
engine := newTopologyEngine(t, mockScriptPlanner(
|
||||
StepSpec{ID: "t0", Role: "researcher", Title: "零", Targets: []string{"output.md"}},
|
||||
StepSpec{ID: "t1", Role: "writer", Title: "一", Needs: []string{"t0"}, Targets: []string{"b.md"}, Critical: &crit},
|
||||
StepSpec{ID: "t2", Role: "reviewer", Title: "二", Needs: []string{"t1"}, Targets: []string{"c.md"}},
|
||||
), failBe, sink)
|
||||
res, err := engine.Run(context.Background(), Goal{Text: "任务板"}, RunOptions{Topology: TopologyTaskBoard})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Status != RunDegraded {
|
||||
t.Fatalf("非关键失败+级联跳过应降级收口:%s %v", res.Status, res.Error)
|
||||
}
|
||||
st, _ := engine.loadRun(engine.runDir(res.RunID))
|
||||
byID := map[string]BoardTask{}
|
||||
for _, task := range st.Board {
|
||||
byID[task.ID] = task
|
||||
}
|
||||
if byID["t0"].Status != "done" || byID["t1"].Status != "failed" || byID["t2"].Status != "skipped" {
|
||||
t.Fatalf("板状态 = %+v", byID)
|
||||
}
|
||||
for _, s := range st.Plan.Workflow.Steps {
|
||||
if s.ID == "t2" && (s.Status != "skipped" || s.Error != boardSkipReason) {
|
||||
t.Fatalf("t2 = %s %q", s.Status, s.Error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// failOnceBackend 对指定目录恒定失败(非瞬态),其余委托 inner。
|
||||
type failOnceBackend struct {
|
||||
inner Executor
|
||||
failKey string
|
||||
}
|
||||
|
||||
func (b *failOnceBackend) Name() string { return b.inner.Name() }
|
||||
|
||||
func (b *failOnceBackend) Execute(ctx context.Context, req ExecRequest) (*ExecResult, error) {
|
||||
key := filepath.Base(filepath.Clean(req.WorkDir))
|
||||
if key == b.failKey {
|
||||
return &ExecResult{Success: false, ErrorReason: "fatal: 拒绝执行"}, nil
|
||||
}
|
||||
return b.inner.Execute(ctx, req)
|
||||
}
|
||||
|
||||
// ---- 12/12 汇总 ----
|
||||
|
||||
// TestTopologyConformance_allTwelve 全 12 内置拓扑汇总跑:每个拓扑名一次
|
||||
// 脚本化最小运行(零模型零网络)。graph/mode/driver 十种断言完成收口;
|
||||
// handoff/mailbox 两个 primitive 断言 fail-closed 拒绝单跑(其原语能力分别
|
||||
// 由 TestTopologyConformance_handoff 与 WP2 mailbox 测试钉住)。
|
||||
func TestTopologyConformance_allTwelve(t *testing.T) {
|
||||
specs := BuiltinTopologies()
|
||||
if len(specs) != 12 {
|
||||
t.Fatalf("内置拓扑 = %d,期望 12", len(specs))
|
||||
}
|
||||
for _, spec := range specs {
|
||||
spec := spec
|
||||
t.Run(spec.Name, func(t *testing.T) {
|
||||
switch spec.Name {
|
||||
case TopologySequential:
|
||||
runAgg(t, spec.Name, mockScriptPlanner(
|
||||
StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}},
|
||||
), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunCompleted)
|
||||
case TopologyParallel:
|
||||
runAgg(t, spec.Name, mockScriptPlanner(
|
||||
StepSpec{ID: "s1", Role: "researcher", Title: "甲", Parallel: true, Targets: []string{"a.md"}},
|
||||
StepSpec{ID: "s2", Role: "writer", Title: "乙", Parallel: true, Targets: []string{"output.md"}},
|
||||
), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunCompleted)
|
||||
case TopologyLoop:
|
||||
runAgg(t, spec.Name, mockScriptPlanner(
|
||||
StepSpec{ID: "s1", Role: "researcher", Title: "循环体", Targets: []string{"output.md", "fields.json"},
|
||||
Loop: &LoopSpec{MaxIterations: 2, UntilField: "done", UntilEquals: "yes"}},
|
||||
), &iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte {
|
||||
if target == "fields.json" {
|
||||
return fieldsJSON(map[string]string{"done": "no"})
|
||||
}
|
||||
return nil
|
||||
}}, RunOptions{}, RunCompleted)
|
||||
case TopologyRouter:
|
||||
runAgg(t, spec.Name, mockScriptPlanner(routerSteps()...),
|
||||
&iterBackend{calls: map[string]int{}, content: func(_ string, _ int, target string) []byte {
|
||||
if target == "fields.json" {
|
||||
return fieldsJSON(map[string]string{"route": "fast"})
|
||||
}
|
||||
return nil
|
||||
}}, RunOptions{}, RunCompleted)
|
||||
case TopologyEvaluatorOptimizer:
|
||||
runAgg(t, spec.Name, mockScriptPlanner(
|
||||
StepSpec{ID: "draft", Role: "writer", Title: "起草", Targets: []string{"output.md"}},
|
||||
StepSpec{ID: "eval", Role: "reviewer", Title: "评审", Targets: []string{"verdict.md", "fields.json"},
|
||||
Loop: &LoopSpec{MaxIterations: 4, UntilField: "verdict", UntilEquals: "pass", Body: []string{"draft"}}},
|
||||
), &iterBackend{calls: map[string]int{}, content: func(_ string, iter int, target string) []byte {
|
||||
if target == "fields.json" {
|
||||
if iter >= 3 {
|
||||
return fieldsJSON(map[string]string{"verdict": "pass"})
|
||||
}
|
||||
return fieldsJSON(map[string]string{"verdict": "fail"})
|
||||
}
|
||||
return nil
|
||||
}}, RunOptions{}, RunCompleted)
|
||||
case TopologyLongRunning:
|
||||
runAgg(t, spec.Name, mockScriptPlanner(
|
||||
StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}},
|
||||
), &iterBackend{calls: map[string]int{}},
|
||||
RunOptions{Deadline: time.Now().Add(5 * time.Second)}, RunCompleted)
|
||||
case TopologyLeader:
|
||||
b := &leaderScriptBackend{
|
||||
decisions: []string{delegateJSON("writer", "撰写简报"), closeJSON},
|
||||
script: map[string]map[string]string{"01_writer": {"output.md": goodReport}},
|
||||
}
|
||||
engine := newLeaderEngine(t, b)
|
||||
res, err := engine.Run(context.Background(), Goal{Text: "汇总"}, RunOptions{Topology: TopologyLeader})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Status != RunCompleted {
|
||||
t.Fatalf("status = %s err = %v", res.Status, res.Error)
|
||||
}
|
||||
case TopologyReview:
|
||||
planner := mockScriptPlanner(
|
||||
StepSpec{ID: "f1", Role: "researcher", Title: "评审", Targets: []string{"findings.json"}},
|
||||
)
|
||||
planner.Agents = []RoleSpec{
|
||||
{Role: "researcher", Name: "评审", Backend: "agg-review"},
|
||||
{Role: "verifier", Name: "核实", Backend: "agg-review"},
|
||||
}
|
||||
engine := newAggEngine(t, planner, &aggReviewBackend{}, []RoleSpec{
|
||||
{Role: "researcher", Name: "评审", Backend: "agg-review"},
|
||||
{Role: "verifier", Name: "核实", Backend: "agg-review"},
|
||||
})
|
||||
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, _ := os.ReadFile(filepath.Join(res.WorkDir, "output.md"))
|
||||
if !strings.Contains(string(data), "已丢弃 0 条未验证") || !strings.Contains(string(data), "样例问题") {
|
||||
t.Fatalf("报告 = %s", data)
|
||||
}
|
||||
case TopologyGroupChat:
|
||||
engine := newAggEngine(t, mockScriptPlanner(
|
||||
StepSpec{ID: "p1", Role: "researcher", Title: "甲"},
|
||||
StepSpec{ID: "p2", Role: "writer", Title: "乙"},
|
||||
), &gcScriptBackend{}, []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: 2})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Status != RunCompleted {
|
||||
t.Fatalf("status = %s err = %v", res.Status, res.Error)
|
||||
}
|
||||
case TopologyTaskBoard:
|
||||
runAgg(t, spec.Name, mockScriptPlanner(
|
||||
StepSpec{ID: "t1", Role: "researcher", Title: "一", Targets: []string{"a.md"}},
|
||||
StepSpec{ID: "t2", Role: "writer", Title: "二", Needs: []string{"t1"}, Targets: []string{"output.md"}},
|
||||
), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunCompleted)
|
||||
case TopologyHandoff, TopologyMailbox:
|
||||
// primitive:不可单独作为运行拓扑(fail-closed);原语能力由
|
||||
// 专项 conformance 钉住(handoff/mailbox 测试)。
|
||||
runAgg(t, spec.Name, mockScriptPlanner(
|
||||
StepSpec{ID: "s1", Role: "researcher", Title: "一", Targets: []string{"output.md"}},
|
||||
), &iterBackend{calls: map[string]int{}}, RunOptions{}, RunFailed)
|
||||
default:
|
||||
t.Fatalf("未覆盖的拓扑 %q", spec.Name)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// runAgg 汇总跑的最小引擎:graph 类拓扑共用 iter 三角色装配。
|
||||
func runAgg(t *testing.T, topo string, planner Planner, be Executor, opts RunOptions, want RunStatus) {
|
||||
t.Helper()
|
||||
opts.Topology = topo
|
||||
engine := newTopologyEngine(t, planner, be, nil)
|
||||
res, err := engine.Run(context.Background(), Goal{Text: "汇总跑 " + topo}, opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if res.Status != want {
|
||||
t.Fatalf("status = %s err = %q,期望 %s", res.Status, res.Error, want)
|
||||
}
|
||||
}
|
||||
|
||||
// aggReviewBackend 汇总跑用的最小 review 后端:finder 出一条发现,verify 判 PASS。
|
||||
type aggReviewBackend struct{}
|
||||
|
||||
func (b *aggReviewBackend) Name() string { return "agg-review" }
|
||||
|
||||
func (b *aggReviewBackend) Execute(_ context.Context, req ExecRequest) (*ExecResult, error) {
|
||||
key := filepath.Base(filepath.Clean(req.WorkDir))
|
||||
if strings.HasPrefix(key, "verify_") {
|
||||
return writeOne(req.WorkDir, "verdict.md", "核对成立。\nVERDICT: PASS"), nil
|
||||
}
|
||||
return writeOne(req.WorkDir, "findings.json",
|
||||
`[{"location":"第1段","severity":"high","issue":"样例问题","evidence":"样例证据"}]`), nil
|
||||
}
|
||||
|
||||
// newAggEngine 自定义角色表的汇总引擎(backend 实例名 = be.Name())。
|
||||
func newAggEngine(t *testing.T, planner Planner, be Executor, 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: NopSink{}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return engine
|
||||
}
|
||||
Reference in New Issue
Block a user