Release orphan boxes hourly and report a missing secrets exchange (DRU-477)

backend/druks/core/tasks.py

@@ -5,6 +5,8 @@
 from druks.harnesses.directory import refresh_added_catalogs
 from druks.harnesses.providers import get_provider, get_providers
 from druks.sandbox import gate
+from druks.sandbox.client import sandbox_client
+from druks.sandbox.models import SandboxIdentity
 from druks.secrets.models import VaultSecret
 from druks.workflows import task
 
@@ -20,12 +22,23 @@ async def reap_deleted_files() -> None:
 
 @task(every="*/15 * * * *")
 async def refresh_tokens() -> None:
-    # Every 15 min. With an ~8h Claude TTL refreshed at <2h remaining (and
-    # codex ~10d at <24h), this keeps both tokens alive with a wide margin
-    # while doing almost nothing on most ticks.
+    # Fifteen minutes keeps every token inside its refresh margin and does
+    # almost nothing on most ticks.
     await _refresh()
 
 
+@task(every="0 * * * *")
+async def release_orphan_boxes() -> None:
+    await _release_orphan_boxes()
+
+
+async def _release_orphan_boxes() -> None:
+    # A run that died without its cleanup leaves its box until the lease ends.
+    for identity in await SandboxIdentity.list_orphans():
+        logger.info("releasing the orphan box %s of run %s", identity.host_id, identity.run_id)
+        await sandbox_client.release(host_id=identity.host_id)
+
+
 @task(every="0 6 * * *")
 async def refresh_catalogs() -> None:
     for provider in get_providers():
@@ -36,10 +49,9 @@ async def refresh_catalogs() -> None:
 async def _refresh() -> dict[str, object]:
     subscriptions = await VaultSecret.list_subscriptions()
 
-    # A rotation ends the token every box holds, so a due rotation runs only
-    # while its subscription is idle, or once urgent. rotate_token no-ops
-    # outside the margin and requests a refresh for every live box. Snapshot
-    # plain values: each refresh commits and expires the session's ORM objects.
+    # A rotation ends the token every box holds, so a due one waits for an
+    # idle subscription unless it is urgent. Plain values: each refresh
+    # commits and expires the session's rows.
     rows = [
         (
             subscription.audience_name,
@@ -106,4 +118,3 @@ def _log_result(result: RotationResult) -> None:
             result.provider,
             result.subscription_id,
         )
-    # "fresh" and "locked" (another worker owns this row's refresh) are quiet no-ops.

backend/druks/doctor.py

@@ -303,6 +303,32 @@ async def _drukbox_doctor(settings: Settings):
         await api.aclose()
 
 
+async def check_secrets_exchange(settings: Settings) -> CheckResult:
+    if not settings.sandbox.service_url:
+        return CheckResult(
+            name="secrets_exchange", ok=True, detail="not configured (sandbox execution is off)"
+        )
+    url = f"{settings.sandbox.exchange_url.rstrip('/')}/healthz"
+    try:
+        async with httpx.AsyncClient(timeout=3.0) as http:
+            status = (await http.get(url)).status_code
+    except httpx.HTTPError as error:
+        return CheckResult(
+            name="secrets_exchange",
+            ok=False,
+            detail=f"drukbox-exchange is unreachable at {url}: {error}. "
+            "Start it: docker compose up -d drukbox-exchange",
+        )
+    if status != 200:
+        return CheckResult(
+            name="secrets_exchange",
+            ok=False,
+            detail=f"drukbox-exchange answered {status} at {url}. "
+            "Read its log: docker compose logs drukbox-exchange",
+        )
+    return CheckResult(name="secrets_exchange", ok=True, detail=url)
+
+
 async def check_sandbox_e2e(settings: Settings) -> CheckResult | list[CheckResult]:
     """Provision a real VM, exercise the acquire and reattach dial paths, and
     probe each registered harness CLI's presence on the image. Costs one
@@ -570,6 +596,7 @@ async def _run_app_check(app_name: str, check) -> CheckResult:
     check_database,
     check_redis,
     check_drukbox,
+    check_secrets_exchange,
     check_capability_modules,
     check_apps,
     check_declared_sandboxes,

backend/druks/sandbox/models.py

@@ -88,7 +88,6 @@ async def create(
         identity = cls(
             run_id=run_id,
             scoped_to=scoped_to,
-            # Own rows: the caller's values stay values.
             secret_refs=[
                 SecretRef(
                     name=ref.name,
@@ -110,9 +109,8 @@ async def create(
         entries = {}
         for ref in secret_refs:
             secret = await VaultSecret.get(ref.secret_id)
-            # A custom entry binds the host the proxy swaps at, the variable
-            # the box exports, and the header the value fills. A catalog entry
-            # leaves those to Drukbox.
+            # A custom entry names its host, variable, and header; a catalog
+            # entry leaves those to Drukbox.
             fields = {}
             if ref.host:
                 header = secret.header or BEARER_HEADER
@@ -188,9 +186,24 @@ async def bind(self, host_id: str) -> None:
         await db_session().commit()
 
     async def revoke(self) -> None:
-        self.revoked_at = Base.utc_now()
+        # A second revoke keeps the first stamp.
+        self.revoked_at = self.revoked_at or Base.utc_now()
         await db_session().commit()
 
+    @classmethod
+    async def list_orphans(cls) -> list["SandboxIdentity"]:
+        """The identities of boxes whose run ended: bound, inside their lease, not revoked."""
+        rows = await db_session().scalars(
+            select(cls)
+            .options(selectinload(cls.run))
+            .where(
+                cls.host_id.is_not(None),
+                cls.revoked_at.is_(None),
+                cls.expires_at > Base.utc_now(),
+            )
+        )
+        return [identity for identity in rows if not identity.run.is_active]
+
     @classmethod
     async def revoke_for_host(cls, engine, host_id: str) -> None:
         # Own transaction: a box release can run outside a step session.

backend/tests/test_doctor.py

@@ -198,6 +198,66 @@ async def test_drukbox_passes_when_unconfigured(tmp_path: Path) -> None:
     assert "not configured" in result.detail
 
 
+_SANDBOX = {"service_url": "https://sb.test", "service_token": "t"}
+
+
+def _exchange(monkeypatch, *, status: int = 200, error: Exception | None = None) -> list[str]:
+    # The exchange as the probe sees it: a status, or a connection error.
+    asked: list[str] = []
+
+    class FakeClient:
+        def __init__(self, **_kwargs) -> None:
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *_exc) -> None:
+            pass
+
+        async def get(self, url: str):
+            asked.append(url)
+            if error:
+                raise error
+            return SimpleNamespace(status_code=status)
+
+    monkeypatch.setattr(doctor.httpx, "AsyncClient", FakeClient)
+    return asked
+
+
+async def test_secrets_exchange_is_not_probed_when_sandbox_execution_is_off(
+    tmp_path: Path, monkeypatch
+) -> None:
+    asked = _exchange(monkeypatch)
+
+    result = await doctor.check_secrets_exchange(make_settings(tmp_path))
+
+    assert result.ok
+    assert "not configured" in result.detail
+    assert asked == []
+
+
+async def test_secrets_exchange_reports_a_healthy_exchange(tmp_path: Path, monkeypatch) -> None:
+    _exchange(monkeypatch, status=200)
+
+    result = await doctor.check_secrets_exchange(make_settings(tmp_path, sandbox=_SANDBOX))
+
+    assert result.ok
+    assert result.detail == "http://127.0.0.1:8781/healthz"
+
+
+async def test_secrets_exchange_names_the_service_and_the_fix_when_unreachable(
+    tmp_path: Path, monkeypatch
+) -> None:
+    _exchange(monkeypatch, error=httpx.ConnectError("refused"))
+
+    result = await doctor.check_secrets_exchange(make_settings(tmp_path, sandbox=_SANDBOX))
+
+    assert not result.ok
+    assert "drukbox-exchange is unreachable at http://127.0.0.1:8781/healthz" in result.detail
+    assert "docker compose up -d drukbox-exchange" in result.detail
+
+
 async def test_declared_sandboxes_pass_when_none_are_declared(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -300,6 +360,7 @@ async def test_run_checks_covers_all_check_names(tmp_path: Path) -> None:
         "database",
         "redis",
         "drukbox",
+        "secrets_exchange",
         "capability_modules",
     }
 

backend/tests/test_orphan_boxes.py

@@ -0,0 +1,78 @@
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+from conftest import connect_provider
+from druks.core import tasks
+from druks.database import db_session
+from druks.durable.engine import _step_engine
+from druks.harnesses.providers import AnthropicProvider
+from druks.sandbox.models import SandboxIdentity, SecretRef
+from druks.testing import seed_run
+from druks_field_notes.workflows import Summarize
+
+
+async def _identity(run_id: str, *, state: str = "running", host_id: str = "") -> SandboxIdentity:
+    await seed_run(db_session(), kind=Summarize.kind, run_id=run_id, state=state)
+    subscription = await connect_provider(
+        AnthropicProvider, {"claudeAiOauth": {"accessToken": "test-token"}}
+    )
+    identity, _ = await SandboxIdentity.create(
+        run_id=run_id,
+        scoped_to="workflow",
+        secret_refs=[SecretRef(name="anthropic", secret_id=subscription.id)],
+    )
+    if host_id:
+        await identity.bind(host_id)
+    return identity
+
+
+def _drukbox(monkeypatch) -> list[str]:
+    # A release ends the identity first, as the real client does.
+    released: list[str] = []
+
+    async def release(*, host_id: str):
+        released.append(host_id)
+        await SandboxIdentity.revoke_for_host(_step_engine(), host_id)
+
+    monkeypatch.setattr(tasks, "sandbox_client", SimpleNamespace(release=release))
+    return released
+
+
+async def test_a_finished_runs_box_is_released_and_a_live_runs_box_stays(druks_db, monkeypatch):
+    await _identity("run-1", state="finished", host_id="host-dead")
+    await _identity("run-2", host_id="host-live")
+    released = _drukbox(monkeypatch)
+
+    await tasks._release_orphan_boxes()
+
+    assert released == ["host-dead"]
+
+
+async def test_a_box_past_its_lease_is_left_to_drukbox(druks_db, monkeypatch):
+    identity = await _identity("run-1", state="finished", host_id="host-old")
+    identity.expires_at = datetime.now(UTC) - timedelta(minutes=1)
+    await db_session().commit()
+    released = _drukbox(monkeypatch)
+
+    await tasks._release_orphan_boxes()
+
+    assert released == []
+
+
+async def test_an_identity_without_a_box_has_nothing_to_release(druks_db, monkeypatch):
+    await _identity("run-1", state="finished")
+    released = _drukbox(monkeypatch)
+
+    await tasks._release_orphan_boxes()
+
+    assert released == []
+
+
+async def test_a_second_tick_finds_nothing_to_do(druks_db, monkeypatch):
+    await _identity("run-1", state="finished", host_id="host-dead")
+    released = _drukbox(monkeypatch)
+
+    await tasks._release_orphan_boxes()
+    await tasks._release_orphan_boxes()
+
+    assert released == ["host-dead"]

docs/concepts.md

@@ -200,7 +200,8 @@ subscription is idle or the token is urgent. One rotator runs at a time, and
 new calls wait for it. After a rotation, Druks requests a refresh from the secrets
 exchange for every live sandbox on that subscription. A provider can revoke
 the previous token at the rotation. Druks revokes the identity when it
-releases the sandbox, and a terminal run denies every fetch. The identity
+releases the sandbox, and a terminal run denies every fetch. An hourly task
+releases the sandbox of a run that ended without its cleanup. The identity
 expires with the sandbox lease.
 
 ## Events, signals, webhooks, and subjects

docs/deployment.md

@@ -212,6 +212,13 @@ identity bearer and nothing else, and it answers `value` and the provider's
 `POST /refresh/<host id>/<service>` on `[sandbox].exchange_url`, one request for
 each live sandbox on the subscription.
 
+An identity dies with its sandbox. Druks revokes it before it deletes the
+sandbox, and the issuer denies a terminal run's identity before any cleanup.
+A run that dies without its cleanup leaves its sandbox until the lease ends.
+The `release_orphan_boxes` task runs every hour and releases such a sandbox
+sooner. Drukbox reaps a sandbox at the end of its lease in any case. `druks doctor` probes the exchange at
+`[sandbox].exchange_url` on `/healthz` and names the corrective action.
+
 Drukbox encrypts the secret entries of each sandbox with `SECRETS_KEY`. The
 installer generates `[secrets].drukbox_secrets_key` and renders it as
 `SECRETS_KEY` for the API and the exchange. Pin the proxy image with

docs/troubleshooting.md

@@ -160,6 +160,19 @@ Make sure that the service listens at `[sandbox].service_url` in `druks.toml`.
 For remote providers, a healthy Drukbox API does not prove SSH access. Then
 follow with `druks doctor --sandbox`.
 
+### The secrets exchange is unreachable
+
+`druks doctor` reports `secrets_exchange` with the URL it probed. Run:
+
+```bash
+docker compose up -d drukbox-exchange
+docker compose logs --tail=200 drukbox-exchange
+```
+
+The exchange binds `127.0.0.1:8781` on the Druks host. Until it answers, a
+sandbox gets no value for its placeholders, and each agent call fails at the
+provider with an authentication error.
+
 ### A sandbox process appears stuck
 
 Druks copies the dashboard transcript from files that a detached VM process writes.