From de4c7fbf9c27a76a4d5cd4d4abf50654dc7bbc8f Mon Sep 17 00:00:00 2001 From: nova28 Date: Sat, 15 Aug 2026 23:00:55 +0800 Subject: [PATCH] fix(routing): retry ACP transport disconnects instead of failing outright (#2680) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- apps/backend/cmd/mock-agent/AGENTS.md | 2 + apps/backend/cmd/mock-agent/handler.go | 88 ++++++++++ apps/backend/cmd/mock-agent/main.go | 8 + .../cmd/mock-agent/transport_lost_test.go | 151 ++++++++++++++++++ .../agent/runtime/routingerr/policy.go | 2 +- .../agent/runtime/routingerr/routingerr.go | 9 ++ .../agent/runtime/routingerr/runtime_rules.go | 42 ++++- .../runtime/routingerr/transport_lost_test.go | 121 ++++++++++++++ .../scheduler/routing_lifecycle_test.go | 22 +++ .../orchestrator/event_handlers_transient.go | 4 + ..._handlers_transient_transport_lost_test.go | 56 +++++++ .../chat/messages/action-message.test.tsx | 16 ++ .../task/chat/messages/action-message.tsx | 2 + .../tests/session/session-recovery.spec.ts | 31 +++- .../transient-retry-transport-lost.spec.ts | 79 +++++++++ apps/web/lib/state/slices/office/types.ts | 3 +- apps/web/src/locales/en/chat.json | 1 + apps/web/src/locales/pseudo/chat.json | 1 + apps/web/src/locales/pt-pt/chat.json | 1 + apps/web/src/locales/zh-cn/chat.json | 1 + apps/web/src/locales/zh-hk/chat.json | 1 + apps/web/src/locales/zh-tw/chat.json | 1 + ...8-provider-neutral-agent-error-recovery.md | 1 + .../specs/platform/provider-error-recovery.md | 14 +- 24 files changed, 645 insertions(+), 12 deletions(-) create mode 100644 apps/backend/cmd/mock-agent/transport_lost_test.go create mode 100644 apps/backend/internal/agent/runtime/routingerr/transport_lost_test.go create mode 100644 apps/backend/internal/orchestrator/event_handlers_transient_transport_lost_test.go create mode 100644 apps/web/e2e/tests/session/transient-retry-transport-lost.spec.ts diff --git a/apps/backend/cmd/mock-agent/AGENTS.md b/apps/backend/cmd/mock-agent/AGENTS.md index bf4bd4f9c2..2a5bc07f5f 100644 --- a/apps/backend/cmd/mock-agent/AGENTS.md +++ b/apps/backend/cmd/mock-agent/AGENTS.md @@ -42,6 +42,8 @@ Scenarios in `scenarioRegistry` can only emit `SessionUpdate` notifications — `/overloaded[:N]` is the reference example (`handler.go: handleOverloaded`): it returns the production `529 Overloaded` error for the first `N` prompts of a session (default 1), then recovers with a normal text response. The orchestrator's backoff retry tears the agent process down and relaunches it between attempts, so the fail-count is persisted in a **temp file** keyed by the session id (`overloadedCounterPath`) rather than an in-memory map — it survives the relaunch (and is cleaned up on recovery and in `CloseSession`). Use `/overloaded` to demo the yellow retry status, or a large `N` like `/overloaded:9` to keep failing so the retry loop stays visible / exhausts to the red recovery banner. +`/transport-lost[:N]` (`handler.go: handleTransportLost`) follows the same shape for the ACP wire-level transport-death signature (`peer disconnected before response` — the acp-go-sdk's own `ErrPeerDisconnected` payload shape), which `routingerr`'s `acp.transport_lost.v1` rule classifies as `CodeAgentTransportLost` and admits to the short-retry ladder. Use `/transport-lost` or a large `N` like `/transport-lost:9` the same way as `/overloaded`. + ## Emitter helpers worth knowing `emitter.go` wraps the ACP `sessionUpdater` with helpers that shield scenarios from SDK plumbing: diff --git a/apps/backend/cmd/mock-agent/handler.go b/apps/backend/cmd/mock-agent/handler.go index 3911d3fd39..738cb5028f 100644 --- a/apps/backend/cmd/mock-agent/handler.go +++ b/apps/backend/cmd/mock-agent/handler.go @@ -116,6 +116,94 @@ func nextOverloadedAttempt(sid acp.SessionId) int { return next } +// transportLostErrorData is the exact data payload the acp-go-sdk emits when +// the peer disconnects before a response (see acp.ErrPeerDisconnected / +// newInternalErrorWithCause in connection.go) — the production signature the +// orchestrator's routingerr classifier matches on ("peer disconnected"). +var transportLostErrorData = map[string]any{"error": "peer disconnected before response"} + +// transportLostCmdRe matches `/transport-lost` or `/e2e:transport-lost`, +// optionally followed by `:N` — the number of consecutive prompts to fail +// with the ACP peer-disconnected signature before recovering (default 1). +// Use a large N (e.g. `/transport-lost:9`) to exhaust the retry budget and +// fall through to the red recovery banner. +var transportLostCmdRe = regexp.MustCompile(`(?i)^/(?:e2e:)?transport-lost(?::(\d+))?$`) + +// parseTransportLostCmd reports whether the prompt is the /transport-lost +// command and, if so, how many consecutive prompts it should fail before +// recovering (default 1). Returns ok=false for any other prompt. +func parseTransportLostCmd(prompt string) (failTimes int, ok bool) { + cmd := stripKandevSystem(strings.TrimSpace(prompt)) + m := transportLostCmdRe.FindStringSubmatch(cmd) + if m == nil { + return 0, false + } + failTimes = 1 + if m[1] != "" { + if n, err := strconv.Atoi(m[1]); err == nil && n >= 0 { + failTimes = n + } + } + return failTimes, true +} + +// handleTransportLost implements the /transport-lost scenario: it returns a +// real prompt-time ACP error carrying the production "peer disconnected +// before response" signature for the first N prompts of a session, then +// recovers with a normal text response. Returns handled=false when the +// prompt is not the transport-lost command. +func (a *mockAgent) handleTransportLost(ctx context.Context, sid acp.SessionId, prompt string) (acp.PromptResponse, error, bool) { + failTimes, ok := parseTransportLostCmd(prompt) + if !ok { + return acp.PromptResponse{}, nil, false + } + + // Same rationale as handleOverloaded: persist the attempt count in a temp + // file keyed by the resume-stable session id so fail-then-recover + // survives the orchestrator's tear-down/relaunch between retries. + attempt := nextTransportLostAttempt(sid) + + if attempt <= failTimes { + _, _ = fmt.Fprintf(logOutput, "mock-agent[%d]: emitting peer-disconnected error for session %s (failure %d/%d)\n", + os.Getpid(), sid, attempt, failTimes) + return acp.PromptResponse{}, &acp.RequestError{ + Code: -32603, + Message: "Internal error", + Data: transportLostErrorData, + }, true + } + + // Recovered: clear the counter and emit a normal success response so the + // turn completes (and the orchestrator clears the retry budget). + _ = os.Remove(transportLostCounterPath(sid)) + e := &emitter{ctx: ctx, conn: a.conn, sid: sid} + e.text(fmt.Sprintf("Agent connection recovered after %d transient failure(s). Here is your response.", failTimes)) + return acp.PromptResponse{StopReason: acp.StopReasonEndTurn}, nil, true +} + +// transportLostCounterPath returns the temp-file path tracking how many times +// the /transport-lost scenario has fired for a session. Keyed by the +// resume-stable ACP session id so the count survives process relaunch across +// backoff retries. +func transportLostCounterPath(sid acp.SessionId) string { + safe := strings.NewReplacer("/", "_", "\\", "_", "..", "_").Replace(string(sid)) + return filepath.Join(os.TempDir(), "kandev-mock-transport-lost-"+safe+".count") +} + +// nextTransportLostAttempt increments and returns the persisted attempt count. +func nextTransportLostAttempt(sid acp.SessionId) int { + path := transportLostCounterPath(sid) + prev := 0 + if raw, err := os.ReadFile(path); err == nil { + if n, convErr := strconv.Atoi(strings.TrimSpace(string(raw))); convErr == nil { + prev = n + } + } + next := prev + 1 + _ = os.WriteFile(path, []byte(strconv.Itoa(next)), 0o600) + return next +} + // bulkCmdRe matches `/bulk` or `/e2e:bulk`, optionally followed by `:N` or ` N` // — the number of agent messages to emit (default 120, capped at 1000). Used to // populate a long chat history for manually testing scrollback and the "Load diff --git a/apps/backend/cmd/mock-agent/main.go b/apps/backend/cmd/mock-agent/main.go index 8f8d184ff2..4e27ddf21c 100644 --- a/apps/backend/cmd/mock-agent/main.go +++ b/apps/backend/cmd/mock-agent/main.go @@ -315,6 +315,12 @@ func (a *mockAgent) Prompt(ctx context.Context, req acp.PromptRequest) (acp.Prom if resp, err, handled := a.handleOverloaded(promptCtx, req.SessionId, prompt); handled { return resp, err } + // Same rationale as /overloaded above: /transport-lost must also surface a + // real prompt-time ACP error, so it is intercepted here rather than routed + // through handlePrompt's emitter. + if resp, err, handled := a.handleTransportLost(promptCtx, req.SessionId, prompt); handled { + return resp, err + } e := &emitter{ctx: promptCtx, conn: a.conn, sid: req.SessionId} handlePrompt(e, prompt, a.model) if promptCtx.Err() != nil { @@ -425,6 +431,7 @@ func (a *mockAgent) CloseSession(_ context.Context, req acp.CloseSessionRequest) delete(a.commandsEmitted, req.SessionId) a.mu.Unlock() _ = os.Remove(overloadedCounterPath(req.SessionId)) + _ = os.Remove(transportLostCounterPath(req.SessionId)) return acp.CloseSessionResponse{}, nil } @@ -496,6 +503,7 @@ func mockAvailableCommands() []acp.AvailableCommand { {Name: "async-subagent-teardown", Description: "Replay async Agent work with a missing completion"}, {Name: toolKeyError, Description: "Simulate an error"}, {Name: "overloaded", Description: "Simulate a transient 529 Overloaded error (fails once, then recovers)"}, + {Name: "transport-lost", Description: "Simulate an ACP transport disconnect (fails once, then recovers)"}, {Name: "thinking", Description: "Emit thinking/reasoning blocks"}, {Name: "crash", Description: "Simulate agent crash"}, {Name: "all", Description: "Demonstrate all message types"}, diff --git a/apps/backend/cmd/mock-agent/transport_lost_test.go b/apps/backend/cmd/mock-agent/transport_lost_test.go new file mode 100644 index 0000000000..6fe253def6 --- /dev/null +++ b/apps/backend/cmd/mock-agent/transport_lost_test.go @@ -0,0 +1,151 @@ +package main + +import ( + "context" + "os" + "strconv" + "strings" + "testing" + + acp "github.com/coder/acp-go-sdk" +) + +func newTransportLostTestAgent() *mockAgent { + return &mockAgent{ + model: "mock-fast", + sessions: make(map[acp.SessionId]bool), + commandsEmitted: make(map[acp.SessionId]bool), + } +} + +func TestParseTransportLostCmd(t *testing.T) { + cases := []struct { + prompt string + wantFail int + wantOK bool + }{ + {"/transport-lost", 1, true}, + {"/TRANSPORT-LOST", 1, true}, + {"/e2e:transport-lost", 1, true}, + {"/transport-lost:3", 3, true}, + {"/e2e:transport-lost:9", 9, true}, + {"/transport-lost:0", 0, true}, + {"ctx/transport-lost:2", 2, true}, + {"hello", 0, false}, + {"/transport-los", 0, false}, + {"transport-lost", 0, false}, + } + for _, tc := range cases { + gotFail, gotOK := parseTransportLostCmd(tc.prompt) + if gotOK != tc.wantOK || (tc.wantOK && gotFail != tc.wantFail) { + t.Errorf("parseTransportLostCmd(%q) = (%d, %v), want (%d, %v)", tc.prompt, gotFail, gotOK, tc.wantFail, tc.wantOK) + } + } +} + +func TestHandleTransportLost_NotTheCommand(t *testing.T) { + a := newTransportLostTestAgent() + _, err, handled := a.handleTransportLost(context.Background(), "s1", "hello there") + if handled { + t.Fatal("handled = true for a non-transport-lost prompt, want false") + } + if err != nil { + t.Fatalf("err = %v, want nil", err) + } +} + +func TestHandleTransportLost_EmitsProductionPeerDisconnected(t *testing.T) { + a := newTransportLostTestAgent() + const sid acp.SessionId = "emit-peer-disconnected-sess" + _ = os.Remove(transportLostCounterPath(sid)) + t.Cleanup(func() { _ = os.Remove(transportLostCounterPath(sid)) }) + _, err, handled := a.handleTransportLost(context.Background(), sid, "/transport-lost") + if !handled { + t.Fatal("handled = false, want true") + } + reqErr, ok := err.(*acp.RequestError) + if !ok { + t.Fatalf("err type = %T, want *acp.RequestError", err) + } + if reqErr.Code != -32603 { + t.Errorf("code = %d, want -32603", reqErr.Code) + } + if !strings.Contains(reqErr.Error(), "peer disconnected before response") { + t.Errorf("error = %q, want it to contain 'peer disconnected before response'", reqErr.Error()) + } +} + +func TestHandleTransportLost_FailsExactlyNTimes(t *testing.T) { + a := newTransportLostTestAgent() + const sid acp.SessionId = "failN-transport-lost-sess" + _ = os.Remove(transportLostCounterPath(sid)) + t.Cleanup(func() { _ = os.Remove(transportLostCounterPath(sid)) }) + + // failTimes=2: the first two prompts must error (the persisted counter + // drives this, so it survives across the simulated relaunches the + // orchestrator performs between retries). + for i := 1; i <= 2; i++ { + _, err, handled := a.handleTransportLost(context.Background(), sid, "/transport-lost:2") + if !handled { + t.Fatalf("attempt %d: handled = false, want true", i) + } + if err == nil { + t.Fatalf("attempt %d: err = nil, want a peer-disconnected error", i) + } + } + // The third attempt would recover (emits text via the live conn) — not + // exercised here to avoid needing a real ACP connection; the recovery path + // is covered end-to-end by the Playwright spec. Assert the persisted count + // reached the fail budget so recovery is next. Read the counter file + // directly rather than calling nextTransportLostAttempt, which would + // increment it as a side effect of the assertion itself. + raw, err := os.ReadFile(transportLostCounterPath(sid)) + if err != nil { + t.Fatalf("failed to read persisted attempt count: %v", err) + } + got, err := strconv.Atoi(strings.TrimSpace(string(raw))) + if err != nil { + t.Fatalf("failed to parse persisted attempt count %q: %v", raw, err) + } + if got != 2 { + t.Errorf("persisted attempt count = %d, want 2", got) + } +} + +func TestHandleTransportLost_RecoversAfterFailureBudget(t *testing.T) { + const sid acp.SessionId = "recover-transport-lost-sess" + _ = os.Remove(transportLostCounterPath(sid)) + t.Cleanup(func() { _ = os.Remove(transportLostCounterPath(sid)) }) + + updater := newCapturingUpdater() + a := newTransportLostTestAgent() + a.conn = updater + + if _, err, handled := a.handleTransportLost(context.Background(), sid, "/transport-lost:1"); !handled || err == nil { + t.Fatalf("first attempt = handled %v, err %v, want handled transport error", handled, err) + } + + response, err, handled := a.handleTransportLost(context.Background(), sid, "/transport-lost:1") + if !handled { + t.Fatal("recovery attempt handled = false, want true") + } + if err != nil { + t.Fatalf("recovery attempt error = %v, want nil", err) + } + if response.StopReason != acp.StopReasonEndTurn { + t.Fatalf("stop reason = %q, want end_turn", response.StopReason) + } + texts := updater.textMessages() + if len(texts) != 1 || !strings.Contains(texts[0], "Agent connection recovered") { + t.Fatalf("recovery text = %q, want one connection-recovered message", texts) + } +} + +func TestMockAvailableCommandsIncludesTransportLost(t *testing.T) { + for _, command := range mockAvailableCommands() { + if command.Name == "transport-lost" { + return + } + } + t.Fatal("mockAvailableCommands does not advertise transport-lost") +} diff --git a/apps/backend/internal/agent/runtime/routingerr/policy.go b/apps/backend/internal/agent/runtime/routingerr/policy.go index f88750a167..f0f519d931 100644 --- a/apps/backend/internal/agent/runtime/routingerr/policy.go +++ b/apps/backend/internal/agent/runtime/routingerr/policy.go @@ -52,7 +52,7 @@ func Decide(context ExecutionContext, e *Error, now time.Time) RecoveryDecision func isShortRetryable(e *Error, now time.Time) bool { switch e.Code { - case CodeNetworkUnavailable, CodeProviderUnavailable, CodeProviderOverloaded, CodeModelCapacity: + case CodeNetworkUnavailable, CodeProviderUnavailable, CodeProviderOverloaded, CodeModelCapacity, CodeAgentTransportLost: return e.Confidence == ConfHigh || e.Confidence == ConfMedium case CodeRateLimited: if e.ResetHint == nil || e.ResetHint.IsZero() { diff --git a/apps/backend/internal/agent/runtime/routingerr/routingerr.go b/apps/backend/internal/agent/runtime/routingerr/routingerr.go index 305b089289..67afde5710 100644 --- a/apps/backend/internal/agent/runtime/routingerr/routingerr.go +++ b/apps/backend/internal/agent/runtime/routingerr/routingerr.go @@ -34,6 +34,7 @@ const ( CodePermissionDeniedByUser Code = "permission_denied_by_user" CodeNpxCacheCorrupted Code = "npx_cache_corrupted" CodeResumeCorrupted Code = "resume_corrupted" + CodeAgentTransportLost Code = "agent_transport_lost" ) // RemediationStartFreshSession is the symbolic RemediationPath value for @@ -283,6 +284,14 @@ func applyInvariants(e *Error) *Error { case CodePermissionDeniedByUser, CodeTask, CodeRepo, CodeAgentRuntime: e.FallbackAllowed = false e.AutoRetryable = false + case CodeAgentTransportLost: + // The ACP pipe dropped mid-turn: a recoverable transport failure, not a + // model-availability problem. Retrying the same provider is expected to + // succeed (the resume token belongs to the current provider), so + // falling back to another provider can't fix it and isn't offered. + e.AutoRetryable = true + e.FallbackAllowed = false + e.UserAction = false } if isPostStartPhase(e.Phase) && e.Code == CodeAgentRuntime { e.FallbackAllowed = false diff --git a/apps/backend/internal/agent/runtime/routingerr/runtime_rules.go b/apps/backend/internal/agent/runtime/routingerr/runtime_rules.go index 9655d96497..d11a1cc8d4 100644 --- a/apps/backend/internal/agent/runtime/routingerr/runtime_rules.go +++ b/apps/backend/internal/agent/runtime/routingerr/runtime_rules.go @@ -59,12 +59,52 @@ var runtimeEnvironmentRules = []runtimeRule{ } }, }, + { + // The ACP transport pipe died mid-turn: the upstream provider service + // dropped the connection before a response arrived. This is + // provider-agnostic wire-level transport death, not a model or + // availability problem — recoverable by retrying the same provider. + // Appended last so the more specific resume-corrupted and + // 529-overloaded rules above still win when signatures co-occur. + id: transportLostRuleID, + pattern: transportLostRe, + build: func(text string) *Error { + // A context cancellation or deadline can be reported alongside the + // same "connection closed" wording the transport-lost signature + // matches on (e.g. "context canceled: connection closed"). Those + // envelopes must keep falling through to manual recovery, so + // reject them here rather than trust the substring match alone. + if cancellationOrDeadlineRe.MatchString(text) { + return nil + } + return &Error{ + Code: CodeAgentTransportLost, + Confidence: ConfHigh, + } + }, + }, } +// cancellationOrDeadlineRe matches context-cancellation and deadline +// signatures that can co-occur with the transport-lost wording in the same +// error string. Kept separate from transportLostRe so the two can be +// combined without the transport-lost pattern itself growing lookahead +// complexity. +var cancellationOrDeadlineRe = regexp.MustCompile(`(?i)\bcontext (?:canceled|deadline exceeded)\b|\bcancel escalated\b`) + const resumeCorruptedRuleID = "anthropic.thinking_blocks.immutable.v1" const overloadedRuleID = "anthropic.overloaded.529.v1" +const transportLostRuleID = "acp.transport_lost.v1" + +// transportLostRe matches the narrow ACP wire-level transport-death +// signatures: the peer disconnecting before a response, or the underlying +// connection closing outright. Deliberately narrow (only these two +// substrings) so it never matches context-cancellation or shutdown-teardown +// error strings, which must keep falling through to manual recovery. +var transportLostRe = regexp.MustCompile(`(?i)peer disconnected|connection closed`) + // overloadedRe matches the transient 529 Overloaded signature: either the // numeric code adjacent to "overloaded" on a single line (in either order), or // the Anthropic "overloaded_error" type token. Kept tight ([^\n] stays within @@ -85,7 +125,7 @@ func IsTransientProviderError(message string) bool { return false } switch e.Code { - case CodeProviderOverloaded, CodeModelCapacity, CodeNetworkUnavailable, CodeProviderUnavailable: + case CodeProviderOverloaded, CodeModelCapacity, CodeNetworkUnavailable, CodeProviderUnavailable, CodeAgentTransportLost: return true default: return false diff --git a/apps/backend/internal/agent/runtime/routingerr/transport_lost_test.go b/apps/backend/internal/agent/runtime/routingerr/transport_lost_test.go new file mode 100644 index 0000000000..3d8a62ee49 --- /dev/null +++ b/apps/backend/internal/agent/runtime/routingerr/transport_lost_test.go @@ -0,0 +1,121 @@ +package routingerr + +import ( + "testing" + "time" +) + +// realPeerDisconnected is the exact ACP transport-death envelope observed in +// production: the upstream provider connection drops mid-turn on prompt-send, +// with no subprocess exit and no session/load failure. +const realPeerDisconnected = `{"code":-32603,"message":"Internal error","data":{"error":"peer disconnected before response"}}` + +func TestMatchRuntimeEnvironmentRules_TransportLost(t *testing.T) { + got, ok := matchRuntimeEnvironmentRules(realPeerDisconnected) + if !ok { + t.Fatalf("expected match, got none") + } + if got.Code != CodeAgentTransportLost { + t.Fatalf("Code = %q, want %q", got.Code, CodeAgentTransportLost) + } + if got.ClassifierRule != transportLostRuleID { + t.Fatalf("ClassifierRule = %q, want %q", got.ClassifierRule, transportLostRuleID) + } + if got.Confidence != ConfHigh { + t.Fatalf("Confidence = %q, want %q", got.Confidence, ConfHigh) + } +} + +func TestMatchRuntimeEnvironmentRules_TransportLost_ConnectionClosed(t *testing.T) { + got, ok := matchRuntimeEnvironmentRules("agent error: connection closed unexpectedly") + if !ok || got.Code != CodeAgentTransportLost { + t.Fatalf("expected agent_transport_lost match, got ok=%v err=%v", ok, got) + } +} + +func TestClassify_TransportLost_Invariants(t *testing.T) { + resetInjection() + e := Classify(Input{ + Phase: PhasePromptSend, + ProviderID: "claude-acp", + Stderr: realPeerDisconnected, + }) + if e == nil { + t.Fatal("expected non-nil Error") + } + if e.Code != CodeAgentTransportLost { + t.Fatalf("Code = %q, want %q", e.Code, CodeAgentTransportLost) + } + if !e.AutoRetryable { + t.Errorf("AutoRetryable = false, want true (recoverable transport drop)") + } + if e.FallbackAllowed { + t.Errorf("FallbackAllowed = true, want false (the resume token belongs to the current provider)") + } + if e.UserAction { + t.Errorf("UserAction = true, want false (no user action needed)") + } + if e.Phase != PhasePromptSend { + t.Errorf("Phase = %q, want %q", e.Phase, PhasePromptSend) + } +} + +func TestDecide_TransportLost_ShortRetry(t *testing.T) { + resetInjection() + e := Classify(Input{Phase: PhasePromptSend, ProviderID: "claude-acp", Stderr: realPeerDisconnected}) + if got := Decide(ContextKanban, e, time.Time{}); got != DecisionShortRetry { + t.Fatalf("Decide(ContextKanban, ...) = %q, want %q", got, DecisionShortRetry) + } + if got := Decide(ContextOffice, e, time.Time{}); got != DecisionShortRetry { + t.Fatalf("Decide(ContextOffice, ...) = %q, want %q", got, DecisionShortRetry) + } +} + +func TestIsTransientProviderError_TransportLost(t *testing.T) { + if !IsTransientProviderError(realPeerDisconnected) { + t.Errorf("IsTransientProviderError(peer disconnected) = false, want true") + } +} + +func TestIsAvailabilityCode_TransportLostExcluded(t *testing.T) { + if IsAvailabilityCode(CodeAgentTransportLost) { + t.Errorf("IsAvailabilityCode(CodeAgentTransportLost) = true, want false (not a model-availability problem)") + } +} + +func TestClassify_TransportLost_ExcludesCancellationSignatures(t *testing.T) { + cases := []string{ + "context canceled", + "context deadline exceeded", + "cancel escalated: prompt abandoned after cancel", + // Composite envelopes: a cancellation/deadline signature co-occurring + // with the transport-lost wording in the same message. Without the + // cancellationOrDeadlineRe guard these would match transportLostRe on + // "connection closed"/"peer disconnected" alone and wrongly become + // auto-retryable, turning a user cancel or timeout into a relaunch loop. + "context canceled: connection closed", + "context deadline exceeded: connection closed", + "context canceled: peer disconnected before response", + } + for _, msg := range cases { + t.Run(msg, func(t *testing.T) { + resetInjection() + e := Classify(Input{Phase: PhasePromptSend, ProviderID: "claude-acp", Stderr: msg}) + if e.Code == CodeAgentTransportLost { + t.Errorf("Classify(%q).Code = %q, want anything but %q", msg, e.Code, CodeAgentTransportLost) + } + if Decide(ContextKanban, e, time.Time{}) == DecisionShortRetry { + t.Errorf("Decide(ContextKanban, Classify(%q)) = DecisionShortRetry, want not short-retry", msg) + } + }) + } +} + +func TestClassify_TransportLost_OverloadedTakesPrecedence(t *testing.T) { + resetInjection() + msg := `API Error: 529 Overloaded. peer disconnected before response` + e := Classify(Input{Phase: PhasePromptSend, ProviderID: "claude-acp", Stderr: msg}) + if e.Code != CodeProviderOverloaded { + t.Fatalf("Code = %q, want %q (529-overloaded rule must win when both signatures co-occur)", e.Code, CodeProviderOverloaded) + } +} diff --git a/apps/backend/internal/office/scheduler/routing_lifecycle_test.go b/apps/backend/internal/office/scheduler/routing_lifecycle_test.go index 2c97f8fd16..d2ff72aa72 100644 --- a/apps/backend/internal/office/scheduler/routing_lifecycle_test.go +++ b/apps/backend/internal/office/scheduler/routing_lifecycle_test.go @@ -111,3 +111,25 @@ func TestHandlePostStartFailure_NonFallbackFailureEscalates(t *testing.T) { t.Fatal("ambiguous runtime failure must not be handled by the routing path (escalate)") } } + +func TestHandlePostStartFailure_TransportLostEscalates(t *testing.T) { + repo := newTestRepoSched(t) + ss := buildScheduler(t, repo, newFakeTaskStarter()) + run := seedRoutedRun(t, repo) + + handled, err := ss.HandlePostStartFailure( + context.Background(), run, makeAgent(), "peer disconnected before response", nil) + if err != nil { + t.Fatalf("HandlePostStartFailure: %v", err) + } + if handled { + t.Fatal("post-start transport loss must escalate instead of entering launch retry routing") + } + got, err := repo.GetRun(context.Background(), run.ID) + if err != nil { + t.Fatalf("get run: %v", err) + } + if got.RoutingBlockedStatus != nil || got.ScheduledRetryAt != nil { + t.Fatalf("post-start transport loss changed routing state: block=%v retry=%v", got.RoutingBlockedStatus, got.ScheduledRetryAt) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_transient.go b/apps/backend/internal/orchestrator/event_handlers_transient.go index 290f47c0c4..32550976f6 100644 --- a/apps/backend/internal/orchestrator/event_handlers_transient.go +++ b/apps/backend/internal/orchestrator/event_handlers_transient.go @@ -409,6 +409,8 @@ func transientFailureLabel(classified *routingerr.Error) string { return "Provider overloaded" case routingerr.CodeRateLimited: return "Rate limited" + case routingerr.CodeAgentTransportLost: + return "Agent connection lost" default: return "Provider temporarily unavailable" } @@ -428,6 +430,8 @@ func transientFailureExhaustedMessage(classified *routingerr.Error) string { condition = "The rate limit remained active" case routingerr.CodeProviderUnavailable: condition = "The provider remained unavailable" + case routingerr.CodeAgentTransportLost: + condition = "The agent connection kept dropping" } } return condition + " after several retries. Resume to try again, or start a fresh session." diff --git a/apps/backend/internal/orchestrator/event_handlers_transient_transport_lost_test.go b/apps/backend/internal/orchestrator/event_handlers_transient_transport_lost_test.go new file mode 100644 index 0000000000..993104a5e3 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_transient_transport_lost_test.go @@ -0,0 +1,56 @@ +package orchestrator + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/agent/runtime/routingerr" + "github.com/kandev/kandev/internal/orchestrator/watcher" +) + +// realPeerDisconnected is the exact ACP transport-death envelope observed in +// production: the upstream provider connection drops mid-turn on prompt-send. +const realPeerDisconnected = `{"code":-32603,"message":"Internal error","data":{"error":"peer disconnected before response"}}` + +func TestHandleTransientFailure_TransportLostSchedulesRetry(t *testing.T) { + svc, mc := newTransientTestService(t) + t.Cleanup(svc.cancelAllTransientRetries) + + took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{ + TaskID: "t1", + SessionID: "s1", + ErrorMessage: realPeerDisconnected, + }) + if !took { + t.Fatal("handleTransientFailure = false, want true (should own the transport-lost failure)") + } + + if _, ok := svc.transientRetries.Load("s1"); !ok { + t.Fatal("expected a transient retry entry for s1") + } + + if len(mc.sessionMessages) != 1 { + t.Fatalf("expected 1 status message, got %d", len(mc.sessionMessages)) + } + msg := mc.sessionMessages[0] + if msg.metadata["failure_code"] != "agent_transport_lost" { + t.Errorf("failure_code = %v, want agent_transport_lost", msg.metadata["failure_code"]) + } + // Must NOT be the red recovery banner. + if msg.metadata["recovery_actions"] == true { + t.Errorf("transient retry must not set recovery_actions=true (that is the red banner)") + } +} + +func TestTransientFailureLabelAndExhaustedMessage_TransportLost(t *testing.T) { + classified := &routingerr.Error{Code: routingerr.CodeAgentTransportLost} + + if got, want := transientFailureLabel(classified), "Agent connection lost"; got != want { + t.Errorf("transientFailureLabel(CodeAgentTransportLost) = %q, want %q", got, want) + } + + want := "The agent connection kept dropping after several retries. Resume to try again, or start a fresh session." + if got := transientFailureExhaustedMessage(classified); got != want { + t.Errorf("transientFailureExhaustedMessage(CodeAgentTransportLost) = %q, want %q", got, want) + } +} diff --git a/apps/web/components/task/chat/messages/action-message.test.tsx b/apps/web/components/task/chat/messages/action-message.test.tsx index f199a4c704..c10c33b8bb 100644 --- a/apps/web/components/task/chat/messages/action-message.test.tsx +++ b/apps/web/components/task/chat/messages/action-message.test.tsx @@ -247,6 +247,22 @@ describe("ActionMessage — transient retry (warning variant)", () => { }); }); +describe("ActionMessage — agent transport lost", () => { + it("renders the agent-transport-lost reason for a dropped ACP connection", () => { + renderAction( + retryMessage({ + content: "Agent connection lost", + metadata: { + ...transientRetryMetadata(1, 5), + failure_code: "agent_transport_lost", + }, + }), + "WAITING_FOR_INPUT", + ); + expect(screen.getByText(/Agent connection lost/i)).toBeTruthy(); + }); +}); + describe("ActionMessage — retry schedule updates", () => { it("resets the fallback countdown when a later retry schedule arrives", () => { vi.useFakeTimers(); diff --git a/apps/web/components/task/chat/messages/action-message.tsx b/apps/web/components/task/chat/messages/action-message.tsx index c2697e4470..c028b3c7f1 100644 --- a/apps/web/components/task/chat/messages/action-message.tsx +++ b/apps/web/components/task/chat/messages/action-message.tsx @@ -198,6 +198,8 @@ function transientRetryReasonKey(failureCode?: string) { return "chat:transientRetryReasonProviderUnavailable"; case "rate_limited": return "chat:transientRetryReasonRateLimited"; + case "agent_transport_lost": + return "chat:transientRetryReasonAgentTransportLost"; default: return "chat:transientRetryReasonGeneric"; } diff --git a/apps/web/e2e/tests/session/session-recovery.spec.ts b/apps/web/e2e/tests/session/session-recovery.spec.ts index 0fd6eeddaa..6cc9beba39 100644 --- a/apps/web/e2e/tests/session/session-recovery.spec.ts +++ b/apps/web/e2e/tests/session/session-recovery.spec.ts @@ -96,6 +96,19 @@ async function createACPProfileWithFailOnResume(apiClient: ApiClient, name: stri }); } +// Worst-case wait for the manual-recovery banner after `/crash`. The mock +// agent's crash is a real subprocess exit whose ACP-level error carries the +// same "peer disconnected before response" text the routingerr classifier +// now treats as a retryable transport-lost signature (see +// docs/decisions/2026-08-08-provider-neutral-agent-error-recovery.md) — every +// genuine production occurrence of that text also means the local subprocess +// died, so the classifier deliberately does not special-case a crash out of +// the retry ladder. `/crash` always re-crashes on relaunch, so recovery +// exhausts the full Kanban backoff (5+10+20+40+60s = 135s) before the manual +// recovery banner replaces the retry-countdown card. Give it headroom above +// that worst case rather than the default 30s. +const CRASH_RECOVERY_TIMEOUT = 170_000; + test.describe("Session recovery", () => { test.describe.configure({ retries: 1 }); @@ -135,6 +148,8 @@ test.describe("Session recovery", () => { }); test("agent crash — start fresh session recovers", async ({ testPage, apiClient, seedData }) => { + test.setTimeout(220_000); + const session = await seedTaskWithSession( testPage, apiClient, @@ -146,7 +161,9 @@ test.describe("Session recovery", () => { await session.sendMessage("/crash"); // Recovery buttons should appear - await expect(session.recoveryFreshButton()).toBeVisible({ timeout: 30_000 }); + await expect(session.recoveryFreshButton()).toBeVisible({ + timeout: CRASH_RECOVERY_TIMEOUT, + }); await expect(session.recoveryResumeButton()).toBeVisible(); // Click "Start fresh session" @@ -172,6 +189,8 @@ test.describe("Session recovery", () => { }); test("agent crash — resume session recovers", async ({ testPage, apiClient, seedData }) => { + test.setTimeout(220_000); + const session = await seedTaskWithSession( testPage, apiClient, @@ -183,7 +202,9 @@ test.describe("Session recovery", () => { await session.sendMessage("/crash"); // Recovery buttons should appear - await expect(session.recoveryResumeButton()).toBeVisible({ timeout: 30_000 }); + await expect(session.recoveryResumeButton()).toBeVisible({ + timeout: CRASH_RECOVERY_TIMEOUT, + }); // Click "Resume session" await session.recoveryResumeButton().click(); @@ -212,7 +233,7 @@ test.describe("Session recovery", () => { apiClient, seedData, }) => { - test.setTimeout(120_000); + test.setTimeout(220_000); // Unique suffix so a swallowed cleanup from a prior run doesn't collide on name. const profile = await createACPProfileWithFailOnResume( @@ -231,7 +252,9 @@ test.describe("Session recovery", () => { // Crash the agent so the recovery message renders with action buttons. await session.sendMessage("/crash"); - await expect(session.recoveryResumeButton()).toBeVisible({ timeout: 30_000 }); + await expect(session.recoveryResumeButton()).toBeVisible({ + timeout: CRASH_RECOVERY_TIMEOUT, + }); // Click "Resume session". The backend may resolve this through several // paths depending on a race (handleAgentStartFailed vs the AgentFailed diff --git a/apps/web/e2e/tests/session/transient-retry-transport-lost.spec.ts b/apps/web/e2e/tests/session/transient-retry-transport-lost.spec.ts new file mode 100644 index 0000000000..9b1aac7a90 --- /dev/null +++ b/apps/web/e2e/tests/session/transient-retry-transport-lost.spec.ts @@ -0,0 +1,79 @@ +import { test, expect } from "../../fixtures/test-base"; +import { seedIdleSession } from "../../helpers/session"; +import { SessionPage } from "../../pages/session-page"; + +test.describe("transient provider error (ACP peer disconnected) retry", () => { + test("shows the yellow retrying card labelled 'Agent connection lost', not the red error banner", async ({ + testPage, + apiClient, + seedData, + }) => { + const session = await seedIdleSession( + testPage, + apiClient, + seedData, + "Transport Lost Retry Test", + ); + + // /transport-lost:9 keeps failing so the retry loop stays visible until cancel. + await session.sendMessage("/transport-lost:9"); + + // The calm yellow "retrying" card + Cancel button must appear, labelled + // for the ACP transport-death signature (not the generic provider copy)... + await expect(session.transientRetryCard()).toBeVisible({ timeout: 30_000 }); + await expect(session.transientRetryCard().getByText("Agent connection lost")).toBeVisible(); + await expect(session.recoveryCancelRetryButton()).toBeVisible(); + + // ...and the red recovery banner must NOT be shown yet (retries in flight). + await expect(session.recoveryResumeButton()).toBeHidden(); + }); + + test("Cancel stops the retry loop and surfaces the recovery banner", async ({ + testPage, + apiClient, + seedData, + }) => { + const session = await seedIdleSession( + testPage, + apiClient, + seedData, + "Transport Lost Cancel Test", + ); + + await session.sendMessage("/transport-lost:9"); + await expect(session.recoveryCancelRetryButton()).toBeVisible({ timeout: 30_000 }); + + await session.recoveryCancelRetryButton().click(); + + // Cancelling falls through to the red Resume / Start-fresh recovery banner. + await expect(session.recoveryResumeButton()).toBeVisible({ timeout: 30_000 }); + await expect(session.recoveryFreshButton()).toBeVisible(); + }); + + test("a peer-disconnected error on the very first turn retries (launch prompt is cached)", async ({ + testPage, + apiClient, + seedData, + }) => { + // Start the task with the failing prompt as the INITIAL turn. Initial + // launches go through LaunchPreparedSession, not PromptTask, so the prompt + // must be cached there too or the retry would park behind a stuck card. + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Initial Transport Lost Test", + seedData.agentProfileId, + { + description: "/transport-lost:9", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + + await expect(session.chat.getByText(/attempt 1 of 5/i)).toBeVisible({ timeout: 30_000 }); + await expect(session.chat.getByText(/attempt 2 of 5/i)).toBeVisible({ timeout: 30_000 }); + }); +}); diff --git a/apps/web/lib/state/slices/office/types.ts b/apps/web/lib/state/slices/office/types.ts index f1f4a0be22..1aea297db0 100644 --- a/apps/web/lib/state/slices/office/types.ts +++ b/apps/web/lib/state/slices/office/types.ts @@ -495,7 +495,8 @@ export type RoutingErrorCode = | "agent_runtime_error" | "task_error" | "repo_error" - | "permission_denied_by_user"; + | "permission_denied_by_user" + | "agent_transport_lost"; export type TierMap = { frontier?: string; diff --git a/apps/web/src/locales/en/chat.json b/apps/web/src/locales/en/chat.json index 11a4fad6fc..c5380898b1 100644 --- a/apps/web/src/locales/en/chat.json +++ b/apps/web/src/locales/en/chat.json @@ -70,6 +70,7 @@ "transientRetryReasonProviderOverloaded": "Provider overloaded", "transientRetryReasonProviderUnavailable": "Provider temporarily unavailable", "transientRetryReasonRateLimited": "Rate limited", + "transientRetryReasonAgentTransportLost": "Agent connection lost", "transientRetryReasonGeneric": "Provider error", "transientRetryIn": "Retrying in {{countdown}}", "transientRetryNow": "Retrying now…", diff --git a/apps/web/src/locales/pseudo/chat.json b/apps/web/src/locales/pseudo/chat.json index 394dc4ec9c..9a7fa85981 100644 --- a/apps/web/src/locales/pseudo/chat.json +++ b/apps/web/src/locales/pseudo/chat.json @@ -70,6 +70,7 @@ "transientRetryReasonProviderOverloaded": "Ƥŕōvĩďēŕ ōvēŕĺōàďēď", "transientRetryReasonProviderUnavailable": "Ƥŕōvĩďēŕ ţēḿƥōŕàŕĩĺŷ ũńàvàĩĺàƀĺē", "transientRetryReasonRateLimited": "Ŕàţē ĺĩḿĩţēď", + "transientRetryReasonAgentTransportLost": "Àĝēńţ ćōńńēćţĩōń ĺōśţ", "transientRetryReasonGeneric": "Ƥŕōvĩďēŕ ēŕŕōŕ", "transientRetryIn": "Ŕēţŕŷĩńĝ ĩń {{countdown}}", "transientRetryNow": "Ŕēţŕŷĩńĝ ńōŵ…", diff --git a/apps/web/src/locales/pt-pt/chat.json b/apps/web/src/locales/pt-pt/chat.json index 5619062f14..ac713a1319 100644 --- a/apps/web/src/locales/pt-pt/chat.json +++ b/apps/web/src/locales/pt-pt/chat.json @@ -129,6 +129,7 @@ "transientRetryReasonProviderOverloaded": "Fornecedor sobrecarregado", "transientRetryReasonProviderUnavailable": "Fornecedor temporariamente indisponível", "transientRetryReasonRateLimited": "Limite de pedidos atingido", + "transientRetryReasonAgentTransportLost": "Ligação ao agente perdida", "transientRetryReasonGeneric": "Erro do fornecedor", "transientRetryIn": "A repetir em {{countdown}}", "transientRetryNow": "A repetir agora…", diff --git a/apps/web/src/locales/zh-cn/chat.json b/apps/web/src/locales/zh-cn/chat.json index 95811c1a96..702a636ec4 100644 --- a/apps/web/src/locales/zh-cn/chat.json +++ b/apps/web/src/locales/zh-cn/chat.json @@ -129,6 +129,7 @@ "transientRetryReasonProviderOverloaded": "服务商过载", "transientRetryReasonProviderUnavailable": "服务商暂时不可用", "transientRetryReasonRateLimited": "请求受限", + "transientRetryReasonAgentTransportLost": "代理连接已断开", "transientRetryReasonGeneric": "服务商错误", "transientRetryIn": "将在 {{countdown}} 后重试", "transientRetryNow": "正在重试…", diff --git a/apps/web/src/locales/zh-hk/chat.json b/apps/web/src/locales/zh-hk/chat.json index 1ede32f878..db08706c5a 100644 --- a/apps/web/src/locales/zh-hk/chat.json +++ b/apps/web/src/locales/zh-hk/chat.json @@ -128,6 +128,7 @@ "transientRetryReasonProviderOverloaded": "服務商過載", "transientRetryReasonProviderUnavailable": "服務商暫時不可用", "transientRetryReasonRateLimited": "請求受限", + "transientRetryReasonAgentTransportLost": "代理連線已中斷", "transientRetryReasonGeneric": "服務商錯誤", "transientRetryIn": "將在 {{countdown}} 後重試", "transientRetryNow": "正在重試…", diff --git a/apps/web/src/locales/zh-tw/chat.json b/apps/web/src/locales/zh-tw/chat.json index 68af5b1013..5537fc4eae 100644 --- a/apps/web/src/locales/zh-tw/chat.json +++ b/apps/web/src/locales/zh-tw/chat.json @@ -128,6 +128,7 @@ "transientRetryReasonProviderOverloaded": "服務商過載", "transientRetryReasonProviderUnavailable": "服務商暫時不可用", "transientRetryReasonRateLimited": "請求受限", + "transientRetryReasonAgentTransportLost": "代理連線已中斷", "transientRetryReasonGeneric": "服務商錯誤", "transientRetryIn": "將在 {{countdown}} 後重試", "transientRetryNow": "正在重試…", diff --git a/docs/decisions/2026-08-08-provider-neutral-agent-error-recovery.md b/docs/decisions/2026-08-08-provider-neutral-agent-error-recovery.md index 57526a3d10..f9793ce66e 100644 --- a/docs/decisions/2026-08-08-provider-neutral-agent-error-recovery.md +++ b/docs/decisions/2026-08-08-provider-neutral-agent-error-recovery.md @@ -70,6 +70,7 @@ The initial policies are: | Semantic cause | Kanban interactive policy | Office unattended policy | | --- | --- | --- | | `network_unavailable`, `provider_overloaded`, `provider_unavailable`, `model_capacity` | Short same-profile retry when high-confidence and replay-safe | Retry the same route for a few seconds; only degrade and fall through after the short budget is exhausted | +| `agent_transport_lost` | Short same-profile retry when high-confidence (the ACP wire-level transport-death signature) | Launch path: short same-profile retry (3 attempts), then routing gives up on the candidate without degrading the provider, since the resume token is provider-specific and fallback is not allowed. Post-start path: the routing handler declines to act (fallback disallowed short-circuits before the retry check), so a mid-turn transport loss is not retried or parked by Office routing | | `rate_limited` | Retry only a confirmed short throttle; honor a bounded retry hint | Retry the same route when a validated wait is at most 60 seconds; otherwise treat it as a long-horizon route block and fall through | | `quota_limited` | Show reset/remediation information; no automatic prompt replay | Fall through to the next configured provider; if all routes are exhausted, park until a validated reset or the Office quota backoff | | `auth_required`, `missing_credentials`, `subscription_required`, `model_unavailable`, `provider_not_configured` | User action; no automatic retry or route change | Mark the failed route user-action-required and try another configured route; if none works, block and surface inbox/settings remediation without a timed retry for that route | diff --git a/docs/specs/platform/provider-error-recovery.md b/docs/specs/platform/provider-error-recovery.md index fc4230bb7e..fe80df9367 100644 --- a/docs/specs/platform/provider-error-recovery.md +++ b/docs/specs/platform/provider-error-recovery.md @@ -48,8 +48,8 @@ failure consistently without sharing the same retry policy. ### Kanban interactive recovery - `network_unavailable`, `provider_overloaded`, `provider_unavailable`, - `model_capacity`, and confirmed short `rate_limited` failures are eligible for - automatic retry. + `model_capacity`, `agent_transport_lost`, and confirmed short `rate_limited` + failures are eligible for automatic retry. - Before replaying, Kandev verifies that the prompt has produced no assistant content or tool activity, unless the adapter provides a resumable retry guarantee. A transient failure after potentially effectful progress remains a @@ -78,9 +78,13 @@ failure consistently without sharing the same retry policy. - Office consumes the shared code through its own routing policy. It never uses Kanban's short prompt timer or retry-attempt budget. -- The short same-route phase applies to every Office run, even when provider - fallback is disabled. Fallback after exhaustion still requires an enabled, - explicitly configured alternative. +- The short same-route phase applies to supported Office retry codes, even when + provider fallback is disabled. Fallback after exhaustion still requires an + enabled, explicitly configured alternative. +- `agent_transport_lost` is launch-only in Office. The launch path retries the + same execution profile three times, then stops routing that candidate without + degrading the provider or selecting a fallback. The post-start path declines + to handle this code, so the existing failure path handles the run. - Network interruption, temporary provider unavailability/overload, model capacity, and a validated rate-limit wait of at most 60 seconds first retry the same execution profile after nominal 5, 10, and 20 second delays with jitter.