fix(sandbox): stamp transient Daytona transport failures at the SDK boundary

src/benchflow/_utils/scoring.py

@@ -4,7 +4,10 @@
 from collections.abc import Iterable, Mapping
 from typing import Any, Literal
 
-from benchflow.diagnostics import DIAGNOSTIC_REASON_IDLE_TIMEOUT
+from benchflow.diagnostics import (
+    DIAGNOSTIC_REASON_IDLE_TIMEOUT,
+    TRANSIENT_SANDBOX_TRANSPORT_MARKER,
+)
 
 # Error category constants
 INSTALL_FAILED = "install_failure"
@@ -174,6 +177,14 @@ def _looks_like_infra_error(error: str) -> bool:
             "connection reset",
             "connection refused",
             "broken pipe",
+            # Stamped by TransientSandboxTransportError at the sandbox-SDK
+            # boundary, where the vendor exception type is still available to
+            # judge. One marker covers every provider call — exec, upload,
+            # download, stat — instead of one entry per vendor message
+            # prefix, which would be endless and silently incomplete.
+            TRANSIENT_SANDBOX_TRANSPORT_MARKER,
+            # Predates the marker above and is kept for strings rebuilt
+            # outside that boundary (e.g. a verifier error re-raised as text).
             "failed to get session command",
             "sandbox not found",
             "workspace not found",

src/benchflow/_utils/text.py

@@ -1,4 +1,9 @@
-"""Plain-text truncation for console and log lines.
+"""Plain-text rendering for console and log lines.
+
+Two helpers live here: :func:`truncate_end`, which shortens a message
+without letting a sliced token read as a whole word, and
+:func:`describe_exception`, which renders an exception so the resulting
+line is never detail-free.
 
 The eval console renders one line per rollout, e.g.::
 
@@ -35,3 +40,47 @@ def truncate_end(message: str, limit: int) -> str:
         if sep:
             kept = head
     return kept.rstrip() + "…"
+
+
+def describe_exception(exc: BaseException) -> str:
+    """Render ``exc`` as a one-line description that is never detail-free.
+
+    ``f"{exc}"`` keeps only ``str(exc)``, which for some SDK errors carries
+    no information at all. The Daytona SDK wraps every toolbox call as
+    ``"<prefix>: " + str(underlying)``, and httpx raises its timeout and
+    connection errors with an *empty* message — so a read timeout on
+    ``execute_session_command`` stringifies to the bare
+    ``"Failed to execute session command: "``, a message whose detail after
+    the colon is empty. The exception *class* (``DaytonaTimeoutError`` vs
+    ``DaytonaConnectionError``) is then the only surviving evidence of what
+    actually went wrong, and plain interpolation discards it.
+
+    Lead with the class name so "the exec timed out" can never again be
+    indistinguishable from "the connection dropped". The trailing
+    ``status_code``/``error_code`` fields serve the *other* shape of SDK
+    failure — an OpenAPI exception carrying an HTTP response — and are
+    absent by construction for the detail-free transport case above, which
+    never reaches a response at all.
+    """
+    name = type(exc).__name__
+    message = str(exc).strip()
+    # Approximation: this tests the whole message, not specifically the
+    # detail after a wrapper prefix — a message that legitimately ends in a
+    # colon is annotated too. That is harmless, and the alternative means
+    # knowing every vendor's prefix.
+    if message.endswith(":"):
+        message = f"{message} (no detail)"
+    described = f"{name}: {message}" if message else f"{name} (no message)"
+    details: list[str] = []
+    status_code = getattr(exc, "status_code", None)
+    # Both fields are "absent" when falsy, but only ``status_code`` has a
+    # meaningful zero-ish value to protect (no HTTP status is 0, yet an
+    # explicit 0 should still surface as evidence of a malformed response).
+    if status_code is not None:
+        details.append(f"status_code={status_code}")
+    error_code = getattr(exc, "error_code", None)
+    if error_code:
+        details.append(f"error_code={error_code}")
+    if details:
+        described = f"{described} [{', '.join(details)}]"
+    return described

src/benchflow/agents/registry.py

@@ -53,6 +53,8 @@
 from dataclasses import dataclass, field
 from pathlib import Path
 
+from benchflow._utils.text import describe_exception
+
 
 def _install_python_script(container_path: str, source: str) -> str:
     """Shell snippet that ensures python3 and writes `source` to container_path.
@@ -1469,7 +1471,7 @@ def _load_agent_plugin_packages() -> None:
     try:
         eps = entry_points(group="benchflow.agents")
     except Exception as exc:  # pragma: no cover - metadata backend quirks
-        FAILED_AGENT_PLUGINS["<entry-point-scan>"] = f"{type(exc).__name__}: {exc}"
+        FAILED_AGENT_PLUGINS["<entry-point-scan>"] = describe_exception(exc)
         logging.getLogger(__name__).warning(
             "benchflow.agents entry-point scan failed; ALL agent plugins are "
             "disabled: %s",
@@ -1483,7 +1485,7 @@ def _load_agent_plugin_packages() -> None:
             if callable(loaded):
                 loaded()
         except Exception as exc:
-            FAILED_AGENT_PLUGINS[ep.name] = f"{type(exc).__name__}: {exc}"
+            FAILED_AGENT_PLUGINS[ep.name] = describe_exception(exc)
             logging.getLogger(__name__).warning(
                 "benchflow.agents plugin %r failed to load (some or all of its "
                 "agents may be unavailable or partially registered): %s",

src/benchflow/continue_run/orchestrator.py

@@ -30,6 +30,7 @@
 from pathlib import Path
 from typing import Any, cast
 
+from benchflow._utils.text import describe_exception
 from benchflow.continue_run.replay_proxy import ReplayProxy, ReplayRouter
 from benchflow.continue_run.run_folder import RunFolder, RunFolderError, load_run_folder
 from benchflow.continue_run.sandbox_proxy import (
@@ -442,7 +443,7 @@ async def _capture(label: str, awaitable: Awaitable[Any]) -> None:
         try:
             await awaitable
         except Exception as exc:
-            message = f"{label}: {exc}"
+            message = f"{label}: {describe_exception(exc)}"
             errors.append(message)
             logger.warning("Continuation teardown step failed: %s", message)
 

src/benchflow/diagnostics.py

@@ -477,6 +477,33 @@ def transport_closed(self) -> TransportClosedDiagnostic | None:
         return d if isinstance(d, TransportClosedDiagnostic) else None
 
 
+TRANSIENT_SANDBOX_TRANSPORT_MARKER = "transient sandbox transport failure"
+
+
+class TransientSandboxTransportError(ConnectionError):
+    """A sandbox-provider API call failed at the transport layer.
+
+    Raised at the SDK boundary, where the vendor exception *type* is still
+    alive, and carrying one stable marker so downstream classification never
+    has to keep a census of vendor message prefixes.
+
+    This exists because the vendor prefix is the wrong thing to match on.
+    The Daytona SDK stamps a different prefix per method — "Failed to
+    execute session command: ", "Failed to upload files: ", "Failed to get
+    file info: " — while the *failure* is identical in every case: an httpx
+    timeout or disconnect against the toolbox API, with an empty message.
+    Enumerating those prefixes in the classifier means one entry per vendor
+    method, silently incomplete the moment the SDK adds or renames one, and
+    it also over-matches: a permanent 401/400/409 shares the prefix but is
+    not transient. Deciding at the raise site, where
+    ``_is_daytona_transient_retry_error`` can inspect the real class, keeps
+    both halves honest.
+    """
+
+    def __init__(self, detail: str) -> None:
+        super().__init__(f"{TRANSIENT_SANDBOX_TRANSPORT_MARKER}: {detail}")
+
+
 # Summary / check_results helpers driven by the registry
 
 
@@ -514,6 +541,8 @@ def format_issue_for_field(
     "IdleTimeoutError",
     "TransportClosedError",
     "RolloutDiagnostics",
+    "TRANSIENT_SANDBOX_TRANSPORT_MARKER",
+    "TransientSandboxTransportError",
     "summary_warning",
     "format_issue_for_field",
 ]

src/benchflow/providers/litellm_runtime.py

@@ -25,6 +25,7 @@
 import httpx
 import yaml
 
+from benchflow._utils.text import describe_exception
 from benchflow.agents.codex_config import apply_codex_provider_config
 from benchflow.agents.env import uses_native_subscription_auth
 from benchflow.agents.registry import AGENTS
@@ -1518,7 +1519,8 @@ async def ensure_litellm_runtime(
         await _raise_litellm_unavailable(
             runtime=None,
             error=(
-                f"LiteLLM proxy failed to start for model {model!r}: {exc}. "
+                f"LiteLLM proxy failed to start for model {model!r}: "
+                f"{describe_exception(exc)}. "
                 "BenchFlow never sends provider traffic directly, so this is fatal."
             ),
         )

src/benchflow/rollout/__init__.py

@@ -74,6 +74,7 @@
 # defined in this module.
 from benchflow._utils.live_activity import ActivitySnapshot, SessionCounters
 from benchflow._utils.scoring import classify_error as classify_error
+from benchflow._utils.text import describe_exception
 from benchflow.acp.types import McpServerSpec
 from benchflow.agents.credentials import upload_credential
 from benchflow.agents.registry import AGENTS
@@ -2177,7 +2178,12 @@ async def _run_lifecycle(self) -> RolloutResult:
             self._error = str(e)
             logger.error(str(e))
         except Exception as e:
-            self._error = str(e)
+            # describe_exception, not str(e): this is the funnel every
+            # unclassified rollout failure lands in, and some SDK errors
+            # stringify to a bare wrapper prefix with no detail behind it.
+            # Persisting those raw leaves an artifact that names neither what
+            # failed nor that the detail was empty.
+            self._error = describe_exception(e)
             logger.error("Run failed", exc_info=True)
         finally:
             await self.cleanup()

src/benchflow/sandbox/daytona.py

@@ -54,11 +54,15 @@ def wait_exponential(*_args: Any, **_kwargs: Any) -> Any:
 # sibling underscore helpers) keep resolving from this module path unchanged.
 # The unused names below are intentional façade re-exports, not dead imports.
 from benchflow.sandbox.daytona_pty import (
+    _DAYTONA_EMPTY_EXIT_CODE_MARKERS,  # noqa: F401
     _DAYTONA_ENV_FILE_PREFIX,  # noqa: F401
+    _DAYTONA_TRANSIENT_RETRY_CLASS_NAMES,  # noqa: F401
     _daytona_preflight,
     _exec_failure_output,  # noqa: F401
+    _is_daytona_transient_retry_error,
     _reject_non_main_service,  # noqa: F401
     _wrap_daytona_command_with_env_file,
+    stamp_transient_transport,
 )
 
 # Re-export the extracted ownership labels + auto-reaper so existing imports of
@@ -232,33 +236,6 @@ def build_sync_client(api_key: str | None = None) -> Any:
 # (``None``, or a non-positive value — both previously meant "no deadline");
 # an explicit positive ``timeout_sec`` is honored byte-for-byte as before.
 _DAYTONA_EXEC_HARD_CAP_SEC = 3600
-_DAYTONA_TRANSIENT_RETRY_CLASS_NAMES = frozenset(
-    {
-        "DaytonaConnectionError",
-        "DaytonaRateLimitError",
-        "DaytonaTimeoutError",
-    }
-)
-_DAYTONA_EMPTY_EXIT_CODE_MARKERS = (
-    "failed to convert exit code to int",
-    'strconv.Atoi: parsing "": invalid syntax',
-)
-
-
-def _is_daytona_transient_retry_error(exc: BaseException) -> bool:
-    if isinstance(exc, (ConnectionError, TimeoutError)):
-        return True
-    exc_type = type(exc)
-    if exc_type.__module__.startswith("daytona.") and all(
-        marker in str(exc) for marker in _DAYTONA_EMPTY_EXIT_CODE_MARKERS
-    ):
-        return True
-    return (
-        exc_type.__module__.startswith("daytona.")
-        and exc_type.__name__ in _DAYTONA_TRANSIENT_RETRY_CLASS_NAMES
-    )
-
-
 _DAYTONA_TRANSIENT_RETRY: Any = retry_if_exception(_is_daytona_transient_retry_error)
 
 # Retry-attempt budgets for transient Daytona failures, named here so the
@@ -533,13 +510,15 @@ async def _stop_sandbox(self) -> None:
         if self._sandbox:
             await self._sandbox.delete()
 
+    @stamp_transient_transport
     @_SDK_RETRY
     async def _get_session_command_with_retry(
         self, session_id: str, command_id: str
     ) -> object:
         sandbox = self._require_sandbox()
         return await sandbox.process.get_session_command(session_id, command_id)
 
+    @stamp_transient_transport
     @_SDK_RETRY
     async def _get_session_command_logs_with_retry(
         self, session_id: str, command_id: str
@@ -618,6 +597,7 @@ async def _poll_response(
             return_code=int(response.exit_code),  # type: ignore[union-attr]
         )
 
+    @stamp_transient_transport
     async def _sandbox_exec(
         self,
         command: str,
@@ -639,6 +619,26 @@ async def _sandbox_exec(
         """
         sandbox = self._require_sandbox()
 
+        # Deliberately NOT wrapped in ``_SDK_RETRY``, unlike every other SDK
+        # call on this corridor: neither of the two calls below is safe to
+        # replay, so a transient here is stamped (see the decorator) and
+        # retried one level up, as a fresh rollout against a fresh sandbox.
+        #
+        # ``create_session``: the id is fixed before the call, so a create
+        # that lands but whose response times out comes back
+        # ``DaytonaConflictError`` on replay — a permanent error, and not in
+        # the transient set, so the retry would convert a blip into a hard
+        # failure.
+        #
+        # ``execute_session_command``: replay is silently destructive rather
+        # than loud. The LiteLLM launcher binds an ephemeral port
+        # (``("127.0.0.1", 0)``), so a second exec does not collide on
+        # EADDRINUSE — it binds a *different* port, and its ``rm -f`` +
+        # rewrite of the state/pid files erases the first launcher's
+        # bookkeeping. The first proxy then survives as an orphan holding the
+        # provider key and LITELLM_MASTER_KEY, invisible to
+        # ``_terminate_sandbox_litellm`` (which only knows the overwritten
+        # pid) for the remaining life of the sandbox.
         session_id = str(uuid4())
         await sandbox.process.create_session(session_id)
 
@@ -695,11 +695,13 @@ async def _sandbox_exec(
                 with contextlib.suppress(Exception):
                     await sandbox.process.delete_session(session_id)
 
+    @stamp_transient_transport
     @_SDK_RETRY
     async def _sdk_upload_file(self, source_path: Path | str, target_path: str) -> None:
         sandbox = self._require_sandbox()
         await sandbox.fs.upload_file(str(source_path), target_path)
 
+    @stamp_transient_transport
     @_SDK_RETRY
     async def _sdk_upload_dir(self, source_dir: Path | str, target_dir: str) -> None:
         sandbox = self._require_sandbox()
@@ -725,13 +727,15 @@ async def _sdk_upload_dir(self, source_dir: Path | str, target_dir: str) -> None
         if file_uploads:
             await sandbox.fs.upload_files(files=file_uploads)
 
+    @stamp_transient_transport
     @_SDK_RETRY
     async def _sdk_download_file(
         self, source_path: str, target_path: Path | str
     ) -> None:
         sandbox = self._require_sandbox()
         await sandbox.fs.download_file(source_path, str(target_path))
 
+    @stamp_transient_transport
     @_SDK_RETRY
     async def _sdk_download_dir(self, source_dir: str, target_dir: Path | str) -> None:
         sandbox = self._require_sandbox()