feat(sandbox): on-demand provisioning via request_sandbox verb (#338)

* feat(sandbox): on-demand request_sandbox verb replaces eager provisioning Sandboxes were provisioned at every agent spawn for opted-in projects, so every role paid the sidecar spin-up and a provisioning failure refused the spawn. Provisioning now happens when an agent asks: the request_sandbox do-verb (dev + QA) reaches the orchestrator through ContentActionsDeps, ensure_sandbox provisions idempotently with an in-memory per-agent cache (evicted at teardown and janitor sweep), and creds return in the envelope payload including ready-to-export ROBOCO_TEST_* values. Spawn now only injects a marker env naming the available services plus a briefing line; sandbox failures can no longer refuse a spawn. Teardown lifecycle unchanged. * feat(sandbox): harden request_sandbox + Phase 3 wiring proof and docs Hardening from adversarial review: ensure_sandbox now provisions the project's full opted-in set on first request (a later superset can never tear down a live sandbox mid-use), serializes per-agent behind an asyncio lock (a client timeout-retry no longer races its own in-flight provision), and verifies container liveness on every cache hit (a dead sandbox evicts and re-provisions instead of serving dead creds). MCP client budget 720->1080s for the full-set cold case. Phase 3: e2e smoke wiring test (manifest grants + guard-chain envelopes over the real API), sandbox-db/tools/map docs and CLAUDE.md rewritten for on-demand. * feat(sandbox): release sandboxes when the agent's work ends CEO directive: sidecars must not dangle once the agent is done. The six work-ending verbs (i_am_done, unclaim, i_am_idle, pass_review, fail_review, i_documented) now release the caller's sandbox best-effort on their success path via release_sandbox (lock + teardown + cache evict; a no-sandbox agent costs a dict lookup). Container removal and the janitor remain the backstop; a re-request provisions fresh. * test(sandbox): monkeypatch the release hook instead of method assignment mypy method-assign rejected the direct AsyncMock assignments; the prior static gate ran before this test file landed. * test(sandbox): guard envelope evidence for mypy in verb tests * chore(prompts): regenerate verb tables for request_sandbox * chore: resolve merge with master (breadcrumbs + statement budget) --------- Co-authored-by: Renn F <rennf93@users.noreply.github.com>
2026-08-03 07:23:24 +02:00 · 2026-07-08 16:40:02 +02:00
parent 9e4025b822
commit 47d78f50ee
30 changed files with 1824 additions and 246 deletions
@@ -381,7 +381,7 @@ Agent backends are pluggable. `roboco/llm/providers/` defines an `AgentProvider`
 **Organizational memory loop (default-off).** Closes the learn→reuse loop so agents stop cold-respawning blind. Three parts, all gated by `ROBOCO_ORG_MEMORY_ENABLED`: ① **capture** — at task completion `TaskService._completion_learnings_for` distills ONE high-signal lesson (Problem→Approach→Gotcha, ≤120 words) via the local model (`MemoryDistiller`, `roboco/services/memory_distiller.py`) instead of the noisy raw-notes/duration capture (flag-off keeps the legacy capture); journal indexing excludes `is_private` reflections from the shared corpus. ② **retrieve (keystone)** — on claim, `_briefing_for` injects `context_briefing["institutional_memory"]`: top-K (`ROBOCO_ORG_MEMORY_TOP_K`) relevance-floored (`ROBOCO_ORG_MEMORY_MIN_SCORE`) lessons + approved playbooks from a role-shaped query (`EvidenceRepo.similar_memory` over the LEARNINGS + PLAYBOOKS pgvector indexes); below the floor nothing is injected (no briefing bloat). ③ **playbooks** — a first-class curated procedure store: `PlaybookTable` (migration 050), the `PLAYBOOKS` OptimalService index, the `draft_playbook` content verb (delivery roles), Auditor `approve_playbook`/`reject_playbook`/`archive_playbook` curation (approval indexes it), and the panel review queue (`playbook-review-queue.tsx`; `/api/playbooks` Auditor/CEO routes). Distillation runs on the local model only — never a cloud LLM in the hot path; every step is best-effort (a failure never blocks completion or the briefing).
-**Sandboxed dev DB/Redis/Mongo (default-off).** Per-project opt-in (`projects.sandbox_services`, migration 057): when armed (`ROBOCO_SANDBOX_DB_ENABLED`), each opted-in project's agent spawn gets orchestrator-provisioned throwaway **sibling containers** (random per-sandbox creds, tmpfs data dir, memory/cpu-capped, labeled `roboco.sandbox=1`) — one per opted-in service — injected as `ROBOCO_TEST_DB_*` / `ROBOCO_TEST_REDIS_*` / `ROBOCO_TEST_MONGO_*` **in place of** the legacy prod-creds gate-env injection (`_append_gate_env`, which points agents at RoboCo's own production Postgres under `ROBOCO_TOOLCHAIN_MATCH_ENABLED`) — sandbox replaces, never coexists with, prod creds. Lifetime tracks the agent container 1:1: teardown at every removal path plus an orphan janitor at startup + each reaper tick (grace-windowed so a sweep can't reap a sandbox whose spawn is still mid-flight; the pre-spawn stale-clear likewise spares the just-provisioned sandbox). Provisioning failure refuses the spawn (fail-loud); docker-in-agent stays structurally absent. `SandboxProvisioner` (`roboco/runtime/sandbox.py`), wired in the orchestrator spawn path. The service set is a **pluggable engine registry** (`roboco/models/sandbox.py`): each engine declares its image, run args, readiness probe, and `ROBOCO_TEST_*` env; `VALID_SANDBOX_SERVICES` is derived from the registry, and the provisioner + orchestrator env injection iterate it, so adding an engine (e.g. mongo) is one class + one registry line — no branch edited in the provisioner or the env emitter.
+**Sandboxed dev DB/Redis/Mongo (default-off).** Per-project opt-in (`projects.sandbox_services`, migration 057); when armed (`ROBOCO_SANDBOX_DB_ENABLED`), provisioning is **on-demand (2026-07-08)**, not eager at spawn: a developer or QA agent calls the `request_sandbox` do-verb (role-scoped to `_DEV_DO`/`_QA_DO` in `role_config.py`; `services` omitted means the project's whole opted-in set) and `ContentActions.request_sandbox` (`roboco/services/gateway/content_actions.py`) walks a guard chain — flag off; no active project-bound task; project not opted into any service; a requested service outside the opted set (remediate names the allowed set); orchestrator handle unavailable (the one **retryable** guard) — before calling `AgentOrchestrator.ensure_sandbox`, which always provisions the project's whole opted-in set regardless of the requested subset (so a later subset/superset request within that set is a guaranteed cache hit and can never trigger a mid-session teardown of a live container the agent is using), verifies a cache hit is still live before trusting it (evicting + re-provisioning on a dead container), serializes concurrent calls for one agent behind a per-slug `asyncio.Lock`, and caches the result in-memory per agent slug (`_sandbox_info`) — the verb filters the returned creds back down to what this call actually asked for. Sibling containers get random per-sandbox creds, tmpfs data dir, memory/cpu caps, labeled `roboco.sandbox=1`; creds return in the verb's envelope `evidence` (`SandboxInfo.as_payload()`), one entry per service including a ready-to-`export` `env` sub-dict (`ROBOCO_TEST_DB_*` / `ROBOCO_TEST_REDIS_*` / `ROBOCO_TEST_MONGO_*`) — never injected as container env, so no spawn-time creds delivery exists at all. Spawn itself only injects a cheap marker env `ROBOCO_SANDBOX_SERVICES_AVAILABLE=<csv>` (never creds) for an opted-in project, plus a briefing line naming `request_sandbox()` explicitly, **in place of** the legacy prod-creds gate-env injection (`_append_gate_env`, which points agents at RoboCo's own production Postgres under `ROBOCO_TOOLCHAIN_MATCH_ENABLED`) — sandbox replaces, never coexists with, prod creds. A provisioning failure now surfaces as a retryable envelope on the verb, never a spawn refusal — sandbox trouble can no longer block dispatch. A sandbox is torn down at end-of-engagement, not just at container removal: `AgentOrchestrator.release_sandbox(agent_slug)` is called (best-effort, never failing the verb; a fast cache-check no-op when nothing was ever requested) by the Choreographer on the SUCCESSFUL exit of `i_am_done` / `unclaim` / `i_am_idle` / `pass_review` / `fail_review` / `i_documented`, so a sidecar doesn't outlive the work that requested it. Lifetime still tracks the agent container 1:1 as the backstop: teardown at every removal path plus an orphan janitor at startup + each reaper tick (grace-windowed so a sweep can't reap a sandbox whose request is still mid-flight; the pre-spawn stale-clear likewise spares a just-requested sandbox) also evicts the `_sandbox_info` cache entry. **Known ceiling:** the cache is in-memory only — an orchestrator restart forgets live sandboxes, so the next `request_sandbox` call re-provisions (the pre-clear tears down any still-running stale container) and returns fresh creds. Docker-in-agent stays structurally absent throughout. The service set is a **pluggable engine registry** (`roboco/models/sandbox.py`): each engine declares its image, run args, readiness probe, and `ROBOCO_TEST_*` env; `VALID_SANDBOX_SERVICES` is derived from the registry, and the provisioner + the verb's payload builder iterate it, so adding an engine (e.g. mongo) is one class + one registry line — no branch edited in the provisioner or the env emitter.
 **Cloud auth via FastAPI Users (default-off).** Lets the panel/API be safely exposed beyond localhost without touching the CEO's local no-login flow while off. Gated by `ROBOCO_CLOUD_AUTH_ENABLED` (+ `ROBOCO_CLOUD_AUTH_EMAIL` / `_PASSWORD` / `_SECRET` / `_COOKIE_MAX_AGE`; `Settings` fails loud at startup if the flag is on with no secret). Off: `get_agent_context` (`roboco/api/deps.py`) and the WS `_require_panel_token` gate (`roboco/api/websocket.py`) are byte-for-byte unchanged (header-trust). On: header-trust is dead for humans — any agent-role claim (`ceo` OR a privileged `main_pm`/`cell_pm`/board role) with no valid HMAC token or session cookie is 401, closing the header-spoof hole on the host-published `:8000` port for every role, not just `ceo` (real agents always carry a signed token, so they're unaffected); the agent-fleet HMAC path (and the orchestrator's `system` self-PATCH) keeps working unmodified in both modes; a valid session cookie authenticates as the single seeded CEO user. New `users` table (migration 058, `UserTable` in `roboco/db/tables.py`) backs FastAPI Users' `SQLAlchemyUserDatabase`; no registration router — `roboco/api/auth/seed.py` idempotently upserts exactly one row from env at startup (by primary key, so an email change renames the row instead of duplicating it). `roboco/api/auth/backend.py` wires a **cookie** transport (httponly, secure, samesite=lax) + a `JWTStrategy` subclass that binds each token to a fingerprint of the current `hashed_password`, so rotating the seeded password invalidates every prior session. Session lifetime is **sliding**: every authenticated request through `get_agent_context` re-mints + re-sets the cookie (`_slide_session_cookie`), so an active session never expires — only genuine inactivity past `cloud_auth_cookie_max_age` (default 30 days) logs out. `GET /api/auth/status` is always mounted (public); `/api/auth/login` + `/api/auth/logout` mount only when armed (`roboco/api/auth/routes.py`, mirroring `apply_guard`'s conditional mount). Panel: `(auth)/login/page.tsx` + `proxy.ts` (the Next 16 rename of `middleware.ts`; probes `/auth/status` over the docker-internal orchestrator URL, not through nginx, and fails open to "off" on any probe error/timeout) gate the `(dashboard)` group; `client.ts` adds `withCredentials` + a 401→`/login` redirect. nginx needs no changes (`/api/auth/*` rides the existing `/api/` proxy location) — but its own static `X-Agent-Token` injection (`ROBOCO_PANEL_AGENT_TOKEN`) is itself a valid HMAC credential that bypasses login when present, so a deployment arming cloud auth for real public exposure should leave that token unset (the two are alternative human-auth tiers, not layered).
@@ -29,6 +29,7 @@
 | `pr_update` | `pr_update(see do_server)` |
 | `draft_playbook` | `draft_playbook(title: str, problem: str, procedure: str, tags: list[str] = PydanticUndefined, source_task_id: UUID | None = None)` |
 | `propose_video` | `propose_video(composition_id: str, x_caption: str, tiktok_caption: str, platforms: list[str], input_props: str | Any | None = None)` |
 | `request_sandbox` | `request_sandbox(services: list[str] | None = None)` |
 | `notify_list` | `notify_list(unread_only: bool = True, pending_ack_only: bool = False, limit: int = 20)` |
 | `notify_get` | `notify_get(notification_id: UUID)` |
 | `notify_ack` | `notify_ack(notification_id: UUID)` |
@@ -24,6 +24,7 @@
 | `dm` | `dm(recipient: str, text: str, task_id: UUID | None = None, skill: str | None = None)` |
 | `evidence` | `evidence(task_id: UUID)` |
 | `draft_playbook` | `draft_playbook(title: str, problem: str, procedure: str, tags: list[str] = PydanticUndefined, source_task_id: UUID | None = None)` |
 | `request_sandbox` | `request_sandbox(services: list[str] | None = None)` |
 | `notify_list` | `notify_list(unread_only: bool = True, pending_ack_only: bool = False, limit: int = 20)` |
 | `notify_get` | `notify_get(notification_id: UUID)` |
 | `notify_ack` | `notify_ack(notification_id: UUID)` |
@@ -38,6 +38,7 @@ real tools live in their agent_sdk drivers, not role_config.
 | `pr_update` | `pr_update(see do_server)` |
 | `draft_playbook` | `draft_playbook(title: str, problem: str, procedure: str, tags: list[str] = PydanticUndefined, source_task_id: UUID | None = None)` |
 | `propose_video` | `propose_video(composition_id: str, x_caption: str, tiktok_caption: str, platforms: list[str], input_props: str | Any | None = None)` |
 | `request_sandbox` | `request_sandbox(services: list[str] | None = None)` |
 | `notify_list` | `notify_list(unread_only: bool = True, pending_ack_only: bool = False, limit: int = 20)` |
 | `notify_get` | `notify_get(notification_id: UUID)` |
 | `notify_ack` | `notify_ack(notification_id: UUID)` |
@@ -67,6 +68,7 @@ real tools live in their agent_sdk drivers, not role_config.
 | `dm` | `dm(recipient: str, text: str, task_id: UUID | None = None, skill: str | None = None)` |
 | `evidence` | `evidence(task_id: UUID)` |
 | `draft_playbook` | `draft_playbook(title: str, problem: str, procedure: str, tags: list[str] = PydanticUndefined, source_task_id: UUID | None = None)` |
 | `request_sandbox` | `request_sandbox(services: list[str] | None = None)` |
 | `notify_list` | `notify_list(unread_only: bool = True, pending_ack_only: bool = False, limit: int = 20)` |
 | `notify_get` | `notify_get(notification_id: UUID)` |
 | `notify_ack` | `notify_ack(notification_id: UUID)` |
@@ -4690,8 +4690,9 @@ The AgentOrchestrator is the runtime brain of RoboCo: it owns the per-agent Dock
 | AgentOrchestrator._should_skip_live_reap | method | roboco/runtime/orchestrator.py:8750 | Spare a live container from reaping UNLESS wedged (grok) or gateway-broken past grace (those kill+evict). |
 | AgentOrchestrator._assignee_is_provider_parked | method | roboco/runtime/orchestrator.py:8527 | True if a task's assignee is provider-parked; reaper skips it so the claim survives for probe-resume. |
 | AgentOrchestrator._reap_with_service | method | roboco/runtime/orchestrator.py:8770 | Inner stale-claim reaper: skip live (unless wedged/broken), skip provider-parked, unclaim_for_reaper the rest. |
-| AgentOrchestrator._maybe_provision_sandbox | method | roboco/runtime/orchestrator.py:2029 | Provision this spawn's sandbox engines (postgres/redis/mongo via `SandboxProvisioner` + the `SANDBOX_ENGINES` registry) when `sandbox_db_enabled` + the project's `sandbox_services` are set; None (byte-for-byte legacy path) otherwise. Fail-loud once opted in — a provisioning failure refuses the spawn. |
+| AgentOrchestrator._sandbox_available_services | method | roboco/runtime/orchestrator.py:2223 | On-demand model (2026-07-08): availability probe only — which services this spawn's project opted into (`sandbox_db_enabled` + `sandbox_services`); `[]` if off/not-opted, byte-for-byte legacy path. No provisioning happens here anymore (see `ensure_sandbox`), so a spawn never fails on sandbox infra. |
-| AgentOrchestrator._append_sandbox_env | staticmethod | roboco/runtime/orchestrator.py:2639 | Inject the sandbox's `ROBOCO_TEST_*` env (postgres `ROBOCO_TEST_DB_*` / redis `ROBOCO_TEST_REDIS_*` / mongo `ROBOCO_TEST_MONGO_*`) via `SandboxInfo.emit_env` over the engine registry, called INSTEAD OF `_append_gate_env` whenever a sandbox was provisioned for this spawn. Emission is registry-driven, so a new engine's env vars land here with no orchestrator change. |
+| AgentOrchestrator.ensure_sandbox | method | roboco/runtime/orchestrator.py:2251 | On-demand model: idempotent provision called by the `request_sandbox` do-verb. Cache hit (`_sandbox_info`, keyed by agent slug) covering the requested services returns the same creds; a miss provisions via `SandboxProvisioner.provision` (pre-clear tears down stale same-named containers) and caches the result. Evicted at teardown and by the janitor sweep; in-memory only — an orchestrator restart forgets it (next call re-provisions). |
 | AgentOrchestrator._append_sandbox_marker_env | staticmethod | roboco/runtime/orchestrator.py:2846 | On-demand model: replaces the old eager `_append_sandbox_env` — injects only a cheap informational marker (`ROBOCO_SANDBOX_SERVICES_AVAILABLE=<csv>`), never creds, naming the services `request_sandbox()` will provision on demand. Called INSTEAD OF `_append_gate_env` for an opted-in project's spawn. |
 | AgentOrchestrator._sandbox_janitor_sweep | method | roboco/runtime/orchestrator.py:9426 | Best-effort: remove sandbox containers whose owner agent is gone; rides the reaper tick, error-isolated. |
 | AgentOrchestrator._x_mentions_poll_loop | method | roboco/runtime/orchestrator.py:7431 | Default-off X-engine mentions-poll tick loop (`x_engine_enabled`); release-post drafts are event-driven, not from this loop. |
 | AgentOrchestrator._run_x_mentions_cycle | method | roboco/runtime/orchestrator.py:7453 | One mentions-poll pass: `get_x_engine(db).run_cycle()` + commit; testable without the sleep. |
@@ -4807,7 +4808,7 @@ stateDiagram-v2
 - `ROBOCO_GROK_MAX_COST_USD` — grok budget kill-switch; `_GROK_RATE_LIMIT_EXIT_CODE=75`, `_GROK_AUTH_EXIT_CODE=78`, `_PROBE_GIVE_UP_THRESHOLD=30`.
 - `ROBOCO_CLAUDE_STUCK_KILL_SECONDS` (default 3600, min 600) — heartbeat-stale kill threshold for non-GROK agents; controls `_maybe_kill_stuck_claude`.
 - `ROBOCO_DISPATCHER_INTERVAL_SECONDS` (30), `ROBOCO_INTERACTIVE_IDLE_REAP_SECONDS`, `ROBOCO_GROK_*` backoff constants.
- `ROBOCO_SANDBOX_DB_ENABLED` (default off) — master switch for the sandboxed per-agent-spawn engine provisioner (`_maybe_provision_sandbox`/`_append_sandbox_env`/`_sandbox_janitor_sweep`); a project participates only when its `sandbox_services` column is also set. The service set is the `SANDBOX_ENGINES` registry (postgres / redis / mongo); adding an engine needs no orchestrator or env-emitter change.
+- `ROBOCO_SANDBOX_DB_ENABLED` (default off) — master switch for the sandboxed per-agent test DB/Redis/Mongo. On-demand model (2026-07-08): nothing is provisioned at spawn — `_sandbox_available_services` only probes+names the project's opted-in set (marker env), and `ensure_sandbox` provisions idempotently when an agent calls the `request_sandbox` do-verb (`ContentActions.request_sandbox`, gateway/content_actions.py). Teardown (`_sandbox_janitor_sweep` + every container-removal path) is unchanged. A project participates only when its `sandbox_services` column is also set. The service set is the `SANDBOX_ENGINES` registry (postgres / redis / mongo); adding an engine needs no orchestrator or env-emitter change.
 - `ROBOCO_DB_NETWORK_ISOLATED` (default off; set by the compose topology that carries the `roboco_data` network) — suppresses the legacy `_append_gate_env` prod-creds injection when postgres/redis are unreachable from the agent mesh.
 - `ROBOCO_CLOUD_AUTH_ENABLED` (+ `_EMAIL`/`_PASSWORD`/`_SECRET`/`_COOKIE_MAX_AGE`, default off) — cloud auth master switch; read by `roboco.api.deps.get_agent_context`/`roboco.api.auth.*`, not the orchestrator itself, but gates whether a spawned agent's own HMAC-token identity path is the sole non-CEO auth route.
 - `ROBOCO_X_ENGINE_ENABLED` (+ `_mentions_interval_seconds` / `_mentions_max_per_cycle` / `_mentions_min_engagement` / `_max_open_posts` / `_account_user_id` / `_request_timeout_seconds`, default off) — gates `_x_mentions_poll_loop`.
@@ -80,8 +80,9 @@ The AgentOrchestrator is the runtime brain of RoboCo: it owns the per-agent Dock
 | AgentOrchestrator._should_skip_live_reap | method | roboco/runtime/orchestrator.py:8750 | Spare a live container from reaping UNLESS wedged (grok) or gateway-broken past grace (those kill+evict). |
 | AgentOrchestrator._assignee_is_provider_parked | method | roboco/runtime/orchestrator.py:8527 | True if a task's assignee is provider-parked; reaper skips it so the claim survives for probe-resume. |
 | AgentOrchestrator._reap_with_service | method | roboco/runtime/orchestrator.py:8770 | Inner stale-claim reaper: skip live (unless wedged/broken), skip provider-parked, unclaim_for_reaper the rest. |
-| AgentOrchestrator._maybe_provision_sandbox | method | roboco/runtime/orchestrator.py:2029 | Provision this spawn's sandbox engines (postgres/redis/mongo via `SandboxProvisioner` + the `SANDBOX_ENGINES` registry) when `sandbox_db_enabled` + the project's `sandbox_services` are set; None (byte-for-byte legacy path) otherwise. Fail-loud once opted in — a provisioning failure refuses the spawn. |
+| AgentOrchestrator._sandbox_available_services | method | roboco/runtime/orchestrator.py:2223 | On-demand model (2026-07-08): availability probe only — which services this spawn's project opted into (`sandbox_db_enabled` + `sandbox_services`); `[]` if off/not-opted, byte-for-byte legacy path. No provisioning happens here anymore (see `ensure_sandbox`), so a spawn never fails on sandbox infra. |
-| AgentOrchestrator._append_sandbox_env | staticmethod | roboco/runtime/orchestrator.py:2639 | Inject the sandbox's `ROBOCO_TEST_*` env (postgres `ROBOCO_TEST_DB_*` / redis `ROBOCO_TEST_REDIS_*` / mongo `ROBOCO_TEST_MONGO_*`) via `SandboxInfo.emit_env` over the engine registry, called INSTEAD OF `_append_gate_env` whenever a sandbox was provisioned for this spawn. Emission is registry-driven, so a new engine's env vars land here with no orchestrator change. |
+| AgentOrchestrator.ensure_sandbox | method | roboco/runtime/orchestrator.py:2251 | On-demand model: idempotent provision called by the `request_sandbox` do-verb. Cache hit (`_sandbox_info`, keyed by agent slug) covering the requested services returns the same creds; a miss provisions via `SandboxProvisioner.provision` (pre-clear tears down stale same-named containers) and caches the result. Evicted at teardown and by the janitor sweep; in-memory only — an orchestrator restart forgets it (next call re-provisions). |
 | AgentOrchestrator._append_sandbox_marker_env | staticmethod | roboco/runtime/orchestrator.py:2846 | On-demand model: replaces the old eager `_append_sandbox_env` — injects only a cheap informational marker (`ROBOCO_SANDBOX_SERVICES_AVAILABLE=<csv>`), never creds, naming the services `request_sandbox()` will provision on demand. Called INSTEAD OF `_append_gate_env` for an opted-in project's spawn. |
 | AgentOrchestrator._sandbox_janitor_sweep | method | roboco/runtime/orchestrator.py:9426 | Best-effort: remove sandbox containers whose owner agent is gone; rides the reaper tick, error-isolated. |
 | AgentOrchestrator._x_mentions_poll_loop | method | roboco/runtime/orchestrator.py:7431 | Default-off X-engine mentions-poll tick loop (`x_engine_enabled`); release-post drafts are event-driven, not from this loop. |
 | AgentOrchestrator._run_x_mentions_cycle | method | roboco/runtime/orchestrator.py:7453 | One mentions-poll pass: `get_x_engine(db).run_cycle()` + commit; testable without the sleep. |
@@ -197,7 +198,7 @@ stateDiagram-v2
 - `ROBOCO_GROK_MAX_COST_USD` — grok budget kill-switch; `_GROK_RATE_LIMIT_EXIT_CODE=75`, `_GROK_AUTH_EXIT_CODE=78`, `_PROBE_GIVE_UP_THRESHOLD=30`.
 - `ROBOCO_CLAUDE_STUCK_KILL_SECONDS` (default 3600, min 600) — heartbeat-stale kill threshold for non-GROK agents; controls `_maybe_kill_stuck_claude`.
 - `ROBOCO_DISPATCHER_INTERVAL_SECONDS` (30), `ROBOCO_INTERACTIVE_IDLE_REAP_SECONDS`, `ROBOCO_GROK_*` backoff constants.
- `ROBOCO_SANDBOX_DB_ENABLED` (default off) — master switch for the sandboxed per-agent-spawn engine provisioner (`_maybe_provision_sandbox`/`_append_sandbox_env`/`_sandbox_janitor_sweep`); a project participates only when its `sandbox_services` column is also set. The service set is the `SANDBOX_ENGINES` registry (postgres / redis / mongo); adding an engine needs no orchestrator or env-emitter change.
+- `ROBOCO_SANDBOX_DB_ENABLED` (default off) — master switch for the sandboxed per-agent test DB/Redis/Mongo. On-demand model (2026-07-08): nothing is provisioned at spawn — `_sandbox_available_services` only probes+names the project's opted-in set (marker env), and `ensure_sandbox` provisions idempotently when an agent calls the `request_sandbox` do-verb (`ContentActions.request_sandbox`, gateway/content_actions.py). Teardown (`_sandbox_janitor_sweep` + every container-removal path) is unchanged. A project participates only when its `sandbox_services` column is also set. The service set is the `SANDBOX_ENGINES` registry (postgres / redis / mongo); adding an engine needs no orchestrator or env-emitter change.
 - `ROBOCO_DB_NETWORK_ISOLATED` (default off; set by the compose topology that carries the `roboco_data` network) — suppresses the legacy `_append_gate_env` prod-creds injection when postgres/redis are unreachable from the agent mesh.
 - `ROBOCO_CLOUD_AUTH_ENABLED` (+ `_EMAIL`/`_PASSWORD`/`_SECRET`/`_COOKIE_MAX_AGE`, default off) — cloud auth master switch; read by `roboco.api.deps.get_agent_context`/`roboco.api.auth.*`, not the orchestrator itself, but gates whether a spawned agent's own HMAC-token identity path is the sole non-CEO auth route.
 - `ROBOCO_X_ENGINE_ENABLED` (+ `_mentions_interval_seconds` / `_mentions_max_per_cycle` / `_mentions_min_engagement` / `_max_open_posts` / `_account_user_id` / `_request_timeout_seconds`, default off) — gates `_x_mentions_poll_loop`.
@@ -2,7 +2,9 @@
 ## What It Is
-A per-agent-spawn throwaway Postgres/Redis/Mongo set, provisioned as **sibling containers** to the agent container (never docker-in-agent — the docker socket/CLI stay structurally absent from agent images). Implemented in `roboco/runtime/sandbox.py` (`SandboxProvisioner`), wired into the orchestrator's spawn path.
+A throwaway Postgres/Redis/Mongo set, provisioned as **sibling containers** to the agent container (never docker-in-agent — the docker socket/CLI stay structurally absent from agent images). Implemented in `roboco/runtime/sandbox.py` (`SandboxProvisioner`) and `AgentOrchestrator.ensure_sandbox`.
 **On-demand since 2026-07-08**: a sandbox is provisioned only when an agent actually asks for one, via the `request_sandbox` do-verb (developer + QA roles) — not eagerly at spawn. See "The `request_sandbox` verb" below; the design rationale lives in `docs/internal/specs/2026-07-08-sandbox-on-demand.md`.
 It replaces — never coexists with — the legacy `_append_gate_env` behavior that hands an agent RoboCo's own production Postgres credentials so its `make quality` gate can run the DB-backed test suite instead of a hollow unit-only subset.
@@ -20,35 +22,84 @@ The service set is a **pluggable engine registry**, not a hardcoded postgres+red
 | Variable | Default | Effect |
 |----------|---------|--------|
-| `ROBOCO_SANDBOX_DB_ENABLED` | `false` | Master switch. Off = spawning behaves exactly as today (the legacy prod-creds gate-env injection, itself gated by `ROBOCO_TOOLCHAIN_MATCH_ENABLED`). Panel-toggleable (Settings → Feature Flags). |
+| `ROBOCO_SANDBOX_DB_ENABLED` | `false` | Master switch. Off = spawning behaves exactly as today (the legacy prod-creds gate-env injection, itself gated by `ROBOCO_TOOLCHAIN_MATCH_ENABLED`), and `request_sandbox` refuses. Panel-toggleable (Settings → Feature Flags). |
-A second, per-project gate applies even when the flag is on: only a project with its `sandbox_services` column set (e.g. `["postgres", "redis", "mongo"]`; migration `057`, nullable/additive) participates. Every other project's spawns are byte-for-byte unaffected. Mongo rides the same column — no new migration, no new feature flag; it is just another registry entry.
+A second, per-project gate applies even when the flag is on: only a project with its `sandbox_services` column set (e.g. `["postgres", "redis", "mongo"]`; migration `057`, nullable/additive) participates. Every other project's spawns are byte-for-byte unaffected. Mongo rides the same column — no new migration, no new feature flag; it is just another registry entry. `request_sandbox` may request any subset of the opted-in set (or omit `services` for the whole set); anything outside it is rejected naming the allowed set.
 ## The `request_sandbox` verb
 Provisioning is **on-demand**: nothing is provisioned at spawn. A developer or QA agent calls the `request_sandbox` content tool (on `roboco-do`) when it actually needs a sandboxed DB, and the orchestrator provisions it inline. Wiring: `roboco/api/schemas/v1/do.py` `RequestSandboxRequest` → `POST /api/v1/do/request_sandbox` → `ContentActions.request_sandbox` (`roboco/services/gateway/content_actions.py`) → `AgentOrchestrator.ensure_sandbox` → `roboco/mcp/do_server.py`'s `request_sandbox()` tool (1080s timeout — `ensure_sandbox` always provisions the project's full opted-in set on first call, so an all-three-engines-cold first request is the norm the timeout must cover, not a rare edge case).
 ```python
 request_sandbox(services: list[str] | None = None)
 ```
 `services` omitted means the project's whole opted-in set. Guards fire in order, each with a clean `invalid_state` envelope + `remediate`:
 1. `ROBOCO_SANDBOX_DB_ENABLED` off → refused before any DB lookup.
 2. No active, project-bound task (agent hasn't `give_me_work`'d) → refused.
 3. Project has no `sandbox_services` opted in → refused.
 4. A requested service outside the project's opted set → refused, remediate **names the allowed set**.
 5. Orchestrator handle unavailable (e.g. mid-restart) → refused, but **retryable** — the one guard that isn't a permanent no.
 Only past all five does it call `ensure_sandbox` and provision. A genuine provisioning failure (image pull, readiness timeout) also surfaces as a retryable `invalid_state`, never a spawn refusal — sandbox trouble can no longer block a spawn or an agent's turn.
 On success, creds return in the ok-envelope's `evidence`, one entry per service — no env injection, no schema change:
 ```json
 {
  "postgres": {
    "host": "roboco-sandbox-pg-be-dev-1",
    "port": 5432,
    "user": "sandbox",
    "password": "<random>",
    "database": "sandbox",
    "env": {
      "ROBOCO_TEST_DB_HOST": "roboco-sandbox-pg-be-dev-1",
      "ROBOCO_TEST_DB_PORT": "5432",
      "ROBOCO_TEST_DB_USER": "sandbox",
      "ROBOCO_TEST_DB_PASSWORD": "<random>",
      "ROBOCO_TEST_DB_ADMIN_DB": "sandbox"
    }
  }
 }
 ```
 The `env` sub-dict (`SandboxInfo.as_payload()`, `roboco/models/sandbox.py`) carries the exact same variable names the legacy env-injection path used, so an agent can `export` them verbatim for gate tooling that reads `ROBOCO_TEST_*`. Networking needs no extra step: sandboxes join `roboco_default` at `docker run`, the same network every agent is on, so DNS resolves the moment the sibling starts.
 `ensure_sandbox` always provisions the project's **whole opted-in set** on first call, regardless of what a given `request_sandbox` call named — so calling it again for any subset or superset of that opted set is a guaranteed cache hit (same creds, no docker calls), and a live container is never torn down mid-session by a later, broader request. `services` only scopes what comes back in this call's `evidence`; the response payload is filtered down to that subset even though the full set was provisioned. A cache hit is also re-verified live (`SandboxProvisioner.is_live`) before being trusted — a container OOM-killed or removed out-of-band evicts the stale entry and triggers a fresh full-set provision with new creds. Concurrent calls for the same agent (e.g. a client timeout + retry) are serialized behind a per-agent-slug `asyncio.Lock` so they can't race `provision()`/`teardown()` against each other. `ensure_sandbox` is always called with the **caller's own** authenticated agent slug — a caller can never reach another agent's sandbox.
 Only `developer` and `qa` roles carry `request_sandbox` in their spawn manifest (`roboco/services/gateway/role_config.py` `_DEV_DO` / `_QA_DO`) — the DB-needing gate roles. It is carried unconditionally on those manifests (declarative); the real gating is the project opt-in check inside the verb itself.
 ## Provisioning
-For an opted-in project's spawn, the orchestrator provisions each requested service through a single generic `_provision_engine` (no per-engine branch): it generates a random 32-hex-char password (`secrets.token_hex(16)`), pre-pulls the image, `docker run`s the sibling container, and polls the engine's readiness probe up to its deadline.
+`ensure_sandbox(agent_slug, requested, opted)` provisions `requested | opted` (in practice the project's whole opted-in set) through a single generic `_provision_engine` (no per-engine branch): it generates a random 32-hex-char password (`secrets.token_hex(16)`), pre-pulls the image, `docker run`s the sibling container, and polls the engine's readiness probe up to its deadline.
 - **Postgres**: named `roboco-sandbox-pg-{agent_id}`, `--tmpfs /var/lib/postgresql/data` (no disk persistence), `--memory 512m --cpus 1`, user/db both `sandbox`, readiness via `pg_isready` up to 60s.
 - **Redis**: named `roboco-sandbox-redis-{agent_id}`, same memory/cpu caps, `redis-server --requirepass`, readiness via `redis-cli -a … ping` up to 15s.
 - **Mongo**: named `roboco-sandbox-mongo-{agent_id}`, `--tmpfs /data/db`, same memory/cpu caps, root user/db `sandbox`/`sandbox`, readiness via `mongosh` ping (auth db `admin`) up to 60s.
-All are labeled `roboco.sandbox=1` plus an owner label (`roboco.sandbox.owner=roboco-agent-{agent_id}`) so the janitor can find them. A provisioning failure is **fail-loud**: the spawn is refused (`AgentReadinessError`) rather than starting an agent whose gate can't run against a broken DB, and any already-provisioned sibling is torn down before re-raising. A stale same-named sandbox left by a crash-missed teardown is pre-cleared before provisioning, so a leftover container can't collide with a fresh `docker run`.
+All are labeled `roboco.sandbox=1` plus an owner label (`roboco.sandbox.owner=roboco-agent-{agent_id}`) so the janitor can find them. A stale same-named sandbox left by a crash-missed teardown is pre-cleared before provisioning, so a leftover container can't collide with a fresh `docker run`. A provisioning failure now surfaces as a retryable envelope on the verb (see above) rather than refusing anything — there is no spawn to refuse, since the sandbox is requested well after the agent is already running.
 ### Image pre-pull
-`_ensure_image` `docker image inspect`s the engine's image and, on absence, `docker pull`s it (300s timeout) **before** `docker run`. Without this a NAS cold pull would hit the 20s run timeout, get killed, and re-pull forever on every respawn. The inspect-then-pull runs per service per spawn, so an already-present image short-circuits in milliseconds.
+`_ensure_image` `docker image inspect`s the engine's image and, on absence, `docker pull`s it (300s timeout) **before** `docker run`. Without this a NAS cold pull would hit the 20s run timeout, get killed, and re-pull forever on every retry. The inspect-then-pull runs per service per call, so an already-present image short-circuits in milliseconds.
-## Injected environment
+## Spawn-time availability probe
-Instead of the legacy `ROBOCO_TEST_DB_*` pointing at RoboCo's own production Postgres, the sandbox's own host/port/user/password are injected. Env var names are preserved per engine so an existing project's conftest needs no change: `ROBOCO_TEST_DB_*` (postgres, incl. `ROBOCO_TEST_DB_ADMIN_DB`), `ROBOCO_TEST_REDIS_*` (redis), and `ROBOCO_TEST_MONGO_*` (mongo, incl. `ROBOCO_TEST_MONGO_AUTH_DB=admin`). The orchestrator's `_append_sandbox_env` is a single `cmd.extend(info.emit_env())` over the registry — a new engine's env lands with no orchestrator edit. It runs **instead of** `_append_gate_env` whenever a sandbox was provisioned for that spawn.
+Spawn itself no longer provisions anything. For an opted-in project, `AgentOrchestrator._sandbox_available_services` is a cheap DB lookup (best-effort — a hiccup degrades to "no sandbox" rather than blocking the spawn) that returns the project's opted-in service list, and the spawn path injects a marker env `ROBOCO_SANDBOX_SERVICES_AVAILABLE=postgres,redis` (never creds) plus one line in the agent's session briefing naming `request_sandbox()` explicitly — cheap and kills a discovery failure mode where an agent doesn't know the tool exists. `AgentConfig.sandbox_info` and the old eager `_append_sandbox_env` are gone; `_append_sandbox_marker_env` runs **instead of** `_append_gate_env` for an opted-in project's spawn (never both).
 ## Lifetime and teardown
-A sandbox's lifetime tracks its owning agent container 1:1: torn down (`stop` → `kill` fallback → `rm -f`, all best-effort and idempotent) at every container-removal path. Teardown iterates **all** registered engines (`SANDBOX_ENGINES.values()`) — a mongo sandbox is reaped by the same path that reaps postgres/redis, with no per-engine teardown branch. An **orphan janitor** also runs at orchestrator startup and on each reaper tick: it lists every `roboco.sandbox=1` container, cross-references live agent containers, and removes any sandbox whose owner is gone.
+A sandbox no longer only dies with its container: `AgentOrchestrator.release_sandbox(agent_slug)` tears it down at the end of the caller's own engagement with its work — the Choreographer calls it (best-effort, never failing the verb) on the SUCCESSFUL exit of `i_am_done`, `unclaim`, `i_am_idle`, `pass_review`/`fail_review`, and `i_documented`, so an agent that finishes then idles or picks up unrelated work doesn't leave a sidecar running for the rest of its session. `release_sandbox` is a fast no-op for the common case of no cached sandbox (a single dict check, before any lock or docker call) and otherwise reuses the same teardown + cache-eviction pairing as container removal. A sandbox's lifetime STILL tracks its owning agent container 1:1 as the backstop: torn down (`stop` → `kill` fallback → `rm -f`, all best-effort and idempotent) at every container-removal path. Teardown iterates **all** registered engines (`SANDBOX_ENGINES.values()`) — a mongo sandbox is reaped by the same path that reaps postgres/redis, with no per-engine teardown branch. An **orphan janitor** also runs at orchestrator startup and on each reaper tick: it lists every `roboco.sandbox=1` container, cross-references live agent containers, and removes any sandbox whose owner is gone.
-The janitor has a **grace window** (`_JANITOR_GRACE_SECONDS`, 180s): a sandbox is provisioned *before* its agent container exists, so a sweep racing a mid-flight spawn would otherwise see "owner not live yet" and reap a fresh sandbox out from under a spawn still starting up. Owners provisioned within the grace window are skipped by that pass. The pre-spawn stale-clear (above) likewise never touches a just-provisioned sandbox.
+The janitor has a **grace window** (`_JANITOR_GRACE_SECONDS`, 180s): a sweep racing a mid-flight `request_sandbox` call would otherwise see "owner not live yet" and reap a sandbox out from under a request still in progress. Owners provisioned within the grace window are skipped by that pass. The pre-spawn stale-clear likewise never touches a sandbox just requested via the verb.
 **Creds cache.** `AgentOrchestrator._sandbox_info` (agent slug → last-provisioned `SandboxInfo`, always covering the project's full opted-in set) is what makes repeat `request_sandbox` calls cheap. It is evicted at every teardown path and by the janitor sweep whenever it reaps an orphan — so a torn-down sandbox's cached creds never outlive the container — and also on a failed on-demand liveness check at cache-hit time (`SandboxProvisioner.is_live`), which catches a container that died without going through any of those explicit teardown paths (OOM-killed, manually removed). A per-agent-slug `asyncio.Lock` (`AgentOrchestrator._sandbox_locks`) wraps the whole check-cache → provision → store section so two concurrent `ensure_sandbox` calls for one agent can't race each other's `provision()`/`teardown()`. **Known ceiling:** the cache is in-memory only; an orchestrator restart forgets it. The next `request_sandbox` call re-provisions (the pre-clear tears down any still-running stale container) and returns fresh creds — agents already treat creds as session-scoped, and a connection to the torn-down sandbox simply fails loudly rather than silently.
 ## Related
 - `docs/internal/specs/2026-07-08-sandbox-on-demand.md` — the on-demand design spec and rationale
 - `docs/rag/tools/task-tools.md` — `request_sandbox` alongside the rest of the dev/QA verb surface
 - `docs/rag/architecture/config-reference.md` — full env var table
 - `docs/rag/architecture/db-network-isolation.md` — the network-topology change this pairs with in a NAS deploy (agents structurally can't reach production Postgres/Redis at all; sandbox is the DB-needing project's alternative)
@@ -147,3 +147,7 @@ progress(task_id, message="API skeleton landed", plan_step="2")
 ```
 Your plan's steps are the progress checklist; the percentage is derived from completed steps — you do not set it.
 ## Sandbox DB/Redis/Mongo (Developer + QA)
 `request_sandbox(services=None)` — a **content tool** on `roboco-do`, not a flow verb — provisions a throwaway sandbox Postgres/Redis/Mongo on demand, for a project that opted in (`projects.sandbox_services`). Only `developer` and `qa` carry it. Omit `services` for the project's whole opted-in set; requesting one outside it is rejected naming the allowed set. Creds come back in the envelope's `evidence`, one entry per service, including ready-to-`export` `ROBOCO_TEST_*` values for gate tooling. Calling it again is a cheap no-op (same creds). See `docs/rag/architecture/sandbox-db.md`.
@@ -744,6 +744,10 @@ async def get_content_actions(
    db_session: DbSession,
 ) -> ContentActions:
    """Build a ContentActions with all service dependencies wired up."""
    # Orchestrator handle for request_sandbox's ensure_sandbox call — same
    # None-safe injection as get_choreographer (the orchestrator may not be
    # initialised yet, e.g. during startup).
    orch: AgentOrchestrator | None = _ServiceHolder.orchestrator
    return ContentActions(
        ContentActionsDeps(
            task=TaskService(db_session),
@@ -754,6 +758,7 @@ async def get_content_actions(
            notifications=NotificationService(),
            notification_delivery=NotificationDeliveryService(db_session),
            evidence_repo=EvidenceRepo(db_session),
            orchestrator=orch,
        )
    )
@@ -31,6 +31,7 @@ from roboco.api.schemas.v1.do import (
    PRUpdateRequest,
    ReadMessagesRequest,
    RejectPlaybookRequest,
    RequestSandboxRequest,
 )
 from roboco.security import (
    guard_deco,
@@ -257,6 +258,17 @@ async def do_evidence(
    return envelope_to_response(env, request)
@router.post("/request_sandbox")
 async def do_request_sandbox(
    request: Request,
    body: RequestSandboxRequest,
    x_agent_id: _AgentIdHeader,
    actions: _ContentActionsDep,
 ) -> dict:
    env = await actions.request_sandbox(agent_id=x_agent_id, services=body.services)
    return envelope_to_response(env, request)
 # ---------------------------------------------------------------------------
 # Wave 1 — pre-gateway parity
 # ---------------------------------------------------------------------------
@@ -168,6 +168,13 @@ class EvidenceRequest(BaseModel):
    task_id: UUID
 class RequestSandboxRequest(BaseModel):
    """On-demand sandbox DB/Redis/Mongo. Omitted `services` = the project's
    whole opted-in set."""
    services: list[str] | None = None
 class ProgressRequest(BaseModel):
    """Progress update; % is DERIVED from the plan checklist.
@@ -42,6 +42,14 @@ _TIMEOUT = 30
 # shared _TIMEOUT above is tuned for fast content-tool calls (note/dm/
 # evidence) and would give up first — client must outlast the server op.
 _COMMIT_TIMEOUT = 190
 # request_sandbox provisions inline; ensure_sandbox now always provisions the
 # project's FULL opted-in set on first call (kills the superset/teardown
 # race — see ensure_sandbox's docstring), so an all-three-cold first request
 # is the norm, not the rare case. Worst case: 3 x 300s cold pulls
 # (SandboxProvisioner._DOCKER_PULL_TIMEOUT_SECONDS) + readiness (~135s) ~=
 # 1035s — images are pre-pulled at startup in practice, so cold pulls here
 # are the exception, but the timeout must cover the worst case anyway.
 _SANDBOX_TIMEOUT = 1080
 # Tight timeout for SDK loopback — local sidecar; gateway path must not stall.
 _SDK_TIMEOUT = 2.0
 # FastAPI's default missing-route status. Every /api/v1/do/* route returns
@@ -694,6 +702,26 @@ def evidence(task_id: str) -> dict[str, Any]:
    return _post("/api/v1/do/evidence", {"task_id": task_id})
 def request_sandbox(services: list[str] | None = None) -> dict[str, Any]:
    """Provision (or reuse) a throwaway sandbox DB/Redis/Mongo for YOUR active task.
    On-demand — nothing is provisioned at spawn. Omit ``services`` to get the
    project's whole opted-in set; requesting a service the project didn't opt
    into is rejected with the allowed set named. Creds come back in
    ``evidence``, one entry per service: ``{host, port, user, password,
    database, env: {ROBOCO_TEST_*: value}}`` — export the ``env`` values
    verbatim for gate tooling that reads them. The whole opted-in set is
    provisioned on first call, so calling this again for any subset or
    superset of it is a cheap no-op (same creds, no re-provisioning); a
    project that never opted into sandbox services will reject this.
    """
    return _post(
        "/api/v1/do/request_sandbox",
        {"services": services},
        timeout=_SANDBOX_TIMEOUT,
    )
 def draft_playbook(
    title: str,
    problem: str,
@@ -891,6 +919,7 @@ _TOOLS: dict[str, Any] = {
    "dm": dm,
    "notify": notify,
    "evidence": evidence,
    "request_sandbox": request_sandbox,
    "progress": progress,
    "notify_list": notify_list,
    "notify_get": notify_get,
@@ -11,8 +11,6 @@ from pathlib import Path
 from typing import Any
 from uuid import UUID, uuid4
 from roboco.models.sandbox import SandboxInfo
 class OrchestratorAgentState(StrEnum):
    """Agent lifecycle states in the orchestrator."""
@@ -66,10 +64,12 @@ class OrchestratorAgentConfig:
    provider_type: str = "anthropic"
    provider_base_url: str | None = None
    provider_auth_token: str | None = None
-    # Set when a sandbox DB/Redis was provisioned for this spawn
+    # Services this spawn's project has opted into (sandbox_db_enabled + the
-    # (sandbox_db_enabled + the project's sandbox_services). Its presence
+    # project's sandbox_services) — an availability probe only. A non-empty
-    # suppresses the legacy `_append_gate_env` prod-creds injection.
+    # list suppresses the legacy `_append_gate_env` prod-creds injection in
-    sandbox_info: SandboxInfo | None = None
+    # favor of a marker env; actual provisioning happens on demand via the
    # `request_sandbox` do-verb (`AgentOrchestrator.ensure_sandbox`), never here.
    sandbox_available_services: list[str] = field(default_factory=list)
@dataclass
@@ -11,6 +11,7 @@ from __future__ import annotations
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Any
@dataclass(frozen=True)
@@ -45,6 +46,26 @@ class SandboxInfo:
            env.extend(SANDBOX_ENGINES[name].emit_env(conn))
        return env
    def as_payload(self) -> dict[str, dict[str, Any]]:
        """Per-service creds dict for the ``request_sandbox`` verb's evidence.
        Same variable names as ``emit_env`` (docs/env parity) but keyed for
        direct agent consumption (JSON) rather than ``docker run -e`` args.
        """
        out: dict[str, dict[str, Any]] = {}
        for name, conn in self.services.items():
            args = SANDBOX_ENGINES[name].emit_env(conn)
            env = dict(pair.split("=", 1) for pair in args[1::2])
            out[name] = {
                "host": conn.host,
                "port": conn.port,
                "user": conn.user,
                "password": conn.password,
                "database": conn.database,
                "env": env,
            }
        return out
 class SandboxEngine(ABC):
    """One sandbox service kind: how to run it, probe it, and feed its creds."""
@@ -795,6 +795,14 @@ class AgentOrchestrator:
    - Cost-efficient on-demand spawning
    """
    # Per-agent-slug lock serializing ensure_sandbox's check-cache -> provision
    # -> store section. Declared here (not in __init__, to keep its statement
    # count under the gate) and lazily allocated on first use.
    _sandbox_locks: dict[str, asyncio.Lock]
    # Expected-stop breadcrumbs (agent_id -> (reason, monotonic ts)); lazily
    # allocated by _record_expected_stop, same statement-budget rationale.
    _expected_stops: dict[str, tuple[str, float]]
    def __init__(
        self,
        mcp_config_dir: Path | None = None,
@@ -811,16 +819,18 @@ class AgentOrchestrator:
        # AGENT_NETWORK itself) so a future network-isolation change only
        # has to flip this constant here — sandboxes ride along.
        self._sandbox = SandboxProvisioner(network=AGENT_NETWORK)
        # On-demand sandbox creds cache (agent slug -> last-provisioned info),
        # consulted by ensure_sandbox / request_sandbox. Evicted at teardown
        # and by the janitor sweep. Known ceiling: in-memory only — an
        # orchestrator restart forgets it; the next request_sandbox call
        # re-provisions (provision()'s pre-clear tears down any stale
        # container) with fresh creds.
        self._sandbox_info: dict[str, SandboxInfo] = {}
        # Gateway-health grace tracker: agent slug -> first time its gateway was
        # seen broken. Tolerates a transient probe miss before the reaper recovers
        # a broken-but-alive agent (see _maybe_recover_broken_gateway).
        self._gateway_broken_since: dict[str, datetime] = {}
        self._waiting_records: dict[str, WaitingRecord] = {}
        # Diagnostics only, in-memory: agent_id -> (reason, monotonic ts) for
        # the most recent orchestrator-initiated stop/kill, so the exit
        # monitor can tell an attributed stop from a truly unexplained one
        # (see _record_expected_stop / _consume_expected_stop).
        self._expected_stops: dict[str, tuple[str, float]] = {}
        # #71: a resumed agent's WaitingRecord is torn down only once liveness is
        # confirmed (not on a bare launch) — a container that launches then dies
        # immediately would otherwise strand its task until the reaper's TTL.
@@ -2094,18 +2104,19 @@ class AgentOrchestrator:
            git_context, project_slug, team, agent_id, task_id
        )
-        # Provision this spawn's sandbox DB/Redis (flag + per-project opt-in),
+        # Availability probe only (flag + per-project opt-in) — sandboxes are
-        # before `docker run` so its connection info can be injected as env.
+        # provisioned on demand via the request_sandbox do-verb
-        # Fail-loud on a provisioning failure (see _maybe_provision_sandbox).
+        # (ensure_sandbox), never here, so a spawn never fails on sandbox
-        sandbox_info = await self._maybe_provision_sandbox(
+        # infrastructure.
-            agent_id, project_slug, task_id
+        sandbox_services = await self._sandbox_available_services(project_slug)
        )
        agent_settings_path = self._generate_agent_settings(
            agent_id, canonical_role, cwd_path, cell_workspace_path
        )
-        briefing_path = await self._write_agent_briefing(agent_id, task_id, cwd_path)
+        briefing_path = await self._write_agent_briefing(
            agent_id, task_id, cwd_path, sandbox_services
        )
        await self._ensure_agent_image(agent_id)
        mcp_config_path = await self._generate_mcp_config(agent_id, git_context)
@@ -2123,7 +2134,7 @@ class AgentOrchestrator:
            provider_type=route.provider_type.value,
            provider_base_url=route.base_url,
            provider_auth_token=route.auth_token,
-            sandbox_info=sandbox_info,
+            sandbox_available_services=sandbox_services,
        )
        instance = AgentInstance(
            agent_id=agent_id,
@@ -2224,22 +2235,19 @@ class AgentOrchestrator:
                f" (task={task_id}): {e}; will retry next tick"
            ) from e
-    async def _maybe_provision_sandbox(
+    async def _sandbox_available_services(self, project_slug: str) -> list[str]:
-        self, agent_id: str, project_slug: str, task_id: str | None
+        """Which sandbox services this project's spawn may request on-demand.
    ) -> SandboxInfo | None:
        """Provision this spawn's sandbox DB/Redis, or None if not opted in.
-        Off (flag or project) => None, byte-for-byte identical to today (the
+        Off (flag or project) => [], byte-for-byte identical to today (the
        legacy `_append_gate_env` prod-creds injection stays active). The
-        project lookup itself is best-effort (a DB hiccup here degrades to
+        project lookup is best-effort (a DB hiccup degrades to "no sandbox"
-        "no sandbox" rather than blocking every spawn on a transient error —
+        rather than blocking the spawn). Provisioning itself no longer
-        `_ensure_worktree_before_spawn` already fails loud on a genuine DB
+        happens here — it is on-demand via the `request_sandbox` do-verb
-        outage). Once a project has opted in, an actual provisioning failure
+        (see `ensure_sandbox`), so a spawn never fails on sandbox
-        (container won't start / never becomes ready) IS fail-loud: an agent
+        infrastructure.
        whose gate can't run must never spawn.
        """
        if not settings.sandbox_db_enabled:
-            return None
+            return []
        from roboco.db.base import get_db_context
        from roboco.services.project import get_project_service
@@ -2248,31 +2256,86 @@ class AgentOrchestrator:
                project = await get_project_service(db).get_by_slug(project_slug)
        except Exception as e:
            logger.warning(
-                "sandbox project lookup failed; skipping sandbox provisioning",
+                "sandbox project lookup failed; no sandbox available this spawn",
                agent_id=agent_id,
                project_slug=project_slug,
                error=str(e),
            )
-            return None
+            return []
-        services = list(project.sandbox_services or []) if project else []
+        return list(project.sandbox_services or []) if project else []
-        if not services:
+
-            return None
+    async def ensure_sandbox(
-        try:
+        self, agent_slug: str, requested: list[str], opted: list[str]
-            return await self._sandbox.provision(agent_id, services)
+    ) -> SandboxInfo:
-        except Exception as e:
+        """Idempotent on-demand provision, called by the `request_sandbox` verb.
-            # str(TimeoutError()) == "" — include the type so a bare timeout
+
-            # (a cold image pull exceeding the run deadline) self-diagnoses.
+        DEVIATION (full-set provisioning): always provisions ``requested |
-            err = f"{type(e).__name__}: {e}"
+        opted`` — effectively the project's whole opted-in set, since
-            logger.error(
+        ``opted`` is already a superset of ``requested`` by the verb's own
-                "sandbox provisioning failed; refusing spawn",
+        guard — rather than only what this particular call named. That makes
-                agent_id=agent_id,
+        any later subset/superset request within the same opted set a
-                task_id=task_id,
+        guaranteed cache hit; it can never fall through to `provision()`,
-                services=services,
+        whose pre-clear `teardown()` would otherwise kill a live, mid-use
-                error=err,
+        container out from under the agent and rotate its creds. The union
-            )
+        (rather than trusting the caller to always pass the full set) is
-            raise AgentReadinessError(
+        belt-and-suspenders — bounded by the project's own opt-in either way.
-                f"sandbox provisioning failed for {agent_id} (task={task_id}): {err}"
+
-            ) from e
+        A cache hit is verified live (`SandboxProvisioner.is_live`) before
        being trusted: a container OOM-killed or removed out-of-band evicts
        the stale entry and falls through to a fresh full-set provision
        (new creds — that's the recovery).
        The whole check-cache -> provision -> store section runs under a
        per-agent-slug lock so two concurrent calls for the same agent can't
        race provision()/teardown() on the same containers.
        """
        full = sorted(set(requested) | set(opted))
        # Lazily-allocated (no __init__ statement) to keep AgentOrchestrator's
        # constructor under the statement-count gate; getattr guards bare
        # __new__() test doubles that never ran __init__ — same convention
        # as _sandbox_info's own test-double guards elsewhere in this class.
        locks = getattr(self, "_sandbox_locks", None)
        if locks is None:
            locks = {}
            self._sandbox_locks = locks
        lock = locks.setdefault(agent_slug, asyncio.Lock())
        async with lock:
            cached = self._sandbox_info.get(agent_slug)
            if cached is not None and set(full) <= set(cached.services):
                if await self._sandbox.is_live(agent_slug, sorted(cached.services)):
                    return cached
                self._sandbox_info.pop(agent_slug, None)
            info = await self._sandbox.provision(agent_slug, full)
            self._sandbox_info[agent_slug] = info
            return info
    async def release_sandbox(self, agent_slug: str) -> None:
        """Best-effort teardown at the end of the caller's task engagement.
        Called by the Choreographer's post-verb hook (i_am_done, unclaim,
        i_am_idle, pass_review/fail_review, i_documented) so a
        `request_sandbox`-provisioned sidecar doesn't outlive the work
        that asked for it, instead of only dying with the agent container.
        Idempotent and never raises (`SandboxProvisioner.teardown`'s own
        contract) — the container-removal teardown + janitor sweep remain
        the backstop.
        The overwhelmingly common call has no sandbox at all, so the cache
        dict is checked BEFORE taking the per-agent lock or touching
        docker — the fast path is a single dict lookup, no lock, no
        subprocess.
        """
        if agent_slug not in self._sandbox_info:
            return
        locks = getattr(self, "_sandbox_locks", None)
        if locks is None:
            locks = {}
            self._sandbox_locks = locks
        lock = locks.setdefault(agent_slug, asyncio.Lock())
        async with lock:
            if agent_slug not in self._sandbox_info:
                return
            await self._sandbox.teardown(agent_slug)
            self._sandbox_info.pop(agent_slug, None)
    async def _launch_spawn(
        self,
@@ -2853,19 +2916,15 @@ class AgentOrchestrator:
        )
    @staticmethod
-    def _append_sandbox_env(cmd: list[str], config: AgentConfig) -> None:
+    def _append_sandbox_marker_env(cmd: list[str], services: list[str]) -> None:
-        """Inject sandbox engine env, in place of the prod-creds gate env.
+        """Cheap availability probe: names the request_sandbox-eligible services.
-        Called INSTEAD OF `_append_gate_env` whenever a sandbox was provisioned
+        Replaces eager sandbox env injection now that provisioning is
-        for this spawn (`config.sandbox_info` set) — sandbox replaces, never
+        on-demand (the `request_sandbox` do-verb) — never prod creds, purely
-        coexists with, the production gate-env injection. Emission is driven by
+        informational so the agent knows the verb will succeed. Called
-        the engine registry via `SandboxInfo.emit_env`, so a new engine's
+        INSTEAD OF `_append_gate_env` for an opted-in project (never both).
        `ROBOCO_TEST_*` vars land here with no orchestrator change.
        """
-        info = config.sandbox_info
+        cmd.extend(["-e", f"ROBOCO_SANDBOX_SERVICES_AVAILABLE={','.join(services)}"])
        if info is None:
            return
        cmd.extend(info.emit_env())
    @staticmethod
    def _default_spawn_prompt() -> str:
@@ -3015,9 +3074,10 @@ class AgentOrchestrator:
            return result.instance_id
        container_name = f"roboco-agent-{config.agent_id}"
-        # teardown_sandbox=False: this spawn's sandbox was provisioned moments
+        # teardown_sandbox=False: nothing is provisioned before spawn anymore
-        # ago in _build_agent_config — the stale-clear must not destroy it.
+        # (sandboxes are on-demand via request_sandbox/ensure_sandbox), so this
-        # Stale sandboxes from a prior crash are cleared by provision() itself.
+        # is now vestigial for THIS spawn — but it still protects a respawn
        # racing a sandbox the agent just requested moments ago via the verb.
        await self._remove_container(
            container_name, teardown_sandbox=False, stop_reason="pre_spawn_stale_clear"
        )
@@ -3029,8 +3089,8 @@ class AgentOrchestrator:
        cmd = self._build_mount_args(container_name, config, hosts)
        self._append_agent_auth_env(cmd, config)
        self._append_git_context_env(cmd, config)
-        if config.sandbox_info is not None:
+        if config.sandbox_available_services:
-            self._append_sandbox_env(cmd, config)
+            self._append_sandbox_marker_env(cmd, config.sandbox_available_services)
        else:
            self._append_gate_env(cmd)
        self._append_image_and_claude_args(cmd, config, initial_prompt)
@@ -3208,6 +3268,13 @@ class AgentOrchestrator:
        if teardown_sandbox and settings.sandbox_db_enabled:
            slug = container_name.removeprefix("roboco-agent-")
            await self._sandbox.teardown(slug)
            # Evict the ensure_sandbox cache so a later request_sandbox call
            # re-provisions instead of handing back creds for a torn-down
            # container. getattr guards bare __new__() test doubles that
            # never ran __init__.
            cache = getattr(self, "_sandbox_info", None)
            if cache is not None:
                cache.pop(slug, None)
    async def _generate_mcp_config(
        self,
@@ -3978,6 +4045,7 @@ class AgentOrchestrator:
        agent_id: str,
        task_id: str | None,
        workspace_path: str,
        sandbox_services: list[str] | None = None,
    ) -> Path | None:
        """Write a compact task briefing to be read by SessionStart hook.
@@ -3987,6 +4055,10 @@ class AgentOrchestrator:
        the task and include title, status, branch, and acceptance criteria.
        On fetch failure we still emit the role-level part (role, escalation
        target, terminal tools, workspace path) — strictly better than nothing.
        `sandbox_services` (when non-empty) names the request_sandbox verb so
        an opted-in project's agent knows it will succeed, rather than relying
        on manifest presence alone to discover it.
        """
        role = get_agent_role(agent_id) or "agent"
        team = get_agent_team(agent_id) or "-"
@@ -3998,6 +4070,12 @@ class AgentOrchestrator:
            task = await self._fetch_task_for_briefing(agent_id, task_id)
            if task is not None:
                task_block = self._format_task_briefing_block(task_id, task)
        sandbox_line = (
            f"- **Sandbox available:** `{', '.join(sandbox_services)}` — call "
            "`request_sandbox()` to provision on demand\n"
            if sandbox_services
            else ""
        )
        content = (
            f"# Session briefing — {agent_id}\n"
@@ -4009,6 +4087,7 @@ class AgentOrchestrator:
            f"- **Team:** {team}\n"
            f"- **Escalate to:** `{escalate_to}`\n"
            f"- **Workspace:** `{workspace_path}`\n"
            f"{sandbox_line}"
            f"{task_block}"
            "\n## Terminal tools (how to exit cleanly)\n"
            "- `i_am_idle()` — no work remaining (every role)\n"
@@ -10038,6 +10117,13 @@ Start now: evidence(task_id="{task_id}")
            return
        with contextlib.suppress(Exception):
            await self._sandbox.janitor_sweep()
        # Evict ensure_sandbox cache entries for agents the sweep just reaped
        # (owner container gone). getattr guards bare __new__() test doubles.
        cache = getattr(self, "_sandbox_info", None)
        if cache:
            live = getattr(self, "_instances", {})
            for slug in set(cache) - set(live):
                cache.pop(slug, None)
    def _assignee_has_active_instance(self, task: Any) -> bool:
        """True if the task's assignee currently holds a live (ACTIVE) container.
@@ -206,6 +206,31 @@ class SandboxProvisioner:
            await asyncio.sleep(_READY_POLL_INTERVAL_SECONDS)
        return False
    async def is_live(self, agent_id: str, services: list[str]) -> bool:
        """True iff every named service's sandbox container is running.
        On-demand liveness check for a cache hit (not the janitor's mass
        sweep): a container OOM-killed or manually removed while the agent
        lives fails this, so the caller can evict the stale cache entry and
        re-provision instead of handing back creds for a dead container.
        """
        run = self._run()
        for service in services:
            engine = SANDBOX_ENGINES.get(service)
            if engine is None:
                return False
            name = engine.container_name(agent_id)
            try:
                rc, stdout, _ = await run(
                    ["inspect", "--format={{.State.Running}}", name],
                    _DOCKER_EXEC_TIMEOUT_SECONDS,
                )
            except Exception:
                return False
            if rc != 0 or stdout.decode().strip() != "true":
                return False
        return True
    async def teardown(self, agent_id: str) -> None:
        """Idempotent: stop+kill+rm every engine's sandbox container. Never raises."""
        for engine in SANDBOX_ENGINES.values():
@@ -674,6 +674,31 @@ class Choreographer:
            logger.warning("audit.log_event failed", error=str(exc), verb=verb)
        return env
    async def _teardown_sandbox_best_effort(self, agent_id: UUID) -> None:
        """Release the caller's request_sandbox sidecar on successful exit.
        Called from the six verbs whose success means the caller's
        engagement with its work has ended (i_am_done, unclaim, i_am_idle,
        pass_review/fail_review, i_documented) — no shared success-emit
        path spans all six (they live across three mixin files, each
        building its own ``Envelope.ok`` at its own site), so this is
        called once at each verb's existing success point rather than
        duplicated teardown logic. No orchestrator (e2e harness, startup)
        is a silent no-op; any other failure is logged, never raised —
        the container-removal teardown + janitor sweep remain the backstop.
        """
        orch = self.orchestrator
        if orch is None:
            return
        from roboco.agents_config import _resolve_to_slug
        try:
            await orch.release_sandbox(_resolve_to_slug(str(agent_id)))
        except Exception as exc:
            logger.warning(
                "sandbox_release_failed", agent_id=str(agent_id), error=str(exc)
            )
    @classmethod
    def _free_text_soup(
        cls, checks: tuple[tuple[str, Any, int], ...]
@@ -2958,6 +2983,7 @@ class Choreographer:
        )
        agent = await self.task.agent_for(agent_id)
        role = str(agent.role) if agent is not None else "developer"
        await self._teardown_sandbox_best_effort(agent_id)
        return Envelope.ok(
            status=str(t.status),
            task_id=str(task_id),
@@ -3544,6 +3570,7 @@ class Choreographer:
        # Deliberately no _touch — unclaim clears assigned_to, so there is
        # no claimant heartbeat to refresh. (Asymmetric with `resume` by
        # design: resume keeps the same claimant active and does heartbeat.)
        await self._teardown_sandbox_best_effort(agent_id)
        return Envelope.ok(
            status=str(after.status),
            task_id=str(task_id),
@@ -4003,6 +4030,10 @@ class Choreographer:
            )
        else:
            next_msg = "container will shut down"
        # The agent truly disengages here (container shuts down) — unlike
        # the idle_with_unread early return above, which sends it right
        # back to work, so no teardown fires there.
        await self._teardown_sandbox_best_effort(agent_id)
        return Envelope.ok(
            status="idle",
            task_id=None,
@@ -53,6 +53,9 @@ class ChoreographerHelpers:
    ) -> Envelope:
        raise NotImplementedError
    async def _teardown_sandbox_best_effort(self, agent_id: UUID) -> None:
        raise NotImplementedError
    async def _toolchain_broken_guard(
        self, agent_id: UUID, task: Any, *, reviewer: bool = False
    ) -> Envelope | None:
@@ -543,6 +543,7 @@ class DocMixin(_Base):
                f"Docs-complete transition committed but the PM handoff "
                f"failed ({exc}). Re-issue the notification via dm."
            )
        await self._teardown_sandbox_best_effort(doc_agent_id)
        env = Envelope.ok(
            status=str(t.status),
            task_id=str(task_id),
@@ -582,6 +582,7 @@ class QAMixin(_Base):
            )
        warning = await self._pass_review_documenter_handoff(qa_agent_id, task_id, t)
        await self._teardown_sandbox_best_effort(qa_agent_id)
        env = Envelope.ok(
            status=str(t.status),
            task_id=str(task_id),
@@ -714,6 +715,7 @@ class QAMixin(_Base):
                task_id=task_id,
                body=f"QA needs changes. Issues:\n{notes}",
            )
        await self._teardown_sandbox_best_effort(qa_agent_id)
        return Envelope.ok(
            status=str(t.status),
            task_id=str(task_id),
@@ -299,6 +299,10 @@ class ContentActionsDeps:
    # context. Matches the choreographer's EvidenceRepo wiring so both
    # paths surface the same shape.
    evidence_repo: Any = None
    # request_sandbox's ensure_sandbox call. Mirrors ChoreographerDeps.orchestrator:
    # optional so tests that don't exercise sandbox provisioning need not plumb
    # it in; None degrades to a retryable "orchestrator unavailable" envelope.
    orchestrator: Any = None
 _VALID_NOTIFY_PRIORITIES: frozenset[str] = frozenset(p.value for p in _comms.Priority)
@@ -412,6 +416,10 @@ class ContentActions:
    def evidence_repo(self) -> Any:
        return self._deps.evidence_repo
    @property
    def orchestrator(self) -> Any:
        return self._deps.orchestrator
    async def _touch_heartbeat(self, task_id: UUID | None) -> None:
        """Best-effort heartbeat refresh on a content-write success path.
@@ -1826,6 +1834,133 @@ class ContentActions:
            context_briefing={},
        )
    async def _sandbox_active_task(self, agent_id: UUID) -> tuple[Any, Envelope | None]:
        """request_sandbox's task guard: an active, project-bound task, or a
        clean invalid_state rejection."""
        t = await self.task.get_active_task_for_agent(agent_id)
        if t is None or t.project_id is None:
            return None, Envelope.invalid_state(
                message="no claimed task with a project — cannot scope a sandbox",
                remediate=(
                    "call give_me_work() first; request_sandbox needs an "
                    "active, project-bound task"
                ),
                context_briefing={},
            )
        return t, None
    @staticmethod
    def _sandbox_scope(
        project: Any, services: list[str] | None
    ) -> tuple[frozenset[str], Envelope | None]:
        """request_sandbox's opt-in guard: the resolved service set, or a
        clean invalid_state rejection naming the project's allowed set."""
        opted = frozenset(project.sandbox_services or []) if project else frozenset()
        if not opted:
            return frozenset(), Envelope.invalid_state(
                message="project has not opted into any sandbox service",
                remediate=(
                    "ask a PM/CEO to set sandbox_services in project "
                    "settings before requesting a sandbox"
                ),
                context_briefing={},
            )
        requested = opted if services is None else frozenset(services)
        unknown = requested - opted
        if unknown:
            return frozenset(), Envelope.invalid_state(
                message=f"requested service(s) {sorted(unknown)} not opted into",
                remediate=(
                    f"this project's opted-in set is {sorted(opted)} — "
                    "request a subset of that"
                ),
                context_briefing={},
            )
        return requested, None
    async def request_sandbox(
        self,
        *,
        agent_id: UUID,
        services: list[str] | None = None,
    ) -> Envelope:
        """On-demand sandbox DB/Redis/Mongo (dev + QA only, see role_config).
        Replaces eager per-spawn provisioning: a sandbox is created only when
        an agent actually asks for one, keyed off the CALLER's authenticated
        slug (never another agent's). ``services`` omitted means the
        project's whole opted-in set.
        Guards, in order: flag off; caller has no claimed/active,
        project-bound task (`_sandbox_active_task`); project not opted into
        any sandbox service, or a requested service outside its opted set
        (`_sandbox_scope`, names the allowed set); orchestrator handle
        unavailable (retryable). `ensure_sandbox` always provisions the
        project's whole opted-in set regardless of ``services`` (so a later
        call can never trigger a mid-session teardown of a live container);
        the evidence payload here is filtered back down to what THIS call
        asked for. Creds come back in the evidence payload, never as
        injected env — see
        ``docs/internal/specs/2026-07-08-sandbox-on-demand.md`` §4.
        """
        if not settings.sandbox_db_enabled:
            return Envelope.invalid_state(
                message="sandbox provisioning is disabled",
                remediate=(
                    "ROBOCO_SANDBOX_DB_ENABLED is off — ask the CEO to arm "
                    "it, or rely on the legacy gate env if this project "
                    "isn't sandboxed"
                ),
                context_briefing={},
            )
        t, rejection = await self._sandbox_active_task(agent_id)
        if rejection is not None:
            return rejection
        from roboco.services.project import get_project_service
        project = await get_project_service(self.task.session).get(t.project_id)
        requested, rejection = self._sandbox_scope(project, services)
        if rejection is not None:
            return rejection
        opted = frozenset(project.sandbox_services or []) if project else frozenset()
        if self.orchestrator is None:
            return Envelope.invalid_state(
                message="orchestrator handle unavailable — cannot provision a sandbox",
                remediate=(
                    "retry request_sandbox shortly; the orchestrator may be restarting"
                ),
                context_briefing={},
            )
        from roboco.agents_config import _resolve_to_slug
        from roboco.runtime.sandbox import SandboxProvisionError
        agent_slug = _resolve_to_slug(str(agent_id))
        try:
            info = await self.orchestrator.ensure_sandbox(
                agent_slug, sorted(requested), sorted(opted)
            )
        except SandboxProvisionError as e:
            return Envelope.invalid_state(
                message=f"sandbox provisioning failed: {e}",
                remediate="retry shortly; escalate to your PM if it keeps failing",
                context_briefing={},
            )
        await self._touch_heartbeat(t.id)
        # ensure_sandbox provisions the project's whole opted-in set (see its
        # docstring); the evidence payload stays scoped to what THIS call
        # asked for.
        payload = info.as_payload()
        filtered = {
            name: payload[name] for name in sorted(requested) if name in payload
        }
        return Envelope.ok(
            status=str(t.status),
            task_id=str(t.id),
            next="use the returned creds for this session; call again anytime",
            evidence=filtered,
            context_briefing={},
        )
    # =========================================================================
    # Wave 1 — pre-gateway parity restoration
    # =========================================================================
@@ -56,6 +56,9 @@ _DEV_DO = (
    # assigned a video-authoring task. The real gate is propose_video's
    # runtime _caller_team check.
    "propose_video",
    # On-demand sandbox DB/Redis/Mongo — carried unconditionally (declarative
    # manifest), gated for real by request_sandbox's project opt-in check.
    "request_sandbox",
    *_NOTIFY_RECEIVER,
 )
@@ -65,6 +68,7 @@ _QA_DO = (
    "dm",
    "evidence",
    "draft_playbook",
    "request_sandbox",
    *_NOTIFY_RECEIVER,
 )
@@ -0,0 +1,121 @@
 """E2E smoke: on-demand sandbox provisioning wiring (2026-07-08 spec, Phase 3).
 Two seams:
 1. The spawn manifest carries ``request_sandbox`` for developer/qa roles only
   (``role_config.py`` -> ``spawn_manifest.build_for_role``) — the class of bug
   that silently strands an agent with no way to ask for a sandbox, or
   silently over-grants the verb to a role never scoped to it.
 2. The verb's HTTP wiring (schema -> route -> ``ContentActions`` ->
   envelope), driven end to end by a scripted dev agent against the REAL
   API. The e2e harness wires no orchestrator (``_ServiceHolder.orchestrator``
   stays unset — no docker here), so the happy-path provision itself is
   covered by the mocked-orchestrator unit suite
   (``tests/unit/gateway/test_request_sandbox_verb.py``); this test proves
   the guard chain up to and including the clean, retryable "orchestrator
   unavailable" envelope every caller sees before docker is ever touched.
 """
 from __future__ import annotations
 from pathlib import Path
 from typing import TYPE_CHECKING
 from uuid import uuid4
 from roboco.config import settings
 from roboco.models.base import TaskStatus
 from tests.e2e_smoke.arcs import seed_company, seed_project, seed_task
 from tests.e2e_smoke.harness import ScriptedAgent, expect_error
 if TYPE_CHECKING:
    import pytest
    from sqlalchemy.ext.asyncio import AsyncSession
    from tests.e2e_smoke.harness import E2EStack
 def _set_sandbox_services(
    stack: E2EStack, project_id: object, services: list[str]
 ) -> None:
    """System-side project opt-in write (mirrors ``arcs.set_branch_name``)."""
    from roboco.db.tables import ProjectTable
    from sqlalchemy import select
    async def _run(session: AsyncSession) -> None:
        row = (
            await session.execute(
                select(ProjectTable).where(ProjectTable.id == project_id)
            )
        ).scalar_one()
        row.sandbox_services = services
    stack.run_db(_run)
 def test_manifest_grants_request_sandbox_to_dev_and_qa_only() -> None:
    """role_config -> spawn_manifest wiring: the verb reaches dev/qa manifests
    and no other role — the exact scope e418a4ca added it under."""
    from roboco.runtime.spawn_manifest import SpawnInputs, build_for_role
    def do_tools(role: str) -> list[str]:
        manifest = build_for_role(
            SpawnInputs(
                agent_id=uuid4(),
                role=role,
                team="backend",
                workspace_path=Path("/tmp/x"),
                agent_model="sonnet",
            )
        )
        return manifest.do_tools
    assert "request_sandbox" in do_tools("developer")
    assert "request_sandbox" in do_tools("qa")
    assert "request_sandbox" not in do_tools("documenter")
 def test_request_sandbox_guard_chain_over_real_api(
    e2e_stack: E2EStack, monkeypatch: pytest.MonkeyPatch
 ) -> None:
    """Full HTTP round-trip proving the guard order the spec's §1 promises:
    flag off refuses before any DB lookup; a requested service outside the
    project's opted set names the allowed set; a request within the opted
    set clears every guard and reaches the (here, absent) orchestrator,
    returning the clean retryable envelope rather than ever refusing a spawn
    or crashing."""
    stack = e2e_stack
    company = seed_company(stack)
    dev = ScriptedAgent(stack, company.dev_id, "be-dev-1", "developer")
    # --- flag off: refused before any task/project lookup -------------------
    monkeypatch.setattr(settings, "sandbox_db_enabled", False)
    env = expect_error(dev.do("request_sandbox"), "invalid_state", "flag off")
    assert "ROBOCO_SANDBOX_DB_ENABLED" in (env.get("remediate") or "")
    # --- flag on, project opted into a subset, request outside it -----------
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    project_id, _project_slug = seed_project(stack, company)
    _set_sandbox_services(stack, project_id, ["redis"])
    seed_task(
        stack,
        title="Sandbox smoke task",
        description="Active project-bound task for request_sandbox to scope to.",
        project_id=project_id,
        created_by=company.cell_pm_id,
        assigned_to=company.dev_id,
        status=TaskStatus.IN_PROGRESS,
    )
    env = expect_error(
        dev.do("request_sandbox", services=["postgres"]),
        "invalid_state",
        "requested service outside opted set",
    )
    assert "redis" in (env.get("remediate") or "")
    # --- same project/task, request within the opted set --------------------
    # Every guard up to provisioning passes; the harness wires no
    # orchestrator, so the verb must return the clean, retryable
    # "unavailable" envelope rather than raising.
    env = expect_error(
        dev.do("request_sandbox"), "invalid_state", "orchestrator unavailable"
    )
    assert "retry" in (env.get("remediate") or "").lower()
@@ -0,0 +1,289 @@
 """ContentActions.request_sandbox — the on-demand sandbox DB/Redis/Mongo verb.
 Guard matrix (flag off / no active task / no project / not opted in / subset
 violation / orchestrator unavailable), the success envelope payload shape, and
 cross-agent isolation (ensure_sandbox is always called with the CALLER's own
 resolved slug, never another agent's).
 """
 from __future__ import annotations
 from unittest.mock import AsyncMock, MagicMock
 from uuid import uuid4
 import pytest
 from roboco.config import settings
 from roboco.models.sandbox import SandboxConnection, SandboxInfo
 from roboco.runtime.sandbox import SandboxProvisionError
 from roboco.services.gateway.content_actions import ContentActions, ContentActionsDeps
 def _make_actions(
    *,
    task_obj: MagicMock | None,
    orchestrator: AsyncMock | None,
 ) -> tuple[ContentActions, MagicMock]:
    task = AsyncMock()
    task.get_active_task_for_agent.return_value = task_obj
    task.session = MagicMock()
    deps = ContentActionsDeps(
        task=task,
        git=MagicMock(),
        a2a=MagicMock(),
        journal=MagicMock(),
        workspace=MagicMock(),
        notifications=MagicMock(),
        orchestrator=orchestrator,
    )
    return ContentActions(deps), task
 def _task(project_id: object | None = uuid4()) -> MagicMock:
    t = MagicMock()
    t.id = uuid4()
    t.project_id = project_id
    t.status = "in_progress"
    return t
 def _stub_project(monkeypatch: pytest.MonkeyPatch, services: list[str] | None) -> None:
    project = MagicMock(sandbox_services=services)
    project_service = MagicMock()
    project_service.get = AsyncMock(return_value=project)
    monkeypatch.setattr(
        "roboco.services.project.get_project_service", lambda _s: project_service
    )
 def _sandbox_info() -> SandboxInfo:
    return SandboxInfo(
        services={
            "postgres": SandboxConnection(
                host="roboco-sandbox-pg-dev-1",
                port=5432,
                password="pw",
                user="sandbox",
                database="sandbox",
            )
        }
    )
 # ---------------------------------------------------------------------------
 # Guard matrix
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_flag_off_refuses_before_task_lookup(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", False)
    actions, task = _make_actions(task_obj=None, orchestrator=None)
    env = await actions.request_sandbox(agent_id=uuid4())
    assert env.error == "invalid_state"
    task.get_active_task_for_agent.assert_not_awaited()
@pytest.mark.asyncio
 async def test_no_active_task_refused(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    actions, _task_svc = _make_actions(task_obj=None, orchestrator=None)
    env = await actions.request_sandbox(agent_id=uuid4())
    assert env.error == "invalid_state"
    assert "give_me_work" in (env.remediate or "")
@pytest.mark.asyncio
 async def test_task_without_project_refused(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    actions, _task_svc = _make_actions(
        task_obj=_task(project_id=None), orchestrator=None
    )
    env = await actions.request_sandbox(agent_id=uuid4())
    assert env.error == "invalid_state"
@pytest.mark.asyncio
 async def test_project_not_opted_in_refused(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=None)
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=None)
    env = await actions.request_sandbox(agent_id=uuid4())
    assert env.error == "invalid_state"
    assert "not opted" in (env.message or "")
@pytest.mark.asyncio
 async def test_requested_service_outside_opted_set_names_allowed_set(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres"])
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=None)
    env = await actions.request_sandbox(agent_id=uuid4(), services=["redis"])
    assert env.error == "invalid_state"
    assert "postgres" in (env.remediate or "")
@pytest.mark.asyncio
 async def test_orchestrator_unavailable_is_retryable(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres"])
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=None)
    env = await actions.request_sandbox(agent_id=uuid4())
    assert env.error == "invalid_state"
    assert "retry" in (env.remediate or "").lower()
@pytest.mark.asyncio
 async def test_provision_failure_surfaces_as_retryable_invalid_state(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres"])
    orch = AsyncMock()
    orch.ensure_sandbox.side_effect = SandboxProvisionError("image pull failed")
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=orch)
    env = await actions.request_sandbox(agent_id=uuid4())
    assert env.error == "invalid_state"
    assert "provisioning failed" in (env.message or "")
 # ---------------------------------------------------------------------------
 # Success path — envelope payload shape
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_success_returns_creds_in_evidence_with_env_subdict(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres"])
    orch = AsyncMock()
    orch.ensure_sandbox.return_value = _sandbox_info()
    actions, task_svc = _make_actions(task_obj=_task(), orchestrator=orch)
    env = await actions.request_sandbox(agent_id=uuid4())
    expected_port = 5432
    assert env.error is None
    assert env.evidence is not None
    payload = env.evidence["postgres"]
    assert payload["host"] == "roboco-sandbox-pg-dev-1"
    assert payload["port"] == expected_port
    assert payload["user"] == "sandbox"
    assert payload["password"] == "pw"
    assert payload["database"] == "sandbox"
    assert payload["env"]["ROBOCO_TEST_DB_HOST"] == "roboco-sandbox-pg-dev-1"
    assert payload["env"]["ROBOCO_TEST_DB_PASSWORD"] == "pw"
    task_svc.heartbeat.assert_awaited_once()
@pytest.mark.asyncio
 async def test_omitted_services_requests_full_opted_set(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres", "redis"])
    orch = AsyncMock()
    orch.ensure_sandbox.return_value = _sandbox_info()
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=orch)
    await actions.request_sandbox(agent_id=uuid4())
    orch.ensure_sandbox.assert_awaited_once()
    called_services = orch.ensure_sandbox.call_args.args[1]
    assert sorted(called_services) == ["postgres", "redis"]
@pytest.mark.asyncio
 async def test_ensure_sandbox_called_with_full_opted_set_not_just_requested(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """DEFECT 1 fix: the verb always passes the project's whole opted-in set
    (not just this call's ``services`` subset) as ensure_sandbox's ``opted``
    argument, so a superset request later in the session can never trigger a
    fresh provision() that tears down the agent's live sandbox."""
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres", "redis"])
    orch = AsyncMock()
    orch.ensure_sandbox.return_value = _sandbox_info()
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=orch)
    await actions.request_sandbox(agent_id=uuid4(), services=["postgres"])
    called_requested = orch.ensure_sandbox.call_args.args[1]
    called_opted = orch.ensure_sandbox.call_args.args[2]
    assert called_requested == ["postgres"]
    assert sorted(called_opted) == ["postgres", "redis"]
@pytest.mark.asyncio
 async def test_response_payload_filtered_to_requested_services(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """`ensure_sandbox` provisions the full opted set under the hood; the
    verb's response only surfaces what THIS call actually asked for."""
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres", "redis"])
    orch = AsyncMock()
    orch.ensure_sandbox.return_value = SandboxInfo(
        services={
            "postgres": SandboxConnection(
                host="h", port=5432, password="pw", user="sandbox", database="sandbox"
            ),
            "redis": SandboxConnection(host="h", port=6379, password="rw"),
        }
    )
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=orch)
    env = await actions.request_sandbox(agent_id=uuid4(), services=["postgres"])
    assert env.error is None
    assert env.evidence is not None
    assert set(env.evidence) == {"postgres"}
 # ---------------------------------------------------------------------------
 # Cross-agent isolation
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_ensure_sandbox_keyed_off_caller_own_slug(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """Two different callers resolve to two different ensure_sandbox slugs —
    a caller can never reach another agent's cached sandbox."""
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    _stub_project(monkeypatch, services=["postgres"])
    orch = AsyncMock()
    orch.ensure_sandbox.return_value = _sandbox_info()
    actions, _task_svc = _make_actions(task_obj=_task(), orchestrator=orch)
    agent_a, agent_b = uuid4(), uuid4()
    await actions.request_sandbox(agent_id=agent_a)
    await actions.request_sandbox(agent_id=agent_b)
    slugs_called = [c.args[0] for c in orch.ensure_sandbox.call_args_list]
    assert slugs_called[0] != slugs_called[1]
    assert slugs_called[0] == str(agent_a)
    assert slugs_called[1] == str(agent_b)
@@ -0,0 +1,513 @@
 """Sandbox release hook — Choreographer._teardown_sandbox_best_effort.
 CEO directive: sandboxes must die when an agent's engagement with its work
 ends, not only when its container is removed. Ties the on-demand
 request_sandbox subsystem's teardown to the SUCCESSFUL exit of the six
 verbs whose completion means exactly that: i_am_done, unclaim, i_am_idle,
 pass_review, fail_review, i_documented. A rejected/failed verb call must
 NOT release (the work isn't done), and a release failure must never fail
 the verb — best-effort, backstopped by the container-removal teardown +
 janitor sweep that already exist.
 Two layers:
  - the helper itself (`_teardown_sandbox_best_effort`), tested directly
    against a fake orchestrator;
  - the six call sites, tested by spying on the helper (patched onto the
    instance) so each verb's fixture stays the minimal one already proven
    by its own dedicated test file (test_choreographer_dev.py,
    test_unclaim.py, test_choreographer_idle_guards.py,
    test_choreographer_qa.py, test_choreographer_doc.py).
 """
 from __future__ import annotations
 from datetime import UTC, datetime
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 from uuid import uuid4
 import pytest
 from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
 from structlog.testing import capture_logs
 def _make_deps(**overrides: Any) -> ChoreographerDeps:
    base: dict[str, Any] = {
        "task": AsyncMock(),
        "work_session": AsyncMock(),
        "git": AsyncMock(),
        "a2a": AsyncMock(),
        "journal": AsyncMock(),
        "audit": AsyncMock(),
        "evidence_repo": AsyncMock(),
    }
    base.update(overrides)
    task = base["task"]
    # Covers both VerbRunner's savepoint (i_am_done/pass_review/fail_review/
    # i_documented) and i_documented's own session.flush() — one shared
    # setup for every verb exercised in this file.
    task.session = MagicMock()
    task.session.begin_nested = MagicMock(
        return_value=MagicMock(
            __aenter__=AsyncMock(return_value=None),
            __aexit__=AsyncMock(return_value=False),
        )
    )
    task.session.flush = AsyncMock()
    repo = base["evidence_repo"]
    for method in (
        "list_unread_a2a",
        "list_unread_mentions",
        "list_pending_notifications",
        "task_metadata_gaps",
        "recent_team_activity",
        "blockers_in_lane",
        "journal_highlights_for_task",
    ):
        getattr(repo, method).return_value = []
    _ldef = base["journal"].latest_decision_at.return_value
    if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
        base["journal"].latest_decision_at.return_value = datetime.now(UTC)
    return ChoreographerDeps(**base)
 # ---------------------------------------------------------------------------
 # The helper itself
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_helper_noop_when_orchestrator_missing() -> None:
    c = Choreographer(_make_deps())  # orchestrator defaults to None
    await c._teardown_sandbox_best_effort(uuid4())  # must not raise
@pytest.mark.asyncio
 async def test_helper_calls_release_sandbox_with_resolved_slug() -> None:
    agent_id = uuid4()
    orch = AsyncMock()
    c = Choreographer(_make_deps(orchestrator=orch))
    await c._teardown_sandbox_best_effort(agent_id)
    # Not seeded in AGENT_UUIDS, so _resolve_to_slug falls back to identity —
    # mirrors test_request_sandbox_verb.py's cross-agent-isolation assertion.
    orch.release_sandbox.assert_awaited_once_with(str(agent_id))
@pytest.mark.asyncio
 async def test_helper_swallows_release_failure_and_logs() -> None:
    agent_id = uuid4()
    orch = AsyncMock()
    orch.release_sandbox.side_effect = RuntimeError("docker down")
    c = Choreographer(_make_deps(orchestrator=orch))
    with capture_logs() as logs:
        await c._teardown_sandbox_best_effort(agent_id)  # must not raise
    assert any("sandbox_release_failed" in str(e.get("event", "")) for e in logs)
 # ---------------------------------------------------------------------------
 # i_am_done
 # ---------------------------------------------------------------------------
 def _ready_i_am_done_task(task_id: Any, agent_id: Any, **overrides: Any) -> MagicMock:
    base = {
        "id": task_id,
        "status": "in_progress",
        "assigned_to": agent_id,
        "plan": {"x": 1},
        "branch_name": "feature/backend/abc",
        "work_session_id": uuid4(),
        "self_verified": False,
        "pr_number": 8,
        "pr_url": "https://x/pr/8",
        "team": "backend",
        "progress_updates": [{"message": "did x"}],
        "acceptance_criteria": [],
        "acceptance_criteria_status": [],
        "commits": [{"sha": "deadbeef"}],
        "documents": [],
        "dev_notes": "Implemented the change and added tests covering the new path.",
        "quick_context": None,
    }
    base.update(overrides)
    return MagicMock(**base)
@pytest.mark.asyncio
 async def test_i_am_done_success_releases_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    agent_id = uuid4()
    task_id = uuid4()
    t = _ready_i_am_done_task(task_id, agent_id)
    after_verify = MagicMock(
        **{**t.__dict__, "self_verified": True, "status": "verifying"}
    )
    after_submit = MagicMock(**{**after_verify.__dict__, "status": "awaiting_qa"})
    task_svc = AsyncMock()
    task_svc.get.return_value = t
    task_svc.agent_for.return_value = MagicMock(
        id=agent_id, role="developer", team="backend", slug=None
    )
    task_svc.submit_verification.return_value = after_verify
    task_svc.submit_qa.return_value = after_submit
    task_svc.qa_agent_for_team.return_value = MagicMock(
        id=uuid4(), skills=[{"id": "code_review"}]
    )
    journal_svc = AsyncMock()
    journal_svc.has_reflect_for_task.return_value = True
    journal_svc.has_decision_for_task.return_value = True
    journal_svc.has_learning_for_task.return_value = False
    journal_svc.has_struggle_for_task.return_value = False
    deps = _make_deps(task=task_svc, journal=journal_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.i_am_done(agent_id, task_id, "done")
    assert env.error is None
    release.assert_awaited_once_with(agent_id)
@pytest.mark.asyncio
 async def test_i_am_done_rejection_does_not_release_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    agent_id = uuid4()
    task_id = uuid4()
    task_svc = AsyncMock()
    task_svc.get.return_value = None
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.i_am_done(agent_id, task_id, "done")
    assert env.error == "not_found"
    release.assert_not_awaited()
 # ---------------------------------------------------------------------------
 # unclaim
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_unclaim_success_releases_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    agent_id = uuid4()
    task_id = uuid4()
    t = MagicMock(id=task_id, status="claimed", assigned_to=agent_id)
    task_svc = AsyncMock()
    task_svc.get.return_value = t
    task_svc.agent_for.return_value = MagicMock(
        id=agent_id, role="developer", team="backend", slug=None
    )
    task_svc.unclaim_for_agent.return_value = MagicMock(
        id=task_id, status="pending", assigned_to=None
    )
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.unclaim(agent_id, task_id)
    assert env.error is None
    release.assert_awaited_once_with(agent_id)
@pytest.mark.asyncio
 async def test_unclaim_rejection_does_not_release_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    agent_id = uuid4()
    task_id = uuid4()
    task_svc = AsyncMock()
    task_svc.get.return_value = None
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.unclaim(agent_id, task_id)
    assert env.error == "not_found"
    release.assert_not_awaited()
 # ---------------------------------------------------------------------------
 # i_am_idle
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_i_am_idle_success_releases_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    agent_id = uuid4()
    task_svc = AsyncMock()
    task_svc.list_assigned_for_agent.return_value = []
    task_svc.list_in_progress_for_agent.return_value = []
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.i_am_idle(agent_id)
    assert env.error is None
    assert env.status == "idle"
    release.assert_awaited_once_with(agent_id)
@pytest.mark.asyncio
 async def test_i_am_idle_with_unread_does_not_release_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """idle_with_unread sends the agent back to work — not a real exit, so
    the container does not shut down and the sandbox must survive."""
    agent_id = uuid4()
    deps = _make_deps()
    deps.evidence_repo.list_unread_a2a.return_value = [{"from": "x", "task_id": "t1"}]
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.i_am_idle(agent_id)
    assert env.status == "idle_with_unread"
    release.assert_not_awaited()
@pytest.mark.asyncio
 async def test_i_am_idle_guard_rejection_does_not_release_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    agent_id = uuid4()
    pending = MagicMock(id=uuid4(), status="pending")
    task_svc = AsyncMock()
    task_svc.list_assigned_for_agent.return_value = [pending]
    task_svc.list_in_progress_for_agent.return_value = []
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.i_am_idle(agent_id)
    assert env.error == "invalid_state"
    release.assert_not_awaited()
 # ---------------------------------------------------------------------------
 # pass_review / fail_review
 # ---------------------------------------------------------------------------
 def _qa_owned_task(task_id: Any, qa_id: Any, **overrides: Any) -> MagicMock:
    base = {
        "id": task_id,
        "status": "awaiting_qa",
        "task_type": "code",
        "team": "backend",
        "assigned_to": qa_id,
        "qa_evidence_inspected": True,
        "quick_context": None,
    }
    base.update(overrides)
    return MagicMock(**base)
 def _qa_agent_mock(qa_id: Any) -> MagicMock:
    return MagicMock(id=qa_id, role="qa", team="backend", slug=None)
@pytest.mark.asyncio
 async def test_pass_review_success_releases_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    qa_id = uuid4()
    task_id = uuid4()
    t = _qa_owned_task(task_id, qa_id)
    after = MagicMock(
        id=task_id,
        status="awaiting_documentation",
        assigned_to=qa_id,
        team="backend",
        pr_url="https://x/pr/8",
        qa_evidence_inspected=True,
    )
    task_svc = AsyncMock()
    task_svc.get.return_value = t
    task_svc.agent_for.return_value = _qa_agent_mock(qa_id)
    task_svc.qa_pass.return_value = after
    task_svc.documenter_for_team.return_value = MagicMock(id=uuid4())
    journal_svc = AsyncMock()
    journal_svc.has_learning_for_task.return_value = True
    deps = _make_deps(task=task_svc, journal=journal_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    notes = (
        "Reviewed PR carefully. Branch convention correct. Commit prefix "
        "verified. README diff matches spec. All acceptance criteria met."
    )
    env = await c.pass_review(qa_id, task_id, notes=notes)
    assert env.error is None
    release.assert_awaited_once_with(qa_id)
@pytest.mark.asyncio
 async def test_pass_review_rejection_does_not_release_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    qa_id = uuid4()
    task_id = uuid4()
    task_svc = AsyncMock()
    task_svc.get.return_value = None
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.pass_review(qa_id, task_id, notes="x")
    assert env.error == "not_found"
    release.assert_not_awaited()
@pytest.mark.asyncio
 async def test_fail_review_success_releases_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    qa_id = uuid4()
    task_id = uuid4()
    dev_id = uuid4()
    t = _qa_owned_task(task_id, qa_id)
    after = MagicMock(
        id=task_id,
        status="needs_revision",
        assigned_to=dev_id,
        team="backend",
    )
    task_svc = AsyncMock()
    task_svc.get.return_value = t
    task_svc.agent_for.return_value = _qa_agent_mock(qa_id)
    task_svc.qa_fail.return_value = after
    journal_svc = AsyncMock()
    journal_svc.has_learning_for_task.return_value = True
    deps = _make_deps(task=task_svc, journal=journal_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    issues = [
        "Missing unit test coverage for /healthz endpoint — add an assertion",
        "Lint errors in /api/foo.py: unused import and missing return type",
    ]
    env = await c.fail_review(qa_id, task_id, issues)
    assert env.error is None
    release.assert_awaited_once_with(qa_id)
@pytest.mark.asyncio
 async def test_fail_review_rejection_does_not_release_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    qa_id = uuid4()
    task_id = uuid4()
    task_svc = AsyncMock()
    task_svc.get.return_value = None
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.fail_review(qa_id, task_id, ["some issue"])
    assert env.error == "not_found"
    release.assert_not_awaited()
 # ---------------------------------------------------------------------------
 # i_documented
 # ---------------------------------------------------------------------------
 def _doc_owned_task(task_id: Any, doc_id: Any, **overrides: Any) -> MagicMock:
    base = {
        "id": task_id,
        "status": "awaiting_documentation",
        "task_type": "code",
        "team": "backend",
        "assigned_to": doc_id,
        "quick_context": None,
    }
    base.update(overrides)
    return MagicMock(**base)
 def _doc_agent_mock(doc_id: Any) -> MagicMock:
    return MagicMock(id=doc_id, role="documenter", team="backend", slug=None)
@pytest.mark.asyncio
 async def test_i_documented_success_releases_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    doc_id = uuid4()
    task_id = uuid4()
    t = _doc_owned_task(task_id, doc_id)
    after = MagicMock(
        id=task_id, status="awaiting_pm_review", assigned_to=doc_id, team="backend"
    )
    task_svc = AsyncMock()
    task_svc.get.return_value = t
    task_svc.agent_for.return_value = _doc_agent_mock(doc_id)
    task_svc.docs_complete.return_value = after
    task_svc.cell_pm_for_team.return_value = MagicMock(id=uuid4())
    journal_svc = AsyncMock()
    journal_svc.has_reflect_for_task.return_value = True
    deps = _make_deps(task=task_svc, journal=journal_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    notes = "Wrote backend/guides/feature-x.md with usage examples and config notes."
    files = ["backend/guides/feature-x.md"]
    env = await c.i_documented(doc_id, task_id, notes=notes, files=files)
    assert env.error is None
    release.assert_awaited_once_with(doc_id)
@pytest.mark.asyncio
 async def test_i_documented_rejection_does_not_release_sandbox(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    doc_id = uuid4()
    task_id = uuid4()
    task_svc = AsyncMock()
    task_svc.get.return_value = None
    deps = _make_deps(task=task_svc)
    c = Choreographer(deps)
    release = AsyncMock()
    monkeypatch.setattr(c, "_teardown_sandbox_best_effort", release)
    env = await c.i_documented(doc_id, task_id, notes="x" * 30, files=["docs.md"])
    assert env.error == "not_found"
    release.assert_not_awaited()
@@ -0,0 +1,40 @@
 """`_write_agent_briefing`'s sandbox availability line.
 Names the `request_sandbox` verb for an opted-in project (spec's "name it —
 cheap and kills a discovery failure mode" default) and is silent otherwise.
 """
 from __future__ import annotations
 from unittest.mock import patch
 import pytest
 from roboco.runtime.orchestrator import AgentOrchestrator
 def _orch() -> AgentOrchestrator:
    with patch.object(AgentOrchestrator, "__init__", return_value=None):
        orch = AgentOrchestrator.__new__(AgentOrchestrator)
    object.__setattr__(orch, "_TOOL_LOAD_CACHE", {})
    return orch
@pytest.mark.asyncio
 async def test_briefing_names_request_sandbox_when_opted_in(tmp_path: object) -> None:
    orch = _orch()
    path = await orch._write_agent_briefing(
        "dev-1", None, str(tmp_path), ["postgres", "redis"]
    )
    assert path is not None
    content = path.read_text()
    assert "request_sandbox()" in content
    assert "postgres, redis" in content
@pytest.mark.asyncio
 async def test_briefing_omits_sandbox_line_when_not_opted_in(tmp_path: object) -> None:
    orch = _orch()
    path = await orch._write_agent_briefing("dev-1", None, str(tmp_path), [])
    assert path is not None
    content = path.read_text()
    assert "request_sandbox()" not in content
@@ -1,8 +1,9 @@
-"""Sandbox env injection: `_append_sandbox_env` + the `_spawn_container` branch.
+"""Sandbox marker env: `_append_sandbox_marker_env` + the `_spawn_container` branch.
-A sandbox-active spawn must inject `ROBOCO_TEST_DB_*` / `ROBOCO_TEST_REDIS_*`
+An opted-in spawn injects a cheap `ROBOCO_SANDBOX_SERVICES_AVAILABLE` marker
-pointed at the sandbox and MUST NOT also run the legacy `_append_gate_env`
+(never prod creds — actual provisioning is on-demand via `request_sandbox`)
-prod-creds injection — sandbox replaces, never coexists with, prod creds.
+and MUST NOT also run the legacy `_append_gate_env` prod-creds injection —
 the marker replaces, never coexists with, prod creds.
 """
 from __future__ import annotations
@@ -13,93 +14,32 @@ from unittest.mock import AsyncMock
 import pytest
 from roboco.models.runtime import OrchestratorAgentConfig
 from roboco.models.sandbox import SandboxConnection, SandboxInfo
 from roboco.runtime.orchestrator import AgentOrchestrator
-def _config(sandbox_info: SandboxInfo | None = None) -> OrchestratorAgentConfig:
+def _config(
    sandbox_available_services: list[str] | None = None,
 ) -> OrchestratorAgentConfig:
    return OrchestratorAgentConfig(
        agent_id="dev-1",
        blueprint_path=Path(),
        mcp_config_path=Path("/tmp/mcp.json"),
-        sandbox_info=sandbox_info,
+        sandbox_available_services=sandbox_available_services or [],
    )
-def test_append_sandbox_env_injects_postgres_and_redis() -> None:
+def test_append_sandbox_marker_env_lists_services() -> None:
    info = SandboxInfo(
        services={
            "postgres": SandboxConnection(
                host="roboco-sandbox-pg-dev-1",
                port=5432,
                password="pgpw",
                user="sandbox",
                database="sandbox",
            ),
            "redis": SandboxConnection(
                host="roboco-sandbox-redis-dev-1", port=6379, password="rdpw"
            ),
        }
    )
    cmd: list[str] = []
-    AgentOrchestrator._append_sandbox_env(cmd, _config(info))
+    AgentOrchestrator._append_sandbox_marker_env(cmd, ["postgres", "redis"])
-    assert "ROBOCO_TEST_DB_HOST=roboco-sandbox-pg-dev-1" in cmd
+    assert "ROBOCO_SANDBOX_SERVICES_AVAILABLE=postgres,redis" in cmd
    assert "ROBOCO_TEST_DB_PORT=5432" in cmd
    assert "ROBOCO_TEST_DB_USER=sandbox" in cmd
    assert "ROBOCO_TEST_DB_PASSWORD=pgpw" in cmd
    assert "ROBOCO_TEST_DB_ADMIN_DB=sandbox" in cmd
    assert "ROBOCO_TEST_REDIS_HOST=roboco-sandbox-redis-dev-1" in cmd
    assert "ROBOCO_TEST_REDIS_PORT=6379" in cmd
    assert "ROBOCO_TEST_REDIS_PASSWORD=rdpw" in cmd
-def test_append_sandbox_env_postgres_only_omits_redis_vars() -> None:
+def test_append_sandbox_marker_env_single_service() -> None:
    info = SandboxInfo(
        services={
            "postgres": SandboxConnection(
                host="roboco-sandbox-pg-dev-1",
                port=5432,
                password="pgpw",
                user="sandbox",
                database="sandbox",
            )
        }
    )
    cmd: list[str] = []
-    AgentOrchestrator._append_sandbox_env(cmd, _config(info))
+    AgentOrchestrator._append_sandbox_marker_env(cmd, ["mongo"])
-    assert "ROBOCO_TEST_DB_HOST=roboco-sandbox-pg-dev-1" in cmd
+    assert "ROBOCO_SANDBOX_SERVICES_AVAILABLE=mongo" in cmd
    assert not any(v.startswith("ROBOCO_TEST_REDIS_") for v in cmd)
 def test_append_sandbox_env_injects_mongo() -> None:
    info = SandboxInfo(
        services={
            "mongo": SandboxConnection(
                host="roboco-sandbox-mongo-dev-1",
                port=27017,
                password="mpw",
                user="sandbox",
                database="admin",
            )
        }
    )
    cmd: list[str] = []
    AgentOrchestrator._append_sandbox_env(cmd, _config(info))
    assert "ROBOCO_TEST_MONGO_HOST=roboco-sandbox-mongo-dev-1" in cmd
    assert "ROBOCO_TEST_MONGO_PORT=27017" in cmd
    assert "ROBOCO_TEST_MONGO_USER=sandbox" in cmd
    assert "ROBOCO_TEST_MONGO_PASSWORD=mpw" in cmd
    assert "ROBOCO_TEST_MONGO_AUTH_DB=admin" in cmd
    assert not any(v.startswith("ROBOCO_TEST_DB_") for v in cmd)
 def test_append_sandbox_env_noop_without_sandbox_info() -> None:
    cmd: list[str] = []
    AgentOrchestrator._append_sandbox_env(cmd, _config(None))
    assert cmd == []
 def _fake_proc() -> AsyncMock:
@@ -125,7 +65,7 @@ def _stub_spawn_container_collaborators(
    monkeypatch.setattr(orch, "_append_gate_env", lambda *_a: calls.append("gate"))
    monkeypatch.setattr(
        orch,
-        "_append_sandbox_env",
+        "_append_sandbox_marker_env",
        lambda *_a: calls.append("sandbox"),
    )
    monkeypatch.setattr(orch, "_append_image_and_claude_args", lambda *_a: None)
@@ -135,27 +75,20 @@ def _stub_spawn_container_collaborators(
@pytest.mark.asyncio
-async def test_spawn_container_uses_sandbox_env_when_sandbox_active(
+async def test_spawn_container_uses_marker_env_when_opted_in(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    orch = AgentOrchestrator.__new__(AgentOrchestrator)
    calls: list[str] = []
    _stub_spawn_container_collaborators(monkeypatch, orch, calls)
-    info = SandboxInfo(
+    await orch._spawn_container(_config(["postgres"]))
        services={
            "postgres": SandboxConnection(
                host="h", port=5432, password="pw", user="sandbox", database="sandbox"
            )
        }
    )
    await orch._spawn_container(_config(info))
    assert calls == ["sandbox"]
@pytest.mark.asyncio
-async def test_spawn_container_uses_legacy_gate_env_without_sandbox(
+async def test_spawn_container_uses_legacy_gate_env_when_not_opted_in(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    orch = AgentOrchestrator.__new__(AgentOrchestrator)
@@ -168,25 +101,19 @@ async def test_spawn_container_uses_legacy_gate_env_without_sandbox(
@pytest.mark.asyncio
-async def test_spawn_container_stale_clear_spares_fresh_sandbox(
+async def test_spawn_container_stale_clear_runs_with_teardown_sandbox_false(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    """The pre-spawn stale-clear must not tear down the sandbox that was
+    """The pre-spawn stale-clear is vestigial now (nothing is provisioned
-    just provisioned for this very spawn (teardown_sandbox=False)."""
+    before spawn) but still passes teardown_sandbox=False — it must not
    tear down a sandbox the agent requested moments ago via the verb."""
    orch = AgentOrchestrator.__new__(AgentOrchestrator)
    calls: list[str] = []
    _stub_spawn_container_collaborators(monkeypatch, orch, calls)
    remove = AsyncMock(return_value=None)
    monkeypatch.setattr(orch, "_remove_container", remove)
-    info = SandboxInfo(
+    await orch._spawn_container(_config(["postgres"]))
        services={
            "postgres": SandboxConnection(
                host="h", port=5432, password="pw", user="sandbox", database="sandbox"
            )
        }
    )
    await orch._spawn_container(_config(info))
    remove.assert_awaited_once_with(
        "roboco-agent-dev-1",
@@ -1,13 +1,18 @@
-"""`AgentOrchestrator._maybe_provision_sandbox` — the spawn-time decision gate.
+"""`AgentOrchestrator._sandbox_available_services` — the spawn-time availability
 probe — and `ensure_sandbox` — the on-demand provision/cache path used by the
 `request_sandbox` do-verb.
-Off (flag or project) => None, byte-for-byte identical to legacy behavior. A
+Off (flag or project) => [], byte-for-byte identical to legacy behavior. A
-project lookup hiccup degrades to "no sandbox" (best-effort, matching the
+project lookup hiccup degrades to "no sandbox available" (best-effort,
-ambient-conventions-resolution convention); an actual provisioning failure
+matching the ambient-conventions-resolution convention). Provisioning itself
-IS fail-loud — an agent whose gate can't run must never spawn.
+no longer happens at spawn time — a spawn never fails on sandbox
 infrastructure; `ensure_sandbox` is the only path that calls
 `SandboxProvisioner.provision`, and it is idempotent via an in-memory cache.
 """
 from __future__ import annotations
 import asyncio
 from contextlib import asynccontextmanager
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -15,15 +20,19 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 from roboco.config import settings
 from roboco.models.sandbox import SandboxConnection, SandboxInfo
-from roboco.runtime.orchestrator import AgentOrchestrator, AgentReadinessError
+from roboco.runtime.orchestrator import AgentOrchestrator
 def _make_orchestrator() -> tuple[AgentOrchestrator, MagicMock]:
    orch = AgentOrchestrator.__new__(AgentOrchestrator)
    orch._bg_tasks = set()
    orch._running = True
    orch._sandbox_info = {}
    sandbox = MagicMock()
    sandbox.provision = AsyncMock()
    # Live by default so cache-hit tests that don't care about liveness pass
    # through; tests exercising DEFECT 3 (dead-container eviction) override.
    sandbox.is_live = AsyncMock(return_value=True)
    orch._sandbox = sandbox
    return orch, sandbox
@@ -33,23 +42,28 @@ async def _fake_db_ctx(db: Any) -> Any:
    yield db
 # ---------------------------------------------------------------------------
 # _sandbox_available_services (spawn-time probe, no provisioning)
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
-async def test_flag_off_returns_none_without_project_lookup(
+async def test_flag_off_returns_empty_without_project_lookup(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", False)
    orch, sandbox = _make_orchestrator()
    with patch("roboco.services.project.get_project_service") as get_svc:
-        result = await orch._maybe_provision_sandbox("dev-1", "roboco-api", "task-1")
+        result = await orch._sandbox_available_services("roboco-api")
-    assert result is None
+    assert result == []
    get_svc.assert_not_called()
    sandbox.provision.assert_not_called()
@pytest.mark.asyncio
-async def test_project_without_sandbox_services_returns_none(
+async def test_project_without_sandbox_services_returns_empty(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
@@ -65,14 +79,14 @@ async def test_project_without_sandbox_services_returns_none(
            return_value=project_service,
        ),
    ):
-        result = await orch._maybe_provision_sandbox("dev-1", "roboco-api", "task-1")
+        result = await orch._sandbox_available_services("roboco-api")
-    assert result is None
+    assert result == []
    sandbox.provision.assert_not_called()
@pytest.mark.asyncio
-async def test_missing_project_returns_none(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_missing_project_returns_empty(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    orch, _sandbox = _make_orchestrator()
    project_service = MagicMock()
@@ -85,13 +99,13 @@ async def test_missing_project_returns_none(monkeypatch: pytest.MonkeyPatch) ->
            return_value=project_service,
        ),
    ):
-        result = await orch._maybe_provision_sandbox("dev-1", "roboco-api", "task-1")
+        result = await orch._sandbox_available_services("roboco-api")
-    assert result is None
+    assert result == []
@pytest.mark.asyncio
-async def test_opted_in_project_provisions_sandbox(
+async def test_opted_in_project_returns_services_without_provisioning(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
@@ -99,14 +113,6 @@ async def test_opted_in_project_provisions_sandbox(
    project = MagicMock(sandbox_services=["postgres"])
    project_service = MagicMock()
    project_service.get_by_slug = AsyncMock(return_value=project)
    info = SandboxInfo(
        services={
            "postgres": SandboxConnection(
                host="h", port=5432, password="pw", user="sandbox", database="sandbox"
            )
        }
    )
    sandbox.provision.return_value = info
    with (
        patch("roboco.db.base.get_db_context", return_value=_fake_db_ctx(MagicMock())),
@@ -115,32 +121,10 @@ async def test_opted_in_project_provisions_sandbox(
            return_value=project_service,
        ),
    ):
-        result = await orch._maybe_provision_sandbox("dev-1", "roboco-api", "task-1")
+        result = await orch._sandbox_available_services("roboco-api")
-    assert result is info
+    assert result == ["postgres"]
-    sandbox.provision.assert_awaited_once_with("dev-1", ["postgres"])
+    sandbox.provision.assert_not_called()
@pytest.mark.asyncio
 async def test_provisioning_failure_raises_readiness_error(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    orch, sandbox = _make_orchestrator()
    project = MagicMock(sandbox_services=["postgres", "redis"])
    project_service = MagicMock()
    project_service.get_by_slug = AsyncMock(return_value=project)
    sandbox.provision.side_effect = RuntimeError("boom")
    with (
        patch("roboco.db.base.get_db_context", return_value=_fake_db_ctx(MagicMock())),
        patch(
            "roboco.services.project.get_project_service",
            return_value=project_service,
        ),
        pytest.raises(AgentReadinessError, match="sandbox provisioning failed"),
    ):
        await orch._maybe_provision_sandbox("dev-1", "roboco-api", "task-1")
@pytest.mark.asyncio
@@ -151,7 +135,138 @@ async def test_project_lookup_failure_degrades_to_no_sandbox(
    orch, sandbox = _make_orchestrator()
    with patch("roboco.db.base.get_db_context", side_effect=RuntimeError("db down")):
-        result = await orch._maybe_provision_sandbox("dev-1", "roboco-api", "task-1")
+        result = await orch._sandbox_available_services("roboco-api")
-    assert result is None
+    assert result == []
    sandbox.provision.assert_not_called()
 # ---------------------------------------------------------------------------
 # ensure_sandbox (on-demand provision + cache, called by request_sandbox)
 # ---------------------------------------------------------------------------
 def _info(services: dict[str, SandboxConnection]) -> SandboxInfo:
    return SandboxInfo(services=services)
@pytest.mark.asyncio
 async def test_ensure_sandbox_miss_provisions_and_caches() -> None:
    orch, sandbox = _make_orchestrator()
    info = _info({"postgres": SandboxConnection(host="h", port=5432, password="pw")})
    sandbox.provision.return_value = info
    result = await orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"])
    assert result is info
    sandbox.provision.assert_awaited_once_with("dev-1", ["postgres"])
    assert orch._sandbox_info["dev-1"] is info
@pytest.mark.asyncio
 async def test_ensure_sandbox_cache_hit_skips_second_provision() -> None:
    orch, sandbox = _make_orchestrator()
    info = _info({"postgres": SandboxConnection(host="h", port=5432, password="pw")})
    sandbox.provision.return_value = info
    first = await orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"])
    second = await orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"])
    assert first is second is info
    sandbox.provision.assert_awaited_once()
@pytest.mark.asyncio
 async def test_ensure_sandbox_first_subset_request_provisions_full_opted_set() -> None:
    """DEFECT 1 fix: a first request for a subset of the project's opted-in
    set provisions the FULL opted set — not just what this call named — so a
    later call for the rest of that set is a guaranteed cache hit and never
    falls through to a fresh provision() (whose pre-clear teardown() would
    otherwise kill the live container the agent is already using)."""
    orch, sandbox = _make_orchestrator()
    info = _info(
        {
            "postgres": SandboxConnection(host="h", port=5432, password="pw"),
            "redis": SandboxConnection(host="h", port=6379, password="rw"),
        }
    )
    sandbox.provision.return_value = info
    first = await orch.ensure_sandbox("dev-1", ["postgres"], ["postgres", "redis"])
    second = await orch.ensure_sandbox(
        "dev-1", ["postgres", "redis"], ["postgres", "redis"]
    )
    assert first is second is info
    sandbox.provision.assert_awaited_once_with("dev-1", ["postgres", "redis"])
    assert orch._sandbox_info["dev-1"] is info
@pytest.mark.asyncio
 async def test_ensure_sandbox_cache_is_per_agent_slug() -> None:
    """Caller A's cache entry never leaks to caller B (cross-agent isolation)."""
    orch, sandbox = _make_orchestrator()
    info_a = _info({"postgres": SandboxConnection(host="a", port=5432, password="pa")})
    info_b = _info({"postgres": SandboxConnection(host="b", port=5432, password="pb")})
    sandbox.provision.side_effect = [info_a, info_b]
    result_a = await orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"])
    result_b = await orch.ensure_sandbox("dev-2", ["postgres"], ["postgres"])
    assert result_a is info_a
    assert result_b is info_b
    assert orch._sandbox_info["dev-1"] is info_a
    assert orch._sandbox_info["dev-2"] is info_b
@pytest.mark.asyncio
 async def test_ensure_sandbox_concurrent_calls_serialize_on_agent_lock() -> None:
    """DEFECT 2 fix: two concurrent ensure_sandbox calls for the same agent
    (e.g. a client timeout + retry) must serialize behind the per-agent lock
    so only one provision() ever runs — never a race between provision() and
    a concurrent teardown()."""
    orch, sandbox = _make_orchestrator()
    info = _info({"postgres": SandboxConnection(host="h", port=5432, password="pw")})
    calls = 0
    async def _slow_provision(_agent_id: str, _services: list[str]) -> SandboxInfo:
        nonlocal calls
        calls += 1
        await asyncio.sleep(0.05)
        return info
    sandbox.provision.side_effect = _slow_provision
    results = await asyncio.gather(
        orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"]),
        orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"]),
    )
    assert results[0] is results[1] is info
    assert calls == 1
@pytest.mark.asyncio
 async def test_ensure_sandbox_cache_hit_with_dead_container_reprovisions() -> None:
    """DEFECT 3 fix: a cache hit whose container is no longer live (OOM-killed,
    manually removed) is evicted and re-provisioned with fresh creds, rather
    than handing back creds for a container that no longer exists."""
    orch, sandbox = _make_orchestrator()
    stale_info = _info(
        {"postgres": SandboxConnection(host="h", port=5432, password="pw-old")}
    )
    fresh_info = _info(
        {"postgres": SandboxConnection(host="h", port=5432, password="pw-new")}
    )
    sandbox.provision.side_effect = [stale_info, fresh_info]
    sandbox.is_live.return_value = False
    first = await orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"])
    second = await orch.ensure_sandbox("dev-1", ["postgres"], ["postgres"])
    expected_provision_calls = 2
    assert first is stale_info
    assert second is fresh_info
    assert sandbox.provision.await_count == expected_provision_calls
    assert orch._sandbox_info["dev-1"] is fresh_info
    sandbox.is_live.assert_awaited_once_with("dev-1", ["postgres"])
@@ -40,6 +40,10 @@ class _FakeRunner:
        # path skips the pull). Tests exercising the pull path override these.
        self.image_present: bool = True
        self.pull_rc: int = 0
        # is_live()'s `docker inspect --format={{.State.Running}}` fake —
        # set post-construction, mirroring image_present/pull_rc above.
        self.inspect_rc: int = 0
        self.inspect_running: bool = True
        self._ps_call_count = 0
    async def __call__(
@@ -48,26 +52,33 @@ class _FakeRunner:
        self.calls.append(args)
        verb = args[0]
        if verb == "run":
-            return self.run_rc, b"container-id\n", b""
+            rc, out, err = self.run_rc, b"container-id\n", b""
-        if verb == "exec":
+        elif verb == "exec":
-            return self.exec_rc, b"", b""
+            rc, out, err = self.exec_rc, b"", b""
-        if verb in ("stop", "kill", "rm"):
+        elif verb in ("stop", "kill", "rm"):
-            return self.teardown_rc, b"", b""
+            rc, out, err = self.teardown_rc, b"", b""
-        if verb == "image":
+        elif verb == "image":
            # `image inspect <img>` — rc 0 means present (skip pull).
            if args[1] != "inspect":
                raise AssertionError(f"unexpected image subverb: {args[1]}")
-            return (0 if self.image_present else 1), b"", b""
+            rc, out, err = (0 if self.image_present else 1), b"", b""
-        if verb == "pull":
+        elif verb == "pull":
-            return self.pull_rc, b"", b"" if self.pull_rc == 0 else b"pull failed\n"
+            rc = self.pull_rc
-        if verb == "ps":
+            out, err = b"", (b"" if rc == 0 else b"pull failed\n")
        elif verb == "ps":
            self._ps_call_count += 1
            # First ps call = the sandbox-labeled listing; second = live agents.
            listing = (
                self.ps_output if self._ps_call_count == 1 else self.ps_live_output
            )
-            return 0, listing, b""
+            rc, out, err = 0, listing, b""
-        raise AssertionError(f"unexpected docker verb: {verb}")
+        elif verb == "inspect":
            rc = self.inspect_rc
            out = b"true\n" if self.inspect_running else b"false\n"
            err = b""
        else:
            raise AssertionError(f"unexpected docker verb: {verb}")
        return rc, out, err
@pytest.fixture(autouse=True)
@@ -296,3 +307,48 @@ async def test_provision_pull_failure_raises() -> None:
        await provisioner.provision("dev-9", ["postgres"])
    # `docker run` never reached — pull failed first.
    assert not any(c[0] == "run" for c in runner.calls)
@pytest.mark.asyncio
 async def test_is_live_true_when_container_running() -> None:
    runner = _FakeRunner()
    provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
    assert await provisioner.is_live("dev-10", ["postgres", "redis"]) is True
    expected_inspect_calls = 2
    inspects = [c for c in runner.calls if c[0] == "inspect"]
    assert len(inspects) == expected_inspect_calls
@pytest.mark.asyncio
 async def test_is_live_false_when_container_stopped() -> None:
    """rc 0 but State.Running == false — container exists but isn't running."""
    runner = _FakeRunner()
    runner.inspect_running = False
    provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
    assert await provisioner.is_live("dev-11", ["postgres"]) is False
@pytest.mark.asyncio
 async def test_is_live_false_when_container_missing() -> None:
    """Nonzero rc — `docker inspect` fails outright on a removed container."""
    runner = _FakeRunner()
    runner.inspect_rc = 1
    provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
    assert await provisioner.is_live("dev-12", ["postgres"]) is False
@pytest.mark.asyncio
 async def test_is_live_short_circuits_on_first_dead_service() -> None:
    """A dead first service skips checking the rest — no need to inspect
    every container once one is already known dead."""
    runner = _FakeRunner()
    runner.inspect_rc = 1
    provisioner = SandboxProvisioner(network=_NETWORK, runner=runner)
    assert await provisioner.is_live("dev-13", ["postgres", "redis"]) is False
    inspects = [c for c in runner.calls if c[0] == "inspect"]
    assert len(inspects) == 1
@@ -118,3 +118,98 @@ async def test_sandbox_janitor_sweep_swallows_errors(
    sandbox.janitor_sweep.side_effect = RuntimeError("boom")
    await orch._sandbox_janitor_sweep()  # must not raise
 # ---------------------------------------------------------------------------
 # ensure_sandbox cache eviction (request_sandbox on-demand provisioning)
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_remove_container_evicts_ensure_sandbox_cache(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    monkeypatch.setattr(asyncio, "create_subprocess_exec", _fake_create_subprocess_exec)
    orch, _sandbox = _make_orchestrator()
    orch._sandbox_info = {"dev-1": MagicMock(), "dev-2": MagicMock()}
    await orch._remove_container("roboco-agent-dev-1")
    assert "dev-1" not in orch._sandbox_info
    assert "dev-2" in orch._sandbox_info
@pytest.mark.asyncio
 async def test_remove_container_teardown_false_spares_cache(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    monkeypatch.setattr(asyncio, "create_subprocess_exec", _fake_create_subprocess_exec)
    orch, _sandbox = _make_orchestrator()
    orch._sandbox_info = {"dev-1": MagicMock()}
    await orch._remove_container("roboco-agent-dev-1", teardown_sandbox=False)
    assert "dev-1" in orch._sandbox_info
@pytest.mark.asyncio
 async def test_janitor_sweep_evicts_cache_for_reaped_agents(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    monkeypatch.setattr(settings, "sandbox_db_enabled", True)
    orch, _sandbox = _make_orchestrator()
    orch._sandbox_info = {"dev-1": MagicMock(), "dev-2": MagicMock()}
    orch._instances = {"dev-2": MagicMock()}  # dev-1's agent instance is gone
    await orch._sandbox_janitor_sweep()
    assert "dev-1" not in orch._sandbox_info
    assert "dev-2" in orch._sandbox_info
 # ---------------------------------------------------------------------------
 # release_sandbox (end-of-engagement teardown, called by the Choreographer's
 # post-verb hook — i_am_done / unclaim / i_am_idle / pass_review / fail_review
 # / i_documented — instead of only at container removal)
 # ---------------------------------------------------------------------------
@pytest.mark.asyncio
 async def test_release_sandbox_no_cache_entry_is_fast_noop() -> None:
    """The overwhelmingly common case: no sandbox for this agent. Must not
    take the per-agent lock or call docker — the cache dict check alone
    decides, before any lock is even allocated."""
    orch, sandbox = _make_orchestrator()
    orch._sandbox_info = {}
    await orch.release_sandbox("dev-1")
    sandbox.teardown.assert_not_called()
    assert not hasattr(orch, "_sandbox_locks")
@pytest.mark.asyncio
 async def test_release_sandbox_tears_down_and_evicts_cache() -> None:
    orch, sandbox = _make_orchestrator()
    orch._sandbox_info = {"dev-1": MagicMock(), "dev-2": MagicMock()}
    await orch.release_sandbox("dev-1")
    sandbox.teardown.assert_awaited_once_with("dev-1")
    assert "dev-1" not in orch._sandbox_info
    assert "dev-2" in orch._sandbox_info
@pytest.mark.asyncio
 async def test_release_sandbox_is_idempotent() -> None:
    """A second release for the same slug (e.g. unclaim right after
    i_am_idle) is a no-op — the first call already evicted the cache."""
    orch, sandbox = _make_orchestrator()
    orch._sandbox_info = {"dev-1": MagicMock()}
    await orch.release_sandbox("dev-1")
    await orch.release_sandbox("dev-1")
    sandbox.teardown.assert_awaited_once_with("dev-1")