From fda569210cfd0c60266a2caddb2f52cb6b9ab8ec Mon Sep 17 00:00:00 2001 From: Codex DX Agent Date: Tue, 14 Jul 2026 22:08:52 +0200 Subject: [PATCH 1/2] fix(mcp): read context text from raw responses (6003e3bb) --- apps/mcp/src/commands/contexts.ts | 29 +++-- apps/mcp/src/lib/api.ts | 26 +++++ apps/mcp/test/contexts-cli.test.js | 10 +- apps/mcp/test/contexts-read.test.js | 167 ++++++++++++++++++++++++++++ 4 files changed, 222 insertions(+), 10 deletions(-) create mode 100644 apps/mcp/test/contexts-read.test.js diff --git a/apps/mcp/src/commands/contexts.ts b/apps/mcp/src/commands/contexts.ts index 1e29616d7..9ff3187e2 100644 --- a/apps/mcp/src/commands/contexts.ts +++ b/apps/mcp/src/commands/contexts.ts @@ -16,7 +16,7 @@ type ContextSummary = { }; type ContextFile = { - path?: string; + path: string; content?: string; is_binary?: boolean; mime_type?: string; @@ -25,6 +25,10 @@ type ContextFile = { note?: string; }; +type ContextDetail = ContextSummary & { + files?: ContextFile[]; +}; + function encodePath(path: string): string { return path.split("/").map(encodeURIComponent).join("/"); } @@ -110,16 +114,23 @@ export async function contextsReadCommand( options: { json?: boolean }, ): Promise { try { + const safeName = requireValue(name, "context name"); + const safePath = requireValue(path, "file path"); const { client } = await createAuthenticatedClient(); - const file = (await client.requestJson("GET", filePath( - requireValue(name, "context name"), - requireValue(path, "file path"), - ))) as ContextFile; + const detail = (await client.requestJson("GET", contextPath(safeName))) as ContextDetail; + const file = detail.files?.find((item) => item.path === safePath); + if (!file) { + log.err(`Context file ${safeName}/${safePath} was not found.`); + return 1; + } + const content = file.is_binary + ? undefined + : await client.requestText("GET", filePath(safeName, safePath)); if (options.json) { - printJson(file); - } else if (typeof file.content === "string" && !file.is_binary) { - process.stdout.write(file.content); - if (!file.content.endsWith("\n")) process.stdout.write("\n"); + printJson(content === undefined ? file : { ...file, content }); + } else if (content !== undefined) { + process.stdout.write(content); + if (!content.endsWith("\n")) process.stdout.write("\n"); } else { log.info(file.note || "Binary file."); if (file.download_url) log.kv("Download", file.download_url); diff --git a/apps/mcp/src/lib/api.ts b/apps/mcp/src/lib/api.ts index ebd27bcd6..419aad606 100644 --- a/apps/mcp/src/lib/api.ts +++ b/apps/mcp/src/lib/api.ts @@ -279,6 +279,32 @@ export class FloomApiClient { return new Uint8Array(await response.arrayBuffer()); } + async requestText(method: string, path: string, options: RequestOptions = {}): Promise { + const auth = options.auth ?? true; + const headers: Record = { + accept: "text/plain, application/json, text/*", + ...(options.headers || {}), + }; + if (auth) { + Object.assign(headers, await this.authHeaders()); + } + const response = await fetchFloom(this.apiBase, buildUrl(this.apiBase, this.resolvePath(path), options.query), { + method, + headers, + }); + if (!response.ok) { + const parsed = await parseResponse(response); + const detail = responseDetail(parsed); + throw new FloomApiError( + `API ${method} ${path} failed with HTTP ${response.status}: ${detail}`, + response.status, + parsed, + response.headers, + ); + } + return response.text(); + } + async uploadFile(inputName: string, filePath: string): Promise { if (!this.credentials) { throw new Error(`Not logged in. Run ${getCommandName()} login first.`); diff --git a/apps/mcp/test/contexts-cli.test.js b/apps/mcp/test/contexts-cli.test.js index 60ae795aa..a122199e5 100644 --- a/apps/mcp/test/contexts-cli.test.js +++ b/apps/mcp/test/contexts-cli.test.js @@ -86,8 +86,16 @@ async function withStubServer(fn) { res.end(JSON.stringify({ name: "crm", files: [] })); return; } + if (req.method === "GET" && req.url === "/contexts/crm") { + res.end(JSON.stringify({ + name: "crm", + files: [{ path: "facts.md", size: 6, mime_type: "text/markdown", is_binary: false }], + })); + return; + } if (req.method === "GET" && req.url === "/contexts/crm/files/facts.md") { - res.end(JSON.stringify({ path: "facts.md", content: "hello\n", is_binary: false })); + res.setHeader("content-type", "text/markdown"); + res.end("hello\n"); return; } if (req.method === "PUT" && req.url === "/contexts/crm/files/facts.md") { diff --git a/apps/mcp/test/contexts-read.test.js b/apps/mcp/test/contexts-read.test.js new file mode 100644 index 000000000..6d2fa576d --- /dev/null +++ b/apps/mcp/test/contexts-read.test.js @@ -0,0 +1,167 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { createServer } from "node:http"; +import { mkdir, mkdtemp, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +async function makeTempHome(apiBase) { + const home = await mkdtemp(join(tmpdir(), "floom-contexts-read-home-")); + const configDir = join(home, ".config", "workeros"); + await mkdir(configDir, { recursive: true }); + await writeFile(join(configDir, "credentials.json"), JSON.stringify({ + api_base: apiBase, + mode: "oss", + api_secret: "test-secret", + authed_at: new Date().toISOString(), + }, null, 2)); + return home; +} + +async function runCli(args, home) { + const child = spawn(process.execPath, ["dist/cli.js", ...args], { + cwd: process.cwd(), + env: { + ...process.env, + HOME: home, + XDG_CONFIG_HOME: join(home, ".config"), + WORKEROS_API_BASE: "", + WORKEROS_API_SECRET: "", + WORKEROS_API_TOKEN: "", + FLOOM_API_BASE: "", + FLOOM_API_SECRET: "", + FLOOM_CLI_TELEMETRY_DISABLED: "1", + }, + stdio: ["ignore", "pipe", "pipe"], + }); + const stdout = []; + const stderr = []; + child.stdout.on("data", (chunk) => stdout.push(chunk)); + child.stderr.on("data", (chunk) => stderr.push(chunk)); + const [code] = await once(child, "exit"); + return { + code, + stdout: Buffer.concat(stdout).toString("utf8"), + stderr: Buffer.concat(stderr).toString("utf8"), + }; +} + +function json(response, status, body) { + response.writeHead(status, { "content-type": "application/json" }); + response.end(JSON.stringify(body)); +} + +async function startMockApi(file, rawBody) { + const seen = []; + const server = createServer((request, response) => { + const url = new URL(request.url || "/", "http://127.0.0.1"); + seen.push(`${request.method} ${url.pathname}`); + if (request.headers["x-floom-secret"] !== "test-secret") { + json(response, 401, { detail: "Unauthorized" }); + return; + } + if (request.method === "GET" && url.pathname === "/contexts/test-pack") { + json(response, 200, { name: "test-pack", files: [file] }); + return; + } + if (request.method === "GET" && url.pathname === `/contexts/test-pack/files/${file.path}`) { + response.writeHead(200, { "content-type": file.mime_type }); + response.end(rawBody); + return; + } + json(response, 404, { detail: "Not found" }); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const address = server.address(); + assert.equal(typeof address, "object"); + return { + server, + seen, + baseUrl: `http://127.0.0.1:${address.port}`, + }; +} + +test("contexts read prints raw JSON text instead of a binary notice", async (t) => { + const content = '{"known_key":"known value"}\n'; + const mock = await startMockApi({ + path: "state.json", + is_binary: false, + mime_type: "application/json", + size: Buffer.byteLength(content), + }, content); + t.after(() => mock.server.close()); + const home = await makeTempHome(mock.baseUrl); + + const result = await runCli(["contexts", "read", "test-pack", "state.json"], home); + + assert.equal(result.code, 0); + assert.match(result.stdout, /"known_key":"known value"/); + assert.doesNotMatch(result.stdout, /Binary file\./); + assert.equal(result.stderr, ""); + assert.deepEqual(mock.seen, [ + "GET /contexts/test-pack", + "GET /contexts/test-pack/files/state.json", + ]); +}); + +test("contexts read prints raw markdown text instead of a binary notice", async (t) => { + const content = "# Known heading\n\nMarkdown body.\n"; + const mock = await startMockApi({ + path: "notes.md", + is_binary: false, + mime_type: "text/markdown", + size: Buffer.byteLength(content), + }, content); + t.after(() => mock.server.close()); + const home = await makeTempHome(mock.baseUrl); + + const result = await runCli(["contexts", "read", "test-pack", "notes.md"], home); + + assert.equal(result.code, 0); + assert.match(result.stdout, /# Known heading/); + assert.match(result.stdout, /Markdown body\./); + assert.doesNotMatch(result.stdout, /Binary file\./); + assert.equal(result.stderr, ""); +}); + +test("contexts read reports genuinely binary files without downloading bytes", async (t) => { + const mock = await startMockApi({ + path: "pixel.png", + is_binary: true, + mime_type: "image/png", + size: 8, + download_url: "https://example.test/download/pixel.png", + }, Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x00, 0xff, 0x01, 0x02])); + t.after(() => mock.server.close()); + const home = await makeTempHome(mock.baseUrl); + + const result = await runCli(["contexts", "read", "test-pack", "pixel.png"], home); + + assert.equal(result.code, 0); + assert.match(result.stdout, /Binary file\./); + assert.match(result.stdout, /https:\/\/example\.test\/download\/pixel\.png/); + assert.match(result.stdout, /image\/png/); + assert.doesNotMatch(result.stdout, /PNG/); + assert.equal(result.stderr, ""); + assert.deepEqual(mock.seen, ["GET /contexts/test-pack"]); +}); + +test("contexts read returns a clear non-zero error when metadata has no matching file", async (t) => { + const mock = await startMockApi({ + path: "other.txt", + is_binary: false, + mime_type: "text/plain", + size: 5, + }, "other"); + t.after(() => mock.server.close()); + const home = await makeTempHome(mock.baseUrl); + + const result = await runCli(["contexts", "read", "test-pack", "missing.txt"], home); + + assert.equal(result.code, 1); + assert.match(result.stderr, /Context file test-pack\/missing\.txt was not found\./); + assert.deepEqual(mock.seen, ["GET /contexts/test-pack"]); +}); From 25a6ca0861881b9dcc84e2655546ed00d669b297 Mon Sep 17 00:00:00 2001 From: Codex DX Agent Date: Tue, 14 Jul 2026 22:22:50 +0200 Subject: [PATCH 2/2] fix(api): isolate transient worker health failures (ab8a68f0) --- apps/api/routers/overview.py | 5 + apps/api/routers/worker_listing.py | 1 + apps/api/run_service.py | 64 +----- apps/api/runner_sandbox/agent_driver.py | 192 ++++++++++++++++-- apps/api/services/retry_classification.py | 63 ++++++ apps/api/services/run_metrics.py | 1 + apps/api/services/worker_serialize.py | 10 +- ...test_worker_health_transient_disconnect.py | 161 +++++++++++++++ 8 files changed, 424 insertions(+), 73 deletions(-) create mode 100644 apps/api/services/retry_classification.py create mode 100644 apps/api/tests/test_worker_health_transient_disconnect.py diff --git a/apps/api/routers/overview.py b/apps/api/routers/overview.py index 18140ccff..374ff2ef1 100644 --- a/apps/api/routers/overview.py +++ b/apps/api/routers/overview.py @@ -37,6 +37,7 @@ _operator_error_message, ) from services.secrets_env import _available_secret_names_for_user +from services.retry_classification import is_infra_retry_error_code from services.worker_access import ( _available_connection_slugs_for_user, _list_operator_workers, @@ -580,6 +581,10 @@ def _is_hidden_worker(row: Dict[str, Any]) -> bool: failure_count = int(latest_failure.get("failure_count") or 0) if worker_id in _consecutive_failure_worker_ids: continue + # One retryable infrastructure blip is not a worker-health problem. + # Multiple failures still surface so genuine clusters remain visible. + if failure_count == 1 and is_infra_retry_error_code(latest_failure.get("error_code")): + continue last_failed_at = latest_failure.get("started_at") or latest_failure.get("completed_at") or latest_failure.get("created_at") cause = _overview_failure_cause(latest_failure) attention_items.append( diff --git a/apps/api/routers/worker_listing.py b/apps/api/routers/worker_listing.py index 37a27cab5..fdb959cf9 100644 --- a/apps/api/routers/worker_listing.py +++ b/apps/api/routers/worker_listing.py @@ -282,6 +282,7 @@ def list_workers( available_secret_names=available_secret_names, last_run_status=last_run.status if last_run else None, has_run=last_run is not None, + last_run_error_code=last_run.error_code if last_run else None, ) triggers = _build_triggers_list(w) diff --git a/apps/api/run_service.py b/apps/api/run_service.py index 8f1847557..024859fe3 100644 --- a/apps/api/run_service.py +++ b/apps/api/run_service.py @@ -236,6 +236,13 @@ def execution_role_enabled() -> bool: _dispatch_terminal_run_alerts, ) from services.db_retry import call_with_deadlock_retry +from services.retry_classification import ( + PERMANENT_RETRY_CATEGORIES as _PERMANENT_RETRY_CATEGORIES, + PERMANENT_RETRY_ERROR_CODES as _PERMANENT_RETRY_ERROR_CODES, + TRANSIENT_RETRY_CATEGORIES as _TRANSIENT_RETRY_CATEGORIES, + TRANSIENT_RETRY_ERROR_CODES as _TRANSIENT_RETRY_ERROR_CODES, + is_infra_retry_error_code as _is_infra_retry_error_code, +) UNKNOWN_RUN_ERROR_CODE = "unknown_error" UNKNOWN_RUN_ERROR_MESSAGE = ( "Run failed before the engine captured a specific failure reason. " @@ -443,59 +450,6 @@ def _do_retry() -> None: t.start() -_PERMANENT_RETRY_ERROR_CODES = { - "cancelled", - "cancelled_before_start", - "cancelled_queued", - "invalid_outputs_shape", - "invalid_worker", - "llm_auth_error", - "llm_model_not_configured", - "llm_quota_exceeded", - "missing_connection", - "missing_required_input", - "missing_secret", - "output_token_limit", - "output_too_large", - "quality_gate_failed", - "schema_violation", - "spend_cap_exceeded", - "token_cap_exceeded", - "user_cancel", - "worker_deleted", - "worker_disabled", - "worker_not_found", -} - -_TRANSIENT_RETRY_ERROR_CODES = { - "agent_runtime_error", - "context_mount_failed", - "e2b_quota_exhausted", - "e2b_sandbox_error", - "llm_provider_error", - "llm_rate_limited", - "interrupted_by_restart", - "mcp_connect_failed", - "orphaned", - "run_abandoned_server_restart", - "run_claimed_without_dispatch", - "timeout", -} - -_PERMANENT_RETRY_CATEGORIES = { - "auth", - "cancelled", - "config", - "quality", - "validation", -} - -_TRANSIENT_RETRY_CATEGORIES = { - "network", - "timeout", -} - - @dataclass(frozen=True) class _RetryDecision: retryable: bool @@ -2131,10 +2085,6 @@ def _dispatch_orphan_timeout_seconds() -> int: return 120 -def _is_infra_retry_error_code(error_code: str | None) -> bool: - return (error_code or "").strip().lower() in _TRANSIENT_RETRY_ERROR_CODES - - def _retryable_driver_max_attempts(error_code: str | None) -> int: raw = os.environ.get("WORKEROS_INFRA_RETRY_MAX_ATTEMPTS", "") default = 3 if _is_infra_retry_error_code(error_code) else 2 diff --git a/apps/api/runner_sandbox/agent_driver.py b/apps/api/runner_sandbox/agent_driver.py index 2669096d7..de349c8eb 100644 --- a/apps/api/runner_sandbox/agent_driver.py +++ b/apps/api/runner_sandbox/agent_driver.py @@ -288,6 +288,48 @@ def _classify_llm_provider_error(exc: BaseException | str) -> str | None: return None +_AGENT_TRANSPORT_ERROR_NAMES = { + "ConnectError", + "ConnectionError", + "NetworkError", + "ReadError", + "RemoteProtocolError", + "TransportError", + "WriteError", +} +_AGENT_TRANSPORT_DISCONNECT_MARKERS = ( + "server disconnected", + "connection reset", + "connection aborted", + "connection terminated", + "remoteprotocolerror", +) + + +def _is_agent_transport_disconnect(exc: BaseException) -> bool: + """Recognize HTTP transport drops across httpx/httpcore/runtime wrappers.""" + + current: BaseException | None = exc + seen: set[int] = set() + while current is not None and id(current) not in seen: + seen.add(id(current)) + module = current.__class__.__module__.lower() + name = current.__class__.__name__ + message = str(current).lower() + if any(marker in message for marker in _AGENT_TRANSPORT_DISCONNECT_MARKERS): + return True + if name in _AGENT_TRANSPORT_ERROR_NAMES and ( + module.startswith("httpx") + or module.startswith("httpcore") + or module.startswith("openai") + or module.startswith("litellm") + ): + return True + cause = current.__cause__ or current.__context__ + current = cause if isinstance(cause, BaseException) else None + return False + + def _llm_error_message(error_code: str, model: str | None = None) -> str: if error_code == "llm_auth_error": return "The configured AI provider credentials were rejected. Update the platform model credentials and retry." @@ -461,6 +503,21 @@ def run( error_code="output_token_limit", retryable=False, ) + if _is_agent_transport_disconnect(exc): + redacted = _redact_provider_message(exc_str, secrets) + logger.warning( + "Agent runtime transport disconnected for worker %s run %s: %s", + worker_id, + run_id, + redacted, + ) + log_fn(f"Agent runtime transport disconnected: {redacted}", "error") + return WorkerResult( + status="error", + error="The agent runtime transport disconnected. The run can be retried.", + error_code="agent_runtime_disconnected", + retryable=True, + ) llm_error_code = _classify_llm_provider_error(exc) if llm_error_code: redacted = _redact_provider_message(exc_str, secrets) @@ -1697,6 +1754,7 @@ def _handle_tool( timeout_seconds=timeout_seconds, user_id=user_id, run_id=run_id, + log_fn=log_fn, ) if name == "remember_learning": return self._remember_learning(args, context_dir, config) @@ -1909,6 +1967,7 @@ def _run_command( timeout_seconds: int, user_id: str | None = None, run_id: str | None = None, + log_fn: Callable[[str, str], None] | None = None, ) -> Dict[str, Any]: cmd = str(args.get("cmd") or "") if not cmd: @@ -1988,6 +2047,9 @@ def _run_command( input_dir=input_dir, output_dir=output_dir, secrets=secrets, + config=config, + user_id=user_id, + log_fn=log_fn, ) # #1000: the local-subprocess path (runner != e2b) runs on the API # HOST, so it is a shell-injection surface (unlike E2B, where @@ -2041,25 +2103,87 @@ def _run_command_e2b( input_dir: Path, output_dir: Path, secrets: Dict[str, str], + config: WorkerConfig, + user_id: str | None = None, + log_fn: Callable[[str, str], None] | None = None, ) -> Dict[str, Any]: api_key = os.environ.get("E2B_API_KEY") if not api_key: return {"ok": False, "error": "E2B_API_KEY is not configured"} from e2b import Sandbox - from runner_sandbox.e2b_driver import _e2b_network_policy + from runner_sandbox.e2b_driver import ( + _WarmSandboxEntry, + _e2b_network_policy, + _e2b_template_for_run, + _warm_pool_enabled, + _warm_pool_key, + _warm_pool_lease, + _warm_pool_return, + ) - sandbox = Sandbox.create( - api_key=api_key, - timeout=max(timeout + 60, 180), - network=_e2b_network_policy(None, api_url=os.environ.get("WORKEROS_API_URL")), + pool_log = log_fn or ( + lambda message, level="info": logger.log( + logging.WARNING if level in {"warning", "error"} else logging.DEBUG, + message, + ) ) + sandbox_template: str | None = None + warm_key: str | None = None + if _warm_pool_enabled(): + sandbox_template, _bundle_baked = _e2b_template_for_run( + bundle_dir, + config, + log_fn=pool_log, + ) + base_key, key_error = _warm_pool_key( + worker_id=str(getattr(config, "id", None) or bundle_dir.name), + user_id=user_id, + worker_dir=bundle_dir, + config=config, + inputs={}, + secrets=secrets, + sandbox_template=sandbox_template, + ) + if key_error: + return {"ok": False, "error": key_error} + if base_key: + # Agent command sandboxes have a different directory lifecycle + # from script-driver sandboxes, so they must not share entries. + warm_key = f"agent-command:{base_key}" + + warm_entry = _warm_pool_lease(warm_key or "", log_fn=pool_log) if warm_key else None + if warm_entry is not None: + sandbox = warm_entry.sandbox + else: + network_config = config if warm_key else None + create_kwargs: Dict[str, Any] = { + "api_key": api_key, + "timeout": max(timeout + 60, 180), + "network": _e2b_network_policy( + network_config, + api_url=os.environ.get("WORKEROS_API_URL"), + ), + } + if sandbox_template and warm_key: + create_kwargs["template"] = sandbox_template + sandbox = Sandbox.create(**create_kwargs) + keep_warm = False + remote_bundle = "/home/user/worker" + remote_inputs = "/home/user/inputs" + remote_outputs = "/home/user/outputs" try: - remote_bundle = "/home/user/worker" - remote_inputs = "/home/user/inputs" - remote_outputs = "/home/user/outputs" - for remote_dir in (remote_bundle, remote_inputs, remote_outputs): - sandbox.files.make_dir(remote_dir) + if warm_entry is not None: + if not self._reset_agent_command_sandbox( + sandbox, + remote_bundle, + remote_inputs, + remote_outputs, + ): + raise RuntimeError("Warm agent command sandbox cleanup failed") + else: + for remote_dir in (remote_bundle, remote_inputs, remote_outputs): + sandbox.files.make_dir(remote_dir) self._upload_tree(sandbox, bundle_dir, remote_bundle) self._upload_tree(sandbox, input_dir, remote_inputs) self._upload_tree(sandbox, output_dir, remote_outputs) @@ -2078,17 +2202,55 @@ def _run_command_e2b( envs=env, timeout=float(timeout), ) - return { + response = { "ok": proc.exit_code == 0, "exit_code": proc.exit_code, "stdout": _truncate(_scrub(proc.stdout or "", secrets), _STDOUT_CAP), "stderr": _truncate(_scrub(proc.stderr or "", secrets), _STDERR_CAP), } + keep_warm = bool(warm_key) and self._reset_agent_command_sandbox( + sandbox, + remote_bundle, + remote_inputs, + remote_outputs, + ) + return response finally: - try: - sandbox.kill() - except Exception as exc: - logger.debug("E2B sandbox already gone (kill suppressed): %s", exc) + pooled = False + if keep_warm and warm_key: + entry = warm_entry or _WarmSandboxEntry( + key=warm_key, + sandbox=sandbox, + workdir=remote_bundle, + mounted_contexts=set(), + ) + pooled = _warm_pool_return(entry, log_fn=pool_log) + if not pooled: + try: + sandbox.kill() + except Exception as exc: + logger.debug("E2B sandbox already gone (kill suppressed): %s", exc) + + def _reset_agent_command_sandbox( + self, + sandbox: Any, + remote_bundle: str, + remote_inputs: str, + remote_outputs: str, + ) -> bool: + """Clear per-call files before an agent command sandbox is pooled.""" + + command = " && ".join( + f"mkdir -p {self._shell_quote(path)} && " + f"find {self._shell_quote(path)} -mindepth 1 -delete" + for path in (remote_bundle, remote_inputs, remote_outputs) + ) + try: + result = sandbox.commands.run(command, timeout=30, request_timeout=45) + except Exception: + logger.debug("Agent command sandbox cleanup failed", exc_info=True) + return False + return getattr(result, "exit_code", 1) == 0 def _upload_tree(self, sandbox: Any, local_root: Path, remote_root: str) -> None: if not local_root.exists(): diff --git a/apps/api/services/retry_classification.py b/apps/api/services/retry_classification.py new file mode 100644 index 000000000..ff9e09f99 --- /dev/null +++ b/apps/api/services/retry_classification.py @@ -0,0 +1,63 @@ +"""Shared run retry and transient-infrastructure classification.""" + +from __future__ import annotations + + +PERMANENT_RETRY_ERROR_CODES = frozenset({ + "cancelled", + "cancelled_before_start", + "cancelled_queued", + "invalid_outputs_shape", + "invalid_worker", + "llm_auth_error", + "llm_model_not_configured", + "llm_quota_exceeded", + "missing_connection", + "missing_required_input", + "missing_secret", + "output_token_limit", + "output_too_large", + "quality_gate_failed", + "schema_violation", + "spend_cap_exceeded", + "token_cap_exceeded", + "user_cancel", + "worker_deleted", + "worker_disabled", + "worker_not_found", +}) + +TRANSIENT_RETRY_ERROR_CODES = frozenset({ + "agent_runtime_disconnected", + "agent_runtime_error", + "context_mount_failed", + "e2b_quota_exhausted", + "e2b_sandbox_error", + "llm_provider_error", + "llm_rate_limited", + "interrupted_by_restart", + "mcp_connect_failed", + "orphaned", + "run_abandoned_server_restart", + "run_claimed_without_dispatch", + "timeout", +}) + +PERMANENT_RETRY_CATEGORIES = frozenset({ + "auth", + "cancelled", + "config", + "quality", + "validation", +}) + +TRANSIENT_RETRY_CATEGORIES = frozenset({ + "network", + "timeout", +}) + + +def is_infra_retry_error_code(error_code: str | None) -> bool: + """Return whether a structured failure code represents transient infra.""" + + return (error_code or "").strip().lower() in TRANSIENT_RETRY_ERROR_CODES diff --git a/apps/api/services/run_metrics.py b/apps/api/services/run_metrics.py index 2196405de..05a977511 100644 --- a/apps/api/services/run_metrics.py +++ b/apps/api/services/run_metrics.py @@ -62,6 +62,7 @@ "tool_iteration_cap_exceeded": "timeout", # process / sandbox crashes "agent_runtime_error": "crash", + "agent_runtime_disconnected": "network", "run_execution_exception": "crash", "execution_error": "crash", "executor_thread_pre_sandbox_exception": "crash", diff --git a/apps/api/services/worker_serialize.py b/apps/api/services/worker_serialize.py index 7749543b8..f643b995f 100644 --- a/apps/api/services/worker_serialize.py +++ b/apps/api/services/worker_serialize.py @@ -29,6 +29,7 @@ from core.urls import _frontend_base_url from services.public_view import _sanitize_operator_text from services.run_serialize import _make_run_summary +from services.retry_classification import is_infra_retry_error_code from services.secrets_env import _available_secret_names_for_user from services.worker_access import ( _normalize_trigger_type, @@ -307,6 +308,7 @@ def _resolve_worker_status( available_secret_names: Iterable[str], last_run_status: Optional[RunStatus], has_run: bool, + last_run_error_code: Optional[str] = None, ) -> WorkerStatus: """Single source of truth for an operator-facing worker status. @@ -315,7 +317,7 @@ def _resolve_worker_status( same worker. The full honesty downgrade ladder, in order: 1. MISSING_SECRET — a required secret is not configured. - 2. NEEDS_ATTENTION — the most recent run FAILED. + 2. NEEDS_ATTENTION — the most recent run FAILED for a non-infra reason. 3. NEEDS_ATTENTION — the worker is durably disabled (``enabled is False``, e.g. smoke-gated on creation). A disabled worker is broken, not healthy. 4. READY — the worker has never run, so "healthy" (which implies a @@ -348,6 +350,7 @@ def _resolve_worker_status( not is_archived and status == WorkerStatus.HEALTHY and last_run_status == RunStatus.FAILED + and not is_infra_retry_error_code(last_run_error_code) ): status = WorkerStatus.NEEDS_ATTENTION @@ -659,6 +662,7 @@ def _build_worker_detail( available_secret_names=available_secret_names, last_run_status=None, has_run=False, + last_run_error_code=None, ) _det_missing_secrets = [] _det_missing_connections = [] @@ -671,6 +675,10 @@ def _build_worker_detail( available_secret_names=available_secret_names, last_run_status=recent_runs[0].status if recent_runs else None, has_run=bool(recent_runs), + # list_for_worker is newest-first. A successful retry therefore + # clears health naturally; while queued, its transient parent is + # exempt from the failed-run downgrade above. + last_run_error_code=recent_runs[0].error_code if recent_runs else None, ) # #556: compute specific missing items. _det_req_secrets = _worker_required_secret_names(worker) if config else [] diff --git a/apps/api/tests/test_worker_health_transient_disconnect.py b/apps/api/tests/test_worker_health_transient_disconnect.py new file mode 100644 index 000000000..715c3effb --- /dev/null +++ b/apps/api/tests/test_worker_health_transient_disconnect.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import sys +from pathlib import Path +from types import SimpleNamespace + +import httpx +import pytest + + +API_DIR = Path(__file__).resolve().parents[1] +if str(API_DIR) not in sys.path: + sys.path.insert(0, str(API_DIR)) + +import run_service # noqa: E402 +from models import RunStatus, WorkerStatus # noqa: E402 +from runner_sandbox.agent_driver import AgentDriver # noqa: E402 +from services.worker_serialize import _resolve_worker_status # noqa: E402 + + +def _status(last_run_status: RunStatus, error_code: str | None) -> WorkerStatus: + return _resolve_worker_status( + {"status": "healthy", "enabled": True, "archived": False}, + config=None, + available_secret_names=set(), + last_run_status=last_run_status, + has_run=True, + last_run_error_code=error_code, + ) + + +@pytest.mark.parametrize("error_code", ["agent_runtime_disconnected", "agent_runtime_error"]) +def test_transient_infra_failure_does_not_downgrade_worker_health(error_code): + assert _status(RunStatus.FAILED, error_code) == WorkerStatus.HEALTHY + + +@pytest.mark.parametrize("error_code", ["schema_violation", "missing_secret", "worker_reported_error"]) +def test_genuine_failure_still_downgrades_worker_health(error_code): + assert _status(RunStatus.FAILED, error_code) == WorkerStatus.NEEDS_ATTENTION + + +def test_successful_retry_as_latest_run_resolves_healthy(): + assert _status(RunStatus.COMPLETED, None) == WorkerStatus.HEALTHY + + +def test_httpx_disconnect_gets_distinct_retryable_infra_code(monkeypatch): + driver = AgentDriver() + + def _raise(coro): + coro.close() + raise httpx.RemoteProtocolError("Server disconnected without sending a response") + + monkeypatch.setattr(driver, "_run_coro_sync", _raise) + result = driver.run( + worker_id="disconnect-worker", + run_id="run-disconnect", + inputs={}, + secrets={}, + log_fn=lambda *_args, **_kwargs: None, + trace_id="trace-disconnect", + ) + + assert result.status == "error" + assert result.error_code == "agent_runtime_disconnected" + assert result.retryable is True + decision = run_service._classify_retry_failure( + error_code=result.error_code, + error=result.error, + result_retryable=result.retryable, + ) + assert decision.retryable is True + assert decision.reason == "transient_failure" + assert run_service._is_infra_retry_error_code(result.error_code) is True + + +class _CommandResult: + exit_code = 0 + stdout = "ok\n" + stderr = "" + + +class _FakeFiles: + def __init__(self): + self.existing: set[str] = set() + + def make_dir(self, path): + self.existing.add(path) + + def exists(self, path, **_kwargs): + return path in self.existing + + +class _FakeCommands: + def __init__(self): + self.commands: list[str] = [] + + def run(self, command, **_kwargs): + self.commands.append(command) + return _CommandResult() + + +class _FakeSandbox: + create_calls = 0 + + def __init__(self): + self.files = _FakeFiles() + self.commands = _FakeCommands() + self.killed = False + + @classmethod + def create(cls, **_kwargs): + cls.create_calls += 1 + return cls() + + def kill(self): + self.killed = True + + +def test_agent_run_command_uses_existing_warm_pool_when_enabled(monkeypatch, tmp_path): + import runner_sandbox.e2b_driver as e2b_driver + + e2b_driver.clear_warm_pool() + _FakeSandbox.create_calls = 0 + monkeypatch.setenv("E2B_API_KEY", "test-key") + monkeypatch.setenv("WORKEROS_E2B_WARM_POOL_ENABLED", "1") + monkeypatch.setitem(sys.modules, "e2b", SimpleNamespace(Sandbox=_FakeSandbox)) + monkeypatch.setattr(e2b_driver, "_warm_pool_key", lambda **_kwargs: ("shared-key", None)) + monkeypatch.setattr(e2b_driver, "_e2b_template_for_run", lambda *_args, **_kwargs: (None, False)) + + driver = AgentDriver() + monkeypatch.setattr(driver, "_upload_tree", lambda *_args, **_kwargs: None) + config = SimpleNamespace( + id="agent-worker", + runtime=SimpleNamespace(runner="e2b", command="python run.py", type="python"), + contexts=[], + connections=[], + capabilities=None, + ) + kwargs = { + "cmd": "echo", + "cmd_args": ["ok"], + "cwd": tmp_path, + "env": {}, + "timeout": 5, + "bundle_dir": tmp_path, + "input_dir": tmp_path, + "output_dir": tmp_path, + "secrets": {}, + "config": config, + "user_id": "user-a", + "log_fn": lambda *_args, **_kwargs: None, + } + + first = driver._run_command_e2b(**kwargs) + second = driver._run_command_e2b(**kwargs) + + assert first["ok"] is True + assert second["ok"] is True + assert _FakeSandbox.create_calls == 1 + assert e2b_driver.warm_pool_size() == 1 + assert e2b_driver.clear_warm_pool() == 1