Fix remote sandbox leaks on reset and dashboard backend validation

plugins/evo/src/evo/backends/remote.py

@@ -12,13 +12,12 @@
 
 import os
 import secrets
-import shutil
 import time
 import uuid
 from pathlib import Path
 from typing import Any
 
-from ..core import utc_now, workspace_path
+from ..core import utc_now
 from . import remote_state
 from .protocol import (
     AllocateCtx,
@@ -148,12 +147,11 @@ def discard(self, ctx: DiscardCtx) -> None:
             return  # nothing to do
         handle = self._handle_for_slot(ctx.root, slot_id)
         if handle is not None:
-            try:
-                self.provider.tear_down(handle)
-            except Exception:
-                # Best-effort; sandbox may already be gone (network blip,
-                # provider-side timeout). State cleanup proceeds regardless.
-                pass
+            # A provider that raises cannot confirm the sandbox is gone.
+            # Keep the record and surface the failure so cleanup can be
+            # retried instead of leaving an untracked, possibly billing
+            # sandbox alive until its provider-side timeout.
+            self.provider.tear_down(handle)
         with remote_state.locked_state(ctx.root, self.state_key) as state:
             # Drop the slot entirely on discard. Re-allocate gets a fresh
             # provision; no half-states left around.
@@ -166,34 +164,48 @@ def discard(self, ctx: DiscardCtx) -> None:
     # ---------------------------------------------------------------- release_lease
 
     def release_lease(self, ctx: DiscardCtx) -> None:
-        """Clear the lease without tearing down the sandbox.
+        """Release a sandbox after an experiment commits.
 
-        POC behavior: ALSO tears down (no warm-reuse yet). When
-        we add `keep_warm` config in alpha.4, this becomes the path that
-        retains the sandbox.
+        POC behavior tears it down (no warm-reuse yet). A failed teardown
+        leaves the lease pinned and marks the record for a later `gc` retry.
+        When we add `keep_warm` config, this becomes the path that retains
+        the sandbox.
         """
         # POC: same as discard.
-        self.discard(ctx)
+        try:
+            self.discard(ctx)
+        except Exception:
+            # The experiment may already be committed. Keep this record pinned
+            # and make it eligible for `gc` retries without letting allocation
+            # reconciliation turn it into a reusable slot.
+            self._mark_cleanup_pending(ctx.root, ctx.node["id"])
+            raise
 
     # ---------------------------------------------------------------- gc
 
     def gc(self, ctx: DiscardCtx) -> bool:
-        """Best-effort cleanup of stale sandboxes whose holders are gone.
+        """Best-effort cleanup of free or cleanup-pending sandboxes.
 
         Returns True if anything got cleaned up so cli.cmd_gc reports it.
         """
         cleaned = False
         with remote_state.locked_state(ctx.root, self.state_key) as state:
             keep: list[dict[str, Any]] = []
             for sandbox in state["sandboxes"]:
-                if sandbox.get("leased_by") is None:
+                lease = sandbox.get("leased_by")
+                pending_for_node = (
+                    sandbox.get("cleanup_pending")
+                    and (lease or {}).get("exp_id") == ctx.node.get("id")
+                )
+                if lease is None or pending_for_node:
                     handle = self._handle_from_record(sandbox)
                     if handle is not None:
                         try:
                             self.provider.tear_down(handle)
                             cleaned = True
                         except Exception:
-                            pass
+                            keep.append(sandbox)
+                            continue
                         self._handles.pop(sandbox["id"], None)
                         self._tokens.pop(sandbox["id"], None)
                 else:
@@ -219,7 +231,8 @@ def sweep_orphans(self, root: Path, live_exp_ids: set[str]) -> list[str]:
                             self.provider.tear_down(handle)
                             torn.append(handle.native_id)
                         except Exception:
-                            pass
+                            keep.append(sandbox)
+                            continue
                         self._handles.pop(sandbox["id"], None)
                         self._tokens.pop(sandbox["id"], None)
                     continue
@@ -230,22 +243,47 @@ def sweep_orphans(self, root: Path, live_exp_ids: set[str]) -> list[str]:
     # ---------------------------------------------------------------- reset_all
 
     def reset_all(self, root: Path) -> None:
-        """Tear down every recorded sandbox and wipe the workspace dir."""
+        """Tear down every recorded sandbox and remove this config's state.
+
+        Only this spec's own state file is removed -- not the run
+        directory. `core.reset_runtime_state` calls reset_all once per
+        distinct backend spec in the run and wipes the run directory
+        itself afterwards; an early rmtree here would destroy the state
+        files the remaining specs need to find their sandboxes.
+
+        A sandbox whose tear_down raises stays in the state file and the
+        error propagates, so reset reports the failure and a retry can
+        finish the cleanup instead of silently dropping the record of a
+        possibly-still-billing sandbox.
+        """
         try:
-            state = remote_state.read_state(root, self.state_key)
+            remote_state.read_state(root, self.state_key)
         except FileNotFoundError:
-            state = {"sandboxes": []}
-        for sandbox in state.get("sandboxes", []):
-            handle = self._handle_from_record(sandbox)
-            if handle is None:
-                continue
-            try:
-                self.provider.tear_down(handle)
-            except Exception:
-                pass
+            self._handles.clear()
+            self._tokens.clear()
+            return
+        errors: list[str] = []
+        with remote_state.locked_state(root, self.state_key) as state:
+            keep: list[dict[str, Any]] = []
+            for sandbox in state.get("sandboxes", []):
+                handle = self._handle_from_record(sandbox)
+                if handle is None:
+                    continue
+                try:
+                    self.provider.tear_down(handle)
+                except Exception as exc:
+                    keep.append(sandbox)
+                    errors.append(f"{handle.native_id}: {exc}")
+            state["sandboxes"] = keep
         self._handles.clear()
         self._tokens.clear()
-        shutil.rmtree(workspace_path(root), ignore_errors=True)
+        if errors:
+            raise RuntimeError(
+                f"could not tear down {len(errors)} sandbox(es): "
+                f"{'; '.join(errors)}. Their records were kept in the "
+                f"backend state so another reset can retry."
+            )
+        remote_state.delete_state(root, self.state_key)
 
     # ---------------------------------------------------------------- internal
 
@@ -492,6 +530,14 @@ def _release_if_matches(self, root: Path, slot_id: int, exp_id: str) -> None:
                         sandbox["leased_by"] = None
                     break
 
+    def _mark_cleanup_pending(self, root: Path, exp_id: str) -> None:
+        with remote_state.locked_state(root, self.state_key) as state:
+            for sandbox in state["sandboxes"]:
+                lease = sandbox.get("leased_by")
+                if lease and lease.get("exp_id") == exp_id:
+                    sandbox["cleanup_pending"] = True
+                    break
+
     def _handle_for_slot(self, root: Path, slot_id: int) -> SandboxHandle | None:
         handle = self._handles.get(slot_id)
         if handle is not None:
@@ -530,7 +576,8 @@ def _reconcile_orphaned(self, root: Path) -> None:
 
         Only acts when the graph has an explicit terminal status. A missing
         node is NOT treated as terminal -- masks real bugs (e.g. a partial
-        graph write would otherwise look like a leaked lease).
+        graph write would otherwise look like a leaked lease). Records marked
+        cleanup-pending stay pinned until `gc` or `reset` confirms teardown.
         """
         from ..core import load_graph
 
@@ -547,5 +594,9 @@ def _reconcile_orphaned(self, root: Path) -> None:
                     continue
                 exp_id = lease.get("exp_id")
                 node = graph["nodes"].get(exp_id)
-                if node is not None and node.get("status") in terminal:
+                if (
+                    node is not None
+                    and node.get("status") in terminal
+                    and not sandbox.get("cleanup_pending")
+                ):
                     sandbox["leased_by"] = None

plugins/evo/src/evo/backends/remote_state.py

@@ -10,6 +10,7 @@
       "native_id": "sb-abc123",
       "base_url": "https://...",
       "leased_by": null | {"exp_id": "exp_NNNN", "pid": 12345, "leased_at": "..."},
+      "cleanup_pending": true | false,
       "last_branch": "evo/run_NNNN/exp_NNNN" | null,
       "provisioned_at": "..."
     }
@@ -121,6 +122,13 @@ def locked_state(root: Path, state_key: str) -> Iterator[dict[str, Any]]:
         atomic_write_json(state_path, _encode_state_for_disk(root, state))
 
 
+def delete_state(root: Path, state_key: str) -> None:
+    """Remove this remote config's state file and its lock file."""
+    state_path = _migrate_legacy_if_needed(root, state_key)
+    state_path.unlink(missing_ok=True)
+    _lock_path(state_path).unlink(missing_ok=True)
+
+
 def read_state(root: Path, state_key: str | None = None) -> dict[str, Any]:
     """Read-only snapshot of this remote config's state file."""
     state_path = _resolve_state_path(root, state_key)

plugins/evo/src/evo/backends/sandbox_providers/__init__.py

@@ -126,7 +126,16 @@ def load_provider(name: str, config: dict[str, Any]) -> SandboxProvider:
     -- keep it actionable.
     """
     if name in _LOADERS:
-        return _LOADERS[name](config)
+        try:
+            return _LOADERS[name](config)
+        except RemoteBackendUnavailable:
+            raise
+        except Exception as exc:
+            # Constructor failures from bad config must surface as the typed
+            # user-facing error, not a raw TypeError 500ing the dashboard.
+            raise RemoteBackendUnavailable(
+                f"Remote provider {name!r} rejected its configuration: {exc}"
+            ) from exc
     if ":" in name or "." in name:
         return _load_dotted_path_provider(name, config)
     known = ", ".join(sorted(_LOADERS)) or "(none)"

plugins/evo/src/evo/cli.py

@@ -1043,7 +1043,9 @@ def cmd_env(args: argparse.Namespace) -> int:
         runtime_env = _ensure_runtime_env_config(config)
 
         if args.env_action == "inherit-shell":
-            runtime_env["inherit_shell"] = args.value == "on"
+            runtime_env["inherit_shell"] = (
+                "always" if args.value == "always" else args.value == "on"
+            )
             atomic_write_json(config_path(root), config)
             print(f"inherit_shell set to {str(runtime_env['inherit_shell']).lower()}")
             return 0
@@ -2602,8 +2604,9 @@ def _runtime_env_for_attempt(
     env_traces_dir: str,
     env_result_path: str,
     env_checkpoint_dir: str,
+    remote: bool = False,
 ) -> dict[str, str]:
-    env = resolve_runtime_env(root, config)
+    env = resolve_runtime_env(root, config, remote=remote)
     env["EVO_TRACES_DIR"] = env_traces_dir
     env["EVO_WORKTREE"] = str(worktree)
     env["EVO_EXPERIMENT_ID"] = exp_id
@@ -2819,6 +2822,7 @@ def _cmd_run_check(
         env_traces_dir=env_traces_dir,
         env_result_path=env_result_path,
         env_checkpoint_dir=env_checkpoint_dir,
+        remote=executor.is_remote,
     )
     gate_records: list[dict] = []
     runtime_records: list[dict] = []
@@ -3121,6 +3125,7 @@ def _mark_active(current_node: dict, _graph: dict) -> None:
         env_traces_dir=env_traces_dir,
         env_result_path=env_result_path,
         env_checkpoint_dir=env_checkpoint_dir,
+        remote=executor.is_remote,
     )
     # Stamp this driver's PID so a re-invocation of `evo run` can detect
     # whether the attempt is actually still in flight (the concurrent-run
@@ -3475,12 +3480,14 @@ def _mark_committed(current_node: dict, _graph: dict) -> None:
             # Release the workspace lease on transition into `committed`.
             # Worktree backend: no-op. Pool backend: returns the slot to the
             # free queue. Failed and evaluated transitions retain the lease.
-            from .backends import DiscardCtx as _DCtx, load_backend as _lb
             committed_node = dict(node)
             committed_node["status"] = "committed"
             committed_node["commit"] = commit
-            _lb(root, node=committed_node, workspace_config=config).release_lease(
-                _DCtx(root=root, node=committed_node)
+            _release_committed_lease(
+                root,
+                committed_node,
+                config,
+                backend=backend,
             )
             delta = "" if parent_score is None else f" ({'+' if metric == 'max' else ''}{score - parent_score:.4f} vs parent)"
             print(f"COMMITTED {args.exp_id} {score}{delta}")
@@ -3579,6 +3586,31 @@ def _mark_failed(current_node: dict, _graph: dict) -> None:
         return 1
 
 
+def _release_committed_lease(
+    root: Path,
+    node: dict[str, Any],
+    config: dict[str, Any],
+    *,
+    backend: Any | None = None,
+) -> None:
+    """Release committed workspace state without changing the run outcome."""
+    from .backends import DiscardCtx, load_backend
+
+    try:
+        resolved = (
+            backend
+            if backend is not None
+            else load_backend(root, node=node, workspace_config=config)
+        )
+        resolved.release_lease(DiscardCtx(root=root, node=node))
+    except Exception as exc:  # noqa: BLE001
+        print(
+            f"WARNING: {node['id']} committed, but workspace cleanup failed: "
+            f"{exc}. Run `evo gc` or `evo reset --yes` to retry.",
+            file=sys.stderr,
+        )
+
+
 def _record_done_result(root: Path, args: argparse.Namespace) -> int:
     config, graph = _require_workspace(root)
     node = _read_node(root, args.exp_id)
@@ -3648,11 +3680,8 @@ def _mark(current_node: dict, _graph: dict) -> None:
     update_node(root, args.exp_id, _mark)
     _finalize_result(root, args.exp_id, node, args.score, status, {"recorded_only": True})
     if status == "committed":
-        from .backends import DiscardCtx as _DCtx, load_backend as _lb
         committed_node = {**node, "status": "committed"}
-        _lb(root, node=committed_node, workspace_config=config).release_lease(
-            _DCtx(root=root, node=committed_node)
-        )
+        _release_committed_lease(root, committed_node, config)
     _emit_experiment_result_telemetry(
         root,
         args.exp_id,
@@ -4719,7 +4748,7 @@ def _cmd_gate_check_impl(
     gates_dir.mkdir(parents=True, exist_ok=True)
     remote = executor.is_remote
     run_cwd: Path | str = worktree if remote else root
-    env = resolve_runtime_env(root, config)
+    env = resolve_runtime_env(root, config, remote=remote)
     runtime_records: list[dict] = []
     gate_records: list[dict] = []
     inherited_gates, gate_origins = _inherited_gate_specs(config, graph, args.exp_id)
@@ -6329,9 +6358,10 @@ def build_parser() -> argparse.ArgumentParser:
     env_show_p.set_defaults(func=cmd_env)
     env_inherit_p = env_sub.add_parser(
         "inherit-shell",
-        help="enable or disable inheriting the orchestrator process environment",
+        help="inherit the orchestrator process environment: on (local runs "
+        "only), off, or always (remote sandboxes too)",
     )
-    env_inherit_p.add_argument("value", choices=["on", "off"])
+    env_inherit_p.add_argument("value", choices=["on", "off", "always"])
     env_inherit_p.set_defaults(func=cmd_env)
     env_load_p = env_sub.add_parser("load", help="add or update a dotenv source")
     env_load_p.add_argument("path")