From c473bb7536afc89a4953443012b62e9571ed86da Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arnaud=20H=C3=A9ritier?= Date: Tue, 1 Sep 2026 20:39:26 +0000 Subject: [PATCH] fix: extend backoff gate to A2A agent-card HTTP errors (#4098) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A2A toolset startup fetches the remote agent's card via agentcard.Resolver, which returns *agentcard.ErrStatusNotOK{StatusCode, Status} on any non-200 response. #4074's deferral rationale ("the agent-card resolver does not expose HTTP status cleanly") was wrong on this point — the resolver already exposes the status; enrichCardError (pkg/tools/a2a/carderror.go) only needs to translate it into *modelerrors.StatusError via modelerrors.WrapHTTPError, mirroring enrichConnectError's handling of remote MCP HTTP errors (pkg/tools/mcp/remote.go). Toolset.Start now returns enrichCardError's result instead of a bare fmt.Errorf, so retryable responses arm the StartableToolSet backoff gate exactly as remote MCP and RAG embedding 429s do. The one thing the resolver genuinely doesn't expose is the Retry-After header: Resolver.Resolve discards the *http.Response after producing ErrStatusNotOK. retryAfterRecorder is a minimal http.RoundTripper installed only in front of the card-resolution GET (never the JSON-RPC transport chain built afterwards) that records the status and Retry-After of the most recent >=400 response; unlike oauthTransport's lastServerErrorSnapshot it carries no mutex, since Start issues exactly one synchronous card GET per attempt. Its header is only forwarded when its recorded status matches ErrStatusNotOK's own StatusCode, so a header from an unrelated response can never be paired with the wrong status. Classifier policy is unchanged from #4062/#4074: the gate arms only on a fixed enumeration (429, 408, 500, 502, 503, 504, 529), not a full 5xx range — 501, 505, and the Cloudflare 520-527 family do not arm it. Deliberately excluded from arming: DNS failures, connection refused, SSRF-blocked private-IP targets, malformed/unparsable agent cards, and other 4xx statuses (bad auth, bad config) — all fail promptly with no pacing. Only the agent-card fetch during startup is paced; per-call SendStreamingMessage failures on an already-started toolset are not. Adds carderror_test.go (retryable/non-retryable status tables incl. 501 to prove the enumeration isn't a full 5xx range, no-status cases for connection-refused/SSRF-block/malformed-card, and Retry-After present/absent) plus backoff_test.go end-to-end tests that drive a real *Toolset through tools.StartableToolSet.TryStart against a mock 503/403 agent-card server, and a recovery test that flips the mock from 503 to a real, working agent card + JSON-RPC handshake after the backoff window elapses. docs/tools/a2a/index.md documents the new "Startup failure behaviour" section with the same precision as MCP's; docs/tools/rag/index.md's cross-reference note now names A2A alongside remote MCP. Refs #4060, #4074, #4098 --- docs/tools/a2a/index.md | 6 ++ docs/tools/rag/index.md | 4 +- pkg/tools/a2a/a2a.go | 15 ++- pkg/tools/a2a/backoff_test.go | 177 ++++++++++++++++++++++++++++++ pkg/tools/a2a/carderror.go | 54 ++++++++++ pkg/tools/a2a/carderror_test.go | 184 ++++++++++++++++++++++++++++++++ 6 files changed, 434 insertions(+), 6 deletions(-) create mode 100644 pkg/tools/a2a/backoff_test.go create mode 100644 pkg/tools/a2a/carderror.go create mode 100644 pkg/tools/a2a/carderror_test.go 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") +}