feat: retry transient provider 529 errors with visible backoff (#1173)
This commit is contained in:
Carlos Florêncio
@@ -26,6 +26,12 @@ When a real-agent message renders poorly (markdown table, tool card, diff panel,
|
||||
|
||||
This avoids burning real-agent tokens, gives you a deterministic payload to iterate against, and the scenario stays around as a permanent regression hook — anyone can re-trigger it after the fix.
|
||||
|
||||
## Special case: emitting a real prompt-time ACP *error*
|
||||
|
||||
Scenarios in `scenarioRegistry` can only emit `SessionUpdate` notifications — they cannot make the prompt itself fail. When you need a real JSON-RPC error response (the kind the backend turns into `agent.failed` with a populated `data.ErrorMessage`), intercept the command in the `Prompt` method (`main.go`) and return a non-nil `error` — typically `&acp.RequestError{Code, Message, Data}`, whose `Error()` serializes the exact `{"code":...,"message":...,"data":...}` envelope a real agent sends.
|
||||
|
||||
`/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.
|
||||
|
||||
## Emitter helpers worth knowing
|
||||
|
||||
`emitter.go` wraps the ACP `sessionUpdater` with helpers that shield scenarios from SDK plumbing:
|
||||
|
||||
@@ -1,16 +1,107 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
acp "github.com/coder/acp-go-sdk"
|
||||
)
|
||||
|
||||
// overloaded529Message is the exact transient 529 message the real Anthropic
|
||||
// API returns and the claude-agent-acp adapter surfaces over ACP. The
|
||||
// orchestrator's routingerr classifier matches the "529 ... Overloaded"
|
||||
// signature in this string and routes it to the retry-with-backoff path.
|
||||
const overloaded529Message = "Internal error: API Error: 529 Overloaded. This is a server-side issue, " +
|
||||
"usually temporary — try again in a moment. If it persists, check https://status.claude.com."
|
||||
|
||||
// overloadedCmdRe matches `/overloaded` or `/e2e:overloaded`, optionally
|
||||
// followed by `:N` — the number of consecutive prompts to fail with a 529
|
||||
// before recovering (default 1, so the turn self-heals on the first retry).
|
||||
// Use a large N (e.g. `/overloaded:9`) to exhaust the retry budget and fall
|
||||
// through to the red recovery banner.
|
||||
var overloadedCmdRe = regexp.MustCompile(`(?i)^/(?:e2e:)?overloaded(?::(\d+))?$`)
|
||||
|
||||
// parseOverloadedCmd reports whether the prompt is the /overloaded command and,
|
||||
// if so, how many consecutive prompts it should fail before recovering
|
||||
// (default 1). Returns ok=false for any other prompt.
|
||||
func parseOverloadedCmd(prompt string) (failTimes int, ok bool) {
|
||||
cmd := stripKandevSystem(strings.TrimSpace(prompt))
|
||||
m := overloadedCmdRe.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
|
||||
}
|
||||
|
||||
// handleOverloaded implements the /overloaded scenario: it returns a real
|
||||
// prompt-time ACP error carrying the production 529 string for the first N
|
||||
// prompts of a session, then recovers with a normal text response. Returns
|
||||
// handled=false when the prompt is not the overloaded command.
|
||||
func (a *mockAgent) handleOverloaded(ctx context.Context, sid acp.SessionId, prompt string) (acp.PromptResponse, error, bool) {
|
||||
failTimes, ok := parseOverloadedCmd(prompt)
|
||||
if !ok {
|
||||
return acp.PromptResponse{}, nil, false
|
||||
}
|
||||
|
||||
// The orchestrator's backoff retry tears the agent process down and
|
||||
// relaunches it via --resume, so an in-memory counter would reset every
|
||||
// attempt and never recover. Persist the attempt count in a file keyed by
|
||||
// the (resume-stable) session id so fail-then-recover survives relaunch.
|
||||
attempt := nextOverloadedAttempt(sid)
|
||||
|
||||
if attempt <= failTimes {
|
||||
_, _ = fmt.Fprintf(logOutput, "mock-agent[%d]: emitting 529 Overloaded for session %s (failure %d/%d)\n",
|
||||
os.Getpid(), sid, attempt, failTimes)
|
||||
return acp.PromptResponse{}, &acp.RequestError{
|
||||
Code: -32603,
|
||||
Message: overloaded529Message,
|
||||
Data: map[string]any{"errorKind": "server_error"},
|
||||
}, true
|
||||
}
|
||||
|
||||
// Recovered: clear the counter and emit a normal success response so the
|
||||
// turn completes (and the orchestrator clears the retry budget).
|
||||
_ = os.Remove(overloadedCounterPath(sid))
|
||||
e := &emitter{ctx: ctx, conn: a.conn, sid: sid}
|
||||
e.text(fmt.Sprintf("Provider recovered after %d transient failure(s) — here is your response.", failTimes))
|
||||
return acp.PromptResponse{StopReason: acp.StopReasonEndTurn}, nil, true
|
||||
}
|
||||
|
||||
// overloadedCounterPath returns the temp-file path tracking how many times the
|
||||
// /overloaded scenario has fired for a session. Keyed by the resume-stable ACP
|
||||
// session id so the count survives process relaunch across backoff retries.
|
||||
func overloadedCounterPath(sid acp.SessionId) string {
|
||||
safe := strings.NewReplacer("/", "_", "\\", "_", "..", "_").Replace(string(sid))
|
||||
return filepath.Join(os.TempDir(), "kandev-mock-overloaded-"+safe+".count")
|
||||
}
|
||||
|
||||
// nextOverloadedAttempt increments and returns the persisted attempt count.
|
||||
func nextOverloadedAttempt(sid acp.SessionId) int {
|
||||
path := overloadedCounterPath(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
|
||||
}
|
||||
|
||||
// delayRange returns min/max delay in milliseconds based on model name.
|
||||
func delayRange(model string) (int, int) {
|
||||
switch model {
|
||||
|
||||
@@ -168,6 +168,12 @@ func (a *mockAgent) LoadSession(_ context.Context, req acp.LoadSessionRequest) (
|
||||
func (a *mockAgent) Prompt(ctx context.Context, req acp.PromptRequest) (acp.PromptResponse, error) {
|
||||
a.emitAvailableCommandsOnce(ctx, req.SessionId)
|
||||
prompt := extractPromptText(req.Prompt)
|
||||
// The /overloaded scenario must surface a real prompt-time ACP *error*
|
||||
// (a JSON-RPC error response), which handlePrompt's emitter cannot do —
|
||||
// so intercept it here and return the error from Prompt directly.
|
||||
if resp, err, handled := a.handleOverloaded(ctx, req.SessionId, prompt); handled {
|
||||
return resp, err
|
||||
}
|
||||
e := &emitter{ctx: ctx, conn: a.conn, sid: req.SessionId}
|
||||
handlePrompt(e, prompt, a.model)
|
||||
return acp.PromptResponse{StopReason: acp.StopReasonEndTurn}, nil
|
||||
@@ -196,6 +202,7 @@ func (a *mockAgent) CloseSession(_ context.Context, req acp.CloseSessionRequest)
|
||||
delete(a.sessions, req.SessionId)
|
||||
delete(a.commandsEmitted, req.SessionId)
|
||||
a.mu.Unlock()
|
||||
_ = os.Remove(overloadedCounterPath(req.SessionId))
|
||||
return acp.CloseSessionResponse{}, nil
|
||||
}
|
||||
|
||||
@@ -254,6 +261,7 @@ func mockAvailableCommands() []acp.AvailableCommand {
|
||||
return []acp.AvailableCommand{
|
||||
{Name: "slow", Description: "Run a slow response (default 5s)", Input: hint("duration (e.g. 10s)")},
|
||||
{Name: "error", Description: "Simulate an error"},
|
||||
{Name: "overloaded", Description: "Simulate a transient 529 Overloaded error (fails once, then recovers)"},
|
||||
{Name: "thinking", Description: "Emit thinking/reasoning blocks"},
|
||||
{Name: "crash", Description: "Simulate agent crash"},
|
||||
{Name: "all", Description: "Demonstrate all message types"},
|
||||
|
||||
@@ -0,0 +1,102 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
acp "github.com/coder/acp-go-sdk"
|
||||
)
|
||||
|
||||
func newOverloadedTestAgent() *mockAgent {
|
||||
return &mockAgent{
|
||||
model: "mock-fast",
|
||||
sessions: make(map[acp.SessionId]bool),
|
||||
commandsEmitted: make(map[acp.SessionId]bool),
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseOverloadedCmd(t *testing.T) {
|
||||
cases := []struct {
|
||||
prompt string
|
||||
wantFail int
|
||||
wantOK bool
|
||||
}{
|
||||
{"/overloaded", 1, true},
|
||||
{"/OVERLOADED", 1, true},
|
||||
{"/e2e:overloaded", 1, true},
|
||||
{"/overloaded:3", 3, true},
|
||||
{"/e2e:overloaded:9", 9, true},
|
||||
{"/overloaded:0", 0, true},
|
||||
{"<kandev-system>ctx</kandev-system>/overloaded:2", 2, true},
|
||||
{"hello", 0, false},
|
||||
{"/overload", 0, false},
|
||||
{"overloaded", 0, false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
gotFail, gotOK := parseOverloadedCmd(tc.prompt)
|
||||
if gotOK != tc.wantOK || (tc.wantOK && gotFail != tc.wantFail) {
|
||||
t.Errorf("parseOverloadedCmd(%q) = (%d, %v), want (%d, %v)", tc.prompt, gotFail, gotOK, tc.wantFail, tc.wantOK)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleOverloaded_NotTheCommand(t *testing.T) {
|
||||
a := newOverloadedTestAgent()
|
||||
_, err, handled := a.handleOverloaded(context.Background(), "s1", "hello there")
|
||||
if handled {
|
||||
t.Fatal("handled = true for a non-overloaded prompt, want false")
|
||||
}
|
||||
if err != nil {
|
||||
t.Fatalf("err = %v, want nil", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleOverloaded_EmitsProduction529(t *testing.T) {
|
||||
a := newOverloadedTestAgent()
|
||||
const sid acp.SessionId = "emit529-sess"
|
||||
_ = os.Remove(overloadedCounterPath(sid))
|
||||
t.Cleanup(func() { _ = os.Remove(overloadedCounterPath(sid)) })
|
||||
_, err, handled := a.handleOverloaded(context.Background(), sid, "/overloaded")
|
||||
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.Message, "529 Overloaded") {
|
||||
t.Errorf("message = %q, want it to contain '529 Overloaded'", reqErr.Message)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleOverloaded_FailsExactlyNTimes(t *testing.T) {
|
||||
a := newOverloadedTestAgent()
|
||||
const sid acp.SessionId = "failN-sess"
|
||||
_ = os.Remove(overloadedCounterPath(sid))
|
||||
t.Cleanup(func() { _ = os.Remove(overloadedCounterPath(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.handleOverloaded(context.Background(), sid, "/overloaded:2")
|
||||
if !handled {
|
||||
t.Fatalf("attempt %d: handled = false, want true", i)
|
||||
}
|
||||
if err == nil {
|
||||
t.Fatalf("attempt %d: err = nil, want a 529 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.
|
||||
if got := nextOverloadedAttempt(sid) - 1; got != 2 {
|
||||
t.Errorf("persisted attempt count = %d, want 2", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package routingerr
|
||||
|
||||
import "testing"
|
||||
|
||||
// realOverloaded529 is the exact transient provider error surfaced by the
|
||||
// claude-agent-acp adapter when the Anthropic API returns 529 Overloaded
|
||||
// mid-turn. Unlike resume-corruption this is a server-side, temporary
|
||||
// condition — retrying after a short backoff is expected to succeed.
|
||||
const realOverloaded529 = `{"code":-32603,"message":"Internal error: API Error: 529 Overloaded. ` +
|
||||
`This is a server-side issue, usually temporary — try again in a moment. If it persists, ` +
|
||||
`check https://status.claude.com.","data":{"errorKind":"server_error"}}`
|
||||
|
||||
func TestMatchRuntimeEnvironmentRules_Overloaded(t *testing.T) {
|
||||
got, ok := matchRuntimeEnvironmentRules(realOverloaded529)
|
||||
if !ok {
|
||||
t.Fatalf("expected match, got none")
|
||||
}
|
||||
if got.Code != CodeProviderOverloaded {
|
||||
t.Fatalf("Code = %q, want %q", got.Code, CodeProviderOverloaded)
|
||||
}
|
||||
if got.ClassifierRule != overloadedRuleID {
|
||||
t.Fatalf("ClassifierRule = %q, want %q", got.ClassifierRule, overloadedRuleID)
|
||||
}
|
||||
if got.Confidence != ConfHigh {
|
||||
t.Fatalf("Confidence = %q, want %q", got.Confidence, ConfHigh)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMatchRuntimeEnvironmentRules_OverloadedErrorToken(t *testing.T) {
|
||||
// The Anthropic error-type token alone (no numeric code) must also fire.
|
||||
msg := `{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}`
|
||||
got, ok := matchRuntimeEnvironmentRules(msg)
|
||||
if !ok || got.Code != CodeProviderOverloaded {
|
||||
t.Fatalf("expected provider_overloaded match, got ok=%v err=%v", ok, got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClassify_Overloaded_Invariants(t *testing.T) {
|
||||
resetInjection()
|
||||
e := Classify(Input{
|
||||
Phase: PhasePromptSend,
|
||||
ProviderID: "claude-acp",
|
||||
Stderr: realOverloaded529,
|
||||
})
|
||||
if e == nil {
|
||||
t.Fatal("expected non-nil Error")
|
||||
}
|
||||
if e.Code != CodeProviderOverloaded {
|
||||
t.Fatalf("Code = %q, want %q", e.Code, CodeProviderOverloaded)
|
||||
}
|
||||
if !e.AutoRetryable {
|
||||
t.Errorf("AutoRetryable = false, want true (transient server-side overload)")
|
||||
}
|
||||
if !e.FallbackAllowed {
|
||||
t.Errorf("FallbackAllowed = false, want true")
|
||||
}
|
||||
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 TestClassify_HTTPStatus529(t *testing.T) {
|
||||
// Callers that surface the status code separately from the body text (e.g.
|
||||
// a structured HTTP error) must also classify 529 as transient/overloaded.
|
||||
resetInjection()
|
||||
e := Classify(Input{Phase: PhasePromptSend, ProviderID: "claude-acp", HTTPStatus: statusOverloaded})
|
||||
if e == nil {
|
||||
t.Fatal("expected non-nil Error")
|
||||
}
|
||||
if e.Code != CodeProviderOverloaded {
|
||||
t.Fatalf("Code = %q, want %q", e.Code, CodeProviderOverloaded)
|
||||
}
|
||||
if !e.AutoRetryable {
|
||||
t.Errorf("AutoRetryable = false, want true")
|
||||
}
|
||||
if e.ClassifierRule != "http.529" {
|
||||
t.Errorf("ClassifierRule = %q, want http.529", e.ClassifierRule)
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsTransientProviderError(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
msg string
|
||||
want bool
|
||||
}{
|
||||
{"real 529 envelope", realOverloaded529, true},
|
||||
{"overloaded_error token", `"type":"overloaded_error"`, true},
|
||||
{"529 then overloaded on one line", "API Error: 529 Overloaded.", true},
|
||||
{"prefixed overloaded error token", `"type":"not_overloaded_error"`, false},
|
||||
{"unrelated 401", "API Error: 401 authentication_error", false},
|
||||
{"plain overloaded word, no code", "the disk is overloaded with files", false},
|
||||
{"bare 529 no overloaded", "got status 529 from upstream", false},
|
||||
{"empty", "", false},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
if got := IsTransientProviderError(tc.msg); got != tc.want {
|
||||
t.Errorf("IsTransientProviderError(%q) = %v, want %v", tc.msg, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -22,6 +22,7 @@ const (
|
||||
CodeQuotaLimited Code = "quota_limited"
|
||||
CodeRateLimited Code = "rate_limited"
|
||||
CodeProviderUnavailable Code = "provider_unavailable"
|
||||
CodeProviderOverloaded Code = "provider_overloaded"
|
||||
CodeModelUnavailable Code = "model_unavailable"
|
||||
CodeProviderNotConfigured Code = "provider_not_configured"
|
||||
CodeUnknownProvider Code = "unknown_provider_error"
|
||||
@@ -96,6 +97,10 @@ type Input struct {
|
||||
|
||||
const exitCodeBinaryMissing = 127
|
||||
|
||||
// statusOverloaded is the non-standard HTTP 529 ("Overloaded") that Anthropic
|
||||
// returns when temporarily overloaded. Not in net/http, so defined here.
|
||||
const statusOverloaded = 529
|
||||
|
||||
// Classify normalizes a failure into a routing-aware Error. See package doc.
|
||||
func Classify(in Input) *Error {
|
||||
excerpt := Sanitize(in.Stderr + "\n" + in.Stdout)
|
||||
@@ -176,6 +181,8 @@ func httpStatusToCode(status int) Code {
|
||||
return CodeRateLimited
|
||||
case http.StatusServiceUnavailable:
|
||||
return CodeProviderUnavailable
|
||||
case statusOverloaded:
|
||||
return CodeProviderOverloaded
|
||||
}
|
||||
return ""
|
||||
}
|
||||
@@ -218,6 +225,12 @@ func applyInvariants(e *Error) *Error {
|
||||
case CodeProviderUnavailable, CodeUnknownProvider:
|
||||
e.AutoRetryable = true
|
||||
e.FallbackAllowed = true
|
||||
case CodeProviderOverloaded:
|
||||
// 529 Overloaded is a transient, server-side condition. Retrying the
|
||||
// same provider after a short backoff is the right move; falling back
|
||||
// to another provider is also fine if backoff keeps failing.
|
||||
e.AutoRetryable = true
|
||||
e.FallbackAllowed = true
|
||||
case CodeNpxCacheCorrupted:
|
||||
e.AutoRetryable = true
|
||||
e.FallbackAllowed = true
|
||||
|
||||
@@ -44,10 +44,42 @@ var runtimeEnvironmentRules = []runtimeRule{
|
||||
}
|
||||
},
|
||||
},
|
||||
{
|
||||
// Anthropic 529 Overloaded surfaced over ACP as a prompt-time error
|
||||
// event: a transient, server-side condition the orchestrator should
|
||||
// retry with backoff rather than tear down. Provider-agnostic because
|
||||
// the "529 ... overloaded" / "overloaded_error" signature is unique
|
||||
// enough not to false-positive on unrelated logs.
|
||||
id: overloadedRuleID,
|
||||
pattern: overloadedRe,
|
||||
build: func(string) *Error {
|
||||
return &Error{
|
||||
Code: CodeProviderOverloaded,
|
||||
Confidence: ConfHigh,
|
||||
}
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
const resumeCorruptedRuleID = "anthropic.thinking_blocks.immutable.v1"
|
||||
|
||||
const overloadedRuleID = "anthropic.overloaded.529.v1"
|
||||
|
||||
// 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
|
||||
// one line) so a stray "529" and "overloaded" in unrelated multi-line logs
|
||||
// can't bridge into a false positive.
|
||||
var overloadedRe = regexp.MustCompile(`(?i)\b529\b[^\n]*overloaded|overloaded[^\n]*\b529\b|\boverloaded_error\b`)
|
||||
|
||||
// IsTransientProviderError reports whether the error message carries the
|
||||
// transient provider-overload (529 Overloaded) signature. Exposed for callers
|
||||
// outside the classify path (e.g. the orchestrator's retry-with-backoff) that
|
||||
// need to branch on transience without re-running full Classify.
|
||||
func IsTransientProviderError(message string) bool {
|
||||
return message != "" && overloadedRe.MatchString(message)
|
||||
}
|
||||
|
||||
// thinkingBlocksImmutableRe matches the Anthropic "thinking blocks cannot be
|
||||
// modified" 400 in either the `thinking` or `redacted_thinking` form. The
|
||||
// gaps stay bounded ([^\n] is non-greedy-friendly within a single line) so a
|
||||
|
||||
@@ -224,6 +224,10 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
|
||||
return
|
||||
}
|
||||
|
||||
// A turn completed successfully — clear any transient retry budget so a
|
||||
// later, unrelated provider overload starts its backoff fresh at attempt 1.
|
||||
s.resetTransientRetry(data.SessionID)
|
||||
|
||||
// Complete the current turn
|
||||
s.completeTurnForSession(ctx, data.SessionID)
|
||||
|
||||
@@ -408,6 +412,9 @@ func (s *Service) handleAgentCompleted(ctx context.Context, data watcher.AgentEv
|
||||
zap.String("session_id", data.SessionID),
|
||||
zap.String("agent_execution_id", data.AgentExecutionID))
|
||||
|
||||
// A successful completion clears any transient retry budget for the session.
|
||||
s.resetTransientRetry(data.SessionID)
|
||||
|
||||
// Update scheduler and remove from queue
|
||||
s.scheduler.HandleTaskCompleted(data.TaskID, true)
|
||||
s.scheduler.RemoveTask(data.TaskID)
|
||||
@@ -501,10 +508,21 @@ func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEvent
|
||||
zap.String("agent_execution_id", data.AgentExecutionID),
|
||||
zap.String("error_message", data.ErrorMessage))
|
||||
|
||||
// Finalize run-mode automation runs first — every other branch below
|
||||
// returns early (resume failure, session-backed recoverable failure,
|
||||
// Transient provider errors (529 Overloaded) get a paced, visible
|
||||
// retry-with-backoff before any red banner. This is the ONLY non-terminal
|
||||
// failure path, so it runs before automation finalization below — otherwise
|
||||
// a transient 529 on a run-mode automation would mark the run failed and
|
||||
// reap its ephemeral worktree out from under the in-flight retry.
|
||||
// handleTransientFailure returns false (falling through) for non-transient
|
||||
// errors, office tasks, or an exhausted budget.
|
||||
if data.SessionID != "" && s.handleTransientFailure(ctx, data) {
|
||||
return
|
||||
}
|
||||
|
||||
// Terminal from here. Finalize run-mode automation runs — every branch
|
||||
// below returns early (resume failure, session-backed recoverable failure,
|
||||
// no-session retry), and run-mode automations need their AutomationRun
|
||||
// flipped + worktree reaped on *every* failure path.
|
||||
// flipped + worktree reaped on *every* terminal failure path.
|
||||
errMsg := data.ErrorMessage
|
||||
if errMsg == "" {
|
||||
errMsg = "agent failed"
|
||||
@@ -683,6 +701,10 @@ func (s *Service) createRecoveryStatusMessage(ctx context.Context, data watcher.
|
||||
statusMsg := fmt.Sprintf("Agent encountered an error: %s", displayMsg)
|
||||
if resumeCorrupted {
|
||||
statusMsg = "This agent session can't be resumed — its saved reasoning state is corrupted. Start a fresh session to continue."
|
||||
} else if routingerr.IsTransientProviderError(data.ErrorMessage) {
|
||||
// Reached after the transient retry budget is exhausted — show friendly
|
||||
// copy instead of dumping the raw 529 JSON envelope.
|
||||
statusMsg = "The provider stayed overloaded after several retries. Resume to try again, or start a fresh session."
|
||||
}
|
||||
hasResumeToken := s.wasResumeAttempt(ctx, data.SessionID)
|
||||
meta := map[string]interface{}{
|
||||
@@ -844,6 +866,12 @@ func (s *Service) handleAgentStopped(ctx context.Context, data watcher.AgentEven
|
||||
zap.String("session_id", data.SessionID),
|
||||
zap.String("agent_execution_id", data.AgentExecutionID))
|
||||
|
||||
// NOTE: we deliberately do NOT resetTransientRetry here — the transient
|
||||
// retry tears down the failed execution via StopExecution as part of its
|
||||
// own re-drive, which surfaces as an agent.stopped event; clearing the loop
|
||||
// on that self-inflicted stop would abort the retry. The loop is freed on
|
||||
// ready/completed (success), cancel, and exhaustion instead.
|
||||
|
||||
// Complete the current turn if there is one
|
||||
s.completeTurnForSession(ctx, data.SessionID)
|
||||
|
||||
|
||||
@@ -0,0 +1,334 @@
|
||||
package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/runtime/routingerr"
|
||||
"github.com/kandev/kandev/internal/orchestrator/watcher"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
v1 "github.com/kandev/kandev/pkg/api/v1"
|
||||
)
|
||||
|
||||
// transientMaxAttempts caps how many times a transient provider error (529
|
||||
// Overloaded) is auto-retried with backoff before falling through to the
|
||||
// manual recovery banner.
|
||||
const transientMaxAttempts = 3
|
||||
|
||||
// recoverActionCancelRetry is the session.recover action that stops an
|
||||
// in-progress transient retry loop and surfaces manual recovery.
|
||||
const recoverActionCancelRetry = "cancel_retry"
|
||||
|
||||
const recoveryCancelRetryButtonTestID = "recovery-cancel-retry-button"
|
||||
|
||||
// Shared status-message metadata keys. Defined as constants because the same
|
||||
// keys are built in more than one place in this package (recovery + retry
|
||||
// status messages), which otherwise trips goconst on new code.
|
||||
const (
|
||||
metaKeyVariant = "variant"
|
||||
metaKeySessionID = "session_id"
|
||||
metaKeyTaskID = "task_id"
|
||||
)
|
||||
|
||||
// metaVariantWarning is the status-message variant that drives the frontend's
|
||||
// yellow (non-alarming) styling, as opposed to the red "error" variant.
|
||||
const metaVariantWarning = "warning"
|
||||
|
||||
// transientRetryBackoff is the per-attempt delay before re-driving a turn that
|
||||
// failed transiently. Index is attempt-1 (5s → 15s → 30s).
|
||||
var transientRetryBackoff = []time.Duration{
|
||||
5 * time.Second,
|
||||
15 * time.Second,
|
||||
30 * time.Second,
|
||||
}
|
||||
|
||||
// transientRetryDelay returns the backoff for a 1-based attempt, clamping to
|
||||
// the longest step so an over-count never panics.
|
||||
func transientRetryDelay(attempt int) time.Duration {
|
||||
if attempt < 1 {
|
||||
attempt = 1
|
||||
}
|
||||
if attempt > len(transientRetryBackoff) {
|
||||
attempt = len(transientRetryBackoff)
|
||||
}
|
||||
return transientRetryBackoff[attempt-1]
|
||||
}
|
||||
|
||||
// capturedPrompt is the minimal context needed to re-drive a failed turn.
|
||||
type capturedPrompt struct {
|
||||
text string
|
||||
model string
|
||||
planMode bool
|
||||
attachments []v1.MessageAttachment
|
||||
}
|
||||
|
||||
// transientRetryEntry tracks one session's in-progress retry loop: the current
|
||||
// attempt count and the cancel func for the armed backoff timer.
|
||||
type transientRetryEntry struct {
|
||||
attempt int
|
||||
cancel func()
|
||||
mu sync.Mutex
|
||||
claimed bool
|
||||
}
|
||||
|
||||
func (e *transientRetryEntry) claim() bool {
|
||||
e.mu.Lock()
|
||||
defer e.mu.Unlock()
|
||||
if e.claimed {
|
||||
return false
|
||||
}
|
||||
e.claimed = true
|
||||
return true
|
||||
}
|
||||
|
||||
// rememberTurnPrompt caches the raw outbound prompt so a transient retry can
|
||||
// re-drive the same turn without the original caller's context.
|
||||
func (s *Service) rememberTurnPrompt(sessionID, text, model string, planMode bool, attachments []v1.MessageAttachment) {
|
||||
if sessionID == "" {
|
||||
return
|
||||
}
|
||||
s.lastTurnPrompt.Store(sessionID, capturedPrompt{
|
||||
text: text,
|
||||
model: model,
|
||||
planMode: planMode,
|
||||
attachments: attachments,
|
||||
})
|
||||
}
|
||||
|
||||
// handleTransientFailure routes a transient provider error (529 Overloaded)
|
||||
// into a paced, visible retry-with-backoff instead of the red recovery banner.
|
||||
// Returns true when it takes ownership (caller must NOT fall through to
|
||||
// handleRecoverableFailure); false for non-transient errors, office tasks,
|
||||
// or an exhausted retry budget.
|
||||
func (s *Service) handleTransientFailure(ctx context.Context, data watcher.AgentEventData) bool {
|
||||
if data.SessionID == "" || !routingerr.IsTransientProviderError(data.ErrorMessage) {
|
||||
return false
|
||||
}
|
||||
// Genuine office tasks (those with an assignee agent profile) render their
|
||||
// own structured error UI — keep them on the existing path rather than the
|
||||
// kanban-style yellow retry card. Note: we intentionally check the task's
|
||||
// assignee, NOT isOfficeSession (session.AgentProfileID), which is also set
|
||||
// for ordinary kanban tasks started with an agent profile.
|
||||
if s.isOfficeTask(ctx, data.TaskID) {
|
||||
return false
|
||||
}
|
||||
|
||||
attempt := s.nextTransientAttempt(data.SessionID)
|
||||
if attempt > transientMaxAttempts {
|
||||
s.logger.Warn("transient retry budget exhausted; falling through to recovery banner",
|
||||
zap.String("task_id", data.TaskID),
|
||||
zap.String("session_id", data.SessionID),
|
||||
zap.Int("attempts", attempt-1))
|
||||
s.resetTransientRetry(data.SessionID)
|
||||
return false
|
||||
}
|
||||
|
||||
delay := transientRetryDelay(attempt)
|
||||
s.logger.Info("scheduling transient provider-error retry",
|
||||
zap.String("task_id", data.TaskID),
|
||||
zap.String("session_id", data.SessionID),
|
||||
zap.Int("attempt", attempt),
|
||||
zap.Int("max_attempts", transientMaxAttempts),
|
||||
zap.Duration("delay", delay))
|
||||
|
||||
// Emit the yellow status (against the failed turn) before completing it.
|
||||
s.createTransientRetryStatusMessage(ctx, data, attempt, delay)
|
||||
s.completeTurnForSession(ctx, data.SessionID)
|
||||
|
||||
// Park the session in WAITING_FOR_INPUT (a calm, banner-less state that
|
||||
// also lets the yellow retry card render — ActionMessage hides itself while
|
||||
// the session is RUNNING). Deliberately NOT FAILED and NOT task→REVIEW.
|
||||
s.updateTaskSessionState(ctx, data.TaskID, data.SessionID, models.TaskSessionStateWaitingForInput, "", false)
|
||||
|
||||
s.scheduleTransientRetry(data.TaskID, data.SessionID, data.AgentExecutionID, attempt, delay)
|
||||
return true
|
||||
}
|
||||
|
||||
// nextTransientAttempt returns the next 1-based attempt number for a session,
|
||||
// cancelling any still-armed timer from a prior attempt.
|
||||
func (s *Service) nextTransientAttempt(sessionID string) int {
|
||||
prev := 0
|
||||
if v, ok := s.transientRetries.Load(sessionID); ok {
|
||||
if entry, ok := v.(*transientRetryEntry); ok {
|
||||
prev = entry.attempt
|
||||
if entry.cancel != nil {
|
||||
entry.cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
return prev + 1
|
||||
}
|
||||
|
||||
// scheduleTransientRetry stores a fresh retry entry and arms its backoff timer.
|
||||
func (s *Service) scheduleTransientRetry(taskID, sessionID, execID string, attempt int, delay time.Duration) {
|
||||
retryCtx, cancel := context.WithCancel(context.Background())
|
||||
entry := &transientRetryEntry{attempt: attempt, cancel: cancel}
|
||||
s.transientRetries.Store(sessionID, entry)
|
||||
go s.runTransientRetry(retryCtx, taskID, sessionID, execID, entry, delay)
|
||||
}
|
||||
|
||||
// runTransientRetry waits out the backoff (or cancellation) then re-drives the
|
||||
// turn. Mirrors the clarification-watchdog goroutine ownership pattern.
|
||||
func (s *Service) runTransientRetry(retryCtx context.Context, taskID, sessionID, execID string, entry *transientRetryEntry, delay time.Duration) {
|
||||
timer := time.NewTimer(delay)
|
||||
defer timer.Stop()
|
||||
|
||||
select {
|
||||
case <-retryCtx.Done():
|
||||
return
|
||||
case <-timer.C:
|
||||
// Only fire if this entry is still the active one for the session.
|
||||
if cur, ok := s.transientRetries.Load(sessionID); !ok || cur != entry || !entry.claim() {
|
||||
return
|
||||
}
|
||||
s.retryTransientPrompt(retryCtx, taskID, sessionID, execID)
|
||||
}
|
||||
}
|
||||
|
||||
// retryTransientPrompt re-drives the failed turn after backoff. The failed
|
||||
// execution is torn down first so PromptTask's ensureSessionRunning resumes a
|
||||
// fresh agent via the resume token (re-establishing the ACP session) rather
|
||||
// than reusing the FAILED execution, which rejects prompts. The session was
|
||||
// parked in WAITING_FOR_INPUT by handleTransientFailure so PromptTask accepts
|
||||
// the re-send straight away.
|
||||
func (s *Service) retryTransientPrompt(ctx context.Context, taskID, sessionID, execID string) {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
v, ok := s.lastTurnPrompt.Load(sessionID)
|
||||
if !ok {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
// No prompt to re-drive (e.g. an uncached launch path). Don't leave the
|
||||
// retry loop parked behind a stuck yellow card — clear it and surface
|
||||
// the manual recovery banner so the user can resume or start fresh.
|
||||
s.logger.Warn("transient retry has no cached prompt; surfacing recovery banner",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID))
|
||||
s.resetTransientRetry(sessionID)
|
||||
s.handleRecoverableFailure(context.Background(), watcher.AgentEventData{
|
||||
TaskID: taskID,
|
||||
SessionID: sessionID,
|
||||
AgentExecutionID: execID,
|
||||
ErrorMessage: "Provider overloaded — automatic retry was not possible. Resume or start fresh to continue.",
|
||||
})
|
||||
return
|
||||
}
|
||||
cp, _ := v.(capturedPrompt)
|
||||
|
||||
if execID != "" {
|
||||
if err := s.executor.StopExecution(ctx, execID, "transient retry: relaunching agent", true); err != nil {
|
||||
s.logger.Debug("failed to stop failed execution before transient retry",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("execution_id", execID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
if _, err := s.PromptTask(ctx, taskID, sessionID, cp.text, cp.model, cp.planMode, cp.attachments, false); err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
s.logger.Error("transient retry prompt failed synchronously; surfacing recovery banner",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Error(err))
|
||||
s.resetTransientRetry(sessionID)
|
||||
s.handleRecoverableFailure(context.Background(), watcher.AgentEventData{
|
||||
TaskID: taskID,
|
||||
SessionID: sessionID,
|
||||
AgentExecutionID: execID,
|
||||
ErrorMessage: "Provider overloaded — automatic retry could not be started. Resume or start fresh to continue.",
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// createTransientRetryStatusMessage emits the calm yellow "retrying" status
|
||||
// (variant=warning) with a Cancel action, driving the frontend's
|
||||
// AgentWarningStatus instead of the red AgentErrorStatus.
|
||||
func (s *Service) createTransientRetryStatusMessage(ctx context.Context, data watcher.AgentEventData, attempt int, delay time.Duration) {
|
||||
if s.messageCreator == nil {
|
||||
return
|
||||
}
|
||||
secs := int(delay.Seconds())
|
||||
content := fmt.Sprintf("Provider overloaded — retrying in %ds (attempt %d/%d)", secs, attempt, transientMaxAttempts)
|
||||
cancelAction := wsRecoveryAction(data.TaskID, data.SessionID, recoverActionCancelRetry,
|
||||
"Cancel", "x", "Stop retrying and choose how to recover", recoveryCancelRetryButtonTestID)
|
||||
meta := map[string]interface{}{
|
||||
metaKeyVariant: metaVariantWarning,
|
||||
"retrying": true,
|
||||
"attempt": attempt,
|
||||
"max_attempts": transientMaxAttempts,
|
||||
"retry_in_seconds": secs,
|
||||
metaKeySessionID: data.SessionID,
|
||||
metaKeyTaskID: data.TaskID,
|
||||
"actions": []map[string]interface{}{cancelAction},
|
||||
}
|
||||
if err := s.messageCreator.CreateSessionMessage(
|
||||
ctx,
|
||||
data.TaskID,
|
||||
content,
|
||||
data.SessionID,
|
||||
string(v1.MessageTypeStatus),
|
||||
s.getActiveTurnID(data.SessionID),
|
||||
meta,
|
||||
false,
|
||||
); err != nil {
|
||||
s.logger.Warn("failed to create transient retry status message",
|
||||
zap.String("task_id", data.TaskID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// resetTransientRetry clears a session's retry entry, cancels its timer, and
|
||||
// drops the cached prompt (which may hold large/sensitive attachment data).
|
||||
// Called on a successful turn, on cancel, and on exhaustion.
|
||||
func (s *Service) resetTransientRetry(sessionID string) {
|
||||
s.lastTurnPrompt.Delete(sessionID)
|
||||
if v, ok := s.transientRetries.LoadAndDelete(sessionID); ok {
|
||||
if entry, ok := v.(*transientRetryEntry); ok && entry.cancel != nil {
|
||||
entry.cancel()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// cancelAllTransientRetries drains every armed retry timer at shutdown.
|
||||
func (s *Service) cancelAllTransientRetries() {
|
||||
s.transientRetries.Range(func(key, _ interface{}) bool {
|
||||
if keyStr, ok := key.(string); ok {
|
||||
s.resetTransientRetry(keyStr)
|
||||
}
|
||||
return true
|
||||
})
|
||||
}
|
||||
|
||||
// CancelTransientRetry stops an in-progress retry loop (user clicked Cancel)
|
||||
// and surfaces the manual recovery banner so they can Resume or Start fresh.
|
||||
// Returns true if a retry loop was active.
|
||||
func (s *Service) CancelTransientRetry(ctx context.Context, taskID, sessionID string) bool {
|
||||
_, active := s.transientRetries.Load(sessionID)
|
||||
s.resetTransientRetry(sessionID)
|
||||
if !active {
|
||||
return false
|
||||
}
|
||||
s.logger.Info("user cancelled transient retry loop",
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("session_id", sessionID))
|
||||
|
||||
execID, _ := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
|
||||
s.handleRecoverableFailure(ctx, watcher.AgentEventData{
|
||||
TaskID: taskID,
|
||||
SessionID: sessionID,
|
||||
AgentExecutionID: execID,
|
||||
ErrorMessage: "Provider overloaded — retries cancelled. Resume or start fresh to continue.",
|
||||
})
|
||||
return true
|
||||
}
|
||||
@@ -0,0 +1,309 @@
|
||||
package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/kandev/kandev/internal/orchestrator/watcher"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
)
|
||||
|
||||
// overloaded529 is the exact transient 529 Overloaded envelope the
|
||||
// claude-agent-acp adapter surfaces over ACP as a prompt-time error event.
|
||||
const overloaded529 = `{"code":-32603,"message":"Internal error: API Error: 529 Overloaded. ` +
|
||||
`This is a server-side issue, usually temporary — try again in a moment.","data":{"errorKind":"server_error"}}`
|
||||
|
||||
func newTransientTestService(t *testing.T) (*Service, *mockMessageCreator) {
|
||||
t.Helper()
|
||||
repo := setupTestRepo(t)
|
||||
seedSession(t, repo, "t1", "s1", "step1")
|
||||
agentMgr := &mockAgentManager{repoForExecutionLookup: repo}
|
||||
svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
mc := &mockMessageCreator{}
|
||||
svc.messageCreator = mc
|
||||
return svc, mc
|
||||
}
|
||||
|
||||
func TestHandleTransientFailure_SchedulesRetryAndEmitsWarning(t *testing.T) {
|
||||
svc, mc := newTransientTestService(t)
|
||||
t.Cleanup(svc.cancelAllTransientRetries)
|
||||
|
||||
took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{
|
||||
TaskID: "t1",
|
||||
SessionID: "s1",
|
||||
ErrorMessage: overloaded529,
|
||||
})
|
||||
if !took {
|
||||
t.Fatal("handleTransientFailure = false, want true (should own the transient failure)")
|
||||
}
|
||||
|
||||
// A retry entry must be armed for the session.
|
||||
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["variant"] != "warning" {
|
||||
t.Errorf("variant = %v, want warning", msg.metadata["variant"])
|
||||
}
|
||||
if msg.metadata["retrying"] != true {
|
||||
t.Errorf("retrying = %v, want true", msg.metadata["retrying"])
|
||||
}
|
||||
if msg.metadata["attempt"] != 1 {
|
||||
t.Errorf("attempt = %v, want 1", msg.metadata["attempt"])
|
||||
}
|
||||
if msg.metadata["max_attempts"] != transientMaxAttempts {
|
||||
t.Errorf("max_attempts = %v, want %d", msg.metadata["max_attempts"], transientMaxAttempts)
|
||||
}
|
||||
if msg.metadata["retry_in_seconds"] != 5 {
|
||||
t.Errorf("retry_in_seconds = %v, want 5", msg.metadata["retry_in_seconds"])
|
||||
}
|
||||
if !strings.Contains(strings.ToLower(msg.content), "retrying") {
|
||||
t.Errorf("content = %q, want a 'retrying' message", msg.content)
|
||||
}
|
||||
// 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)")
|
||||
}
|
||||
// Must carry a Cancel action.
|
||||
actions, ok := msg.metadata["actions"].([]map[string]interface{})
|
||||
if !ok || actionByTestID(actions, recoveryCancelRetryButtonTestID) == nil {
|
||||
t.Errorf("expected a cancel action with test_id %q, got %v", recoveryCancelRetryButtonTestID, msg.metadata["actions"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleTransientFailure_NonTransientReturnsFalse(t *testing.T) {
|
||||
svc, mc := newTransientTestService(t)
|
||||
t.Cleanup(svc.cancelAllTransientRetries)
|
||||
|
||||
took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{
|
||||
TaskID: "t1",
|
||||
SessionID: "s1",
|
||||
ErrorMessage: "agent crashed with exit code 1",
|
||||
})
|
||||
if took {
|
||||
t.Fatal("handleTransientFailure = true for a non-transient error, want false")
|
||||
}
|
||||
if len(mc.sessionMessages) != 0 {
|
||||
t.Errorf("expected no status message for non-transient error, got %d", len(mc.sessionMessages))
|
||||
}
|
||||
if _, ok := svc.transientRetries.Load("s1"); ok {
|
||||
t.Error("expected no retry entry for a non-transient error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleTransientFailure_ExhaustedFallsThrough(t *testing.T) {
|
||||
svc, mc := newTransientTestService(t)
|
||||
t.Cleanup(svc.cancelAllTransientRetries)
|
||||
|
||||
// Pre-seed an entry that has already used the full budget.
|
||||
svc.transientRetries.Store("s1", &transientRetryEntry{attempt: transientMaxAttempts, cancel: func() {}})
|
||||
|
||||
took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{
|
||||
TaskID: "t1",
|
||||
SessionID: "s1",
|
||||
ErrorMessage: overloaded529,
|
||||
})
|
||||
if took {
|
||||
t.Fatal("handleTransientFailure = true after budget exhausted, want false (fall through to recovery)")
|
||||
}
|
||||
if len(mc.sessionMessages) != 0 {
|
||||
t.Errorf("expected no new retry status message on exhaustion, got %d", len(mc.sessionMessages))
|
||||
}
|
||||
if _, ok := svc.transientRetries.Load("s1"); ok {
|
||||
t.Error("expected retry entry cleared on exhaustion")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancelTransientRetry_StopsLoopAndShowsRecovery(t *testing.T) {
|
||||
svc, mc := newTransientTestService(t)
|
||||
t.Cleanup(svc.cancelAllTransientRetries)
|
||||
|
||||
svc.scheduleTransientRetry("t1", "s1", "", 1, 5*time.Second)
|
||||
if _, ok := svc.transientRetries.Load("s1"); !ok {
|
||||
t.Fatal("expected an armed retry entry before cancel")
|
||||
}
|
||||
|
||||
cancelled := svc.CancelTransientRetry(context.Background(), "t1", "s1")
|
||||
if !cancelled {
|
||||
t.Fatal("CancelTransientRetry = false, want true (a loop was active)")
|
||||
}
|
||||
if _, ok := svc.transientRetries.Load("s1"); ok {
|
||||
t.Error("expected retry entry cleared after cancel")
|
||||
}
|
||||
|
||||
// Cancel surfaces the red recovery banner so the user can recover manually.
|
||||
if len(mc.sessionMessages) != 1 {
|
||||
t.Fatalf("expected 1 recovery status message after cancel, got %d", len(mc.sessionMessages))
|
||||
}
|
||||
if mc.sessionMessages[0].metadata["recovery_actions"] != true {
|
||||
t.Errorf("expected recovery_actions=true after cancel, got %v", mc.sessionMessages[0].metadata["recovery_actions"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancelTransientRetry_NoActiveLoop(t *testing.T) {
|
||||
svc, _ := newTransientTestService(t)
|
||||
if svc.CancelTransientRetry(context.Background(), "t1", "s1") {
|
||||
t.Error("CancelTransientRetry = true with no active loop, want false")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResetTransientRetry_ClearsEntry(t *testing.T) {
|
||||
svc, _ := newTransientTestService(t)
|
||||
svc.scheduleTransientRetry("t1", "s1", "", 1, 5*time.Second)
|
||||
svc.resetTransientRetry("s1")
|
||||
if _, ok := svc.transientRetries.Load("s1"); ok {
|
||||
t.Error("expected entry cleared after resetTransientRetry")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResetTransientRetry_DropsCachedPrompt(t *testing.T) {
|
||||
svc, _ := newTransientTestService(t)
|
||||
// The cached prompt can hold large/sensitive attachment data — it must be
|
||||
// released when the retry loop ends, not retained for the backend's life.
|
||||
svc.rememberTurnPrompt("s1", "hello", "", false, nil)
|
||||
if _, ok := svc.lastTurnPrompt.Load("s1"); !ok {
|
||||
t.Fatal("expected a cached prompt")
|
||||
}
|
||||
svc.resetTransientRetry("s1")
|
||||
if _, ok := svc.lastTurnPrompt.Load("s1"); ok {
|
||||
t.Error("expected cached prompt cleared after resetTransientRetry")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetryTransientPrompt_NoCachedPromptSurfacesRecovery(t *testing.T) {
|
||||
svc, mc := newTransientTestService(t)
|
||||
t.Cleanup(svc.cancelAllTransientRetries)
|
||||
|
||||
// Arm a retry entry but never cache a prompt (an uncached launch path).
|
||||
svc.scheduleTransientRetry("t1", "s1", "", 1, time.Hour)
|
||||
|
||||
// The timer firing with no cached prompt must NOT leave the loop parked —
|
||||
// it clears the entry and surfaces the manual recovery banner.
|
||||
svc.retryTransientPrompt(context.Background(), "t1", "s1", "")
|
||||
|
||||
if _, ok := svc.transientRetries.Load("s1"); ok {
|
||||
t.Error("expected retry entry cleared when no prompt is cached")
|
||||
}
|
||||
hasRecovery := false
|
||||
for _, m := range mc.sessionMessages {
|
||||
if m.metadata["recovery_actions"] == true {
|
||||
hasRecovery = true
|
||||
}
|
||||
}
|
||||
if !hasRecovery {
|
||||
t.Error("expected a recovery_actions banner after an uncached retry")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetryTransientPrompt_SynchronousPromptErrorSurfacesRecovery(t *testing.T) {
|
||||
repo := setupTestRepo(t)
|
||||
seedSession(t, repo, "t1", "s1", "step1")
|
||||
session, err := repo.GetTaskSession(context.Background(), "s1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to load session: %v", err)
|
||||
}
|
||||
session.State = models.TaskSessionStateWaitingForInput
|
||||
if err := repo.UpdateTaskSession(context.Background(), session); err != nil {
|
||||
t.Fatalf("failed to update session: %v", err)
|
||||
}
|
||||
seedExecutorRunning(t, repo, "s1", "t1", "exec-1")
|
||||
|
||||
agentMgr := &mockAgentManager{
|
||||
repoForExecutionLookup: repo,
|
||||
promptErr: errors.New("session rejected prompt synchronously"),
|
||||
}
|
||||
svc := createTestServiceWithScheduler(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
|
||||
mc := &mockMessageCreator{}
|
||||
svc.messageCreator = mc
|
||||
t.Cleanup(svc.cancelAllTransientRetries)
|
||||
|
||||
svc.rememberTurnPrompt("s1", "hello", "", false, nil)
|
||||
svc.transientRetries.Store("s1", &transientRetryEntry{attempt: 1, cancel: func() {}})
|
||||
|
||||
svc.retryTransientPrompt(context.Background(), "t1", "s1", "exec-1")
|
||||
|
||||
if _, ok := svc.transientRetries.Load("s1"); ok {
|
||||
t.Error("expected retry entry cleared after synchronous PromptTask failure")
|
||||
}
|
||||
if _, ok := svc.lastTurnPrompt.Load("s1"); ok {
|
||||
t.Error("expected cached prompt cleared after synchronous PromptTask failure")
|
||||
}
|
||||
hasRecovery := false
|
||||
for _, m := range mc.sessionMessages {
|
||||
if m.metadata["recovery_actions"] == true {
|
||||
hasRecovery = true
|
||||
}
|
||||
}
|
||||
if !hasRecovery {
|
||||
t.Error("expected recovery banner after synchronous PromptTask failure")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTransientRetryEntryClaimPreventsDoubleFire(t *testing.T) {
|
||||
entry := &transientRetryEntry{}
|
||||
if !entry.claim() {
|
||||
t.Fatal("first claim = false, want true")
|
||||
}
|
||||
if entry.claim() {
|
||||
t.Fatal("second claim = true, want false")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCancelAllTransientRetries_DropsCachedPrompt(t *testing.T) {
|
||||
svc, _ := newTransientTestService(t)
|
||||
svc.rememberTurnPrompt("s1", "hello", "", false, nil)
|
||||
svc.scheduleTransientRetry("t1", "s1", "", 1, time.Hour)
|
||||
|
||||
svc.cancelAllTransientRetries()
|
||||
|
||||
if _, ok := svc.transientRetries.Load("s1"); ok {
|
||||
t.Error("expected retry entry cleared after cancelAllTransientRetries")
|
||||
}
|
||||
if _, ok := svc.lastTurnPrompt.Load("s1"); ok {
|
||||
t.Error("expected cached prompt cleared after cancelAllTransientRetries")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTransientRetryDelay(t *testing.T) {
|
||||
cases := []struct {
|
||||
attempt int
|
||||
want time.Duration
|
||||
}{
|
||||
{0, 5 * time.Second}, // clamps up
|
||||
{1, 5 * time.Second},
|
||||
{2, 15 * time.Second},
|
||||
{3, 30 * time.Second},
|
||||
{4, 30 * time.Second}, // clamps to last
|
||||
}
|
||||
for _, tc := range cases {
|
||||
if got := transientRetryDelay(tc.attempt); got != tc.want {
|
||||
t.Errorf("transientRetryDelay(%d) = %v, want %v", tc.attempt, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleTransientFailure_AttemptCounterIncrements(t *testing.T) {
|
||||
svc, _ := newTransientTestService(t)
|
||||
t.Cleanup(svc.cancelAllTransientRetries)
|
||||
|
||||
data := watcher.AgentEventData{TaskID: "t1", SessionID: "s1", ErrorMessage: overloaded529}
|
||||
|
||||
svc.handleTransientFailure(context.Background(), data)
|
||||
svc.handleTransientFailure(context.Background(), data)
|
||||
|
||||
v, ok := svc.transientRetries.Load("s1")
|
||||
if !ok {
|
||||
t.Fatal("expected retry entry")
|
||||
}
|
||||
entry, _ := v.(*transientRetryEntry)
|
||||
if entry.attempt != 2 {
|
||||
t.Errorf("attempt = %d, want 2 after two transient failures", entry.attempt)
|
||||
}
|
||||
}
|
||||
@@ -180,7 +180,7 @@ func (h *Handlers) wsSetPlanMode(ctx context.Context, msg *ws.Message) (*ws.Mess
|
||||
type wsRecoverSessionRequest struct {
|
||||
TaskID string `json:"task_id"`
|
||||
SessionID string `json:"session_id"`
|
||||
Action string `json:"action"` // "resume" or "fresh_start"
|
||||
Action string `json:"action"` // "resume", "fresh_start", or "cancel_retry"
|
||||
}
|
||||
|
||||
func (h *Handlers) wsRecoverSession(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
|
||||
@@ -194,8 +194,16 @@ func (h *Handlers) wsRecoverSession(ctx context.Context, msg *ws.Message) (*ws.M
|
||||
if req.SessionID == "" {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil)
|
||||
}
|
||||
// Cancel an in-progress transient (529 Overloaded) retry loop and surface
|
||||
// the manual recovery banner. Distinct from resume/fresh_start: it does not
|
||||
// relaunch the agent, it stops the backoff timer.
|
||||
if req.Action == "cancel_retry" {
|
||||
cancelled := h.service.CancelTransientRetry(ctx, req.TaskID, req.SessionID)
|
||||
return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{"cancelled": cancelled})
|
||||
}
|
||||
|
||||
if req.Action != "resume" && req.Action != "fresh_start" {
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "action must be 'resume' or 'fresh_start'", nil)
|
||||
return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "action must be 'resume', 'fresh_start', or 'cancel_retry'", nil)
|
||||
}
|
||||
|
||||
resp, err := h.service.RecoverSession(ctx, req.TaskID, req.SessionID, req.Action)
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/orchestrator"
|
||||
ws "github.com/kandev/kandev/pkg/websocket"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func setupOrchestratorHandlers(t *testing.T) *Handlers {
|
||||
t.Helper()
|
||||
log, err := logger.NewLogger(logger.LoggingConfig{
|
||||
Level: "error",
|
||||
Format: "console",
|
||||
OutputPath: "stderr",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
return NewHandlers(&orchestrator.Service{}, log)
|
||||
}
|
||||
|
||||
func TestWsRecoverSessionCancelRetryReportsServiceResult(t *testing.T) {
|
||||
handlers := setupOrchestratorHandlers(t)
|
||||
response, err := handlers.wsRecoverSession(context.Background(), createTestMessage(t, ws.ActionSessionRecover, map[string]interface{}{
|
||||
"task_id": "t1",
|
||||
"session_id": "s1",
|
||||
"action": "cancel_retry",
|
||||
}))
|
||||
require.NoError(t, err)
|
||||
|
||||
var payload struct {
|
||||
Cancelled bool `json:"cancelled"`
|
||||
}
|
||||
require.NoError(t, json.Unmarshal(response.Payload, &payload))
|
||||
require.False(t, payload.Cancelled)
|
||||
}
|
||||
@@ -334,6 +334,17 @@ type Service struct {
|
||||
// phantom turn just to host it.
|
||||
cancelInFlight sync.Map
|
||||
|
||||
// transientRetries tracks in-progress transient-provider-error (529
|
||||
// Overloaded) retry loops. key: sessionID, value: *transientRetryEntry.
|
||||
// A backoff timer per session re-drives the failed prompt; cancelled on
|
||||
// success, user-cancel, or service shutdown.
|
||||
transientRetries sync.Map
|
||||
|
||||
// lastTurnPrompt caches the most recent outbound prompt per session so a
|
||||
// transient-failure retry can re-drive the same turn without the caller's
|
||||
// context. key: sessionID, value: capturedPrompt. Replaced every turn.
|
||||
lastTurnPrompt sync.Map
|
||||
|
||||
// Service state
|
||||
mu sync.RWMutex
|
||||
running bool
|
||||
@@ -844,6 +855,7 @@ func (s *Service) Stop() error {
|
||||
}
|
||||
|
||||
s.cancelAllClarificationWatchdogs()
|
||||
s.cancelAllTransientRetries()
|
||||
|
||||
if len(errs) > 0 {
|
||||
return errs[0]
|
||||
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/runtime/agentctl"
|
||||
"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
|
||||
"github.com/kandev/kandev/internal/agent/runtime/routingerr"
|
||||
"github.com/kandev/kandev/internal/orchestrator/dto"
|
||||
"github.com/kandev/kandev/internal/orchestrator/executor"
|
||||
"github.com/kandev/kandev/internal/orchestrator/queue"
|
||||
@@ -380,6 +381,10 @@ func (s *Service) StartCreatedSession(ctx context.Context, taskID, sessionID, ag
|
||||
|
||||
executorID := session.ExecutorID
|
||||
|
||||
// Cache the raw prompt so a transient-provider-error (529) retry can
|
||||
// re-drive this first turn — initial launches bypass PromptTask.
|
||||
s.rememberTurnPrompt(sessionID, prompt, "", planMode, attachments)
|
||||
|
||||
execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, executor.LaunchOptions{AgentProfileID: effectiveProfileID, ExecutorID: executorID, Prompt: effectivePrompt, StartAgent: true, Attachments: attachments})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -606,6 +611,10 @@ func (s *Service) startTask(ctx context.Context, taskID string, agentProfileID s
|
||||
mcpMode = executor.McpModeOffice
|
||||
}
|
||||
|
||||
// Cache the raw prompt so a transient-provider-error (529) retry can
|
||||
// re-drive this first turn — initial launches bypass PromptTask.
|
||||
s.rememberTurnPrompt(sessionID, prompt, "", planMode, attachments)
|
||||
|
||||
execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, executor.LaunchOptions{
|
||||
AgentProfileID: agentProfileID,
|
||||
ExecutorID: executorID,
|
||||
@@ -1999,6 +2008,11 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom
|
||||
|
||||
previousSessionState := session.State
|
||||
|
||||
// Cache the raw prompt so a transient-provider-error (529) retry can
|
||||
// re-drive this turn after backoff without the caller's context. Stores
|
||||
// the pre-injection prompt; PromptTask re-applies config/plan transforms.
|
||||
s.rememberTurnPrompt(sessionID, prompt, model, planMode, attachments)
|
||||
|
||||
s.setSessionRunning(ctx, taskID, sessionID, session)
|
||||
s.startTurnForSession(ctx, sessionID)
|
||||
|
||||
@@ -2076,7 +2090,11 @@ func (s *Service) handlePromptError(ctx context.Context, taskID, sessionID strin
|
||||
// force-unblock a hung agent. That is not a "review this failure" condition —
|
||||
// Service.CancelAgent reconciles DB state (WAITING_FOR_INPUT, cancel message,
|
||||
// complete turn) already.
|
||||
if !isTransientPromptError(err) && !errors.Is(err, lifecycle.ErrCancelEscalated) {
|
||||
// A transient provider error (529 Overloaded) is owned by the async
|
||||
// retry-with-backoff path (handleTransientFailure), which keeps the task
|
||||
// in progress while it retries — so don't flap it to REVIEW here.
|
||||
if !isTransientPromptError(err) && !errors.Is(err, lifecycle.ErrCancelEscalated) &&
|
||||
!routingerr.IsTransientProviderError(err.Error()) {
|
||||
_ = s.taskRepo.UpdateTaskState(ctx, taskID, v1.TaskStateReview)
|
||||
}
|
||||
s.completeTurnForSession(ctx, sessionID)
|
||||
|
||||
@@ -0,0 +1,108 @@
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import type { ReactNode } from "react";
|
||||
import { cleanup, fireEvent, render, screen, waitFor } from "@testing-library/react";
|
||||
import { StateProvider } from "@/components/state-provider";
|
||||
import { ActionMessage } from "./action-message";
|
||||
import { sessionId as toSessionId, taskId as toTaskId, type Message } from "@/lib/types/http";
|
||||
|
||||
const requestMock = vi.fn().mockResolvedValue({});
|
||||
|
||||
vi.mock("@/lib/ws/connection", () => ({
|
||||
getWebSocketClient: () => ({ request: requestMock }),
|
||||
}));
|
||||
|
||||
afterEach(() => {
|
||||
cleanup();
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
const CANCEL_TEST_ID = "recovery-cancel-retry-button";
|
||||
|
||||
function retryMessage(overrides: Partial<Message> = {}): Message {
|
||||
return {
|
||||
id: "msg-1",
|
||||
session_id: toSessionId("sess-1"),
|
||||
task_id: toTaskId("task-1"),
|
||||
author_type: "system",
|
||||
content: "Provider overloaded — retrying in 5s (attempt 1/3)",
|
||||
type: "status",
|
||||
created_at: "2026-05-30T00:00:00Z",
|
||||
metadata: {
|
||||
variant: "warning",
|
||||
retrying: true,
|
||||
attempt: 1,
|
||||
max_attempts: 3,
|
||||
retry_in_seconds: 5,
|
||||
session_id: "sess-1",
|
||||
task_id: "task-1",
|
||||
actions: [
|
||||
{
|
||||
type: "ws_request",
|
||||
label: "Cancel",
|
||||
icon: "x",
|
||||
test_id: CANCEL_TEST_ID,
|
||||
params: {
|
||||
method: "session.recover",
|
||||
payload: { task_id: "task-1", session_id: "sess-1", action: "cancel_retry" },
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
...overrides,
|
||||
} as Message;
|
||||
}
|
||||
|
||||
function Wrapper({ children }: { children: ReactNode }) {
|
||||
return <StateProvider initialState={{}}>{children}</StateProvider>;
|
||||
}
|
||||
|
||||
describe("ActionMessage — transient retry (warning variant)", () => {
|
||||
it("renders the retrying copy in amber, not red", () => {
|
||||
render(<ActionMessage comment={retryMessage()} sessionState="WAITING_FOR_INPUT" />, {
|
||||
wrapper: Wrapper,
|
||||
});
|
||||
const text = screen.getByText(/retrying in 5s \(attempt 1\/3\)/i);
|
||||
expect(text.className).toContain("text-amber-600");
|
||||
expect(text.className).not.toContain("text-red-600");
|
||||
});
|
||||
|
||||
it("Cancel fires a session.recover ws_request with action cancel_retry", async () => {
|
||||
render(<ActionMessage comment={retryMessage()} sessionState="WAITING_FOR_INPUT" />, {
|
||||
wrapper: Wrapper,
|
||||
});
|
||||
fireEvent.click(screen.getByTestId(CANCEL_TEST_ID));
|
||||
await waitFor(() => expect(requestMock).toHaveBeenCalledTimes(1));
|
||||
expect(requestMock).toHaveBeenCalledWith("session.recover", {
|
||||
task_id: "task-1",
|
||||
session_id: "sess-1",
|
||||
action: "cancel_retry",
|
||||
});
|
||||
});
|
||||
|
||||
it("hides while the session is RUNNING (retry in flight) to avoid a stale card", () => {
|
||||
const { container } = render(
|
||||
<ActionMessage comment={retryMessage()} sessionState="RUNNING" />,
|
||||
{ wrapper: Wrapper },
|
||||
);
|
||||
expect(container.firstChild).toBeNull();
|
||||
});
|
||||
|
||||
it("renders the red variant for a non-warning recovery banner", () => {
|
||||
const errorMsg = retryMessage({
|
||||
content: "Agent encountered an error",
|
||||
metadata: {
|
||||
variant: "error",
|
||||
recovery_actions: true,
|
||||
actions: [
|
||||
{ type: "ws_request", label: "Resume session", test_id: "recovery-resume-button" },
|
||||
],
|
||||
},
|
||||
} as Partial<Message>);
|
||||
render(<ActionMessage comment={errorMsg} sessionState="WAITING_FOR_INPUT" />, {
|
||||
wrapper: Wrapper,
|
||||
});
|
||||
const text = screen.getByText(/Agent encountered an error/i);
|
||||
expect(text.className).toContain("text-red-600");
|
||||
expect(text.className).not.toContain("text-amber-600");
|
||||
});
|
||||
});
|
||||
@@ -9,6 +9,7 @@ import {
|
||||
IconPlayerPlay,
|
||||
IconSparkles,
|
||||
IconGitCommit,
|
||||
IconX,
|
||||
} from "@tabler/icons-react";
|
||||
import { cn } from "@/lib/utils";
|
||||
import { Button } from "@kandev/ui/button";
|
||||
@@ -31,6 +32,7 @@ const ICON_MAP: Record<string, React.ElementType> = {
|
||||
sparkles: IconSparkles,
|
||||
"git-commit": IconGitCommit,
|
||||
"alert-triangle": IconAlertTriangle,
|
||||
x: IconX,
|
||||
};
|
||||
|
||||
type ActionMeta = {
|
||||
|
||||
@@ -130,6 +130,12 @@ export type StatusMetadata = {
|
||||
message?: string;
|
||||
variant?: "default" | "warning" | "error";
|
||||
cancelled?: boolean;
|
||||
// Transient provider-error (529 Overloaded) retry state. Present on the
|
||||
// yellow "retrying" status message the orchestrator emits during backoff.
|
||||
retrying?: boolean;
|
||||
attempt?: number;
|
||||
max_attempts?: number;
|
||||
retry_in_seconds?: number;
|
||||
};
|
||||
|
||||
export type RecoveryAuthMethod = {
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import { expect } from "@playwright/test";
|
||||
import { expect, type Page } from "@playwright/test";
|
||||
import type { SeedData } from "../fixtures/test-base";
|
||||
import type { ApiClient } from "./api-client";
|
||||
import { SessionPage } from "../pages/session-page";
|
||||
|
||||
const DONE_STATES = ["COMPLETED", "WAITING_FOR_INPUT"];
|
||||
|
||||
@@ -64,3 +66,33 @@ export async function waitForSessionEnvironment(
|
||||
)
|
||||
.toBe(options.expectedEnvironmentId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Seed a task + session and navigate to it, waiting for the first (normal)
|
||||
* turn to complete. Follow-up prompts can then exercise retry flows from a
|
||||
* clean idle state.
|
||||
*/
|
||||
export async function seedIdleSession(
|
||||
testPage: Page,
|
||||
apiClient: ApiClient,
|
||||
seedData: SeedData,
|
||||
title: string,
|
||||
): Promise<SessionPage> {
|
||||
const task = await apiClient.createTaskWithAgent(
|
||||
seedData.workspaceId,
|
||||
title,
|
||||
seedData.agentProfileId,
|
||||
{
|
||||
description: "/e2e:simple-message",
|
||||
workflow_id: seedData.workflowId,
|
||||
workflow_step_id: seedData.startStepId,
|
||||
repository_ids: [seedData.repositoryId],
|
||||
},
|
||||
);
|
||||
if (!task.session_id) throw new Error("createTaskWithAgent did not return a session_id");
|
||||
await testPage.goto(`/t/${task.id}`);
|
||||
const session = new SessionPage(testPage);
|
||||
await session.waitForLoad();
|
||||
await expect(session.idleInput()).toBeVisible({ timeout: 30_000 });
|
||||
return session;
|
||||
}
|
||||
|
||||
@@ -419,6 +419,16 @@ export class SessionPage {
|
||||
return this.page.getByTestId("recovery-fresh-button");
|
||||
}
|
||||
|
||||
/** "Cancel" button shown on the yellow transient-retry (529 Overloaded) card. */
|
||||
recoveryCancelRetryButton(): Locator {
|
||||
return this.page.getByTestId("recovery-cancel-retry-button");
|
||||
}
|
||||
|
||||
/** The yellow "Provider overloaded — retrying…" status card text. */
|
||||
transientRetryCard(): Locator {
|
||||
return this.chat.getByText(/Provider overloaded — retrying/i);
|
||||
}
|
||||
|
||||
/** Context reset divider shown in chat after resetting agent context. */
|
||||
contextResetDivider(): Locator {
|
||||
return this.chat.getByText("Context reset");
|
||||
@@ -741,6 +751,17 @@ export class SessionPage {
|
||||
await editor.press(`${modifier}+Enter`);
|
||||
}
|
||||
|
||||
/**
|
||||
* Type and submit a chat message via the Send button. Mobile (touch) layouts
|
||||
* don't submit on Ctrl/Cmd+Enter, so mobile specs use this instead.
|
||||
*/
|
||||
async sendMessageViaButton(text: string) {
|
||||
const editor = this.page.locator(".tiptap.ProseMirror").first();
|
||||
await editor.click();
|
||||
await editor.fill(text);
|
||||
await this.page.getByTestId("submit-message-button").click();
|
||||
}
|
||||
|
||||
/** Toggle plan mode on/off by clicking the plan mode toggle button in the toolbar.
|
||||
*
|
||||
* Waits for the button to advertise `data-plan-available="true"` before clicking.
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
// Filename starts with "mobile-" so it runs on the mobile-chrome Playwright
|
||||
// project (Pixel 5 emulation) — see e2e/playwright.config.ts. Mobile parity for
|
||||
// the transient provider-error (529 Overloaded) retry flow: the yellow retry
|
||||
// card and its Cancel button must render and work on a narrow touch viewport.
|
||||
import { test, expect } from "../../fixtures/test-base";
|
||||
import { seedIdleSession } from "../../helpers/session";
|
||||
|
||||
test.describe("mobile: transient provider error retry", () => {
|
||||
test("yellow retry card + Cancel works on mobile and surfaces recovery", async ({
|
||||
testPage,
|
||||
apiClient,
|
||||
seedData,
|
||||
}) => {
|
||||
const session = await seedIdleSession(testPage, apiClient, seedData, "Mobile Overloaded Test");
|
||||
|
||||
await session.sendMessageViaButton("/overloaded:9");
|
||||
|
||||
// Yellow retry card + Cancel button render on the narrow viewport.
|
||||
await expect(session.transientRetryCard()).toBeVisible({ timeout: 30_000 });
|
||||
await expect(session.recoveryCancelRetryButton()).toBeVisible();
|
||||
await expect(session.recoveryResumeButton()).toBeHidden();
|
||||
|
||||
// Tap Cancel → red recovery banner.
|
||||
await session.recoveryCancelRetryButton().click();
|
||||
await expect(session.recoveryResumeButton()).toBeVisible({ timeout: 30_000 });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,88 @@
|
||||
import { test, expect } from "../../fixtures/test-base";
|
||||
import { seedIdleSession } from "../../helpers/session";
|
||||
import { SessionPage } from "../../pages/session-page";
|
||||
|
||||
test.describe("transient provider error (529 Overloaded) retry", () => {
|
||||
test("shows the yellow retrying card, not the red error banner", async ({
|
||||
testPage,
|
||||
apiClient,
|
||||
seedData,
|
||||
}) => {
|
||||
const session = await seedIdleSession(testPage, apiClient, seedData, "Overloaded Retry Test");
|
||||
|
||||
// /overloaded:9 keeps failing so the retry loop stays visible until cancel.
|
||||
await session.sendMessage("/overloaded:9");
|
||||
|
||||
// The calm yellow "retrying" card + Cancel button must appear...
|
||||
await expect(session.transientRetryCard()).toBeVisible({ timeout: 30_000 });
|
||||
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, "Overloaded Cancel Test");
|
||||
|
||||
await session.sendMessage("/overloaded: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("retries are paced — the attempt counter advances across the backoff", async ({
|
||||
testPage,
|
||||
apiClient,
|
||||
seedData,
|
||||
}) => {
|
||||
const session = await seedIdleSession(testPage, apiClient, seedData, "Overloaded Backoff Test");
|
||||
|
||||
// /overloaded:9 keeps failing, so each backoff retry re-drives the prompt
|
||||
// and the orchestrator advances the attempt counter. The yellow card must
|
||||
// progress from attempt 1/3 to attempt 2/3 — proving the retry loop is live
|
||||
// and paced rather than an instant, silent resume.
|
||||
await session.sendMessage("/overloaded:9");
|
||||
|
||||
await expect(session.chat.getByText(/retrying in 5s \(attempt 1\/3\)/i)).toBeVisible({
|
||||
timeout: 30_000,
|
||||
});
|
||||
await expect(session.chat.getByText(/attempt 2\/3/i)).toBeVisible({ timeout: 30_000 });
|
||||
});
|
||||
|
||||
test("a 529 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 Overload Test",
|
||||
seedData.agentProfileId,
|
||||
{
|
||||
description: "/overloaded: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(/retrying in 5s \(attempt 1\/3\)/i)).toBeVisible({
|
||||
timeout: 30_000,
|
||||
});
|
||||
await expect(session.chat.getByText(/attempt 2\/3/i)).toBeVisible({ timeout: 30_000 });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,80 @@
|
||||
# 0011: Transient provider errors (529 Overloaded) auto-retry with visible backoff
|
||||
|
||||
**Status:** accepted
|
||||
**Date:** 2026-05-30
|
||||
**Area:** backend, frontend
|
||||
|
||||
## Context
|
||||
|
||||
When the Anthropic API returns a transient `529 Overloaded` mid-turn, the agent
|
||||
surfaces it over ACP as an error event that the backend turns into `agent.failed`.
|
||||
Previously this was handled identically to a terminal failure: a red "Agent has
|
||||
encountered an error" banner, the execution torn down, the task dropped to
|
||||
`REVIEW`, and (in some paths) an instant silent re-resume — reading as a flickery
|
||||
glitch rather than a deliberate retry. The classifier had no rule for 529, so it
|
||||
fell through to `phase.poststart.unknown` → `CodeAgentRuntime` (`AutoRetryable=false`).
|
||||
|
||||
## Decision
|
||||
|
||||
Transient provider overload is now a first-class, auto-retryable condition with a
|
||||
calm, paced, visible retry — separate from generic agent failures.
|
||||
|
||||
- **Classification** (`internal/agent/runtime/routingerr/`): new runtime rule
|
||||
`anthropic.overloaded.529.v1` matches the `529 … overloaded` / `overloaded_error`
|
||||
signature and maps it to a dedicated `CodeProviderOverloaded`
|
||||
(`AutoRetryable=true, FallbackAllowed=true`). Chose a dedicated code over reusing
|
||||
`CodeProviderUnavailable` (HTTP-503 semantics) for distinct UX copy and clean
|
||||
`routing_code: provider_overloaded` metrics. Exposes
|
||||
`IsTransientProviderError(msg) bool`, mirroring `IsResumeCorrupted`.
|
||||
- **Orchestrator** (`internal/orchestrator/event_handlers_transient.go`):
|
||||
`handleAgentFailed` routes transient errors to `handleTransientFailure` before
|
||||
the red recoverable-failure path. It schedules a backoff retry (5s → 15s → 30s,
|
||||
max 3 attempts) via a Service-owned cancellable timer goroutine (mirroring the
|
||||
clarification-watchdog pattern; drained in `Stop()`), parks the session in
|
||||
`WAITING_FOR_INPUT` (calm, banner-less) without moving the task to `REVIEW` or
|
||||
cleaning up, and emits a `variant: "warning"` status message with a Cancel
|
||||
action. On each retry it tears down the failed execution and re-drives the
|
||||
cached prompt via `PromptTask` (resuming a fresh agent). The prompt is cached
|
||||
both in `PromptTask` (follow-ups) and in the `LaunchPreparedSession` initial
|
||||
paths (`startTask`, `StartCreatedSession`) so a 529 on the very first turn can
|
||||
retry too; if no prompt is cached, the retry clears itself and surfaces the
|
||||
recovery banner rather than parking. After the budget is exhausted, or on
|
||||
Cancel (`session.recover` action `cancel_retry`), it falls through to the
|
||||
existing red Resume / Start-fresh banner (with friendly "stayed overloaded
|
||||
after several retries" copy instead of the raw 529). Gated to non-office tasks
|
||||
(`isOfficeTask`, i.e. `AssigneeAgentProfileID`) so genuine office tasks keep
|
||||
their structured error UI. The transient branch runs *before* run-mode
|
||||
automation finalization in `handleAgentFailed` — it's the only non-terminal
|
||||
failure path, so finalizing (which reaps the ephemeral worktree) must wait
|
||||
until the budget is actually spent. The synchronous prompt-error path
|
||||
(`handlePromptError`) also skips the `REVIEW` transition for transient errors.
|
||||
`httpStatusToCode` maps a structured `529` to `CodeProviderOverloaded` too,
|
||||
for adapters that surface the status separately from the body.
|
||||
- **Frontend** (`components/task/chat/messages/action-message.tsx`): the existing
|
||||
`variant: "warning"` rendering (amber, not red) drives the retry card; the
|
||||
bottom `AgentStatus` banner is never `FAILED` during retry, so there is no red
|
||||
flicker. The Cancel button dispatches the `cancel_retry` `ws_request`.
|
||||
- **Repro** (`cmd/mock-agent/`): a `/overloaded[:N]` scenario returns a real
|
||||
prompt-time ACP error carrying the production 529 string for the first N
|
||||
prompts (file-backed counter), then recovers.
|
||||
|
||||
## Consequences
|
||||
|
||||
- A 529 mid-turn now reads as "Provider overloaded — retrying in Ns (attempt X/Y)"
|
||||
with a Cancel button instead of a red error flash, and self-heals when the
|
||||
provider recovers, without user intervention.
|
||||
- Adds a per-session retry-loop owner the Service must drain on shutdown
|
||||
(goleak-covered).
|
||||
- Re-driving via `PromptTask` re-sends the user prompt to the resumed agent; for a
|
||||
real agent this could append a duplicate user turn in agent-side history — an
|
||||
accepted limitation for the transient-recovery case.
|
||||
|
||||
## Alternatives Considered
|
||||
|
||||
- **Reuse `CodeProviderUnavailable`** — rejected: conflates HTTP-503 "unavailable"
|
||||
with "temporarily overloaded, retry"; muddies metrics and UX copy.
|
||||
- **Keep the agent process alive and re-prompt in place** — rejected: the lifecycle
|
||||
marks the execution FAILED on a prompt error and rejects further prompts, so the
|
||||
retry must tear down and resume a fresh execution.
|
||||
- **Office-style FAILED state for transient errors** — rejected: the goal is an
|
||||
uninterrupted working→retrying→working chat, not a terminal failure surface.
|
||||
@@ -16,3 +16,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or
|
||||
| 0008 | [DB upgrade safety - meta table, pre-migration backup, migration logging](0008-db-upgrade-safety.md) | accepted | backend | 2026-05-16 |
|
||||
| 0009 | [Fail-closed GC semantics for filesystem and container cleanup](0009-fail-closed-gc-semantics.md) | accepted | backend | 2026-05-16 |
|
||||
| 0010 | [Worktree copy-files — per-repo, idempotent, host-local](0010-worktree-copy-files.md) | accepted | backend, frontend | 2026-05-19 |
|
||||
| 0011 | [Transient provider errors (529 Overloaded) auto-retry with visible backoff](0011-transient-provider-error-retry.md) | accepted | backend, frontend | 2026-05-30 |
|
||||
|
||||
Reference in New Issue
Block a user