diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index a7d9e8c7f..0ca024318 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -319,6 +319,7 @@ jobs: OAC_HARNESSES: ${{ vars.OAC_HARNESSES }} OAC_DEFAULT_HARNESS: ${{ vars.OAC_DEFAULT_HARNESS }} OAC_EXECUTION_CONCURRENCY: ${{ vars.OAC_EXECUTION_CONCURRENCY }} + OAC_TURN_AUTO_RETRIES: ${{ vars.OAC_TURN_AUTO_RETRIES }} OAC_WRITE_AUDIT_RETENTION: ${{ vars.OAC_WRITE_AUDIT_RETENTION }} OAC_LOG_LEVEL: ${{ vars.OAC_LOG_LEVEL }} OAC_DATABASE_PASSWORD: ${{ secrets.OAC_DATABASE_PASSWORD }} @@ -352,7 +353,7 @@ jobs: envsubst '$OAC_PUBLIC_URL $OAC_DATABASE_URL' \ < deploy/kubernetes/prod/core-env.yaml.tpl > "$work/core-env.yaml" for name in OAC_HARNESSES OAC_DEFAULT_HARNESS OAC_EXECUTION_CONCURRENCY \ - OAC_WRITE_AUDIT_RETENTION OAC_LOG_LEVEL; do + OAC_TURN_AUTO_RETRIES OAC_WRITE_AUDIT_RETENTION OAC_LOG_LEVEL; do value="${!name:-}" if [ -n "$value" ]; then printf ' %s: "%s"\n' "$name" "$value" >> "$work/core-env.yaml"; fi done diff --git a/deploy/compose/compose.yaml b/deploy/compose/compose.yaml index 408ad1c56..d2742fa78 100644 --- a/deploy/compose/compose.yaml +++ b/deploy/compose/compose.yaml @@ -61,6 +61,7 @@ services: OAC_EXECUTION_CONCURRENCY: ${OAC_EXECUTION_CONCURRENCY:-} OAC_SANDBOX_MAX_ACTIVE: ${OAC_SANDBOX_MAX_ACTIVE:-} OAC_SANDBOX_MAX_RETAINED: ${OAC_SANDBOX_MAX_RETAINED:-} + OAC_TURN_AUTO_RETRIES: ${OAC_TURN_AUTO_RETRIES:-} OAC_DEFAULT_HARNESS: ${OAC_DEFAULT_HARNESS:-} OAC_HARNESSES: ${OAC_HARNESSES:-} OAC_WRITE_AUDIT_RETENTION: ${OAC_WRITE_AUDIT_RETENTION:-} diff --git a/deploy/compose/test_compose.py b/deploy/compose/test_compose.py index c4e2e0b93..38d675e6d 100644 --- a/deploy/compose/test_compose.py +++ b/deploy/compose/test_compose.py @@ -81,7 +81,7 @@ def test_compose_uses_private_services_and_ordered_initialization(self): self.assertEqual(services['web']['healthcheck']['test'], ['CMD', '/usr/local/bin/oac-web', 'healthcheck']) self.assertNotIn('python3', json.dumps(self.compose)) self.assertEqual(services['init']['environment']['OAC_REVISION'], 'd' * 40) - for name in ('OAC_SANDBOX_MAX_ACTIVE', 'OAC_SANDBOX_MAX_RETAINED', 'OAC_EXECUTION_CONCURRENCY', 'OAC_DEFAULT_HARNESS', 'OAC_HARNESSES', 'OAC_WRITE_AUDIT_RETENTION', 'OAC_LOG_LEVEL'): + for name in ('OAC_SANDBOX_MAX_ACTIVE', 'OAC_SANDBOX_MAX_RETAINED', 'OAC_EXECUTION_CONCURRENCY', 'OAC_TURN_AUTO_RETRIES', 'OAC_DEFAULT_HARNESS', 'OAC_HARNESSES', 'OAC_WRITE_AUDIT_RETENTION', 'OAC_LOG_LEVEL'): self.assertEqual(services['core']['environment'][name], '', name) def test_direct_sandbox_capacity_reaches_core(self): diff --git a/deploy/kubernetes/README.md b/deploy/kubernetes/README.md index 54ccfd467..f7842e566 100644 --- a/deploy/kubernetes/README.md +++ b/deploy/kubernetes/README.md @@ -64,7 +64,7 @@ Restrict the repository’s `production` GitHub Environment to deployments from | `OAC_K8S_INSECURE` | `true` skips API server certificate verification, for an address the certificate does not name. Leave it unset otherwise | | `OAC_DEPLOY_RUNNER` | Runner label; `ubuntu-22.04` by default. Set this one at repository level: the job that pins the commit runs before the environment is entered, so an environment-scoped value does not reach it | | `OAC_CORE_CPU_REQUEST`, `OAC_CORE_CPU_LIMIT`, `OAC_CORE_MEMORY_REQUEST`, `OAC_CORE_MEMORY_LIMIT` | Core's resources; `500m`, `2`, `1Gi` and `4Gi` by default | -| `OAC_HARNESSES`, `OAC_DEFAULT_HARNESS`, `OAC_EXECUTION_CONCURRENCY`, `OAC_WRITE_AUDIT_RETENTION`, `OAC_LOG_LEVEL` | The matching [process settings](../../docs/configuration.md#settings), which define their values and defaults | +| `OAC_HARNESSES`, `OAC_DEFAULT_HARNESS`, `OAC_EXECUTION_CONCURRENCY`, `OAC_TURN_AUTO_RETRIES`, `OAC_WRITE_AUDIT_RETENTION`, `OAC_LOG_LEVEL` | The matching [process settings](../../docs/configuration.md#settings), which define their values and defaults | A change to any of these takes effect on the next run: a digest of the environment and the secrets enters both Pod templates, so the Pods are replaced even when the images are unchanged. diff --git a/docs/configuration.md b/docs/configuration.md index a6f1788de..06d7649c3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -53,6 +53,7 @@ Model providers are not process settings; see [Default models](#default-models). | `OAC_EXECUTION_CONCURRENCY` | `4` | Concurrent execution work, from 1 to 1024 | | `OAC_SANDBOX_MAX_ACTIVE` | `100` | Active sandbox limit for direct Providers with suspension, from 1 to 100000. Independent of execution concurrency and node capacity | | `OAC_SANDBOX_MAX_RETAINED` | `400` | Retained sandbox limit for direct Providers, from 1 to 100000, including active, suspended and unconfirmed cleanup. Must be at least the active limit | +| `OAC_TURN_AUTO_RETRIES` | `0` | Automatic retries of a Turn that failed on a recoverable model or tool error, from 0 to 5. `0` turns it off | | `OAC_DEFAULT_HARNESS` | `codex` | Harness used when a request does not name one | | `OAC_HARNESSES` | Every registered Harness | Comma-separated Harnesses to enable besides the default one. Unknown names stop startup | | `OAC_WRITE_AUDIT_RETENTION` | `2160h` | Minimum `1h` | @@ -272,7 +273,7 @@ Core reads its process environment. Compose interpolates `.env` into it and moun | `OAC_CREDENTIAL_KEY_FILE` | Required. `/run/oac/credential.key`: a base64-encoded random 32-byte key. Core seals stored credentials with it | | `OAC_CORE_KEY_DIGESTS_FILE` | Required. `/run/oac/core-key-digests.json`: a JSON array with the SHA-256 of the Core key | | `OAC_INSTALLATION_ID_FILE` | Required. `/run/oac/installation.id`: the installation ID, a canonical UUID. Core refuses an ID other than the one its database recorded | -| `OAC_SANDBOX_MAX_ACTIVE`, `OAC_SANDBOX_MAX_RETAINED`, `OAC_EXECUTION_CONCURRENCY`, `OAC_DEFAULT_HARNESS`, `OAC_HARNESSES`, `OAC_WRITE_AUDIT_RETENTION`, `OAC_OAUTH_TRUSTED_ORIGINS`, `OAC_HISTORY_SETTINGS_FILE`, `OAC_LOG_LEVEL`, `OAC_LOG_FORMAT`, `OAC_LOG_ADD_SOURCE` | The matching [process settings](#settings). Web reads the three log settings too | +| `OAC_SANDBOX_MAX_ACTIVE`, `OAC_SANDBOX_MAX_RETAINED`, `OAC_EXECUTION_CONCURRENCY`, `OAC_TURN_AUTO_RETRIES`, `OAC_DEFAULT_HARNESS`, `OAC_HARNESSES`, `OAC_WRITE_AUDIT_RETENTION`, `OAC_OAUTH_TRUSTED_ORIGINS`, `OAC_HISTORY_SETTINGS_FILE`, `OAC_LOG_LEVEL`, `OAC_LOG_FORMAT`, `OAC_LOG_ADD_SOURCE` | The matching [process settings](#settings). Web reads the three log settings too | | `OAC_PROVIDER_ROOT` | Absolute adapter artifact root. The Core image sets `/opt/oac`. Each adapter owns its helper paths beneath this root. Core serves self-hosted daemon installers from its `native-installers/` directory when that holds a `catalog.json`, after checking the catalog against its own release. Adapter state lives at `/state`, the data volume's [`state/`](#compose-installations) | Core logs the history file path it loads, never environment values or file contents. diff --git a/docs/zh/configuration.md b/docs/zh/configuration.md index 01ec93be4..cca22759c 100644 --- a/docs/zh/configuration.md +++ b/docs/zh/configuration.md @@ -1,7 +1,7 @@ --- title: "配置参考" source: docs/configuration.md -source_hash: 04cde38505f1f09ea913ab2a966e09bb571350ecca8c6e100dbd2e77f143547f +source_hash: 9b3a2b1e7dcc2e742b252ebd700a088be56416ee0a293e78a79730df026b3576 --- Core 安装的每项设置都恰好只有一个归属位置，分属以下三类： @@ -57,6 +57,7 @@ Core 安装的每项设置都恰好只有一个归属位置，分属以下三类 | `OAC_EXECUTION_CONCURRENCY` | `4` | Concurrent execution work, from 1 to 1024 | | `OAC_SANDBOX_MAX_ACTIVE` | `100` | 启用暂停能力的直接 Provider 的活跃沙箱上限，范围 1–100000；独立于执行并发度和节点容量 | | `OAC_SANDBOX_MAX_RETAINED` | `400` | 直接 Provider 的保留沙箱上限，范围 1–100000，包括活跃、已暂停以及尚未确认清理完成的分配；必须不小于活跃上限 | +| `OAC_TURN_AUTO_RETRIES` | `0` | Turn 因可恢复的模型或工具错误失败后的自动重试次数，0 到 5。`0` 表示关闭 | | `OAC_DEFAULT_HARNESS` | `codex` | Harness used when a request does not name one | | `OAC_HARNESSES` | Every registered Harness | Comma-separated Harnesses to enable besides the default one. Unknown names stop startup | | `OAC_WRITE_AUDIT_RETENTION` | `2160h` | Minimum `1h` | @@ -276,7 +277,7 @@ Core 读取进程环境。Compose 将 `.env` 插值到环境中，并把机密 | `OAC_CREDENTIAL_KEY_FILE` | 必填。`/run/oac/credential.key`：Base64 编码的 32 字节随机密钥。Core 用它加密存储的凭据 | | `OAC_CORE_KEY_DIGESTS_FILE` | 必填。`/run/oac/core-key-digests.json`：一个包含 Core 密钥 SHA-256 的 JSON 数组 | | `OAC_INSTALLATION_ID_FILE` | 必填。`/run/oac/installation.id`：安装 ID，采用规范 UUID 格式。如果 ID 与数据库记录的 ID 不一致，Core 会拒绝它 | -| `OAC_SANDBOX_MAX_ACTIVE`、`OAC_SANDBOX_MAX_RETAINED`、`OAC_EXECUTION_CONCURRENCY`、`OAC_DEFAULT_HARNESS`、`OAC_HARNESSES`、`OAC_WRITE_AUDIT_RETENTION`、`OAC_OAUTH_TRUSTED_ORIGINS`、`OAC_HISTORY_SETTINGS_FILE`、`OAC_LOG_LEVEL`、`OAC_LOG_FORMAT`、`OAC_LOG_ADD_SOURCE` | 对应的[进程设置](#settings)。Web 也读取三个日志设置 | +| `OAC_SANDBOX_MAX_ACTIVE`、`OAC_SANDBOX_MAX_RETAINED`、`OAC_EXECUTION_CONCURRENCY`、`OAC_TURN_AUTO_RETRIES`、`OAC_DEFAULT_HARNESS`、`OAC_HARNESSES`、`OAC_WRITE_AUDIT_RETENTION`、`OAC_OAUTH_TRUSTED_ORIGINS`、`OAC_HISTORY_SETTINGS_FILE`、`OAC_LOG_LEVEL`、`OAC_LOG_FORMAT`、`OAC_LOG_ADD_SOURCE` | 对应的[进程设置](#settings)。Web 也读取三个日志设置 | | `OAC_PROVIDER_ROOT` | 适配器构件的绝对根目录。Core 镜像设置为 `/opt/oac`。每个适配器都拥有此根目录下的辅助路径。当其中的 `native-installers/` 目录包含 `catalog.json` 时，Core 在核对该目录清单与自身发行版后提供自托管守护进程安装程序。适配器状态位于 `/state`，即数据卷的 [`state/`](#compose-installations) | Core 会记录所加载的历史文件路径，但绝不记录环境变量的值或文件内容。 diff --git a/services/core/cmd/server/main.go b/services/core/cmd/server/main.go index 20f2c4b4f..27a697cb5 100644 --- a/services/core/cmd/server/main.go +++ b/services/core/cmd/server/main.go @@ -264,7 +264,7 @@ func run(config processconfig.Config) error { Credentials: vaultService, Observer: modelConfigurationStore, Deployment: deploymentService, DeploymentReader: deploymentStore, Sessions: sessionService, SessionsReader: sessionStore, - ManagedRuntimes: managedNodes.runtime, MaxConcurrentExecutions: config.ExecutionConcurrency} + ManagedRuntimes: managedNodes.runtime, MaxConcurrentExecutions: config.ExecutionConcurrency, TurnAutoRetries: config.TurnAutoRetries} lease, err := pgunit.AcquireLease(ctx, pool) if err != nil { return err diff --git a/services/core/internal/execution/dispatcher.go b/services/core/internal/execution/dispatcher.go index 7a49301cc..8516f2b21 100644 --- a/services/core/internal/execution/dispatcher.go +++ b/services/core/internal/execution/dispatcher.go @@ -56,6 +56,9 @@ type Dispatcher struct { // MaxConcurrentExecutions bounds work admitted by this Core execution owner. // Zero uses DefaultExecutionConcurrency. It is independent of sandbox capacity. MaxConcurrentExecutions int + // TurnAutoRetries is how many times a Turn that failed on a recoverable model + // or tool error is retried automatically. Zero disables retries. + TurnAutoRetries int } // Credentials opens the bearer token of a Session's frozen MCP credential diff --git a/services/core/internal/execution/turn_retry.go b/services/core/internal/execution/turn_retry.go new file mode 100644 index 000000000..dbc00005f --- /dev/null +++ b/services/core/internal/execution/turn_retry.go @@ -0,0 +1,250 @@ +package execution + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" +) + +// MaxTurnAutoRetries bounds OAC_TURN_AUTO_RETRIES; zero turns the feature off. +const MaxTurnAutoRetries = 5 + +const retryPrefix = "[auto-retry " + +// retryBackoff is the wait before attempt n (1-based); later attempts reuse the last value. +var retryBackoff = []time.Duration{5 * time.Second, 15 * time.Second, 30 * time.Second} + +// retryableEngineCodes are the engine failure classes a new attempt can plausibly fix. +// An empty code is an unclassified failure. Authentication, quota, missing model, +// context length and policy failures are never retried. +var retryableEngineCodes = map[string]bool{ + "": true, "invalid_request": true, "server_error": true, "server_overloaded": true, + "rate_limit_exceeded": true, "request_timeout": true, "connection_failed": true, +} + +// retryFingerprints match the lowercased executor error text. Entries without retry +// are failures stored in the executor's native history: a new user message cannot +// remove them, so retrying would only spend tokens. The first match wins. +var retryFingerprints = []struct { + name string + all []string + retry bool + adjust bool // the model can fix it by changing the tool call +}{ + {"encrypted_reasoning", []string{"encrypted content", "could not be verified"}, false, false}, + {"image_unsupported", []string{"do not support image input"}, false, false}, + {"tool_result_image", []string{"tool_result", "input_image"}, false, false}, + {"missing_messages", []string{"field messages is required"}, true, false}, + {"ends_with_model_turn", []string{"requests ending with a model turn"}, true, false}, + {"tool_args", []string{"invalid_tool"}, true, true}, + {"tool_args", []string{"tool_use"}, true, true}, + {"tool_args", []string{"tool input"}, true, true}, + {"tool_args", []string{"invalid arguments"}, true, true}, + {"tool_args", []string{"invalid tool"}, true, true}, +} + +type retryDecision struct { + Retry bool + Fingerprint string + Reason string // why not, for logs + Hint string +} + +// decideRetry judges a failed Turn. attempt is the number of retries already made. +func decideRetry(turn sessions.Turn, attempt, limit int) retryDecision { + skip := func(reason string) retryDecision { return retryDecision{Reason: reason} } + var result Result + switch { + case limit <= 0: + return skip("disabled") + case attempt >= limit: + return skip("attempts_exhausted") + case turn.Status != sessions.TurnFailed: + return skip("not_failed") + case !turn.CancelRequestedAt.IsZero(): + return skip("cancel_requested") + case json.Unmarshal(turn.Outcome, &result) != nil: + return skip("outcome_unreadable") + case result.ErrorCode != "engine_failed": + return skip("error_code_not_retryable") + case !retryableEngineCodes[result.EngineErrorCode]: + return skip("engine_code_not_retryable") + case permanentHTTPStatus(result.EngineHTTPStatus): + return skip("engine_http_status_not_retryable") + } + name := cmpName(result.EngineErrorCode, "unclassified") + adjust := false + text := strings.ToLower(result.Error) + for _, fp := range retryFingerprints { + matched := true + for _, part := range fp.all { + matched = matched && strings.Contains(text, part) + } + if !matched { + continue + } + if !fp.retry { + return skip("unrecoverable_history:" + fp.name) + } + name, adjust = fp.name, fp.adjust + break + } + return retryDecision{Retry: true, Fingerprint: name, Hint: retryHint(attempt+1, adjust)} +} + +// permanentHTTPStatus reports upstream statuses a new attempt cannot fix, which beta +// can carry on a connection_failed classification. +func permanentHTTPStatus(status *int) bool { + if status == nil { + return false + } + switch *status { + case 401, 402, 403, 404: + return true + } + return false +} + +func cmpName(value, fallback string) string { + if value == "" { + return fallback + } + return value +} + +// retryHint escalates with the attempt number. It never carries the original error text. +func retryHint(attempt int, adjust bool) string { + switch { + case attempt <= 1 && adjust: + return "Check the parameters of your last tool call and correct them before calling it again." + case attempt == 2: + return "If the same error happens again, complete the step a different way and do not repeat the same tool call arguments." + case attempt >= 3: + return "Do not call the tool that just failed again; use the information you already have to give a result." + } + return "" +} + +func retryMessage(attempt, limit int, d retryDecision) string { + text := fmt.Sprintf("%s%d/%d] The previous attempt failed (%s). Continue the task from where you stopped.", retryPrefix, attempt, limit, d.Fingerprint) + if d.Hint != "" { + text += "\n" + d.Hint + } + return text +} + +// retryAttempt reads how many retries led to this Turn from its first input. +func retryAttempt(inputs []sessions.TurnInput) int { + for _, input := range inputs { + var payload struct { + Input []struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + } `json:"input"` + } + if input.Kind != "message" || json.Unmarshal(input.Payload, &payload) != nil { + continue + } + for _, message := range payload.Input { + for _, part := range message.Content { + var n, m int + if strings.HasPrefix(part.Text, retryPrefix) { + if _, err := fmt.Sscanf(part.Text, retryPrefix+"%d/%d]", &n, &m); err == nil && n > 0 { + return n + } + } + return 0 + } + } + } + return 0 +} + +// retryHost is what a retry needs from Core; Worker plus the Sessions reader provide it. +type retryHost interface { + ListTurns(ctx context.Context, tenant, session, cursor string, limit int, ascending bool) (sessions.TurnPage, error) + ListTurnInputs(ctx context.Context, tenant, session, turn string, after int64, limit int) ([]sessions.TurnInput, error) + SubmitInputs(ctx context.Context, tenant, session, key string, inputs []sessions.Input) ([]sessions.InputReceipt, error) +} + +type workerRetryHost struct { + sessions.Reader + worker *Worker +} + +func (h workerRetryHost) SubmitInputs(ctx context.Context, tenant, session, key string, inputs []sessions.Input) ([]sessions.InputReceipt, error) { + return h.worker.SubmitInputs(ctx, tenant, session, key, inputs) +} + +// maybeAutoRetry schedules a retry for a Turn that just failed. It never blocks the caller. +func (w *Worker) maybeAutoRetry(ctx context.Context, tenant string, turn sessions.Turn) { + limit := w.dispatcher.TurnAutoRetries + if limit <= 0 || turn.Status != sessions.TurnFailed { + return + } + host := workerRetryHost{Reader: w.dispatcher.SessionsReader, worker: w} + go autoRetry(ctx, host, tenant, turn, limit, sleepContext) +} + +func sleepContext(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +// autoRetry waits, rechecks that the failed Turn is still the newest and not +// cancelled, then submits one retry message. The idempotency key is derived from +// the failed Turn, so a repeated call cannot start a second Turn. +func autoRetry(ctx context.Context, host retryHost, tenant string, turn sessions.Turn, limit int, wait func(context.Context, time.Duration) error) { + logger := log.Ctx(ctx) + read, cancel := context.WithTimeout(ctx, 10*time.Second) + inputs, err := host.ListTurnInputs(read, tenant, turn.SessionID, turn.ID, 0, 5) + cancel() + if err != nil { + logger.Warn("oac-core turn auto-retry skipped", "session_id", turn.SessionID, "turn_id", turn.ID, "decision", "skip:inputs_unavailable") + return + } + attempt := retryAttempt(inputs) + decision := decideRetry(turn, attempt, limit) + fields := []any{"session_id", turn.SessionID, "turn_id", turn.ID, "attempt", attempt + 1, "max", limit, "fingerprint", decision.Fingerprint} + if !decision.Retry { + logger.Info("oac-core turn auto-retry skipped", append(fields, "decision", "skip:"+decision.Reason)...) + return + } + if wait(ctx, retryBackoff[min(attempt, len(retryBackoff)-1)]) != nil { + return + } + submit, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + page, err := host.ListTurns(submit, tenant, turn.SessionID, "", 1, false) + if err != nil || len(page.Turns) != 1 || page.Turns[0].ID != turn.ID { + logger.Info("oac-core turn auto-retry skipped", append(fields, "decision", "skip:newer_turn_or_session_gone")...) + return + } + if !decideRetry(page.Turns[0], attempt, limit).Retry { + logger.Info("oac-core turn auto-retry skipped", append(fields, "decision", "skip:cancel_requested")...) + return + } + payload, _ := json.Marshal(map[string]any{ + "type": "agent.session.input.message", + "input": []any{map[string]any{"role": "user", "content": []any{ + map[string]any{"type": "input_text", "text": retryMessage(attempt+1, limit, decision)}, + }}}, + }) + if _, err := host.SubmitInputs(submit, tenant, turn.SessionID, "auto-retry:"+turn.ID, []sessions.Input{{Kind: "message", Payload: payload}}); err != nil { + logger.Warn("oac-core turn auto-retry submit failed", append(fields, "decision", "skip:submit_failed")...) + return + } + logger.Info("oac-core turn auto-retry submitted", append(fields, "decision", "retry")...) +} diff --git a/services/core/internal/execution/turn_retry_test.go b/services/core/internal/execution/turn_retry_test.go new file mode 100644 index 000000000..9476842c1 --- /dev/null +++ b/services/core/internal/execution/turn_retry_test.go @@ -0,0 +1,202 @@ +package execution + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" +) + +func failedTurn(t *testing.T, result Result) sessions.Turn { + t.Helper() + outcome, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + return sessions.Turn{ID: "turn-1", SessionID: "session-1", Status: sessions.TurnFailed, Outcome: outcome} +} + +func engineFailure(code, text string) Result { + return Result{ErrorCode: "engine_failed", EngineErrorCode: code, Error: text} +} + +func TestDecideRetryClasses(t *testing.T) { + for _, tc := range []struct { + name string + result Result + retry bool + fp string + }{ + {"unclassified", engineFailure("", "claudesdk: execution_failed"), true, "unclassified"}, + {"invalid request", engineFailure("invalid_request", "boom"), true, "invalid_request"}, + {"server error", engineFailure("server_error", ""), true, "server_error"}, + {"rate limit", engineFailure("rate_limit_exceeded", ""), true, "rate_limit_exceeded"}, + {"connection", engineFailure("connection_failed", ""), true, "connection_failed"}, + {"authentication", engineFailure("authentication_error", ""), false, ""}, + {"usage limit", engineFailure("usage_limit_exceeded", ""), false, ""}, + {"model missing", engineFailure("resource_not_found", ""), false, ""}, + {"context length", engineFailure("context_length_exceeded", ""), false, ""}, + {"cancelled", Result{ErrorCode: "cancelled"}, false, ""}, + {"disconnected", Result{ErrorCode: "device_disconnected"}, false, ""}, + {"A", engineFailure("invalid_request", "400: Field messages is required"), true, "missing_messages"}, + {"E", engineFailure("invalid_request", "Requests ending with a model turn are not supported"), true, "ends_with_model_turn"}, + {"tool args", engineFailure("invalid_request", "tool_use ids were found without tool_result"), true, "tool_args"}, + {"B", engineFailure("invalid_request", "The encrypted content could not be verified"), false, ""}, + {"C", engineFailure("invalid_request", "Models do not support image input"), false, ""}, + {"D", engineFailure("invalid_request", "tool_result contains input_image"), false, ""}, + } { + t.Run(tc.name, func(t *testing.T) { + d := decideRetry(failedTurn(t, tc.result), 0, 3) + if d.Retry != tc.retry || (tc.retry && d.Fingerprint != tc.fp) { + t.Fatalf("%+v", d) + } + }) + } +} + +func TestDecideRetrySkipsPermanentHTTPStatus(t *testing.T) { + for status, retry := range map[int]bool{401: false, 402: false, 403: false, 404: false, 502: true, 429: true} { + status := status + result := engineFailure("connection_failed", "") + result.EngineHTTPStatus = &status + if got := decideRetry(failedTurn(t, result), 0, 3).Retry; got != retry { + t.Errorf("status %d: retry = %v, want %v", status, got, retry) + } + } +} +func TestDecideRetryLimitsAndCancel(t *testing.T) { + turn := failedTurn(t, engineFailure("server_error", "")) + if decideRetry(turn, 0, 0).Retry || decideRetry(turn, 3, 3).Retry || !decideRetry(turn, 2, 3).Retry { + t.Fatal("attempt limits not honored") + } + turn.CancelRequestedAt = time.Now() + if decideRetry(turn, 0, 3).Retry { + t.Fatal("cancel must win") + } + turn.CancelRequestedAt = time.Time{} + turn.Status = sessions.TurnCompleted + if decideRetry(turn, 0, 3).Retry { + t.Fatal("completed turn retried") + } +} + +func TestRetryHintEscalatesAndDoesNotLeak(t *testing.T) { + secret := "sk-synthetic-secret-body" + turn := failedTurn(t, engineFailure("invalid_request", "invalid tool arguments: "+secret)) + seen := map[string]bool{} + for attempt := 0; attempt < 3; attempt++ { + d := decideRetry(turn, attempt, 3) + text := retryMessage(attempt+1, 3, d) + if strings.Contains(text, secret) || !strings.HasPrefix(text, "[auto-retry ") { + t.Fatal(text) + } + seen[d.Hint] = true + } + if len(seen) != 3 { + t.Fatalf("hints did not escalate: %v", seen) + } +} + +func messageInputs(text string) []sessions.TurnInput { + payload, _ := json.Marshal(map[string]any{"input": []any{map[string]any{"role": "user", "content": []any{map[string]any{"type": "input_text", "text": text}}}}}) + return []sessions.TurnInput{{Kind: "message", Payload: payload}} +} + +func TestRetryAttemptParsing(t *testing.T) { + if retryAttempt(messageInputs("hello")) != 0 || retryAttempt(nil) != 0 { + t.Fatal("plain input counted as retry") + } + if got := retryAttempt(messageInputs("[auto-retry 2/3] failed")); got != 2 { + t.Fatal(got) + } +} + +type fakeRetryHost struct { + mu sync.Mutex + latest []sessions.Turn + inputs []sessions.TurnInput + submitted map[string]int + keys []string + texts []string + err error +} + +func (f *fakeRetryHost) ListTurns(context.Context, string, string, string, int, bool) (sessions.TurnPage, error) { + return sessions.TurnPage{Turns: f.latest}, f.err +} + +func (f *fakeRetryHost) ListTurnInputs(context.Context, string, string, string, int64, int) ([]sessions.TurnInput, error) { + return f.inputs, nil +} + +func (f *fakeRetryHost) SubmitInputs(_ context.Context, _, _, key string, inputs []sessions.Input) ([]sessions.InputReceipt, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.keys = append(f.keys, key) + f.texts = append(f.texts, string(inputs[0].Payload)) + return nil, nil +} + +func noWait(context.Context, time.Duration) error { return nil } + +func TestAutoRetrySubmitsOnceWithStableKey(t *testing.T) { + turn := failedTurn(t, engineFailure("server_error", "secret-body")) + host := &fakeRetryHost{latest: []sessions.Turn{turn}} + var waited time.Duration + autoRetry(t.Context(), host, "tenant", turn, 3, func(_ context.Context, d time.Duration) error { waited = d; return nil }) + autoRetry(t.Context(), host, "tenant", turn, 3, noWait) + if len(host.keys) != 2 || host.keys[0] != "auto-retry:turn-1" || host.keys[1] != host.keys[0] || waited != retryBackoff[0] { + t.Fatal(host.keys, waited) + } + if strings.Contains(host.texts[0], "secret-body") || !strings.Contains(host.texts[0], "[auto-retry 1/3]") { + t.Fatal(host.texts[0]) + } +} + +func TestAutoRetryBackoffGrowsAndStopsAtLimit(t *testing.T) { + turn := failedTurn(t, engineFailure("server_error", "")) + host := &fakeRetryHost{latest: []sessions.Turn{turn}, inputs: messageInputs("[auto-retry 2/3] x")} + var waited time.Duration + autoRetry(t.Context(), host, "t", turn, 3, func(_ context.Context, d time.Duration) error { waited = d; return nil }) + if waited != retryBackoff[2] || len(host.keys) != 1 || !strings.Contains(host.texts[0], "[auto-retry 3/3]") { + t.Fatal(waited, host.keys) + } + host = &fakeRetryHost{latest: []sessions.Turn{turn}, inputs: messageInputs("[auto-retry 3/3] x")} + autoRetry(t.Context(), host, "t", turn, 3, noWait) + if len(host.keys) != 0 { + t.Fatal("retried past the limit") + } +} + +func TestAutoRetrySkips(t *testing.T) { + turn := failedTurn(t, engineFailure("server_error", "")) + newer := sessions.Turn{ID: "turn-2", Status: sessions.TurnQueued} + cancelled := turn + cancelled.CancelRequestedAt = time.Now() + for name, host := range map[string]*fakeRetryHost{ + "newer turn": {latest: []sessions.Turn{newer}}, + "session gone": {err: errors.New("not found")}, + "cancel after wait": {latest: []sessions.Turn{cancelled}}, + } { + autoRetry(t.Context(), host, "t", turn, 3, noWait) + if len(host.keys) != 0 { + t.Fatal(name) + } + } + host := &fakeRetryHost{latest: []sessions.Turn{turn}} + ctx, cancel := context.WithCancel(t.Context()) + cancel() + autoRetry(ctx, host, "t", turn, 3, sleepContext) + if len(host.keys) != 0 { + t.Fatal("submitted after context cancel") + } + autoRetry(t.Context(), host, "t", turn, 0, noWait) + if len(host.keys) != 0 { + t.Fatal("disabled switch still retried") + } +} diff --git a/services/core/internal/execution/worker.go b/services/core/internal/execution/worker.go index f6d1ef913..78d00f92f 100644 --- a/services/core/internal/execution/worker.go +++ b/services/core/internal/execution/worker.go @@ -389,8 +389,12 @@ func (w *Worker) reconcile(ctx context.Context) error { } func (w *Worker) runClaim(ctx context.Context, item sessions.ExecutionWork) error { - _, err := w.dispatcher.Run(ctx, item.TenantID, item.SessionID, item.TurnID) - if err == nil || errors.Is(err, sessions.ErrTurnConflict) { + ran, err := w.dispatcher.Run(ctx, item.TenantID, item.SessionID, item.TurnID) + if err == nil { + w.maybeAutoRetry(ctx, item.TenantID, ran) + return nil + } + if errors.Is(err, sessions.ErrTurnConflict) { return nil } var rejection *preparationRejection diff --git a/services/core/internal/execution/worker_schedule.go b/services/core/internal/execution/worker_schedule.go index aeb7f8845..c7a6307c4 100644 --- a/services/core/internal/execution/worker_schedule.go +++ b/services/core/internal/execution/worker_schedule.go @@ -113,6 +113,7 @@ func (w *Worker) runEnvironmentInput(ctx context.Context, item scheduledWork) er w.schedulingObservations.finish(ctx, item.observation, outcome) } if err == nil { + w.maybeAutoRetry(ctx, item.TenantID, run.Turn) return nil } if errors.Is(err, errPreparationFailed) && run.Reservation.State == sessions.EnvironmentInputPending { diff --git a/services/core/internal/processconfig/config.go b/services/core/internal/processconfig/config.go index 11cd644ab..2dcf27a91 100644 --- a/services/core/internal/processconfig/config.go +++ b/services/core/internal/processconfig/config.go @@ -57,6 +57,7 @@ type Config struct { // CoreKeys authenticates the Core key from OAC_CORE_KEY_DIGESTS_FILE. CoreKeys *api.DeploymentAuthenticator ExecutionConcurrency int + TurnAutoRetries int DefaultHarness string // Harnesses is the sorted set of enabled Harnesses, the default included. Harnesses []string @@ -135,6 +136,9 @@ func Load() (Config, error) { if c.ExecutionConcurrency, err = executionConcurrency(); err != nil { return Config{}, err } + if c.TurnAutoRetries, err = turnAutoRetries(); err != nil { + return Config{}, err + } c.DefaultHarness = cmp.Or(os.Getenv("OAC_DEFAULT_HARNESS"), defaultHarness) if _, known := (engine.Catalog{}).Lookup(c.DefaultHarness); !known { return Config{}, configError("OAC_DEFAULT_HARNESS is not a known harness") @@ -176,6 +180,7 @@ func (c Config) Settings() []api.InstallationSetting { setting("OAC_LOG_FORMAT", format, "auto", []api.InstallationService{api.InstallationCore, api.InstallationWeb}), setting("OAC_LOG_ADD_SOURCE", c.Log.AddSource, false, []api.InstallationService{api.InstallationCore, api.InstallationWeb}), setting("OAC_EXECUTION_CONCURRENCY", c.ExecutionConcurrency, execution.DefaultExecutionConcurrency, []api.InstallationService{api.InstallationCore}), + setting("OAC_TURN_AUTO_RETRIES", c.TurnAutoRetries, 0, []api.InstallationService{api.InstallationCore}), setting("OAC_HARNESSES", c.Harnesses, (engine.Catalog{}).Kinds(), []api.InstallationService{api.InstallationCore}), setting("OAC_DEFAULT_HARNESS", c.DefaultHarness, defaultHarness, []api.InstallationService{api.InstallationCore}), setting("OAC_WRITE_AUDIT_RETENTION", duration(c.WriteAuditRetention), duration(defaultWriteAuditRetention), []api.InstallationService{api.InstallationCore}), @@ -271,6 +276,18 @@ func executionConcurrency() (int, error) { return limit, nil } +func turnAutoRetries() (int, error) { + value := os.Getenv("OAC_TURN_AUTO_RETRIES") + if value == "" { + return 0, nil + } + limit, err := strconv.Atoi(value) + if err != nil || limit < 0 || limit > execution.MaxTurnAutoRetries { + return 0, configError("OAC_TURN_AUTO_RETRIES must be an integer between 0 and 5") + } + return limit, nil +} + // harnesses reads OAC_HARNESSES. Unset enables every Harness this build // supports; a list supplements the default Harness. func harnesses(defaultEngine string) ([]string, error) { diff --git a/services/core/internal/processconfig/config_test.go b/services/core/internal/processconfig/config_test.go index 931c414f3..c745f3bb8 100644 --- a/services/core/internal/processconfig/config_test.go +++ b/services/core/internal/processconfig/config_test.go @@ -231,3 +231,20 @@ func TestSettingsReportEffectiveValuesAndHideHistory(t *testing.T) { t.Fatal(found) } } + +func TestTurnAutoRetries(t *testing.T) { + required(t) + if c, err := Load(); err != nil || c.TurnAutoRetries != 0 { + t.Fatal(c.TurnAutoRetries, err) + } + for value, want := range map[string]int{"0": 0, "3": 3, "5": 5} { + t.Setenv("OAC_TURN_AUTO_RETRIES", value) + if c, err := Load(); err != nil || c.TurnAutoRetries != want { + t.Fatal(value, c.TurnAutoRetries, err) + } + } + for _, value := range []string{"-1", "6", "1.5", "synthetic-secret"} { + t.Setenv("OAC_TURN_AUTO_RETRIES", value) + rejects(t, "OAC_TURN_AUTO_RETRIES", "synthetic-secret") + } +}