fix: handle invalidated sandboxes in daytona gracefully

application/sandbox/artifacts_capture.py

@@ -122,7 +122,10 @@ def snapshot_signatures(manager: Any, session_id: str) -> Dict[str, Tuple[int, O
     try:
         files = manager.list_files(session_id)
     except Exception:
-        logger.exception("artifacts_capture: pre-exec listing failed")
+        # Best-effort: a listing failure (sandbox auto-stopped/deleted) just means no
+        # pre-image, so change detection falls back to "capture everything". Swallowed
+        # and recoverable -> WARN, not ERROR (an ERROR here false-alarms monitoring).
+        logger.warning("artifacts_capture: pre-exec listing failed", exc_info=True)
         return signatures
     candidates = sorted(f for f in files if not f.startswith("inputs/") and not _is_scratch(f))
     if len(candidates) > MAX_SCANNED_FILES:
@@ -164,7 +167,9 @@ def capture_artifacts(
     try:
         post_files = set(manager.list_files(session_id))
     except Exception:
-        logger.exception("artifacts_capture: post-exec listing failed")
+        # Best-effort: without a post-image there is nothing to capture, so return
+        # empty. Swallowed and recoverable -> WARN, not ERROR (see snapshot_signatures).
+        logger.warning("artifacts_capture: post-exec listing failed", exc_info=True)
         return []
 
     produced = (f for f in post_files if not f.startswith("inputs/"))

application/sandbox/daytona.py

@@ -376,14 +376,44 @@ class DaytonaSandbox(CodeSandbox):
                     return self._to_result(handle.sandbox.process.code_run(wrapped, timeout=wall))
                 except Exception:  # noqa: BLE001 - second failure -> error result below
                     pass
-            return ExecResult(
+            result = ExecResult(
                 status="error",
                 error_name=type(exc).__name__,
                 error_value=str(exc) or "code_run failed",
                 exit_code=-1,
             )
+            # An auto-DELETED sandbox (vs a merely stopped one, handled above) can't be
+            # woken and would fail every retry on this cached handle. Drop the dead
+            # handle and flag the runtime invalid so SandboxManager cold-starts a fresh
+            # sandbox on the next call, rather than looping on the tombstone until the
+            # idle-TTL reap. Mirrors the Jupyter backend's kernel-death handling.
+            if self._sandbox_gone(handle):
+                self._forget_handle(session_id, handle)
+                result.runtime_invalidated = True
+            return result
         return self._to_result(response)
 
+    def _sandbox_gone(self, handle: "_Handle") -> bool:
+        """True when the sandbox behind ``handle`` no longer exists in the cloud.
+
+        Separates an auto-DELETED sandbox (unrecoverable — a fresh one must be created)
+        from a transient transport blip on a still-live sandbox (retryable on the same
+        handle). A ``get`` that cannot resolve the id, or a terminal state, means gone.
+        """
+        try:
+            fresh = self._client.get(handle.sandbox_id)
+        except Exception:  # noqa: BLE001 - unresolvable id => the sandbox is gone
+            return True
+        state = getattr(fresh, "state", None)
+        state_value = getattr(state, "value", state)
+        return state_value in (None, "destroyed", "deleted", "error", "archived")
+
+    def _forget_handle(self, session_id: str, handle: "_Handle") -> None:
+        """Drop the cached handle if it is still the registered one (ABA-safe)."""
+        with self._lock:
+            if self._handles.get(session_id) is handle:
+                self._handles.pop(session_id, None)
+
     @staticmethod
     def _with_workspace_cwd(workspace: str, code: str) -> str:
         """Prepend a chdir into the session workspace so relative paths resolve there.

tests/sandbox/test_artifacts_capture.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import hashlib
+import logging
 
 import pytest
 
@@ -127,3 +128,46 @@ class TestReadSweepCap:
         captured = ac.capture_artifacts(mgr, "sid", pre, user_id="u")
         assert captured == []  # nothing changed -> nothing persisted
         assert mgr.reads == ac.MAX_SCANNED_FILES  # but the sweep is still bounded
+
+
+class _ListingRaisesMgr:
+    """A manager whose workspace listing fails (sandbox auto-stopped/deleted)."""
+
+    def list_files(self, _sid):
+        # Mirrors the generic IOError the Daytona/Jupyter backends raise when the
+        # workspace can't be listed because the runtime is gone.
+        raise IOError("list_files failed: RuntimeError")
+
+
+@pytest.mark.unit
+class TestListingFailureLogLevel:
+    """Regression: a swallowed, recoverable workspace-listing failure must log at
+    WARNING, not ERROR.
+
+    ``snapshot_signatures`` / ``capture_artifacts`` catch a failed listing, log it,
+    and return empty (best-effort — the exec result is still delivered). Logging it
+    via ``logger.exception`` stamps it ERROR, which inflated the production error
+    stream: on 2026-07-20 these two lines were 7 of the day's ~43 real-app ERRORs
+    and false-alarmed the error review (ledger #28, Bug B). A no-op capture is not
+    an error.
+    """
+
+    def test_pre_exec_listing_failure_logs_warning_not_error(self, caplog):
+        with caplog.at_level(logging.DEBUG, logger="application.sandbox.artifacts_capture"):
+            sigs = ac.snapshot_signatures(_ListingRaisesMgr(), "sid")
+        assert sigs == {}  # swallowed, best-effort
+        recs = [r for r in caplog.records if "pre-exec listing failed" in r.getMessage()]
+        assert recs, "expected a pre-exec listing-failed log record"
+        assert all(r.levelno == logging.WARNING for r in recs), (
+            "a swallowed, recoverable listing failure must not log at ERROR"
+        )
+
+    def test_post_exec_listing_failure_logs_warning_not_error(self, caplog):
+        with caplog.at_level(logging.DEBUG, logger="application.sandbox.artifacts_capture"):
+            captured = ac.capture_artifacts(_ListingRaisesMgr(), "sid", {}, user_id="u")
+        assert captured == []  # swallowed, best-effort
+        recs = [r for r in caplog.records if "post-exec listing failed" in r.getMessage()]
+        assert recs, "expected a post-exec listing-failed log record"
+        assert all(r.levelno == logging.WARNING for r in recs), (
+            "a swallowed, recoverable listing failure must not log at ERROR"
+        )

tests/sandbox/test_daytona.py

@@ -753,3 +753,72 @@ def test_factory_forwards_max_sandboxes(fake_sdk, monkeypatch):
     backend = sc.SandboxCreator.create_backend("daytona")
     assert backend._max_sandboxes == 7
     sc.SandboxCreator.reset()
+
+
+# --- Regression: a DELETED sandbox must invalidate the runtime (Bug A) ----
+#
+# Daytona auto-STOPS an idle sandbox (auto_stop_interval) and later auto-DELETES
+# it (auto_delete_interval). A stop is recoverable — ``_ensure_started`` wakes it
+# and retries (covered above). A delete is NOT: the cached handle points at a
+# tombstone, ``code_run`` raises "container not found", and ``client.get`` can no
+# longer resolve it. The Jupyter backend handles the equivalent kernel-death by
+# dropping its handle and setting ``runtime_invalidated=True`` so the manager
+# cold-starts on the next call; the Daytona backend must do the same.
+#
+# Without this the manager keeps a dead handle and EVERY subsequent call in the
+# session fails until the idle-TTL reap — observed in prod as one sandbox failing
+# five times across ~18 minutes (error-review ledger #28, Bug A).
+
+
+def test_exec_invalidates_runtime_when_sandbox_deleted(sandbox):
+    """A code_run against an auto-DELETED sandbox drops the handle and flags the
+    runtime invalid so the next open cold-starts."""
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    # Deleted under the still-cached handle: exec fails AND the sandbox can no
+    # longer be resolved (get raises => gone, not merely stopped).
+    created.process.code_run.side_effect = RuntimeError("No such container: sbx-1")
+    sandbox._client.get = mock.Mock(side_effect=KeyError("sbx-1"))
+
+    res = sandbox.exec("conv-1", "print('x')")
+
+    assert res.status == "error"
+    assert res.runtime_invalidated is True  # signals the manager to cold-start
+    assert "conv-1" not in sandbox._handles  # tombstone dropped; next open re-creates
+
+
+def test_exec_does_not_invalidate_live_sandbox_on_transient_error(sandbox):
+    """A blip on a still-live (started) sandbox is NOT an invalidation: the warm
+    handle stays so a later call can reuse it."""
+    sandbox.open("conv-1")
+    _, created = sandbox._client.created[0]
+    created.process.code_run.side_effect = RuntimeError("transient transport blip")
+
+    res = sandbox.exec("conv-1", "print('x')")
+
+    assert res.status == "error"
+    assert res.runtime_invalidated is False
+    assert "conv-1" in sandbox._handles
+
+
+def test_manager_cold_starts_after_daytona_sandbox_deleted(fake_sdk):
+    """End-to-end (SandboxManager + DaytonaSandbox): once the cloud sandbox is
+    deleted, the next run transparently creates a FRESH sandbox instead of reusing
+    the dead handle. This is the user-facing recovery contract Bug A breaks."""
+    from application.sandbox.daytona import DaytonaSandbox
+    from application.sandbox.manager import SandboxManager
+
+    backend = DaytonaSandbox(api_key="k")
+    mgr = SandboxManager(backend, max_ttl=600)
+    mgr.open("conv-1")
+    _, created = backend._client.created[0]
+    created.process.code_run.side_effect = RuntimeError("No such container: sbx-1")
+    backend._client.get = mock.Mock(side_effect=KeyError("sbx-1"))  # deleted => gone
+
+    res = mgr.exec("conv-1", "print('x')")
+
+    assert res.status == "error"
+    assert not mgr.has_session("conv-1")  # manager dropped the invalidated session
+
+    mgr.open("conv-1")  # recovery
+    assert backend._client.create.call_count == 2  # a NEW sandbox, not the tombstone