diff --git a/docs/tools/a2a/index.md b/docs/tools/a2a/index.md index 9cdc367d9..29a58bbf9 100644 --- a/docs/tools/a2a/index.md +++ b/docs/tools/a2a/index.md @@ -39,6 +39,12 @@ The `Authorization` header shown above authenticates to endpoints served with `d When Docker Desktop is running, eligible requests use its PAC adapter before environment proxy settings. Set `DOCKER_AGENT_DISABLE_DESKTOP_PROXY=1` (or `true`, `yes`, or `on`) to restore standard `HTTP_PROXY`, `HTTPS_PROXY`, `ALL_PROXY`, and `NO_PROXY` routing; `NO_PROXY` does not bypass Docker Desktop PAC selection. Docker Agent does not evaluate PAC files or URLs directly—see [Docker Desktop proxy](../fetch/index.md#docker-desktop-proxy). +## Startup failure behaviour + +Starting the toolset fetches the remote agent's card (`/.well-known/agent-card.json`). When that request fails with one of a fixed set of retryable HTTP statuses — 429 Too Many Requests, 408 Request Timeout, 500/502/503/504, or 529 (Anthropic-style "overloaded") — Docker Agent paces the next start attempt with the same [bounded exponential backoff gate](../rag/index.md#indexing-failures-retries-and-backoff) that remote MCP and RAG embedding calls use, instead of re-fetching the card on every agent turn. This is a fixed enumeration, not a full 5xx range: less-common codes such as 501, 505, or the Cloudflare 520–527 family do not arm the gate. A server-supplied `Retry-After` header, when present, is honored. + +Everything else fails fast, retried every turn with no artificial delay: DNS failures, connection refused, SSRF-blocked private-IP targets (unless `allow_private_ips: true` is set), a malformed or unparsable agent card, and non-retryable HTTP statuses such as 400/401/403/404/501. Only the agent-card fetch during startup is paced; failures from an already-started toolset's per-call `SendStreamingMessage` requests are not. + > [!TIP] > **See also** > diff --git a/docs/tools/rag/index.md b/docs/tools/rag/index.md index 333e10f8d..ba90830e3 100644 --- a/docs/tools/rag/index.md +++ b/docs/tools/rag/index.md @@ -204,7 +204,9 @@ current gate triggers for RAG indexing specifically: > This 429-only trigger set is specific to the RAG/embedding path. Other toolset > types have their own trigger sets against the same gate — for example, remote > MCP toolsets also pace on 408 and a fixed set of 5xx-family statuses (see -> [MCP startup failure behaviour](../mcp/index.md#lifecycle-auto-restart-profiles)). +> [MCP startup failure behaviour](../mcp/index.md#lifecycle-auto-restart-profiles)), +> and the A2A toolset paces its agent-card fetch on the same fixed set (see +> [A2A startup failure behaviour](../a2a/index.md#startup-failure-behaviour)). ### Retry policy and parameters diff --git a/pkg/tools/a2a/a2a.go b/pkg/tools/a2a/a2a.go index dc866a777..869aaa08b 100644 --- a/pkg/tools/a2a/a2a.go +++ b/pkg/tools/a2a/a2a.go @@ -186,17 +186,22 @@ func (t *Toolset) Start(ctx context.Context) error { // `allow_private_ips: true` opt-in disables this for legitimate // internal-service use. client := httpclient.ClientForAllowPrivateIPs(t.timeout, t.allowPrivateIPs) + base := client.Transport + if base == nil { + base = http.DefaultTransport + } + // Recorder sits only in front of the card-resolution GET(s); it never + // reaches the JSON-RPC transport chain built from base below. + rec := &retryAfterRecorder{base: base} + client.Transport = rec + resolver := agentcard.NewResolver(client) card, err := resolver.Resolve(ctx, t.url) if err != nil { - return fmt.Errorf("failed to fetch A2A agent card: %w", err) + return enrichCardError(err, rec) } httpClient := client - base := httpClient.Transport - if base == nil { - base = http.DefaultTransport - } endpointOrigin := t.url if card.URL != "" { diff --git a/pkg/tools/a2a/backoff_test.go b/pkg/tools/a2a/backoff_test.go new file mode 100644 index 000000000..5731cc500 --- /dev/null +++ b/pkg/tools/a2a/backoff_test.go @@ -0,0 +1,177 @@ +package a2a + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + goa2a "github.com/a2aproject/a2a-go/a2a" + "github.com/a2aproject/a2a-go/a2asrv" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/modelerrors" + "github.com/docker/docker-agent/pkg/tools" +) + +// TestBackoffGate_A2ARetryableStatusPacesStart is an end-to-end regression +// test: it drives a REAL *Toolset (via NewToolset, matching production +// wiring) through tools.StartableToolSet.TryStart against a mock +// agent-card server that always answers 503. It proves the whole chain — +// enrichCardError -> Toolset.Start -> the backoff gate in tryStartLocked — +// stays intact end to end, mirroring +// TestBackoffGate_RemoteMCPRetryableStatusPacesReconnect +// (pkg/tools/mcp/remote_test.go). +func TestBackoffGate_A2ARetryableStatusPacesStart(t *testing.T) { + t.Parallel() + + var attempts atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + + toolset := NewToolset("test", srv.URL, nil, WithAllowPrivateIPs(true)) + + now := time.Now() + clock := func() time.Time { return now } + identityJitter := func(d time.Duration) time.Duration { return d } + + s := tools.NewStartable(toolset, tools.WithStartRetryClock(clock), tools.WithStartRetryJitter(identityJitter)) + + // Attempt 1: gate is idle, the real card-resolution attempt runs and + // fails, arming the gate. + started, err := s.TryStart(t.Context()) + assert.False(t, started) + require.Error(t, err) + afterFirst := attempts.Load() + assert.Positive(t, afterFirst, "first TryStart must hit the server") + + var se *modelerrors.StatusError + require.ErrorAs(t, err, &se, "the gate-arming error must carry the StatusError") + assert.Equal(t, http.StatusServiceUnavailable, se.StatusCode) + + // Immediately after: gate armed, TryStart returns without a new + // resolution attempt reaching the server. + started, err = s.TryStart(t.Context()) + assert.False(t, started) + require.Error(t, err) + assert.Equal(t, afterFirst, attempts.Load(), "gate must block the retry from reaching the server") + + // Advance the fake clock past the backoff window (comfortably beyond + // the documented 5-minute cap): gate opens, a new attempt reaches the + // server. + now = now.Add(6 * time.Minute) + started, err = s.TryStart(t.Context()) + assert.False(t, started) + require.Error(t, err) + assert.Greater(t, attempts.Load(), afterFirst, "gate must open and retry once the window elapses") +} + +// TestBackoffGate_A2ANonRetryableStatusFailsPromptly is the negative +// counterpart: a 403 (bad auth / bad config) must fail every turn without +// any pacing, through the same real TryStart path. +func TestBackoffGate_A2ANonRetryableStatusFailsPromptly(t *testing.T) { + t.Parallel() + + var attempts atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + w.WriteHeader(http.StatusForbidden) + })) + defer srv.Close() + + toolset := NewToolset("test", srv.URL, nil, WithAllowPrivateIPs(true)) + s := tools.NewStartable(toolset) + + var prev int32 + for range 3 { + started, err := s.TryStart(t.Context()) + assert.False(t, started) + require.Error(t, err) + cur := attempts.Load() + assert.Greater(t, cur, prev, "403 must reach the server every turn, no pacing") + prev = cur + } +} + +// TestBackoffGate_A2ARecoversAfterBackoffWindow closes the "repeated +// failures then eventual success" recovery criterion at the A2A integration +// layer (mirrored from TestBackoffGate_RemoteMCPRecoversAfterBackoffWindow): +// an agent-card endpoint that answers 503 arms the gate, then recovers to a +// real, working agent card + JSON-RPC handshake — the next TryStart after +// the window elapses must actually start the toolset, not merely stop +// erroring. +func TestBackoffGate_A2ARecoversAfterBackoffWindow(t *testing.T) { + t.Parallel() + + rpcServer := httptest.NewServer(a2asrv.NewJSONRPCHandler(a2asrv.NewHandler(testA2AHandler{}))) + defer rpcServer.Close() + + var failing atomic.Bool + failing.Store(true) + + var attempts atomic.Int32 + cardServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + attempts.Add(1) + if failing.Load() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(goa2a.AgentCard{ + Name: "test", + Description: "test", + URL: rpcServer.URL, + Version: "1.0.0", + ProtocolVersion: string(goa2a.Version), + PreferredTransport: goa2a.TransportProtocolJSONRPC, + Capabilities: goa2a.AgentCapabilities{Streaming: true}, + DefaultInputModes: []string{"text/plain"}, + DefaultOutputModes: []string{"text/plain"}, + Skills: []goa2a.AgentSkill{{ + ID: "test", + Name: "test", + Description: "test", + Tags: []string{"test"}, + }}, + }) + })) + defer cardServer.Close() + + toolset := NewToolset("test", cardServer.URL, nil, WithAllowPrivateIPs(true)) + + now := time.Now() + clock := func() time.Time { return now } + identityJitter := func(d time.Duration) time.Duration { return d } + + s := tools.NewStartable(toolset, tools.WithStartRetryClock(clock), tools.WithStartRetryJitter(identityJitter)) + + // Attempt 1: server failing, resolution fails, gate arms. + started, err := s.TryStart(t.Context()) + assert.False(t, started) + require.Error(t, err) + + var se *modelerrors.StatusError + require.ErrorAs(t, err, &se) + + // Attempt 2, still within the window: gate blocks, no new request. + afterFirst := attempts.Load() + started, err = s.TryStart(t.Context()) + assert.False(t, started) + require.Error(t, err) + assert.Equal(t, afterFirst, attempts.Load(), "gate must block the retry from reaching the server") + + // Server recovers and the backoff window elapses: the next TryStart + // must actually start the toolset. + failing.Store(false) + now = now.Add(6 * time.Minute) + + started, err = s.TryStart(t.Context()) + require.NoError(t, err) + assert.True(t, started, "toolset must actually start once the server recovers, not merely stop erroring") +} diff --git a/pkg/tools/a2a/carderror.go b/pkg/tools/a2a/carderror.go new file mode 100644 index 000000000..cf9927ba6 --- /dev/null +++ b/pkg/tools/a2a/carderror.go @@ -0,0 +1,54 @@ +package a2a + +import ( + "errors" + "fmt" + "net/http" + + "github.com/a2aproject/a2a-go/a2aclient/agentcard" + + "github.com/docker/docker-agent/pkg/modelerrors" +) + +// retryAfterRecorder records the status and Retry-After header of the most +// recent >=400 response, since agentcard.Resolver discards the *http.Response +// on error. No mutex: Start issues one synchronous GET per attempt. +type retryAfterRecorder struct { + base http.RoundTripper + + status int + retryAfter string +} + +func (r *retryAfterRecorder) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := r.base.RoundTrip(req) + if err == nil && resp != nil && resp.StatusCode >= 400 { + r.status = resp.StatusCode + r.retryAfter = resp.Header.Get("Retry-After") + } + return resp, err +} + +func (r *retryAfterRecorder) snapshot() (status int, retryAfter string) { + return r.status, r.retryAfter +} + +// enrichCardError wraps a failed card resolution as *modelerrors.StatusError +// when err is an *agentcard.ErrStatusNotOK, forwarding rec's Retry-After only +// when its recorded status matches, mirroring enrichConnectError (remote.go). +func enrichCardError(err error, rec *retryAfterRecorder) error { + wrapped := fmt.Errorf("failed to fetch A2A agent card: %w", err) + + var statusErr *agentcard.ErrStatusNotOK + if !errors.As(err, &statusErr) { + return wrapped + } + + resp := &http.Response{Header: http.Header{}} + if rec != nil { + if status, retryAfter := rec.snapshot(); status == statusErr.StatusCode && retryAfter != "" { + resp.Header.Set("Retry-After", retryAfter) + } + } + return modelerrors.WrapHTTPError(statusErr.StatusCode, resp, wrapped) +} diff --git a/pkg/tools/a2a/carderror_test.go b/pkg/tools/a2a/carderror_test.go new file mode 100644 index 000000000..361043cda --- /dev/null +++ b/pkg/tools/a2a/carderror_test.go @@ -0,0 +1,184 @@ +package a2a + +import ( + "fmt" + "net" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/docker/docker-agent/pkg/modelerrors" +) + +// TestEnrichCardError_RetryableStatusArmsGate verifies that a card-resolution +// failure carrying one of the fixed retryable HTTP statuses surfaces as a +// *modelerrors.StatusError, arming the StartableToolSet backoff gate exactly +// as remote MCP's enrichConnectError does (pkg/tools/mcp/remote.go). +func TestEnrichCardError_RetryableStatusArmsGate(t *testing.T) { + t.Parallel() + + for _, status := range []int{429, 408, 500, 502, 503, 504, 529} { + t.Run(fmt.Sprintf("status_%d", status), func(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(status) + })) + defer srv.Close() + + toolSet := NewToolset("test", srv.URL, nil, WithAllowPrivateIPs(true)) + err := toolSet.Start(t.Context()) + require.Error(t, err) + + var se *modelerrors.StatusError + require.ErrorAs(t, err, &se, "a %d response must surface as *StatusError", status) + assert.Equal(t, status, se.StatusCode) + assert.True(t, modelerrors.RetryableHTTPStatus(se), + "status %d must be classified retryable by the backoff gate", status) + }) + } +} + +// TestEnrichCardError_NonRetryableStatusDoesNotArm verifies that a +// card-resolution failure carrying a non-retryable status still surfaces as +// *StatusError (for structured access) but is not classified retryable, so +// bad-config / auth / malformed-request failures fail promptly. 501 proves +// the arming set is a fixed enumeration and not a full 5xx range. +func TestEnrichCardError_NonRetryableStatusDoesNotArm(t *testing.T) { + t.Parallel() + + for _, status := range []int{400, 401, 403, 404, 501} { + t.Run(fmt.Sprintf("status_%d", status), func(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(status) + })) + defer srv.Close() + + toolSet := NewToolset("test", srv.URL, nil, WithAllowPrivateIPs(true)) + err := toolSet.Start(t.Context()) + require.Error(t, err) + + var se *modelerrors.StatusError + require.ErrorAs(t, err, &se, "%d must still surface as *StatusError (wrapped for structured access)", status) + assert.Equal(t, status, se.StatusCode) + assert.False(t, modelerrors.RetryableHTTPStatus(se), + "status %d must NOT be classified retryable", status) + }) + } +} + +// TestEnrichCardError_NoStatusNoStatusError verifies that failures which +// never produce an HTTP response — connection refused, SSRF-blocked dial, +// or a 200 response whose body doesn't parse as an agent card — never carry +// a *StatusError, so the backoff gate does not arm on them. +func TestEnrichCardError_NoStatusNoStatusError(t *testing.T) { + t.Parallel() + + t.Run("connection refused", func(t *testing.T) { + t.Parallel() + + ln, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := ln.Addr().String() + require.NoError(t, ln.Close()) + + toolSet := NewToolset("test", "http://"+addr, nil, WithAllowPrivateIPs(true)) + err = toolSet.Start(t.Context()) + require.Error(t, err) + + var se *modelerrors.StatusError + assert.NotErrorAs(t, err, &se, "a connection-refused failure must not carry a *StatusError") + }) + + t.Run("SSRF-blocked private IP", func(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + + // allow_private_ips is left at its default (false): the loopback + // httptest server must never be reached, so any 503 it would have + // served is irrelevant to the error the toolset produces. + toolSet := NewToolset("test", srv.URL, nil) + err := toolSet.Start(t.Context()) + require.Error(t, err) + + var se *modelerrors.StatusError + assert.NotErrorAs(t, err, &se, "an SSRF-blocked dial must not carry a *StatusError") + }) + + t.Run("malformed agent card", func(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = fmt.Fprint(w, "not valid json") + })) + defer srv.Close() + + toolSet := NewToolset("test", srv.URL, nil, WithAllowPrivateIPs(true)) + err := toolSet.Start(t.Context()) + require.Error(t, err) + + var se *modelerrors.StatusError + assert.NotErrorAs(t, err, &se, "a malformed 200 response must not carry a *StatusError") + }) +} + +// TestEnrichCardError_RetryAfterHonoured verifies that a server-supplied +// Retry-After header on the agent-card response is parsed through to the +// resulting *modelerrors.StatusError, matching the handling already in +// place for remote MCP (pkg/tools/mcp/oauth.go) and model-provider adapters. +func TestEnrichCardError_RetryAfterHonoured(t *testing.T) { + t.Parallel() + + for _, status := range []int{503, 429} { + t.Run(fmt.Sprintf("status_%d", status), func(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Retry-After", "120") + w.WriteHeader(status) + })) + defer srv.Close() + + toolSet := NewToolset("test", srv.URL, nil, WithAllowPrivateIPs(true)) + err := toolSet.Start(t.Context()) + require.Error(t, err) + + var se *modelerrors.StatusError + require.ErrorAs(t, err, &se) + assert.Equal(t, status, se.StatusCode) + assert.Equal(t, 120*time.Second, se.RetryAfter, + "the server's Retry-After header must be parsed onto the StatusError") + }) + } +} + +// TestEnrichCardError_NoRetryAfterHeaderLeavesZero verifies that when the +// server does not send a Retry-After header, RetryAfter stays zero so the +// gate falls back to its own computed delay. +func TestEnrichCardError_NoRetryAfterHeaderLeavesZero(t *testing.T) { + t.Parallel() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusTooManyRequests) + })) + defer srv.Close() + + toolSet := NewToolset("test", srv.URL, nil, WithAllowPrivateIPs(true)) + err := toolSet.Start(t.Context()) + require.Error(t, err) + + var se *modelerrors.StatusError + require.ErrorAs(t, err, &se) + assert.Zero(t, se.RetryAfter, "no Retry-After header means the gate computes its own delay") +}