From 5f0c8107f9018d8ff47cb079cca24012eace39a5 Mon Sep 17 00:00:00 2001
From: Guzman Pintos <guzman@luxor.tech>
Date: Wed, 22 Jul 2026 14:18:58 -0300
Subject: [PATCH 1/2] Fix remote sandbox leaks on reset and dashboard backend
 validation

evo reset only reset the workspace-default backend, so sandboxes created
through per-experiment overrides were never torn down. reset now runs
reset_all for every distinct backend spec, remote specs first.

- Propagate teardown failures from reset and preserve state for retry:
  failed sandboxes stay in the state file and the error propagates; core
  keeps the run directory when any remote teardown fails so a second
  reset can retry cleanup.
- Stop inheriting the host shell env into remote sandboxes: inherit_shell
  true now applies to local runs only; a new 'always' value opts remote
  sandboxes back in.
- Validate the dashboard execution endpoint payload shape, reject
  non-scalar values, and map BackendError to 400; load_provider wraps
  constructor crashes in RemoteBackendUnavailable.
---
 plugins/evo/src/evo/backends/remote.py        |  69 ++++---
 plugins/evo/src/evo/backends/remote_state.py  |   7 +
 .../backends/sandbox_providers/__init__.py    |  11 +-
 plugins/evo/src/evo/cli.py                    |  16 +-
 plugins/evo/src/evo/core.py                   |  80 +++++++-
 plugins/evo/src/evo/dashboard.py              |  19 +-
 .../test_dashboard_execution_validation.py    | 106 +++++++++++
 tests/unit/test_remote_teardown.py            | 113 ++++++++++++
 tests/unit/test_reset_backends.py             | 172 ++++++++++++++++++
 tests/unit/test_runtime_env.py                |  32 ++++
 10 files changed, 589 insertions(+), 36 deletions(-)
 create mode 100644 tests/unit/test_dashboard_execution_validation.py
 create mode 100644 tests/unit/test_remote_teardown.py
 create mode 100644 tests/unit/test_reset_backends.py

diff --git a/plugins/evo/src/evo/backends/remote.py b/plugins/evo/src/evo/backends/remote.py
index e91871dc..a2a306f0 100644
--- a/plugins/evo/src/evo/backends/remote.py
+++ b/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.
@@ -193,7 +191,8 @@ def gc(self, ctx: DiscardCtx) -> bool:
                             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 +218,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 +230,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
 
diff --git a/plugins/evo/src/evo/backends/remote_state.py b/plugins/evo/src/evo/backends/remote_state.py
index da688edb..f1e52aaa 100644
--- a/plugins/evo/src/evo/backends/remote_state.py
+++ b/plugins/evo/src/evo/backends/remote_state.py
@@ -121,6 +121,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)
diff --git a/plugins/evo/src/evo/backends/sandbox_providers/__init__.py b/plugins/evo/src/evo/backends/sandbox_providers/__init__.py
index e5d6a439..cd094923 100644
--- a/plugins/evo/src/evo/backends/sandbox_providers/__init__.py
+++ b/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)"
diff --git a/plugins/evo/src/evo/cli.py b/plugins/evo/src/evo/cli.py
index 6132091e..61b469c5 100644
--- a/plugins/evo/src/evo/cli.py
+++ b/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
@@ -4719,7 +4724,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 +6334,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")
diff --git a/plugins/evo/src/evo/core.py b/plugins/evo/src/evo/core.py
index 11ad408c..cdd5d561 100644
--- a/plugins/evo/src/evo/core.py
+++ b/plugins/evo/src/evo/core.py
@@ -415,15 +415,27 @@ def _dotenv_path(root: Path, raw_path: str) -> Path:
     return path
 
 
-def resolve_runtime_env(root: Path, config: dict[str, Any]) -> dict[str, str]:
+def resolve_runtime_env(
+    root: Path,
+    config: dict[str, Any],
+    *,
+    remote: bool = False,
+) -> dict[str, str]:
     """Resolve benchmark/gate runtime env fresh for each attempt.
 
     Config stores source metadata only; values are loaded from the orchestrator
     process environment and configured dotenv files when this function runs.
+
+    `inherit_shell: true` forwards the orchestrator's environment only when
+    the attempt runs locally: host vars like HOME or PATH override sane
+    values inside a remote sandbox and break anything that reads them
+    (npm's cache dir, for one). `inherit_shell: "always"` forwards it to
+    remote sandboxes too, for callers who know their env translates.
     """
     runtime_env = dict(config.get("runtime_env") or {})
     resolved: dict[str, str] = {}
-    if runtime_env.get("inherit_shell", True):
+    inherit = runtime_env.get("inherit_shell", True)
+    if inherit == "always" or (inherit and not remote):
         resolved.update(os.environ)
 
     for source in runtime_env.get("dotenv", []) or []:
@@ -881,10 +893,68 @@ def delete_discarded_experiment(root: Path, node: dict[str, Any]) -> None:
 
 
 def reset_runtime_state(root: Path) -> None:
-    """Remove the active run's worktrees, branches, and directory."""
-    from .backends import load_backend
+    """Remove the active run's worktrees, branches, and directory.
+
+    Every distinct backend spec in the run gets its own `reset_all` -- the
+    workspace default plus any per-experiment overrides (`evo new --remote
+    ...` on a worktree-default workspace), so remote sandboxes are torn
+    down no matter which backend is the default. Remote specs reset first
+    because local backends wipe the run directory that still holds the
+    remote state files. If any remote teardown fails, the run directory is
+    preserved (local resets and the final cleanup are skipped) so the
+    surviving state can drive a retry, and the error propagates.
+    """
+    import shutil
+
+    from .backends import (
+        backend_spec_for_node,
+        backend_spec_from_config,
+        load_backend,
+    )
+
+    config = load_config(root)
+    specs: list[tuple[str, dict[str, Any]]] = [backend_spec_from_config(config)]
+    try:
+        graph = load_graph(root)
+    except (FileNotFoundError, RuntimeError):
+        graph = {"nodes": {}}
+    for node_id, node in graph.get("nodes", {}).items():
+        if node_id == "root":
+            continue
+        spec = backend_spec_for_node(root, node, workspace_config=config)
+        if spec not in specs:
+            specs.append(spec)
+    remote_specs = [spec for spec in specs if spec[0] == "remote"]
+    local_specs = [spec for spec in specs if spec[0] != "remote"]
+
+    failures: list[str] = []
+    for name, backend_config in remote_specs:
+        try:
+            backend = load_backend(
+                root, explicit_name=name, explicit_config=backend_config
+            )
+            backend.reset_all(root)
+        except Exception as exc:
+            failures.append(f"{name} ({backend_config.get('provider', '-')}): {exc}")
+    if failures:
+        raise RuntimeError(
+            "reset could not tear down every remote sandbox; runtime state "
+            "was kept so `evo reset` can retry the cleanup: "
+            + "; ".join(failures)
+        )
 
-    load_backend(root).reset_all(root)
+    for name, backend_config in local_specs:
+        try:
+            load_backend(
+                root, explicit_name=name, explicit_config=backend_config
+            ).reset_all(root)
+        except Exception as exc:
+            failures.append(f"{name}: {exc}")
+    shutil.rmtree(workspace_path(root), ignore_errors=True)
+    if failures:
+        raise RuntimeError(
+            "reset could not clean up every backend: " + "; ".join(failures)
+        )
 
 
 def git_status_porcelain(path: Path) -> str:
diff --git a/plugins/evo/src/evo/dashboard.py b/plugins/evo/src/evo/dashboard.py
index ff40583a..9faddd53 100644
--- a/plugins/evo/src/evo/dashboard.py
+++ b/plugins/evo/src/evo/dashboard.py
@@ -9,6 +9,7 @@
 
 from flask import Flask, Response, jsonify, request, send_from_directory
 
+from .backends.protocol import BackendError
 from .core import (
     PRUNE_KIND_EXHAUSTED,
     PRUNE_KINDS,
@@ -304,6 +305,10 @@ def _provider_readiness(config: dict[str, Any]) -> dict[str, Any]:
 def _clean_provider_config(data: dict[str, Any]) -> dict[str, Any]:
     cleaned: dict[str, Any] = {}
     for key, value in (data or {}).items():
+        if isinstance(value, dict):
+            raise ValueError(
+                f"provider_config.{key} must be a scalar or list, not an object"
+            )
         if isinstance(value, str):
             value = value.strip()
         if value in ("", None):
@@ -407,7 +412,14 @@ def _validate_and_save_execution_settings(root: Path, body: dict[str, Any]) -> d
         provider = str(body.get("provider", "")).strip()
         if not provider:
             raise ValueError("remote backend requires a provider")
-        provider_config = _clean_provider_config(body.get("provider_config") or {})
+        raw_provider_config = body.get("provider_config")
+        if raw_provider_config is None:
+            raw_provider_config = {}
+        if not isinstance(raw_provider_config, dict):
+            raise ValueError(
+                "provider_config must be a JSON object of key/value settings"
+            )
+        provider_config = _clean_provider_config(raw_provider_config)
         if old_name == "remote" and old_config.get("provider") == provider:
             provider_config = _preserve_secret_fields(
                 provider_config,
@@ -463,7 +475,8 @@ def _validate_and_save_execution_settings(root: Path, body: dict[str, Any]) -> d
 
 
 def _normalize_runtime_env_settings(body: dict[str, Any]) -> dict[str, Any]:
-    inherit_shell = bool(body.get("inherit_shell", True))
+    raw_inherit = body.get("inherit_shell", True)
+    inherit_shell = "always" if raw_inherit == "always" else bool(raw_inherit)
     sources = []
     for raw in body.get("dotenv", []) or []:
         if not isinstance(raw, dict):
@@ -817,7 +830,7 @@ def workspace_execution():
         body = request.get_json(silent=True) or {}
         try:
             summary = _validate_and_save_execution_settings(_root(), body)
-        except (ValueError, RuntimeError) as exc:
+        except (ValueError, RuntimeError, BackendError) as exc:
             return jsonify({"error": str(exc)}), 400
         return jsonify(summary)
 
diff --git a/tests/unit/test_dashboard_execution_validation.py b/tests/unit/test_dashboard_execution_validation.py
new file mode 100644
index 00000000..35445a41
--- /dev/null
+++ b/tests/unit/test_dashboard_execution_validation.py
@@ -0,0 +1,106 @@
+"""Dashboard API tests: POST /api/workspace/execution with malformed
+payloads must return 400 and leave the existing configuration untouched,
+instead of 500ing or silently resetting saved settings."""
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from evo.core import init_workspace, load_config, save_config
+from evo.dashboard import create_app
+
+
+class TestExecutionSettingsValidation(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmp.name)
+        init_workspace(
+            self.root,
+            target="t.py",
+            benchmark="python bench.py",
+            metric="max",
+            gate=None,
+        )
+        self.app = create_app(self.root)
+        self.client = self.app.test_client()
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def _seed_remote_config(self) -> dict:
+        cfg = load_config(self.root)
+        cfg["execution_backend"] = "remote"
+        cfg["execution_backend_config"] = {
+            "provider": "manual",
+            "provider_config": {
+                "base_url": "http://127.0.0.1:9",
+                "bearer_token": "seed-secret",
+            },
+        }
+        save_config(self.root, cfg)
+        return cfg["execution_backend_config"]
+
+    def _post(self, payload: dict):
+        return self.client.post("/api/workspace/execution", json=payload)
+
+    def test_provider_config_list_is_rejected_and_config_untouched(self):
+        seeded = self._seed_remote_config()
+        resp = self._post(
+            {"backend": "remote", "provider": "manual", "provider_config": []}
+        )
+        self.assertEqual(resp.status_code, 400, resp.get_data(as_text=True))
+        cfg = load_config(self.root)
+        self.assertEqual(cfg["execution_backend_config"], seeded)
+
+    def test_provider_config_string_is_rejected_not_500(self):
+        resp = self._post(
+            {"backend": "remote", "provider": "manual", "provider_config": "oops"}
+        )
+        self.assertEqual(resp.status_code, 400, resp.get_data(as_text=True))
+        self.assertIn("provider_config", resp.get_json()["error"])
+
+    def test_provider_config_nested_object_value_is_rejected_not_500(self):
+        resp = self._post(
+            {
+                "backend": "remote",
+                "provider": "manual",
+                "provider_config": {"base_url": {"nested": 1}},
+            }
+        )
+        self.assertEqual(resp.status_code, 400, resp.get_data(as_text=True))
+        self.assertIn("base_url", resp.get_json()["error"])
+
+    def test_provider_construction_failure_is_400_not_500(self):
+        resp = self._post(
+            {
+                "backend": "remote",
+                "provider": "definitely-not-a-provider",
+                "provider_config": {},
+            }
+        )
+        self.assertEqual(resp.status_code, 400, resp.get_data(as_text=True))
+        self.assertIn("provider", resp.get_json()["error"].lower())
+
+    def test_valid_manual_save_still_works(self):
+        resp = self._post(
+            {
+                "backend": "remote",
+                "provider": "manual",
+                "provider_config": {
+                    "base_url": "http://127.0.0.1:9",
+                    "bearer_token": "tok",
+                },
+            }
+        )
+        self.assertEqual(resp.status_code, 200, resp.get_data(as_text=True))
+        cfg = load_config(self.root)
+        self.assertEqual(cfg["execution_backend"], "remote")
+        self.assertEqual(
+            cfg["execution_backend_config"]["provider_config"]["base_url"],
+            "http://127.0.0.1:9",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/unit/test_remote_teardown.py b/tests/unit/test_remote_teardown.py
new file mode 100644
index 00000000..78004b1b
--- /dev/null
+++ b/tests/unit/test_remote_teardown.py
@@ -0,0 +1,113 @@
+"""Remote sandbox records survive failed teardown attempts."""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from evo.backends import remote_state
+from evo.backends.protocol import DiscardCtx, SandboxHandle
+from evo.backends.remote import RemoteSandboxBackend
+
+
+class _TeardownProvider:
+    name = "teardown-test"
+
+    def __init__(self) -> None:
+        self.fail = True
+        self.calls: list[str] = []
+
+    def provision(self, spec: Any) -> SandboxHandle:
+        raise NotImplementedError
+
+    def tear_down(self, handle: SandboxHandle) -> None:
+        self.calls.append(handle.native_id)
+        if self.fail:
+            raise RuntimeError(f"close failed for {handle.native_id}")
+
+    def is_alive(self, handle: SandboxHandle) -> bool:
+        return True
+
+    def build_client(self, handle: SandboxHandle) -> Any:
+        raise NotImplementedError
+
+
+def _backend_with_record(
+    root: Path,
+    provider: _TeardownProvider,
+    *,
+    exp_id: str | None,
+) -> RemoteSandboxBackend:
+    backend = RemoteSandboxBackend(provider)
+    remote_state.init_state(
+        root,
+        provider=provider.name,
+        provider_config={},
+        state_key=backend.state_key,
+    )
+    with remote_state.locked_state(root, backend.state_key) as state:
+        state["sandboxes"].append({
+            "id": 0,
+            "native_id": "sandbox-1",
+            "base_url": "https://sandbox-1.example.test",
+            "bearer_token": "test-token",
+            "leased_by": (
+                {"exp_id": exp_id, "pid": 1, "leased_at": "now"}
+                if exp_id is not None
+                else None
+            ),
+            "last_branch": None,
+            "provisioned_at": "now",
+        })
+    return backend
+
+
+def _native_ids(root: Path, backend: RemoteSandboxBackend) -> list[str]:
+    state = remote_state.read_state(root, backend.state_key)
+    return [sandbox["native_id"] for sandbox in state["sandboxes"]]
+
+
+@pytest.mark.parametrize("method_name", ["discard", "release_lease"])
+def test_direct_cleanup_retains_record_and_surfaces_failure(
+    tmp_path: Path,
+    method_name: str,
+) -> None:
+    provider = _TeardownProvider()
+    backend = _backend_with_record(tmp_path, provider, exp_id="exp_0001")
+    ctx = DiscardCtx(root=tmp_path, node={"id": "exp_0001"})
+
+    with pytest.raises(RuntimeError, match="close failed for sandbox-1"):
+        getattr(backend, method_name)(ctx)
+    assert _native_ids(tmp_path, backend) == ["sandbox-1"]
+
+    provider.fail = False
+    getattr(backend, method_name)(ctx)
+    assert _native_ids(tmp_path, backend) == []
+
+
+def test_gc_retains_record_until_teardown_succeeds(tmp_path: Path) -> None:
+    provider = _TeardownProvider()
+    backend = _backend_with_record(tmp_path, provider, exp_id=None)
+    ctx = DiscardCtx(root=tmp_path, node={"id": "exp_0001"})
+
+    assert backend.gc(ctx) is False
+    assert _native_ids(tmp_path, backend) == ["sandbox-1"]
+
+    provider.fail = False
+    assert backend.gc(ctx) is True
+    assert _native_ids(tmp_path, backend) == []
+
+
+def test_orphan_sweep_retains_record_until_teardown_succeeds(
+    tmp_path: Path,
+) -> None:
+    provider = _TeardownProvider()
+    backend = _backend_with_record(tmp_path, provider, exp_id=None)
+
+    assert backend.sweep_orphans(tmp_path, set()) == []
+    assert _native_ids(tmp_path, backend) == ["sandbox-1"]
+
+    provider.fail = False
+    assert backend.sweep_orphans(tmp_path, set()) == ["sandbox-1"]
+    assert _native_ids(tmp_path, backend) == []
diff --git a/tests/unit/test_reset_backends.py b/tests/unit/test_reset_backends.py
new file mode 100644
index 00000000..f7f732e5
--- /dev/null
+++ b/tests/unit/test_reset_backends.py
@@ -0,0 +1,172 @@
+"""`evo reset` must reset every distinct backend spec in the run -- the
+workspace default plus per-experiment overrides -- so remote sandboxes
+created via `evo new --remote ...` are torn down even when the workspace
+default is a local backend."""
+from __future__ import annotations
+
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+from evo.backends import backend_state_key, remote_state
+from evo.core import (
+    atomic_write_json,
+    graph_path,
+    init_workspace,
+    load_graph,
+    reset_runtime_state,
+    workspace_path,
+)
+
+
+FAKE_PROVIDER_MODULE = '''\
+from pathlib import Path
+
+
+class FakeProvider:
+    name = "fake"
+
+    def __init__(self, config):
+        self.config = dict(config)
+
+    def provision(self, spec):
+        raise RuntimeError("not used in this test")
+
+    def tear_down(self, handle):
+        marker_dir = Path(self.config["marker_dir"])
+        if (marker_dir / "fail-teardown").exists():
+            raise RuntimeError("teardown boom")
+        (marker_dir / f"torn-{handle.native_id}").touch()
+
+    def is_alive(self, handle):
+        return False
+
+    def build_client(self, handle):
+        raise RuntimeError("not used in this test")
+'''
+
+
+class TestResetTearsDownOverrideBackends(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmp.name)
+        subprocess.run(["git", "init", "-q"], cwd=self.root, check=True)
+        subprocess.run(
+            ["git", "config", "user.email", "t@t"], cwd=self.root, check=True
+        )
+        subprocess.run(["git", "config", "user.name", "t"], cwd=self.root, check=True)
+        (self.root / "t.py").write_text("x = 1\n", encoding="utf-8")
+        subprocess.run(["git", "add", "."], cwd=self.root, check=True)
+        subprocess.run(["git", "commit", "-qm", "init"], cwd=self.root, check=True)
+        init_workspace(
+            self.root,
+            target="t.py",
+            benchmark="python bench.py",
+            metric="max",
+            gate=None,
+        )
+        self.marker_dir = self.root / "markers"
+        self.marker_dir.mkdir()
+        self.modules_dir = self.root / "fake_modules"
+        self.modules_dir.mkdir()
+        (self.modules_dir / "evo_test_fake_provider.py").write_text(
+            FAKE_PROVIDER_MODULE, encoding="utf-8"
+        )
+        sys.path.insert(0, str(self.modules_dir))
+
+    def tearDown(self):
+        sys.path.remove(str(self.modules_dir))
+        sys.modules.pop("evo_test_fake_provider", None)
+        self._tmp.cleanup()
+
+    def _seed_remote_override_node(self) -> str:
+        provider_ref = "evo_test_fake_provider:FakeProvider"
+        provider_config = {"marker_dir": str(self.marker_dir)}
+        backend_config = {
+            "provider": provider_ref,
+            "provider_config": provider_config,
+        }
+
+        graph = load_graph(self.root)
+        graph["nodes"]["exp_0000"] = {
+            "id": "exp_0000",
+            "status": "pending",
+            "backend": "remote",
+            "backend_config": backend_config,
+        }
+        atomic_write_json(graph_path(self.root), graph)
+
+        state_key = backend_state_key("remote", backend_config)
+        remote_state.init_state(
+            self.root,
+            provider=provider_ref,
+            provider_config=provider_config,
+            state_key=state_key,
+        )
+        with remote_state.locked_state(self.root, state_key) as state:
+            state["sandboxes"].append(
+                {
+                    "id": 0,
+                    "native_id": "sb-test",
+                    "base_url": "http://127.0.0.1:9",
+                    "bearer_token": "",
+                    "metadata": {},
+                    "leased_by": None,
+                }
+            )
+        return state_key
+
+    def test_reset_tears_down_per_node_remote_sandboxes(self):
+        self._seed_remote_override_node()
+
+        reset_runtime_state(self.root)
+
+        self.assertTrue(
+            (self.marker_dir / "torn-sb-test").exists(),
+            "per-node remote sandbox was not torn down by reset",
+        )
+        self.assertFalse(workspace_path(self.root).exists())
+
+    def test_failed_teardown_preserves_state_and_raises_then_retry_succeeds(self):
+        state_key = self._seed_remote_override_node()
+        (self.marker_dir / "fail-teardown").touch()
+
+        with self.assertRaises(RuntimeError) as ctx:
+            reset_runtime_state(self.root)
+        self.assertIn("sb-test", str(ctx.exception))
+
+        # The failure must not erase the record of the (possibly still
+        # billing) sandbox: run dir and its state entry survive for retry.
+        self.assertTrue(workspace_path(self.root).exists())
+        state = remote_state.read_state(self.root, state_key)
+        self.assertEqual(
+            [s["native_id"] for s in state["sandboxes"]], ["sb-test"]
+        )
+
+        (self.marker_dir / "fail-teardown").unlink()
+        reset_runtime_state(self.root)
+        self.assertTrue((self.marker_dir / "torn-sb-test").exists())
+        self.assertFalse(workspace_path(self.root).exists())
+
+    def test_reset_survives_unloadable_override_backend(self):
+        graph = load_graph(self.root)
+        graph["nodes"]["exp_0000"] = {
+            "id": "exp_0000",
+            "status": "pending",
+            "backend": "remote",
+            "backend_config": {"provider": "definitely-not-a-provider"},
+        }
+        atomic_write_json(graph_path(self.root), graph)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            reset_runtime_state(self.root)
+        self.assertIn("definitely-not-a-provider", str(ctx.exception))
+        # An unloadable provider can't tear down its sandboxes; the run
+        # dir must survive so a later reset can finish the cleanup.
+        self.assertTrue(workspace_path(self.root).exists())
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/unit/test_runtime_env.py b/tests/unit/test_runtime_env.py
index 00a8fd42..797b03ba 100644
--- a/tests/unit/test_runtime_env.py
+++ b/tests/unit/test_runtime_env.py
@@ -108,6 +108,38 @@ def test_resolve_runtime_env_overlays_dotenv_over_shell(tmp_path: Path, monkeypa
     assert "OTHER" not in resolved
 
 
+def test_resolve_runtime_env_remote_skips_shell_inheritance(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HOST_ONLY_VAR", "macos-value")
+    write(tmp_path / ".env", "TOKEN=dotenv\n")
+    config = {
+        "runtime_env": {
+            "inherit_shell": True,
+            "dotenv": [{"path": ".env", "mode": "all"}],
+        }
+    }
+    resolved = resolve_runtime_env(tmp_path, config, remote=True)
+    assert "HOST_ONLY_VAR" not in resolved
+    assert "HOME" not in resolved
+    assert resolved["TOKEN"] == "dotenv"
+
+    local = resolve_runtime_env(tmp_path, config, remote=False)
+    assert local["HOST_ONLY_VAR"] == "macos-value"
+
+
+def test_resolve_runtime_env_always_inherits_for_remote(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HOST_ONLY_VAR", "macos-value")
+    config = {"runtime_env": {"inherit_shell": "always", "dotenv": []}}
+    resolved = resolve_runtime_env(tmp_path, config, remote=True)
+    assert resolved["HOST_ONLY_VAR"] == "macos-value"
+
+
+def test_resolve_runtime_env_off_stays_off_everywhere(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("HOST_ONLY_VAR", "macos-value")
+    config = {"runtime_env": {"inherit_shell": False, "dotenv": []}}
+    assert "HOST_ONLY_VAR" not in resolve_runtime_env(tmp_path, config, remote=False)
+    assert "HOST_ONLY_VAR" not in resolve_runtime_env(tmp_path, config, remote=True)
+
+
 def test_resolve_runtime_env_overlays_dashboard_variables(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("TOKEN", "shell")
     write(tmp_path / ".evo" / "run_0000" / "runtime_env_values.json", '{"variables":{"TOKEN":"dashboard"}}')

From 23628bf3e230c4e8aedb02809b5b626a00b7cc6a Mon Sep 17 00:00:00 2001
From: Capsule Agent <capsule-agent@tenki.cloud>
Date: Mon, 27 Jul 2026 20:44:14 +0000
Subject: [PATCH 2/2] Fix remote cleanup review feedback

---
 plugins/evo/src/evo/backends/remote.py       |  44 ++++++--
 plugins/evo/src/evo/backends/remote_state.py |   1 +
 plugins/evo/src/evo/cli.py                   |  38 +++++--
 plugins/evo/src/evo/core.py                  |  21 +++-
 tests/unit/test_committed_cleanup.py         | 111 +++++++++++++++++++
 tests/unit/test_remote_teardown.py           |  22 ++++
 tests/unit/test_reset_backends.py            |  70 +++++++++++-
 7 files changed, 286 insertions(+), 21 deletions(-)
 create mode 100644 tests/unit/test_committed_cleanup.py

diff --git a/plugins/evo/src/evo/backends/remote.py b/plugins/evo/src/evo/backends/remote.py
index a2a306f0..1a7d7ed7 100644
--- a/plugins/evo/src/evo/backends/remote.py
+++ b/plugins/evo/src/evo/backends/remote.py
@@ -164,19 +164,27 @@ 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.
         """
@@ -184,7 +192,12 @@ def gc(self, ctx: DiscardCtx) -> bool:
         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:
@@ -517,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:
@@ -555,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
 
@@ -572,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
diff --git a/plugins/evo/src/evo/backends/remote_state.py b/plugins/evo/src/evo/backends/remote_state.py
index f1e52aaa..db4fd4be 100644
--- a/plugins/evo/src/evo/backends/remote_state.py
+++ b/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": "..."
     }
diff --git a/plugins/evo/src/evo/cli.py b/plugins/evo/src/evo/cli.py
index 61b469c5..5efb550e 100644
--- a/plugins/evo/src/evo/cli.py
+++ b/plugins/evo/src/evo/cli.py
@@ -3480,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}")
@@ -3584,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)
@@ -3653,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,
diff --git a/plugins/evo/src/evo/core.py b/plugins/evo/src/evo/core.py
index cdd5d561..ae0d159e 100644
--- a/plugins/evo/src/evo/core.py
+++ b/plugins/evo/src/evo/core.py
@@ -902,14 +902,17 @@ def reset_runtime_state(root: Path) -> None:
     because local backends wipe the run directory that still holds the
     remote state files. If any remote teardown fails, the run directory is
     preserved (local resets and the final cleanup are skipped) so the
-    surviving state can drive a retry, and the error propagates.
+    surviving state can drive a retry, and the error propagates. A remote
+    spec with no sandbox records is skipped before provider construction.
     """
     import shutil
 
     from .backends import (
+        backend_state_key,
         backend_spec_for_node,
         backend_spec_from_config,
         load_backend,
+        remote_state,
     )
 
     config = load_config(root)
@@ -930,6 +933,22 @@ def reset_runtime_state(root: Path) -> None:
     failures: list[str] = []
     for name, backend_config in remote_specs:
         try:
+            state_key = backend_state_key(
+                name,
+                {
+                    "provider": backend_config.get("provider"),
+                    "provider_config": dict(
+                        backend_config.get("provider_config", {}) or {}
+                    ),
+                },
+            )
+            try:
+                state = remote_state.read_state(root, state_key)
+            except FileNotFoundError:
+                continue
+            if not state.get("sandboxes"):
+                remote_state.delete_state(root, state_key)
+                continue
             backend = load_backend(
                 root, explicit_name=name, explicit_config=backend_config
             )
diff --git a/tests/unit/test_committed_cleanup.py b/tests/unit/test_committed_cleanup.py
new file mode 100644
index 00000000..41c30a50
--- /dev/null
+++ b/tests/unit/test_committed_cleanup.py
@@ -0,0 +1,111 @@
+"""Cleanup failures must not overwrite successful experiment outcomes."""
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from evo import cli
+from evo.backends.worktree import WorktreeBackend
+from evo.core import (
+    allocate_experiment,
+    attempt_dir,
+    attempt_outcome_path,
+    experiment_result_path,
+    init_workspace,
+    load_graph,
+)
+
+
+def _init_experiment(root: Path) -> None:
+    subprocess.run(["git", "init", "-q", "-b", "main"], cwd=root, check=True)
+    subprocess.run(
+        ["git", "config", "user.email", "test@example.test"],
+        cwd=root,
+        check=True,
+    )
+    subprocess.run(["git", "config", "user.name", "test"], cwd=root, check=True)
+    (root / "agent.py").write_text("STATE = 1\n", encoding="utf-8")
+    (root / "eval.py").write_text(
+        "import json, os\n"
+        "from pathlib import Path\n"
+        "Path(os.environ['EVO_RESULT_PATH']).write_text("
+        "json.dumps({'score': 1.0}))\n",
+        encoding="utf-8",
+    )
+    subprocess.run(["git", "add", "."], cwd=root, check=True)
+    subprocess.run(["git", "commit", "-qm", "init"], cwd=root, check=True)
+    init_workspace(
+        root,
+        target="agent.py",
+        benchmark="python3 eval.py",
+        metric="max",
+        gate=None,
+        host="generic",
+        per_exp_timeout=30,
+    )
+    allocate_experiment(root, "root", "baseline")
+
+
+def _fail_release(self: WorktreeBackend, ctx: object) -> None:
+    raise RuntimeError("synthetic cleanup failure")
+
+
+def test_run_cleanup_failure_keeps_committed_outcome(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    _init_experiment(tmp_path)
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setattr(WorktreeBackend, "release_lease", _fail_release)
+
+    args = cli.build_parser().parse_args(["run", "exp_0000"])
+    assert args.func(args) == 0
+
+    node = load_graph(tmp_path)["nodes"]["exp_0000"]
+    result = json.loads(
+        experiment_result_path(tmp_path, "exp_0000").read_text(encoding="utf-8")
+    )
+    outcome = json.loads(
+        attempt_outcome_path(tmp_path, "exp_0000", 1).read_text(encoding="utf-8")
+    )
+    attempt_state = json.loads(
+        (attempt_dir(tmp_path, "exp_0000", 1) / "attempt_state.json").read_text(
+            encoding="utf-8"
+        )
+    )
+
+    assert node["status"] == "committed"
+    assert result["status"] == "committed"
+    assert outcome["outcome"] == "committed"
+    assert attempt_state["status"] == "committed"
+    assert "workspace cleanup failed" in capsys.readouterr().err
+
+
+def test_done_cleanup_failure_keeps_committed_result(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    _init_experiment(tmp_path)
+    monkeypatch.setattr(WorktreeBackend, "release_lease", _fail_release)
+    args = argparse.Namespace(
+        exp_id="exp_0000",
+        traces=None,
+        no_compare=False,
+        score=1.0,
+    )
+
+    assert cli._record_done_result(tmp_path, args) == 0
+
+    node = load_graph(tmp_path)["nodes"]["exp_0000"]
+    result = json.loads(
+        experiment_result_path(tmp_path, "exp_0000").read_text(encoding="utf-8")
+    )
+    assert node["status"] == "committed"
+    assert result["status"] == "committed"
+    assert "workspace cleanup failed" in capsys.readouterr().err
diff --git a/tests/unit/test_remote_teardown.py b/tests/unit/test_remote_teardown.py
index 78004b1b..ca4e4e8a 100644
--- a/tests/unit/test_remote_teardown.py
+++ b/tests/unit/test_remote_teardown.py
@@ -68,6 +68,10 @@ def _native_ids(root: Path, backend: RemoteSandboxBackend) -> list[str]:
     return [sandbox["native_id"] for sandbox in state["sandboxes"]]
 
 
+def _records(root: Path, backend: RemoteSandboxBackend) -> list[dict[str, Any]]:
+    return remote_state.read_state(root, backend.state_key)["sandboxes"]
+
+
 @pytest.mark.parametrize("method_name", ["discard", "release_lease"])
 def test_direct_cleanup_retains_record_and_surfaces_failure(
     tmp_path: Path,
@@ -80,6 +84,8 @@ def test_direct_cleanup_retains_record_and_surfaces_failure(
     with pytest.raises(RuntimeError, match="close failed for sandbox-1"):
         getattr(backend, method_name)(ctx)
     assert _native_ids(tmp_path, backend) == ["sandbox-1"]
+    if method_name == "release_lease":
+        assert _records(tmp_path, backend)[0]["cleanup_pending"] is True
 
     provider.fail = False
     getattr(backend, method_name)(ctx)
@@ -99,6 +105,22 @@ def test_gc_retains_record_until_teardown_succeeds(tmp_path: Path) -> None:
     assert _native_ids(tmp_path, backend) == []
 
 
+def test_gc_retries_failed_release_for_committed_node(tmp_path: Path) -> None:
+    provider = _TeardownProvider()
+    backend = _backend_with_record(tmp_path, provider, exp_id="exp_0001")
+    ctx = DiscardCtx(
+        root=tmp_path,
+        node={"id": "exp_0001", "status": "committed"},
+    )
+
+    with pytest.raises(RuntimeError, match="close failed for sandbox-1"):
+        backend.release_lease(ctx)
+    provider.fail = False
+
+    assert backend.gc(ctx) is True
+    assert _native_ids(tmp_path, backend) == []
+
+
 def test_orphan_sweep_retains_record_until_teardown_succeeds(
     tmp_path: Path,
 ) -> None:
diff --git a/tests/unit/test_reset_backends.py b/tests/unit/test_reset_backends.py
index f7f732e5..99e1caea 100644
--- a/tests/unit/test_reset_backends.py
+++ b/tests/unit/test_reset_backends.py
@@ -150,22 +150,84 @@ def test_failed_teardown_preserves_state_and_raises_then_retry_succeeds(self):
         self.assertTrue((self.marker_dir / "torn-sb-test").exists())
         self.assertFalse(workspace_path(self.root).exists())
 
-    def test_reset_survives_unloadable_override_backend(self):
+    def test_reset_skips_unloadable_override_without_live_sandboxes(self):
+        backend_config = {"provider": "definitely-not-a-provider"}
         graph = load_graph(self.root)
         graph["nodes"]["exp_0000"] = {
             "id": "exp_0000",
             "status": "pending",
             "backend": "remote",
-            "backend_config": {"provider": "definitely-not-a-provider"},
+            "backend_config": backend_config,
         }
         atomic_write_json(graph_path(self.root), graph)
 
+        state_key = backend_state_key(
+            "remote",
+            {
+                "provider": backend_config["provider"],
+                "provider_config": {},
+            },
+        )
+        remote_state.init_state(
+            self.root,
+            provider="definitely-not-a-provider",
+            provider_config={},
+            state_key=state_key,
+        )
+
+        reset_runtime_state(self.root)
+
+        self.assertFalse(workspace_path(self.root).exists())
+
+    def test_reset_preserves_unloadable_backend_with_sandbox_record(self):
+        backend_config = {"provider": "definitely-not-a-provider"}
+        graph = load_graph(self.root)
+        graph["nodes"]["exp_0000"] = {
+            "id": "exp_0000",
+            "status": "pending",
+            "backend": "remote",
+            "backend_config": backend_config,
+        }
+        atomic_write_json(graph_path(self.root), graph)
+
+        state_key = backend_state_key(
+            "remote",
+            {
+                "provider": backend_config["provider"],
+                "provider_config": {},
+            },
+        )
+        remote_state.init_state(
+            self.root,
+            provider="definitely-not-a-provider",
+            provider_config={},
+            state_key=state_key,
+        )
+        with remote_state.locked_state(self.root, state_key) as state:
+            state["sandboxes"].append(
+                {
+                    "id": 0,
+                    "native_id": "still-live",
+                    "base_url": "https://still-live.example.test",
+                    "bearer_token": "",
+                    "metadata": {},
+                    "leased_by": None,
+                }
+            )
+
         with self.assertRaises(RuntimeError) as ctx:
             reset_runtime_state(self.root)
         self.assertIn("definitely-not-a-provider", str(ctx.exception))
-        # An unloadable provider can't tear down its sandboxes; the run
-        # dir must survive so a later reset can finish the cleanup.
         self.assertTrue(workspace_path(self.root).exists())
+        self.assertEqual(
+            [
+                sandbox["native_id"]
+                for sandbox in remote_state.read_state(self.root, state_key)[
+                    "sandboxes"
+                ]
+            ],
+            ["still-live"],
+        )
 
 
 if __name__ == "__main__":