From fe72fdc23474ba58a0295afbaacc7ee22ee3372b Mon Sep 17 00:00:00 2001 From: Jason HONG <136784169+hongzexin@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:13:01 +0800 Subject: [PATCH] fix(runtime): preserve progress across provider and storage failures --- server/cmd/multica/cmd_config.go | 29 ++++- server/cmd/multica/cmd_config_test.go | 31 +++++ server/internal/cli/config.go | 12 ++ server/internal/cli/config_test.go | 26 ++++ server/internal/daemon/config.go | 25 ++++ server/internal/daemon/config_test.go | 78 ++++++++++++ server/internal/handler/file.go | 16 ++- server/internal/handler/file_test.go | 52 ++++++++ .../internal/service/retry_deferred_test.go | 113 ++++++++++++++++++ server/internal/service/task.go | 25 ++-- .../service/task_complete_race_test.go | 20 ++-- 11 files changed, 404 insertions(+), 23 deletions(-) diff --git a/server/cmd/multica/cmd_config.go b/server/cmd/multica/cmd_config.go index 5f611a0ffb0..758fe9cc166 100644 --- a/server/cmd/multica/cmd_config.go +++ b/server/cmd/multica/cmd_config.go @@ -3,6 +3,7 @@ package main import ( "fmt" "os" + "path/filepath" "strconv" "strings" "time" @@ -31,6 +32,7 @@ var configSetSupportedKeys = []string{ "server_url", "app_url", "workspace_id", + "codex_path", "device_name", "runtime_name", "max_concurrent_tasks", @@ -48,7 +50,7 @@ var configSetCmd = &cobra.Command{ Use: "set ", Short: "Set a CLI configuration value", Long: "Supported keys: " + - "server_url, app_url, workspace_id, " + + "server_url, app_url, workspace_id, codex_path, " + "device_name, runtime_name, max_concurrent_tasks, poll_interval, " + "heartbeat_interval, agent_timeout, " + "codex_semantic_inactivity_timeout, codex_handshake_timeout, " + @@ -95,6 +97,7 @@ func runConfigShow(cmd *cobra.Command, _ []string) error { fmt.Fprintf(os.Stdout, "%-34s %s\n", "server_url:", valueOrDefault(cfg.ServerURL, "(not set)")) fmt.Fprintf(os.Stdout, "%-34s %s\n", "app_url:", valueOrDefault(cfg.AppURL, "(not set)")) fmt.Fprintf(os.Stdout, "%-34s %s\n", "workspace_id:", valueOrDefault(cfg.WorkspaceID, "(not set)")) + fmt.Fprintf(os.Stdout, "%-34s %s\n", "codex_path:", valueOrDefault(codexPath(cfg), "(not set)")) fmt.Fprintf(os.Stdout, "%-34s %s\n", "device_name:", valueOrDefault(cfg.DeviceName, "(not set)")) fmt.Fprintf(os.Stdout, "%-34s %s\n", "runtime_name:", valueOrDefault(cfg.RuntimeName, "(not set)")) fmt.Fprintf(os.Stdout, "%-34s %s\n", "max_concurrent_tasks:", intOrDefault(cfg.MaxConcurrentTasks, "(not set)")) @@ -151,6 +154,23 @@ func applyConfigSet(cfg *cli.CLIConfig, key, value string) error { cfg.AppURL = value case "workspace_id": cfg.WorkspaceID = value + case "codex_path": + if value == "" { + if cfg.Backends != nil { + cfg.Backends.Codex = nil + if cfg.Backends.OpenClaw == nil { + cfg.Backends = nil + } + } + return nil + } + if !filepath.IsAbs(value) { + return fmt.Errorf("codex_path must be an absolute path (got %q)", value) + } + if cfg.Backends == nil { + cfg.Backends = &cli.BackendOverrides{} + } + cfg.Backends.Codex = &cli.CodexOverride{BinaryPath: filepath.Clean(value)} case "device_name": cfg.DeviceName = value case "runtime_name": @@ -236,6 +256,13 @@ func applyConfigSet(cfg *cli.CLIConfig, key, value string) error { return nil } +func codexPath(cfg cli.CLIConfig) string { + if cfg.Backends == nil || cfg.Backends.Codex == nil { + return "" + } + return cfg.Backends.Codex.BinaryPath +} + // assignBool parses value as a strict bool into dst. Shared by the // disable_* toggles. Empty string clears the field (false). func assignBool(dst *bool, key, value string) error { diff --git a/server/cmd/multica/cmd_config_test.go b/server/cmd/multica/cmd_config_test.go index eec152c1bd7..1b726af4356 100644 --- a/server/cmd/multica/cmd_config_test.go +++ b/server/cmd/multica/cmd_config_test.go @@ -62,6 +62,7 @@ func TestRunConfigShowIncludesProfileAndDefaults(t *testing.T) { "server_url:", "app_url:", "workspace_id:", + "codex_path:", "device_name:", "runtime_name:", "max_concurrent_tasks:", @@ -196,6 +197,7 @@ func TestApplyConfigSetSupportsDaemonKeys(t *testing.T) { cfg := cli.CLIConfig{} pairs := []struct{ key, val string }{ + {"codex_path", "/opt/company/bin/mcodex"}, {"device_name", "vm-1-custom-name"}, {"runtime_name", "worker-a"}, {"max_concurrent_tasks", "4"}, @@ -213,6 +215,7 @@ func TestApplyConfigSetSupportsDaemonKeys(t *testing.T) { } } if cfg.DeviceName != "vm-1-custom-name" || + codexPath(cfg) != "/opt/company/bin/mcodex" || cfg.RuntimeName != "worker-a" || cfg.MaxConcurrentTasks != 4 || cfg.PollInterval != "10s" || @@ -226,6 +229,34 @@ func TestApplyConfigSetSupportsDaemonKeys(t *testing.T) { } } +func TestApplyConfigSetCodexPathValidationAndClear(t *testing.T) { + t.Parallel() + + cfg := cli.CLIConfig{ + Backends: &cli.BackendOverrides{ + OpenClaw: &cli.OpenClawOverride{StateDir: "/var/lib/openclaw"}, + }, + } + if err := applyConfigSet(&cfg, "codex_path", "relative/mcodex"); err == nil { + t.Fatal("relative codex_path should be rejected") + } + if err := applyConfigSet(&cfg, "codex_path", "/opt/company/bin/../bin/mcodex"); err != nil { + t.Fatalf("set codex_path: %v", err) + } + if got := codexPath(cfg); got != "/opt/company/bin/mcodex" { + t.Fatalf("codex_path = %q, want cleaned absolute path", got) + } + if err := applyConfigSet(&cfg, "codex_path", ""); err != nil { + t.Fatalf("clear codex_path: %v", err) + } + if got := codexPath(cfg); got != "" { + t.Fatalf("codex_path after clear = %q", got) + } + if cfg.Backends == nil || cfg.Backends.OpenClaw == nil { + t.Fatal("clearing codex_path should preserve unrelated backend overrides") + } +} + func TestApplyConfigSetPositiveDurationRoundTripsToDaemonResolver(t *testing.T) { const envName = "TEST_MULTICA_PERSISTED_DURATION" t.Setenv(envName, "") diff --git a/server/internal/cli/config.go b/server/internal/cli/config.go index ca40535bf98..4168fa0ce5a 100644 --- a/server/internal/cli/config.go +++ b/server/internal/cli/config.go @@ -157,9 +157,21 @@ type CLIConfig struct { // Go's encoding/json drops fields that are not represented in this struct on // load/save round-trip (see TestCLIConfig_UnknownFieldsArePreserved). type BackendOverrides struct { + Codex *CodexOverride `json:"codex,omitempty"` OpenClaw *OpenClawOverride `json:"openclaw,omitempty"` } +// CodexOverride configures the Codex executable used by daemon starts on this +// machine. Persisting the path closes the gap where an environment-only +// MULTICA_CODEX_PATH override is lost after a manual daemon restart. +// +// Resolution precedence: +// +// MULTICA_CODEX_PATH (env) > backends.codex.binary_path > PATH lookup +type CodexOverride struct { + BinaryPath string `json:"binary_path,omitempty"` +} + // OpenClawOverride configures the OpenClaw backend. All fields are optional; // empty values fall through to the existing discovery path (PATH lookup for // BinaryPath, default `~/.openclaw/` for StateDir). diff --git a/server/internal/cli/config_test.go b/server/internal/cli/config_test.go index 1e643a572da..09e0d89363d 100644 --- a/server/internal/cli/config_test.go +++ b/server/internal/cli/config_test.go @@ -121,6 +121,32 @@ func TestCLIConfig_OpenClawOverride_RoundTrip(t *testing.T) { } } +func TestCLIConfig_CodexOverride_RoundTrip(t *testing.T) { + tmp := t.TempDir() + t.Setenv("HOME", tmp) + + original := CLIConfig{ + ServerURL: "https://api.multica.ai", + Backends: &BackendOverrides{ + Codex: &CodexOverride{BinaryPath: "/opt/company/bin/mcodex"}, + }, + } + if err := SaveCLIConfig(original); err != nil { + t.Fatal(err) + } + + loaded, err := LoadCLIConfig() + if err != nil { + t.Fatal(err) + } + if loaded.Backends == nil || loaded.Backends.Codex == nil { + t.Fatalf("Backends.Codex should be non-nil after round-trip, got %+v", loaded.Backends) + } + if got := loaded.Backends.Codex.BinaryPath; got != "/opt/company/bin/mcodex" { + t.Errorf("Codex BinaryPath round-trip: got %q", got) + } +} + // TestCLIConfig_OpenClawOverride_PartialFieldsOmitted verifies that an // override with only one field set does not emit empty strings for the // unset field. Important so users can intentionally set only BinaryPath diff --git a/server/internal/daemon/config.go b/server/internal/daemon/config.go index d94dd1aab7d..3846b0ab4b1 100644 --- a/server/internal/daemon/config.go +++ b/server/internal/daemon/config.go @@ -207,6 +207,10 @@ func LoadConfig(overrides Overrides) (Config, error) { slog.Warn("could not load CLI config for backend overrides; proceeding without", "profile", overrides.Profile, "err", err) } else { + if codex := codexOverrideFrom(cliCfg); codex != nil { + restore := applyCodexOverride(codex) + defer restore() + } if oc := openclawOverrideFrom(cliCfg); oc != nil { applyOpenclawOverride(oc) } @@ -1011,6 +1015,27 @@ func openclawOverrideFrom(cfg cli.CLIConfig) *cli.OpenClawOverride { return cfg.Backends.OpenClaw } +func codexOverrideFrom(cfg cli.CLIConfig) *cli.CodexOverride { + if cfg.Backends == nil { + return nil + } + return cfg.Backends.Codex +} + +// applyCodexOverride translates the persisted Codex path into the existing +// environment-based probe contract. An explicitly exported environment value +// remains authoritative for backward compatibility. +func applyCodexOverride(codex *cli.CodexOverride) func() { + if codex == nil || codex.BinaryPath == "" { + return func() {} + } + if _, set := os.LookupEnv("MULTICA_CODEX_PATH"); set { + return func() {} + } + _ = os.Setenv("MULTICA_CODEX_PATH", codex.BinaryPath) + return func() { _ = os.Unsetenv("MULTICA_CODEX_PATH") } +} + // applyOpenclawOverride translates the config-file overrides into process // env vars, which the existing probe() / buildEnv code paths already honor. // Env-set-by-user wins over config-set-by-file: we only Setenv when the var diff --git a/server/internal/daemon/config_test.go b/server/internal/daemon/config_test.go index 6e8275229c3..d1c8c9db378 100644 --- a/server/internal/daemon/config_test.go +++ b/server/internal/daemon/config_test.go @@ -844,6 +844,7 @@ func TestLoadConfig_SkipsLoginShellWhenLookPathSucceeds(t *testing.T) { } func TestLoadConfig_UsesCodexDesktopAppBundleFallback(t *testing.T) { + t.Setenv("HOME", t.TempDir()) pathDir := t.TempDir() fakeCodex := filepath.Join(pathDir, "Codex.app", "Contents", "Resources", "codex") if err := os.MkdirAll(filepath.Dir(fakeCodex), 0o755); err != nil { @@ -886,6 +887,7 @@ func TestLoadConfig_UsesCodexDesktopAppBundleFallback(t *testing.T) { // Multica must resolve the bundled CLI under ChatGPT.app (and prefer it over // the legacy Codex.app path when both exist). func TestLoadConfig_UsesChatGPTAppBundleCodexPath(t *testing.T) { + t.Setenv("HOME", t.TempDir()) pathDir := t.TempDir() fakeChatGPT := filepath.Join(pathDir, "ChatGPT.app", "Contents", "Resources", "codex") fakeLegacy := filepath.Join(pathDir, "Codex.app", "Contents", "Resources", "codex") @@ -1141,6 +1143,82 @@ func TestOpenclawOverrideFrom_NavigationCases(t *testing.T) { } } +func TestApplyCodexOverride_EnvWinsOverConfig(t *testing.T) { + t.Setenv("MULTICA_CODEX_PATH", "/from/env/codex") + + restore := applyCodexOverride(&cli.CodexOverride{BinaryPath: "/from/config/mcodex"}) + restore() + + if got := os.Getenv("MULTICA_CODEX_PATH"); got != "/from/env/codex" { + t.Errorf("MULTICA_CODEX_PATH: env should win, got %q", got) + } +} + +func TestApplyCodexOverride_RestoresPreviouslyUnsetEnvironment(t *testing.T) { + os.Unsetenv("MULTICA_CODEX_PATH") + t.Cleanup(func() { os.Unsetenv("MULTICA_CODEX_PATH") }) + + restore := applyCodexOverride(&cli.CodexOverride{BinaryPath: "/from/config/mcodex"}) + if got := os.Getenv("MULTICA_CODEX_PATH"); got != "/from/config/mcodex" { + t.Fatalf("MULTICA_CODEX_PATH during probe = %q", got) + } + restore() + if _, set := os.LookupEnv("MULTICA_CODEX_PATH"); set { + t.Fatal("MULTICA_CODEX_PATH should be restored to unset after probe") + } +} + +func TestCodexOverrideFrom_NavigationCases(t *testing.T) { + if got := codexOverrideFrom(cli.CLIConfig{}); got != nil { + t.Errorf("nil Backends should produce nil override, got %+v", got) + } + if got := codexOverrideFrom(cli.CLIConfig{Backends: &cli.BackendOverrides{}}); got != nil { + t.Errorf("nil Codex inside Backends should produce nil override, got %+v", got) + } + want := &cli.CodexOverride{BinaryPath: "/x/mcodex"} + got := codexOverrideFrom(cli.CLIConfig{Backends: &cli.BackendOverrides{Codex: want}}) + if got != want { + t.Errorf("happy path should return inner pointer; got %p want %p", got, want) + } +} + +func TestLoadConfig_AppliesCodexOverrideFromConfigFile(t *testing.T) { + stageFakeAgent(t) + customDir := t.TempDir() + customCodex := filepath.Join(customDir, "mcodex") + if err := os.WriteFile(customCodex, []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil { + t.Fatalf("write fake codex: %v", err) + } + + os.Unsetenv("MULTICA_CODEX_PATH") + t.Cleanup(func() { os.Unsetenv("MULTICA_CODEX_PATH") }) + homeForCLIConfig := t.TempDir() + t.Setenv("HOME", homeForCLIConfig) + if err := cli.SaveCLIConfig(cli.CLIConfig{ + ServerURL: "http://localhost:8080", + Backends: &cli.BackendOverrides{ + Codex: &cli.CodexOverride{BinaryPath: customCodex}, + }, + }); err != nil { + t.Fatalf("save cli config: %v", err) + } + + loaded, err := LoadConfig(Overrides{ + ServerURL: "http://localhost:8080", + WorkspacesRoot: t.TempDir(), + }) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + codex, ok := loaded.Agents["codex"] + if !ok { + t.Fatalf("agents map missing codex; got keys=%v", agentKeys(loaded.Agents)) + } + if codex.Path != customCodex { + t.Errorf("codex.Path: got %q, want %q", codex.Path, customCodex) + } +} + // TestLoadConfig_AppliesBackendOverridesFromConfigFile is the integration // test that ties commit 1's schema to commit 2's wire-up: write a config // file with backends.openclaw.{binary_path,state_dir}, call LoadConfig diff --git a/server/internal/handler/file.go b/server/internal/handler/file.go index 6bf47fe0128..d25585b4b7f 100644 --- a/server/internal/handler/file.go +++ b/server/internal/handler/file.go @@ -548,8 +548,7 @@ func (h *Handler) UploadFile(w http.ResponseWriter, r *http.Request) { link, err := h.Storage.Upload(r.Context(), key, data, contentType, header.Filename) if err != nil { - slog.Error("file upload failed", "error", err) - writeError(w, http.StatusInternalServerError, "upload failed") + writeStorageUploadError(w, err) return } params.Url = link @@ -575,8 +574,7 @@ func (h *Handler) UploadFile(w http.ResponseWriter, r *http.Request) { // No workspace context (e.g. avatar upload) — upload directly. link, err := h.Storage.Upload(r.Context(), key, data, contentType, header.Filename) if err != nil { - slog.Error("file upload failed", "error", err) - writeError(w, http.StatusInternalServerError, "upload failed") + writeStorageUploadError(w, err) return } writeJSON(w, http.StatusOK, map[string]string{ @@ -586,6 +584,16 @@ func (h *Handler) UploadFile(w http.ResponseWriter, r *http.Request) { }) } +func writeStorageUploadError(w http.ResponseWriter, err error) { + slog.Error("file upload failed", "error", err) + writeErrorCode( + w, + http.StatusBadGateway, + "storage_upload_failed", + "file storage rejected the upload; contact an administrator", + ) +} + // --------------------------------------------------------------------------- // ListAttachments — GET /api/issues/{id}/attachments // --------------------------------------------------------------------------- diff --git a/server/internal/handler/file_test.go b/server/internal/handler/file_test.go index d97f4da1411..9adfefa25bb 100644 --- a/server/internal/handler/file_test.go +++ b/server/internal/handler/file_test.go @@ -59,6 +59,18 @@ type mockStorage struct { presignDispositions []string } +type failingUploadStorage struct{ mockStorage } + +func (m *failingUploadStorage) Upload( + _ context.Context, + _ string, + _ []byte, + _ string, + _ string, +) (string, error) { + return "", fmt.Errorf("simulated object storage rejection") +} + func (m *mockStorage) Upload(_ context.Context, key string, data []byte, _ string, _ string) (string, error) { m.mu.Lock() defer m.mu.Unlock() @@ -233,6 +245,46 @@ func TestUploadFileForeignWorkspace(t *testing.T) { } } +func TestUploadFileStorageFailureReturnsDiagnosableGatewayError(t *testing.T) { + origStorage := testHandler.Storage + testHandler.Storage = &failingUploadStorage{} + defer func() { testHandler.Storage = origStorage }() + + var body bytes.Buffer + writer := multipart.NewWriter(&body) + part, err := writer.CreateFormFile("file", "test.txt") + if err != nil { + t.Fatal(err) + } + if _, err := part.Write([]byte("hello world")); err != nil { + t.Fatal(err) + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + + req := httptest.NewRequest("POST", "/api/upload-file", &body) + req.Header.Set("Content-Type", writer.FormDataContentType()) + req.Header.Set("X-User-ID", testUserID) + + w := httptest.NewRecorder() + testHandler.UploadFile(w, req) + if w.Code != http.StatusBadGateway { + t.Fatalf("storage failure: expected 502, got %d: %s", w.Code, w.Body.String()) + } + + var response map[string]string + if err := json.Unmarshal(w.Body.Bytes(), &response); err != nil { + t.Fatalf("decode response: %v", err) + } + if response["code"] != "storage_upload_failed" { + t.Fatalf("storage failure code: got %q", response["code"]) + } + if strings.Contains(response["error"], "simulated object storage rejection") { + t.Fatalf("storage error leaked internal details: %q", response["error"]) + } +} + // TestUploadFileResolvesWorkspaceViaSlugHeader is a regression test for the // v2 workspace URL refactor (#1141). The frontend switched from sending // X-Workspace-ID (UUID) to X-Workspace-Slug. For endpoints that sit outside diff --git a/server/internal/service/retry_deferred_test.go b/server/internal/service/retry_deferred_test.go index 5bf6ea41373..b3de65e6ecb 100644 --- a/server/internal/service/retry_deferred_test.go +++ b/server/internal/service/retry_deferred_test.go @@ -167,3 +167,116 @@ func TestFailTaskProviderNetworkBudget(t *testing.T) { }) } } + +// TestFailTaskProviderCapacityRetry verifies that a terminal provider 429 keeps +// the completed run state available and creates exactly one delayed retry that +// resumes the same session and work directory. +func TestFailTaskProviderCapacityRetry(t *testing.T) { + pool := newResolveOriginatorPool(t) + ctx := context.Background() + q := db.New(pool) + _, _, agentID, issueID := seedAttributionFixture(t, pool) + svc := &TaskService{Queries: q, TxStarter: pool, Bus: events.New()} + + var runtimeID string + if err := pool.QueryRow(ctx, `SELECT runtime_id::text FROM agent WHERE id = $1`, agentID).Scan(&runtimeID); err != nil { + t.Fatalf("read agent runtime: %v", err) + } + + const ( + sourceSession = "capacity-source-session" + sourceWorkDir = "/tmp/capacity-source-workdir" + reason = "agent_error.provider_capacity_or_rate_limit" + ) + var parentID pgtype.UUID + if err := pool.QueryRow(ctx, ` + INSERT INTO agent_task_queue ( + agent_id, runtime_id, issue_id, status, priority, attempt, + max_attempts, session_id, work_dir + ) + VALUES ($1, $2, $3, 'running', 0, 1, 2, $4, $5) + RETURNING id + `, agentID, runtimeID, issueID, sourceSession, sourceWorkDir).Scan(&parentID); err != nil { + t.Fatalf("insert parent task: %v", err) + } + t.Cleanup(func() { + pool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE parent_task_id = $1 OR id = $1`, parentID) + }) + + failureTime := time.Now() + if _, err := svc.FailTask( + ctx, + parentID, + "429 Too Many Requests", + sourceSession, + sourceWorkDir, + reason, + false, + "", + ); err != nil { + t.Fatalf("FailTask: %v", err) + } + + var ( + childStatus string + childAttempt int32 + childMaxAttempts int32 + childSession string + childWorkDir string + childForceFresh bool + childFireAt pgtype.Timestamptz + parentStatus string + parentFailReason pgtype.Text + childCount int + ) + if err := pool.QueryRow(ctx, ` + SELECT count(*), coalesce(max(status), ''), coalesce(max(attempt), 0), + coalesce(max(max_attempts), 0), coalesce(max(session_id), ''), + coalesce(max(work_dir), ''), coalesce(bool_or(force_fresh_session), false), + max(fire_at) + FROM agent_task_queue + WHERE parent_task_id = $1 + `, parentID).Scan( + &childCount, + &childStatus, + &childAttempt, + &childMaxAttempts, + &childSession, + &childWorkDir, + &childForceFresh, + &childFireAt, + ); err != nil { + t.Fatalf("read retry child: %v", err) + } + if err := pool.QueryRow(ctx, ` + SELECT status, failure_reason FROM agent_task_queue WHERE id = $1 + `, parentID).Scan(&parentStatus, &parentFailReason); err != nil { + t.Fatalf("read failed parent: %v", err) + } + + if childCount != 1 { + t.Fatalf("retry children = %d, want exactly 1", childCount) + } + if parentStatus != "failed" || !parentFailReason.Valid || parentFailReason.String != reason { + t.Errorf("parent = status %q reason %q, want failed/%q", parentStatus, parentFailReason.String, reason) + } + if childStatus != "deferred" || !childFireAt.Valid { + t.Fatalf("child = status %q fire_at %v, want deferred with fire_at", childStatus, childFireAt) + } + if childAttempt != 2 || childMaxAttempts != 2 { + t.Errorf("child budget = %d/%d, want attempt 2/max 2", childAttempt, childMaxAttempts) + } + if childSession != sourceSession || childWorkDir != sourceWorkDir || childForceFresh { + t.Errorf( + "child resume state = session %q workdir %q force_fresh %v", + childSession, + childWorkDir, + childForceFresh, + ) + } + minimumFireAt := failureTime.Add(providerCapacityRetryWait - 2*time.Second) + maximumFireAt := time.Now().Add(providerCapacityRetryWait + 2*time.Second) + if childFireAt.Time.Before(minimumFireAt) || childFireAt.Time.After(maximumFireAt) { + t.Errorf("child fire_at = %v, want about %s after failure", childFireAt.Time, providerCapacityRetryWait) + } +} diff --git a/server/internal/service/task.go b/server/internal/service/task.go index cd50133e07e..3e94e1fc60a 100644 --- a/server/internal/service/task.go +++ b/server/internal/service/task.go @@ -4122,12 +4122,12 @@ func (s *TaskService) FailTask(ctx context.Context, taskID pgtype.UUID, errMsg, // etc.) are intentionally excluded — those are real problems that the user // should see, not infrastructure flakiness. // -// The one agent_error.* exception is provider_network: a mid-stream provider -// disconnect (e.g. Claude Code's "API Error: Connection closed mid-response") -// is transient infrastructure flakiness, not an agent decision. Unattended +// The agent_error.* exceptions are provider_network and +// provider_capacity_or_rate_limit: both are transient provider failures, not +// agent decisions. Unattended // issue runs otherwise terminate on it, while interactive chat only survives // because the CLI's own in-process retry happens to recover first — so we make -// the platform retry it directly (MUL-4910). It is resume-safe (not in +// the platform retry them directly (MUL-4910). They are resume-safe (not in // resumeUnsafeFailureReason), so the retry child inherits the session and // continues the truncated conversation rather than restarting from scratch. // skill_bundle_unavailable is retryable for the same reason: the agent process @@ -4139,8 +4139,9 @@ var retryableReasons = map[string]bool{ "runtime_recovery": true, "timeout": true, "codex_semantic_inactivity": true, - string(taskfailure.ReasonAgentProviderNetwork): true, - string(taskfailure.ReasonSkillBundleUnavailable): true, + string(taskfailure.ReasonAgentProviderNetwork): true, + string(taskfailure.ReasonAgentProviderCapacityOrRateLimit): true, + string(taskfailure.ReasonSkillBundleUnavailable): true, } // Transient provider stream cuts (provider_network) get a bespoke three-tier @@ -4151,6 +4152,7 @@ var retryableReasons = map[string]bool{ const ( providerNetworkMaxAttempts = 3 providerNetworkFinalRetryWait = 5 * time.Second + providerCapacityRetryWait = 15 * time.Second ) // retryAttemptCeiling reports how many attempts the auto-retry path allows for @@ -4175,11 +4177,14 @@ func retryAttemptCeiling(reason string, taskMaxAttempts int32) int32 { } // retryDelayForAttempt reports how long to defer the NEXT attempt after a -// failure at failedAttempt. Only provider_network's final attempt is deferred -// (~5s); every other retry — including provider_network's first — is immediate -// (zero delay → the child is created 'queued', claimable at once). Callers pass -// the returned delay to CreateRetryTask via fire_at. +// failure at failedAttempt. Provider capacity/rate-limit failures always wait +// briefly so a second request does not immediately hit the same exhausted +// provider window. provider_network keeps its existing immediate second attempt +// and ~5s final attempt. A zero delay creates an immediately claimable child. func retryDelayForAttempt(reason string, failedAttempt int32) time.Duration { + if reason == string(taskfailure.ReasonAgentProviderCapacityOrRateLimit) { + return providerCapacityRetryWait + } if reason == string(taskfailure.ReasonAgentProviderNetwork) && failedAttempt >= providerNetworkMaxAttempts-1 { return providerNetworkFinalRetryWait diff --git a/server/internal/service/task_complete_race_test.go b/server/internal/service/task_complete_race_test.go index 5489a27273b..2673244da38 100644 --- a/server/internal/service/task_complete_race_test.go +++ b/server/internal/service/task_complete_race_test.go @@ -169,12 +169,12 @@ func TestFailTask_AlreadyFinalized(t *testing.T) { } } -// TestProviderNetworkRetrySchedule locks in the three-tier schedule for a -// transient provider stream cut (MUL-4910): first run + immediate retry + one -// retry deferred ~5s, and only for provider_network — other retryable reasons -// keep their generic max_attempts=2 (single, immediate retry). -func TestProviderNetworkRetrySchedule(t *testing.T) { +// TestProviderTransientRetrySchedule locks in the retry schedules for transient +// provider failures: provider_network gets its three-tier sequence, while +// provider capacity/rate-limit failures get one delayed, resume-safe retry. +func TestProviderTransientRetrySchedule(t *testing.T) { const provNet = "agent_error.provider_network" + const provCapacity = "agent_error.provider_capacity_or_rate_limit" // Attempt ceiling: provider_network is raised to 3, but only ever WIDENS the // budget and never overrides the max_attempts<=1 "retry disabled" contract. @@ -195,15 +195,16 @@ func TestProviderNetworkRetrySchedule(t *testing.T) { } } - // Backoff: only provider_network's final attempt (after the 2nd failure) is - // deferred; its first retry and every other reason are immediate. + // Backoff: provider_network's final attempt (after the 2nd failure) is + // deferred; provider capacity gets a cooldown before its single retry. delayCases := []struct { reason string failedAttempt int32 want time.Duration }{ {provNet, 1, 0}, // first failure → immediate retry - {provNet, 2, providerNetworkFinalRetryWait}, // second failure → 5s-deferred retry + {provNet, 2, providerNetworkFinalRetryWait}, // second failure → 5s-deferred retry + {provCapacity, 1, providerCapacityRetryWait}, // capacity failure → bounded cooldown {"timeout", 2, 0}, // unrelated reason → never deferred } for _, tc := range delayCases { @@ -232,6 +233,8 @@ func TestProviderNetworkRetrySchedule(t *testing.T) { {"provider_network second run still retries (deferred tier)", provNet, 2, 2, true}, {"provider_network third run is the ceiling", provNet, 3, 2, false}, {"provider_network with retry disabled (max_attempts=1) never retries", provNet, 1, 1, false}, + {"provider capacity first run retries", provCapacity, 1, 2, true}, + {"provider capacity exhausts at attempt 2", provCapacity, 2, 2, false}, {"timeout keeps single immediate retry", "timeout", 1, 2, true}, {"timeout exhausts at attempt 2", "timeout", 2, 2, false}, {"non-retryable reason never retries", "agent_error.unknown", 1, 2, false}, @@ -255,6 +258,7 @@ func TestTaskFailureClassifiers(t *testing.T) { // Transient mid-stream provider disconnect (MUL-4910): retryable, and // resume-safe so the retry continues the truncated conversation. {reason: "agent_error.provider_network", wantType: "agent_error", wantResumeOK: true, wantRetry: true}, + {reason: "agent_error.provider_capacity_or_rate_limit", wantType: "agent_error", wantResumeOK: true, wantRetry: true}, {reason: "runtime_recovery", wantType: "runtime", wantResumeOK: true, wantRetry: true}, {reason: "iteration_limit", wantType: "agent_output", wantResumeOK: false, wantRetry: false}, {reason: "api_invalid_request", wantType: "agent_error", wantResumeOK: false, wantRetry: false},