fix: contexts read binary-detection + transient-disconnect worker health (6003e3bb, ab8a68f0)

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(

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)

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

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():