diff --git a/adapters/a2a/card.go b/adapters/a2a/card.go new file mode 100644 index 0000000..d4c9b97 --- /dev/null +++ b/adapters/a2a/card.go @@ -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 +} diff --git a/adapters/a2a/conformance_test.go b/adapters/a2a/conformance_test.go new file mode 100644 index 0000000..b9bf1e0 --- /dev/null +++ b/adapters/a2a/conformance_test.go @@ -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) + } +} diff --git a/adapters/a2a/executor.go b/adapters/a2a/executor.go new file mode 100644 index 0000000..6c016cb --- /dev/null +++ b/adapters/a2a/executor.go @@ -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() +} diff --git a/adapters/a2a/go.mod b/adapters/a2a/go.mod new file mode 100644 index 0000000..141ea0b --- /dev/null +++ b/adapters/a2a/go.mod @@ -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