weichen
424bfc0e29
Why: Q3 WP4 requires A2A 1.0 interop from zero: publish this team's capabilities as an Agent Card and consume remote agents as squad.Executors with the A2A task state machine mapped onto ExecResult. Evidence: adapters/a2a conformance with a local fake A2A server (both directions): card build from role/blueprint registries, well-known JSON serving, FetchCard fail-closed validation, message/send -> tasks/get polling through submitted/working -> completed with TextPart/FilePart artifacts mapped to patches/targets, failed/canceled/rejected/input-required mapped to Success=false fail-closed (no cross-process hold), direct Message result, JSON-RPC protocol and ctx-cancel error propagation. Module green with -race; squad root suite unaffected (SELF). CI authoritative (GATE) for the root module; the adapter module follows the adapters/intent and adapters/native separate-go.mod precedent. Impact: new module agent-squad/adapters/a2a (card.go, executor.go); no changes to the squad root module. Refs: Q3 plan WP4 (2026-10-07-agent-stack-q2q3-plans.md); task card .task/feature-agent-stack-q3.md
332 lines
9.5 KiB
Go
332 lines
9.5 KiB
Go
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()
|
||
}
|