feat(P01): sandboxes lifecycle API + G-5 abuse control (Wave 3)

Task 1-3-01: /v1/sandboxes endpoints (create/list/get/snapshot/delete) over the manager
singleton via DI; lifespan boots the startup orphan reaper (a-1) + destroys all on shutdown.
Abuse control (G-5): learner allowlist (403 unknown id), per-learner active cap (429),
global create-rate cap (429). CORS gains DELETE. 174 full-suite tests green; manual probe
POST /v1/sandboxes -> 201 verified live; ruff clean.

---ci---
phase: 1
milestone: v0.3
status: execute
requirements: {covered: [REQ-3-001, REQ-3-002], partial: []}
---/ci---

apps/ai-service/.env.example

@@ -15,5 +15,13 @@ AI_SANDBOX_MAX_CONCURRENT=5
 AI_SANDBOX_TIMEOUT_S=900
 AI_SANDBOX_MAX_WORKDIR_MB=512
 
+# G-5 abuse control (NOT auth — KYC/identity deferred):
+# comma-separated learner allowlist; unknown ids can't create sandboxes (403)
+AI_LEARNER_ALLOWLIST=pilot-learner
+# max ACTIVE sandboxes per learner → 429 when exceeded
+AI_SANDBOX_MAX_PER_LEARNER=1
+# global creates per rolling 60s window (in-memory) → 429 when exceeded
+AI_SANDBOX_CREATES_PER_MIN=10
+
 # Persistence (SQLite)
 AI_DB_PATH=ai_service/data/nextcraft.db

apps/ai-service/ai_service/api/__init__.py

@@ -8,6 +8,7 @@ from .chat import router as chat_router
 from .lab import router as lab_router
 from .mentor import router as mentor_router
 from .proctor import router as proctor_router
+from .sandboxes import router as sandboxes_router
 
 __all__ = [
     "assessment_router",
@@ -15,4 +16,5 @@ __all__ = [
     "lab_router",
     "mentor_router",
     "proctor_router",
+    "sandboxes_router",
 ]

apps/ai-service/ai_service/api/deps.py

@@ -6,6 +6,8 @@ from ..agents.registry import AgentRegistry
 from ..agents.session import SessionStore
 from ..config import Settings
 from ..llm.base import LLMProvider
+from ..sandbox.manager import SandboxManager
+from ..sandbox.workdir import SandboxDir
 
 
 def get_settings(request: Request) -> Settings:
@@ -22,3 +24,12 @@ def get_session_store(request: Request) -> SessionStore:
 
 def get_agent_registry(request: Request) -> AgentRegistry:
     return request.app.state.agent_registry
+
+
+def get_sandbox_manager(request: Request) -> SandboxManager:
+    return request.app.state.sandbox_manager
+
+
+def get_sandbox_test_layout(request: Request) -> SandboxDir | None:
+    """Optional test seam (app.state.sandbox_test_layout); always None in prod."""
+    return getattr(request.app.state, "sandbox_test_layout", None)

apps/ai-service/ai_service/api/sandboxes.py

@@ -0,0 +1,219 @@
+"""/v1/sandboxes — sandbox lifecycle API with G-5 abuse control (REQ-3-001).
+
+The sandbox fabric (`sandbox/manager.py`) owns lifecycle; this module owns the
+HTTP contract and the abuse gates, which are MIDDLEWARE-LAYER concerns and
+therefore live here, never in the manager:
+
+  allowlist (403)        G-5: `learner_id` must be in
+                         `settings.learner_allowlist`. This is NOT auth —
+                         KYC/identity is deferred; the allowlist only keeps
+                         unvetted ids from spawning namespaces on this box.
+  per-learner cap (429)  `settings.sandbox_max_per_learner` ACTIVE sandboxes
+                         per learner (default 1 — one pilot, one box).
+  global create cap (429) `settings.sandbox_creates_per_min` creates per
+                         rolling 60s window across all learners; in-memory,
+                         process-local (matches the handle registry, D-019).
+  pool full (503)        D-032 capacity guard (`PoolFullError`), no queue.
+
+Response models are local to the api/ surface. The manager returns
+`SandboxHandleInfo` rows (handle fields + learner_id); the response `pid`
+field is typed `int | None` and excluded — a host-process detail that is
+never part of the API contract.
+"""
+
+import time
+from collections import deque
+
+from fastapi import APIRouter, Depends, HTTPException, Response
+from pydantic import BaseModel, ConfigDict, Field
+
+from ..config import Settings
+from ..sandbox.manager import (
+    PoolFullError,
+    SandboxHandleInfo,
+    SandboxManager,
+    SandboxNotFoundError,
+)
+from ..sandbox.workdir import SandboxDir
+from .deps import get_sandbox_manager, get_sandbox_test_layout, get_settings
+
+router = APIRouter(prefix="/v1/sandboxes", tags=["sandboxes"])
+
+# In-memory create-rate window (monotonic timestamps, process-local). Module
+# state is acceptable here for the same reason the handle registry is: one
+# process, one box, no store (D-019/D-027 precedent).
+_CREATE_TIMES: deque[float] = deque()
+
+
+# -- contracts ----------------------------------------------------------------
+
+
+class SandboxCreateRequest(BaseModel):
+    learner_id: str = Field(min_length=1)
+
+
+class SandboxResponse(BaseModel):
+    """Public sandbox handle. `workdir` is the absolute host path."""
+
+    model_config = ConfigDict(frozen=True)
+
+    id: str
+    learner_id: str
+    workdir: str
+    created_at: str
+    pid: int | None = Field(
+        default=None,
+        exclude=True,  # host-process detail; never part of the API contract
+    )
+
+
+def _to_response(info: SandboxHandleInfo) -> SandboxResponse:
+    return SandboxResponse(
+        id=info.id,
+        learner_id=info.learner_id,
+        workdir=str(info.workdir),
+        created_at=info.created_at.isoformat(),
+        pid=info.pid,
+    )
+
+
+class SandboxListResponse(BaseModel):
+    sandboxes: list[SandboxResponse]
+
+
+class SnapshotResponse(BaseModel):
+    sandbox_id: str
+    snapshot_path: str
+    files: list[str]
+
+
+# -- abuse control (G-5; middleware layer, not auth) ---------------------------
+
+
+def _enforce_allowlist(learner_id: str, settings: Settings) -> None:
+    if learner_id not in settings.learner_allowlist:
+        raise HTTPException(
+            status_code=403,
+            detail=f"learner_id {learner_id!r} is not on the sandbox allowlist (G-5)",
+        )
+
+
+def _check_per_learner_cap(
+    infos: list[SandboxHandleInfo], learner_id: str, settings: Settings
+) -> None:
+    active = sum(1 for info in infos if info.learner_id == learner_id)
+    if active >= settings.sandbox_max_per_learner:
+        raise HTTPException(
+            status_code=429,
+            detail=(
+                f"learner {learner_id!r} already has {active} active "
+                f"sandbox(es); per-learner cap is {settings.sandbox_max_per_learner}"
+            ),
+        )
+
+
+def _check_global_create_rate(settings: Settings) -> None:
+    """Sliding-window global create cap. Admitted only after ALL checks pass,
+    so a rejected create never consumes budget."""
+    now = time.monotonic()
+    while _CREATE_TIMES and now - _CREATE_TIMES[0] > 60.0:
+        _CREATE_TIMES.popleft()
+    if len(_CREATE_TIMES) >= settings.sandbox_creates_per_min:
+        raise HTTPException(
+            status_code=429,
+            detail=(
+                f"global sandbox create rate exceeded "
+                f"({settings.sandbox_creates_per_min}/min); retry shortly"
+            ),
+        )
+    _CREATE_TIMES.append(now)
+
+
+# -- endpoints ---------------------------------------------------------------
+
+
+@router.post("", status_code=201, response_model=SandboxResponse)
+async def create_sandbox(
+    body: SandboxCreateRequest,
+    manager: SandboxManager = Depends(get_sandbox_manager),
+    settings: Settings = Depends(get_settings),
+) -> SandboxResponse:
+    _enforce_allowlist(body.learner_id, settings)
+    _check_per_learner_cap(await manager.list(), body.learner_id, settings)
+    _check_global_create_rate(settings)
+    try:
+        info = await manager.create(body.learner_id)
+    except PoolFullError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    return _to_response(info)
+
+
+@router.get("", response_model=SandboxListResponse)
+async def list_sandboxes(
+    manager: SandboxManager = Depends(get_sandbox_manager),
+) -> SandboxListResponse:
+    return SandboxListResponse(
+        sandboxes=[_to_response(info) for info in await manager.list()]
+    )
+
+
+@router.get("/{sandbox_id}", response_model=SandboxResponse)
+async def get_sandbox(
+    sandbox_id: str,
+    manager: SandboxManager = Depends(get_sandbox_manager),
+) -> SandboxResponse:
+    try:
+        info = await manager.get(sandbox_id)
+    except SandboxNotFoundError:
+        raise HTTPException(
+            status_code=404, detail=f"unknown sandbox {sandbox_id!r}"
+        ) from None
+    return _to_response(info)
+
+
+@router.post("/{sandbox_id}/snapshot", response_model=SnapshotResponse)
+async def snapshot_sandbox(
+    sandbox_id: str,
+    response: Response,
+    manager: SandboxManager = Depends(get_sandbox_manager),
+    test_layout: SandboxDir | None = Depends(get_sandbox_test_layout),
+) -> SnapshotResponse:
+    try:
+        snapshot_path = await manager.snapshot(sandbox_id)
+    except SandboxNotFoundError:
+        raise HTTPException(
+            status_code=404, detail=f"unknown sandbox {sandbox_id!r}"
+        ) from None
+    # Test seam: a copied workspace keeps the `files` assertion honest for
+    # fast stub-backend tests; production always hits the real snapshot above.
+    if test_layout is not None and test_layout.snapshots.is_dir():
+        copies = sorted(test_layout.snapshots.iterdir())
+        if copies:
+            response.headers["X-Snapshot-Copy"] = str(copies[-1])
+    return SnapshotResponse(
+        sandbox_id=sandbox_id,
+        snapshot_path=str(snapshot_path),
+        files=sorted(p.name for p in snapshot_path.iterdir()),
+    )
+
+
+@router.delete("/{sandbox_id}", status_code=204)
+async def delete_sandbox(
+    sandbox_id: str,
+    manager: SandboxManager = Depends(get_sandbox_manager),
+    test_layout: SandboxDir | None = Depends(get_sandbox_test_layout),
+) -> Response:
+    try:
+        await manager.get(sandbox_id)
+    except SandboxNotFoundError:
+        raise HTTPException(
+            status_code=404, detail=f"unknown sandbox {sandbox_id!r}"
+        ) from None
+    # Workdir is KEPT (purge_workdir=False): snapshots must survive destroy so
+    # a learner's last state can be restored. The periodic G-2 reaper owns
+    # quota; explicit purge is an ops action, not an API verb.
+    await manager.destroy(sandbox_id, purge_workdir=False)
+    result = Response(status_code=204)
+    if test_layout is not None:
+        result.headers["X-Workspace-Copy"] = str(test_layout.workspace)
+    return result

apps/ai-service/ai_service/config.py

@@ -1,8 +1,10 @@
 """Service settings — pydantic-settings, env prefix AI_, .env support."""
 
 from pathlib import Path
+from typing import Annotated
 
-from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic import field_validator
+from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 # apps/ai-service/  (sandbox dir default is relative to the app, not the CWD)
 _SERVICE_ROOT = Path(__file__).resolve().parent.parent
@@ -37,5 +39,28 @@ class Settings(BaseSettings):
     # manager sweep (NOT kernel-enforced — no cgroup delegation/sudo here).
     sandbox_max_workdir_mb: int = 512
 
+    # G-5 abuse control (NOT auth — KYC/auth is deferred; these keep the
+    # single-box pilot from melting down before identity lands):
+    #
+    # Server-side learner allowlist. Env form is a COMMA-SEPARATED string
+    # (e.g. AI_LEARNER_ALLOWLIST="pilot-learner,learner-2"); NoDecode skips
+    # pydantic-settings' JSON decoding of complex types and the validator
+    # below splits/strips/drops empties. Default: the single mock pilot id.
+    learner_allowlist: Annotated[list[str], NoDecode] = ["pilot-learner"]
+
+    # Max ACTIVE sandboxes per learner → API maps excess to 429.
+    sandbox_max_per_learner: int = 1
+
+    # Global create-rate ceiling (creates per rolling 60s window, shared
+    # across learners) → API maps excess to 429. In-memory, process-local.
+    sandbox_creates_per_min: int = 10
+
+    @field_validator("learner_allowlist", mode="before")
+    @classmethod
+    def _split_allowlist_csv(cls, value: object) -> object:
+        if isinstance(value, str):
+            return [item.strip() for item in value.split(",") if item.strip()]
+        return value
+
     # D-027: SQLite path for telemetry/grades/variants/defenses stores.
     db_path: Path = _SERVICE_ROOT / "ai_service" / "data" / "nextcraft.db"

apps/ai-service/ai_service/main.py

@@ -1,5 +1,8 @@
 """FastAPI app factory — lifespan, CORS, health, routers."""
 
+import asyncio
+import contextlib
+import logging
 from contextlib import asynccontextmanager
 
 import httpx
@@ -14,9 +17,17 @@ from .api import (
     lab_router,
     mentor_router,
     proctor_router,
+    sandboxes_router,
 )
 from .config import Settings
 from .llm import create_provider
+from .sandbox import SandboxManager, UnshareBackend
+
+logger = logging.getLogger(__name__)
+
+#: Interval between wall-clock/G-2 reaper passes (the manager owns the pass;
+#: the lifespan owns the loop). 60s against a 900s default timeout → ≤6.7% lag.
+REAPER_INTERVAL_S = 60.0
 
 
 def create_app(settings: Settings | None = None) -> FastAPI:
@@ -32,16 +43,45 @@ def create_app(settings: Settings | None = None) -> FastAPI:
         app.state.session_store = InMemorySessionStore()
         app.state.agent_registry = AgentRegistry()
         register_builtin_agents(app.state.agent_registry)
-        yield
-        await app.state.http_client.aclose()
 
-    app = FastAPI(title="Nextcraft AI Service", version="0.2.0", lifespan=lifespan)
+        # v0.3 sandbox fabric (REQ-3-001): singleton manager, DI'd via
+        # app.state. Tests may pre-set app.state.sandbox_manager (dependency
+        # override by state injection) to swap the backend; the lifespan then
+        # adopts it instead of constructing the real UnshareBackend one.
+        manager = getattr(app.state, "sandbox_manager", None)
+        if manager is None:
+            manager = SandboxManager(backend=UnshareBackend(), settings=settings)
+            app.state.sandbox_manager = manager
+        await manager.start()  # a-1: reap on-disk orphans from a previous process
+
+        async def _reaper_loop() -> None:
+            # Wall-clock timeout + G-2 workdir-size sweep, one pass per tick.
+            while True:
+                await asyncio.sleep(REAPER_INTERVAL_S)
+                try:
+                    await manager.reap_expired()
+                except Exception:
+                    logger.exception("sandbox reaper pass failed; retrying next tick")
+
+        reaper = asyncio.create_task(_reaper_loop())
+        try:
+            yield
+        finally:
+            reaper.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await reaper
+            # No orphans outlive the process (a-1, shutdown half): destroy
+            # everything live; workdirs stay on disk for snapshot restore.
+            await manager.destroy_all()
+            await app.state.http_client.aclose()
+
+    app = FastAPI(title="Nextcraft AI Service", version="0.3.0", lifespan=lifespan)
 
     # A-008: localhost-only CORS, no credentials
     app.add_middleware(
         CORSMiddleware,
         allow_origins=["http://localhost:3000", "http://127.0.0.1:3000"],
-        allow_methods=["GET", "POST", "OPTIONS"],
+        allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
         allow_headers=["Content-Type"],
         allow_credentials=False,
     )
@@ -59,6 +99,7 @@ def create_app(settings: Settings | None = None) -> FastAPI:
     app.include_router(assessment_router)
     app.include_router(mentor_router)
     app.include_router(proctor_router)
+    app.include_router(sandboxes_router)
     return app
 
 

apps/ai-service/ai_service/sandbox/__init__.py

@@ -5,6 +5,7 @@ Public surface:
     SandboxBackend                  — the protocol every backend implements (D-024 port).
     UnshareBackend                  — util-linux `unshare` backend (D-024 backend).
     SandboxManager                  — lifecycle + pool guard (D-032) + reapers (G-2, a-1).
+    SandboxHandleInfo               — manager return row: handle fields + learner_id.
     PoolFullError / SandboxNotFoundError / SandboxIntegrityEvent — manager surface.
     SandboxUnavailableError         — raised when namespaces are not usable on this host.
     SandboxDir / workspace_path / create_layout / snapshot — per-sandbox workdir layout.
@@ -15,6 +16,7 @@ Boundary rule: `sandbox/` never imports `api/` or `agents/`; it owns subprocess
 from .backend import ExecResult, ResourceLimits, SandboxBackend, SandboxHandle, SandboxSpec
 from .manager import (
     PoolFullError,
+    SandboxHandleInfo,
     SandboxIntegrityEvent,
     SandboxManager,
     SandboxNotFoundError,
@@ -29,6 +31,7 @@ __all__ = [
     "SandboxBackend",
     "SandboxDir",
     "SandboxHandle",
+    "SandboxHandleInfo",
     "SandboxIntegrityEvent",
     "SandboxManager",
     "SandboxNotFoundError",

apps/ai-service/ai_service/sandbox/manager.py

@@ -2,7 +2,9 @@
 
 Responsibilities (D-032, G-2, a-1):
 
-- create/list/get/snapshot/destroy over a `SandboxBackend` port.
+- create/list/get/snapshot/destroy over a `SandboxBackend` port. create/list/
+  get return `SandboxHandleInfo` rows — the handle fields plus the owning
+  `learner_id` — so the API layer never re-asks "who owns this id?".
 - Capacity guard (D-032): `create` raises `PoolFullError` when the active
   count reaches `settings.sandbox_max_concurrent`. No queue — the API layer
   maps this to 503.
@@ -87,6 +89,23 @@ class SandboxIntegrityEvent(BaseModel):
     observed_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
 
 
+class SandboxHandleInfo(BaseModel):
+    """A `SandboxHandle` plus its owning `learner_id` (manager return row).
+
+    Handles alone don't carry the learner — the registry side-table does —
+    and every API read/list needs it, so the manager joins the two ONCE here
+    instead of exposing `_learner_ids` internals.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    id: str
+    learner_id: str
+    workdir: Path
+    created_at: datetime
+    pid: int | None = None
+
+
 class SandboxManager:
     """Lifecycle owner for learner sandboxes.
 
@@ -123,7 +142,7 @@ class SandboxManager:
 
     # -- lifecycle ----------------------------------------------------------
 
-    async def create(self, learner_id: str) -> SandboxHandle:
+    async def create(self, learner_id: str) -> SandboxHandleInfo:
         """Spawn a sandbox for `learner_id`, or raise `PoolFullError` (D-032)."""
         async with self._lock:
             if len(self._handles) >= self._settings.sandbox_max_concurrent:
@@ -139,23 +158,33 @@ class SandboxManager:
             self._learner_ids[handle.id] = learner_id
             self._write_pid_marker(handle, learner_id)
             logger.info("sandbox created: id=%s learner=%s", handle.id, learner_id)
-            return handle
+            return self._info_for(handle)
 
-    async def list(self) -> list[SandboxHandle]:
-        """All live handles (idle + busy; the backend has no busy flag)."""
+    async def list(self) -> list[SandboxHandleInfo]:
+        """All live sandboxes (idle + busy; the backend has no busy flag)."""
         async with self._lock:
-            return list(self._handles.values())
+            return [self._info_for(h) for h in self._handles.values()]
 
-    async def get(self, sandbox_id: str) -> SandboxHandle:
+    async def get(self, sandbox_id: str) -> SandboxHandleInfo:
+        async with self._lock:
+            handle = self._handles.get(sandbox_id)
+            learner_id = self._learner_ids.get(sandbox_id, "unknown")
+        if handle is None:
+            raise SandboxNotFoundError(sandbox_id)
+        return SandboxHandleInfo(
+            id=handle.id,
+            learner_id=learner_id,
+            workdir=handle.workdir,
+            created_at=handle.created_at,
+            pid=handle.pid,
+        )
+
+    async def snapshot(self, sandbox_id: str) -> Path:
+        """Copy the workspace into `<workdir>/snapshots/<utc-ts>/`; return it."""
         async with self._lock:
             handle = self._handles.get(sandbox_id)
         if handle is None:
             raise SandboxNotFoundError(sandbox_id)
-        return handle
-
-    async def snapshot(self, sandbox_id: str) -> Path:
-        """Copy the workspace into `<workdir>/snapshots/<utc-ts>/`; return it."""
-        handle = await self.get(sandbox_id)
         return await self._backend.snapshot(handle)
 
     async def destroy(self, sandbox_id: str, *, purge_workdir: bool = False) -> None:
@@ -203,10 +232,13 @@ class SandboxManager:
         cap_bytes = int(self._settings.sandbox_max_workdir_mb) * 1024 * 1024
 
         async with self._lock:
-            handles = list(self._handles.values())
+            rows = [
+                (handle, self._learner_ids.get(handle.id, "unknown"), handle.created_at)
+                for handle in self._handles.values()
+            ]
 
-        for handle in handles:
-            age_s = (now - handle.created_at).total_seconds()
+        for handle, learner_id, created_at in rows:
+            age_s = (now - created_at).total_seconds()
             if age_s > timeout_s:
                 await self.destroy(handle.id)
                 destroyed.append(handle.id)
@@ -219,7 +251,7 @@ class SandboxManager:
                 continue  # already gone; no size sweep needed on a dead handle
             size = _tree_size_bytes(workdir_mod.workspace_path_from_workdir(handle.workdir))
             if size > cap_bytes:
-                await self._reap_oversized(handle, size, cap_bytes)
+                await self._reap_oversized(handle, learner_id, size, cap_bytes)
                 destroyed.append(handle.id)
         return destroyed
 
@@ -261,8 +293,30 @@ class SandboxManager:
                 )
             )
 
+    async def destroy_all(self) -> None:
+        """Shutdown hook: destroy every live sandbox (no orphans on exit).
+
+        Workdirs (and their snapshots) are kept on disk — destroy semantics
+        here match `destroy(purge_workdir=False)`; the next boot's startup
+        reaper (a-1) decides what to clean based on pid markers.
+        """
+        async with self._lock:
+            handles = list(self._handles.values())
+        for handle in handles:
+            await self.destroy(handle.id)
+
     # -- internals ------------------------------------------------------------
 
+    def _info_for(self, handle: SandboxHandle) -> SandboxHandleInfo:
+        # Caller holds the lock (create/list) — the side-table read is atomic.
+        return SandboxHandleInfo(
+            id=handle.id,
+            learner_id=self._learner_ids.get(handle.id, "unknown"),
+            workdir=handle.workdir,
+            created_at=handle.created_at,
+            pid=handle.pid,
+        )
+
     def _write_pid_marker(self, handle: SandboxHandle, learner_id: str) -> None:
         marker = handle.workdir / PID_MARKER
         try:
@@ -280,10 +334,13 @@ class SandboxManager:
             logger.warning("could not write pid marker %s", marker)
 
     async def _reap_oversized(
-        self, handle: SandboxHandle, size_bytes: int, cap_bytes: int
+        self,
+        handle: SandboxHandle,
+        learner_id: str,
+        size_bytes: int,
+        cap_bytes: int,
     ) -> None:
         """G-2 sweep step: snapshot evidence → destroy → record the signal."""
-        learner_id = self._learner_ids.get(handle.id, "unknown")
         snapshot_path: Path | None = None
         try:
             snapshot_path = await self._backend.snapshot(handle)

apps/ai-service/tests/api/test_sandboxes.py

@@ -0,0 +1,361 @@
+"""/v1/sandboxes API tests — lifecycle, pool guard, G-5 abuse control (REQ-3-001).
+
+Every test here runs against `StubBackend` (workdir layout without namespace
+spawning) EXCEPT the final probe-guarded real-backend path; Wave 1/2 already
+covers boundary isolation, so the API layer only needs the manager contract.
+
+DI/lifespan wiring note: tests build the manager themselves, pre-set
+`app.state.sandbox_manager` BEFORE the TestClient lifespan runs, and the
+lifespan adopts that instance (state-injection override) instead of
+constructing the real UnshareBackend one. The lifespan still `start()`s it,
+runs the periodic reaper, and `destroy_all()`s on shutdown — the same
+production path, just with a stub at the port.
+
+Snapshot/destroy content plumbing for stub tests uses the explicit test seam
+`app.state.sandbox_test_layout` → `X-Snapshot-Copy` / `X-Workspace-Copy`
+response headers (production never sets them).
+"""
+
+from __future__ import annotations
+
+import shutil
+import time
+from collections.abc import Iterator
+from datetime import UTC, datetime
+from pathlib import Path
+
+import pytest
+from fastapi.testclient import TestClient
+
+from ai_service.config import Settings
+from ai_service.main import create_app
+from ai_service.sandbox import SandboxHandle, SandboxManager, UnshareBackend
+from ai_service.sandbox.backend import ExecResult, SandboxSpec
+from ai_service.sandbox.workdir import SandboxDir, create_layout
+from ai_service.sandbox.workdir import snapshot as workdir_snapshot
+from tests.sandbox.test_isolation import requires_userns
+
+PILOT = "pilot-learner"
+OTHER = "pilot-learner-2"
+
+BASE_SETTINGS: dict = {
+    "provider": "mock",
+    "sandbox_max_concurrent": 5,
+    "sandbox_max_per_learner": 1,
+    "sandbox_creates_per_min": 10,
+}
+
+
+class StubBackend:
+    """Structural SandboxBackend: lays out the workdir, spawns nothing.
+
+    Tracks spawn/destroy calls so lifespan shutdown behaviour is assertable.
+    """
+
+    def __init__(self) -> None:
+        self.spawned_specs: list[SandboxSpec] = []
+        self.destroyed_ids: list[str] = []
+
+    async def spawn(self, spec: SandboxSpec) -> SandboxHandle:
+        create_layout(spec)
+        self.spawned_specs.append(spec)
+        return SandboxHandle(
+            id=spec.sandbox_id, pid=None, workdir=spec.workdir,
+            created_at=datetime.now(UTC),
+        )
+
+    async def exec(self, handle: SandboxHandle, cmd: list[str]) -> ExecResult:
+        raise NotImplementedError("API tests never exec")
+
+    async def snapshot(self, handle: SandboxHandle) -> Path:
+        return workdir_snapshot(handle.workdir)
+
+    async def destroy(self, handle: SandboxHandle) -> None:
+        self.destroyed_ids.append(handle.id)
+        handle.pid = None
+
+
+def _seed_workspace(layout: SandboxDir, name: str, content: str) -> None:
+    layout.workspace.mkdir(parents=True, exist_ok=True)
+    (layout.workspace / name).write_text(content)
+
+
+@pytest.fixture(autouse=True)
+def _clear_create_window() -> Iterator[None]:
+    """Module-global rate window must start empty in every test."""
+    from ai_service.api import sandboxes as sandboxes_api
+
+    sandboxes_api._CREATE_TIMES.clear()
+    yield
+    sandboxes_api._CREATE_TIMES.clear()
+
+
+@pytest.fixture()
+def stub_backend() -> StubBackend:
+    return StubBackend()
+
+
+@pytest.fixture()
+def client(
+    tmp_path: Path, stub_backend: StubBackend, monkeypatch: pytest.MonkeyPatch
+) -> Iterator[TestClient]:
+    """App + adopted stub manager, sandbox dir rooted in a per-test tmp_path."""
+    monkeypatch.setenv("AI_SANDBOX_CREATES_PER_MIN", "150")  # shared-window headroom
+    settings = Settings(**{**BASE_SETTINGS, "sandbox_dir": tmp_path / "sandboxes"})
+    app = create_app(settings)
+    app.state.sandbox_manager = SandboxManager(backend=stub_backend, settings=settings)
+    with TestClient(app) as c:
+        yield c
+
+
+# -- lifecycle roundtrip --------------------------------------------------------
+
+
+def test_create_list_get_snapshot_delete_roundtrip(client: TestClient) -> None:
+    created = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+    assert created.status_code == 201
+    body = created.json()
+    assert body["id"].startswith("sbx-")
+    assert body["learner_id"] == PILOT
+    assert body["created_at"]
+    assert "pid" not in body  # host-process detail, excluded from the contract
+    sandbox_id = body["id"]
+
+    listed = client.get("/v1/sandboxes")
+    assert listed.status_code == 200
+    rows = listed.json()["sandboxes"]
+    assert [r["id"] for r in rows] == [sandbox_id]
+    assert rows[0]["learner_id"] == PILOT
+
+    got = client.get(f"/v1/sandboxes/{sandbox_id}")
+    assert got.status_code == 200
+    assert got.json()["id"] == sandbox_id
+
+    # Seed workspace content on the host side, then snapshot it away.
+    layout = SandboxDir(
+        root=Path(body["workdir"]),
+        workspace=Path(body["workdir"]) / "workspace",
+        snapshots=Path(body["workdir"]) / "snapshots",
+    )
+    _seed_workspace(layout, "solution.py", "print(42)\n")
+
+    snap = client.post(f"/v1/sandboxes/{sandbox_id}/snapshot")
+    assert snap.status_code == 200
+    snap_body = snap.json()
+    assert snap_body["sandbox_id"] == sandbox_id
+    assert snap_body["files"] == ["solution.py"]
+    snapshot_path = Path(snap_body["snapshot_path"])
+    assert (snapshot_path / "solution.py").read_text() == "print(42)\n"
+
+    # Destroy: 204, registry empties, but the workdir (and snapshot) survives.
+    deleted = client.delete(f"/v1/sandboxes/{sandbox_id}")
+    assert deleted.status_code == 204
+    assert deleted.content == b""
+    assert client.get("/v1/sandboxes").json()["sandboxes"] == []
+    assert client.get(f"/v1/sandboxes/{sandbox_id}").status_code == 404
+    assert layout.root.is_dir()  # kept for restore; not purged (API semantic)
+    assert (snapshot_path / "solution.py").is_file()
+
+
+def test_create_validates_body_422(client: TestClient) -> None:
+    assert client.post("/v1/sandboxes", json={}).status_code == 422
+    assert client.post("/v1/sandboxes", json={"learner_id": ""}).status_code == 422
+
+
+def test_snapshot_unknown_id_404(client: TestClient) -> None:
+    response = client.post("/v1/sandboxes/sbx-nope/snapshot")
+    assert response.status_code == 404
+    assert "sbx-nope" in response.json()["detail"]
+
+
+def test_delete_unknown_id_404(client: TestClient) -> None:
+    response = client.delete("/v1/sandboxes/sbx-nope")
+    assert response.status_code == 404
+    assert "sbx-nope" in response.json()["detail"]
+
+
+def test_get_unknown_id_404(client: TestClient) -> None:
+    assert client.get("/v1/sandboxes/sbx-nope").status_code == 404
+
+
+# -- D-032 pool guard → 503 ------------------------------------------------------
+
+
+def test_pool_full_returns_503(tmp_path: Path, stub_backend: StubBackend) -> None:
+    settings = Settings(
+        provider="mock",
+        sandbox_dir=tmp_path / "sandboxes",
+        sandbox_max_concurrent=2,
+        sandbox_max_per_learner=5,  # cap disabled for this test
+        sandbox_creates_per_min=150,
+    )
+    app = create_app(settings)
+    app.state.sandbox_manager = SandboxManager(backend=stub_backend, settings=settings)
+    with TestClient(app) as client:
+        ids = [
+            client.post("/v1/sandboxes", json={"learner_id": PILOT}).json()["id"]
+            for _ in range(2)
+        ]
+        overflow = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+        assert overflow.status_code == 503
+        assert "pool full" in overflow.json()["detail"]
+        # Capacity frees on destroy: the next create succeeds.
+        client.delete(f"/v1/sandboxes/{ids[0]}")
+        refilled = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+        assert refilled.status_code == 201
+
+
+# -- G-5 abuse control ------------------------------------------------------------
+
+
+def test_non_allowlisted_learner_403(client: TestClient) -> None:
+    response = client.post("/v1/sandboxes", json={"learner_id": "intruder-7"})
+    assert response.status_code == 403
+    assert "allowlist" in response.json()["detail"]
+    assert client.get("/v1/sandboxes").json()["sandboxes"] == []  # nothing spawned
+
+
+def test_second_active_sandbox_for_same_learner_429(
+    tmp_path: Path, stub_backend: StubBackend
+) -> None:
+    settings = Settings(
+        provider="mock",
+        sandbox_dir=tmp_path / "sandboxes",
+        learner_allowlist=[PILOT, OTHER],
+        sandbox_max_concurrent=10,
+        sandbox_max_per_learner=1,
+        sandbox_creates_per_min=150,
+    )
+    app = create_app(settings)
+    app.state.sandbox_manager = SandboxManager(backend=stub_backend, settings=settings)
+    with TestClient(app) as client:
+        first = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+        assert first.status_code == 201
+        second = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+        assert second.status_code == 429
+        assert "per-learner cap" in second.json()["detail"]
+        # Cap counts ACTIVE sandboxes: after destroy the learner can create again.
+        client.delete(f"/v1/sandboxes/{first.json()['id']}")
+        again = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+        assert again.status_code == 201
+        # A different allowlisted learner was never blocked by the pilot's sandbox.
+        other = client.post("/v1/sandboxes", json={"learner_id": OTHER})
+        assert other.status_code == 201
+
+
+def test_burst_over_global_create_rate_429(tmp_path: Path, stub_backend: StubBackend) -> None:
+    settings = Settings(
+        provider="mock",
+        sandbox_dir=tmp_path / "sandboxes",
+        sandbox_max_concurrent=10,
+        sandbox_max_per_learner=10,  # per-learner cap disabled for this test
+        sandbox_creates_per_min=3,
+    )
+    app = create_app(settings)
+    manager = SandboxManager(backend=stub_backend, settings=settings)
+    app.state.sandbox_manager = manager
+    with TestClient(app) as client:
+        codes = [
+            client.post("/v1/sandboxes", json={"learner_id": PILOT}).status_code
+            for _ in range(3)
+        ]
+        assert codes == [201, 201, 201]
+        burst = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+        assert burst.status_code == 429
+        assert "create rate" in burst.json()["detail"]
+        assert manager.active_count == 3  # the 429 spawned nothing
+
+        # Rejected creates never consume budget; still full one moment later.
+        assert client.post("/v1/sandboxes", json={"learner_id": PILOT}).status_code == 429
+
+
+def test_create_rate_window_slides(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The window is monotonic-time based; fake the clock, skip real sleeping."""
+    from fastapi import HTTPException
+
+    from ai_service.api import sandboxes as sandboxes_api
+
+    sandboxes_api._CREATE_TIMES.clear()
+    fake_now = 1_000.0
+    monkeypatch.setattr(time, "monotonic", lambda: fake_now)
+    settings = Settings(provider="mock", sandbox_creates_per_min=2)
+    sandboxes_api._check_global_create_rate(settings)
+    sandboxes_api._check_global_create_rate(settings)
+    with pytest.raises(HTTPException) as excinfo:
+        sandboxes_api._check_global_create_rate(settings)
+    assert excinfo.value.status_code == 429
+
+    fake_now += 61.0  # window slides: oldest entries age out
+    sandboxes_api._check_global_create_rate(settings)  # admitted again
+
+
+# -- lifespan wiring --------------------------------------------------------------
+
+
+def test_lifespan_start_and_shutdown_destroy(
+    tmp_path: Path, stub_backend: StubBackend
+) -> None:
+    """a-1 at boot: an orphan workdir with a dead recorded pid is reaped;
+    at shutdown every live sandbox is destroyed (no orphans outlive us)."""
+    orphan_root = tmp_path / "sandboxes" / "sbx-orphaned"
+    (orphan_root / "workspace").mkdir(parents=True)
+    (orphan_root / "sandbox.json").write_text(
+        '{"sandbox_id": "sbx-orphaned", "learner_id": "?", "pid": 4194304}'
+    )
+    settings = Settings(
+        provider="mock",
+        sandbox_dir=tmp_path / "sandboxes",
+        sandbox_creates_per_min=150,
+    )
+    manager = SandboxManager(backend=stub_backend, settings=settings)
+    app = create_app(settings)
+    app.state.sandbox_manager = manager
+    with TestClient(app) as client:
+        assert not orphan_root.exists()  # startup reaper ran during lifespan boot
+        assert any(e.kind == "orphan_reaped" for e in manager.integrity_events)
+        created = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+        sandbox_id = created.json()["id"]
+        assert manager.active_count == 1
+    # Exiting the TestClient runs the shutdown half of the lifespan.
+    assert stub_backend.destroyed_ids == [sandbox_id]
+    assert manager.active_count == 0
+
+
+def test_lifespan_constructs_real_manager_when_not_overridden(tmp_path: Path) -> None:
+    """No override → the lifespan builds the production UnshareBackend manager."""
+    settings = Settings(provider="mock", sandbox_dir=tmp_path / "sandboxes")
+    app = create_app(settings)
+    with TestClient(app):
+        manager = app.state.sandbox_manager
+        assert isinstance(manager, SandboxManager)
+
+
+# -- real-backend path (probe-guarded; Wave 1/2 owns deep isolation) --------------
+
+#: Repo-anchored, gitignored scratch root (tests/sandboxes/ like isolation
+#: tests) — /tmp is off-limits: the in-namespace tmpfs shadows host /tmp.
+REAL_SANDBOXES_ROOT = Path(__file__).resolve().parents[1] / "sandboxes"
+
+
+@requires_userns
+def test_real_backend_create_path_runs(tmp_path: Path) -> None:
+    """One cheap real-namespace pass through the API (create only, no exec)."""
+    sandbox_root = REAL_SANDBOXES_ROOT / f"api-{tmp_path.name}"
+    settings = Settings(
+        provider="mock",
+        sandbox_dir=sandbox_root,
+        sandbox_creates_per_min=150,
+    )
+    app = create_app(settings)  # no override → lifespan wires UnshareBackend
+    try:
+        with TestClient(app) as client:
+            created = client.post("/v1/sandboxes", json={"learner_id": PILOT})
+            assert created.status_code == 201
+            sandbox_id = created.json()["id"]
+            assert (sandbox_root / sandbox_id / "workspace").is_dir()
+            assert isinstance(app.state.sandbox_manager._backend, UnshareBackend)
+        # Shutdown destroyed the live sandbox; the kept workdir still exists.
+        assert app.state.sandbox_manager.active_count == 0
+        assert (sandbox_root / sandbox_id / "workspace").is_dir()
+    finally:
+        shutil.rmtree(sandbox_root, ignore_errors=True)

apps/ai-service/tests/sandbox/test_manager.py

@@ -1,10 +1,8 @@
 """SandboxManager tests (REQ-3-001, REQ-3-002).
 
-Manager-logic tests run against a fake in-tree `SandboxBackend` (no real
-namespaces — spawning is the backend's concern, lifecycle is the manager's).
-One end-to-end smoke test uses the REAL UnshareBackend, guarded by the same
-userns probe as test_isolation.py, so it skips cleanly where namespaces are
-unavailable (on this box it RUNS).
+The manager's create/list/get return `SandboxHandleInfo` rows (handle fields
++ learner_id); `backend.exec` and workdir helpers still take the raw handle,
+so handle-mutating tests reach the registry via `manager._handles`.
 """
 
 from __future__ import annotations
@@ -81,11 +79,13 @@ def manager(settings: Settings, backend: FakeBackend) -> SandboxManager:
 
 
 async def test_create_registers_handle(manager: SandboxManager) -> None:
-    handle = await manager.create("learner-1")
-    assert handle.id.startswith("sbx-")
-    assert (handle.workdir / "workspace").is_dir()
+    info = await manager.create("learner-1")
+    assert info.id.startswith("sbx-")
+    assert info.learner_id == "learner-1"
+    assert info.pid is None
+    assert (info.workdir / "workspace").is_dir()
     assert manager.active_count == 1
-    marker = json.loads((handle.workdir / PID_MARKER).read_text())
+    marker = json.loads((info.workdir / PID_MARKER).read_text())
     assert marker["pid"] == os.getpid()
     assert marker["learner_id"] == "learner-1"
 
@@ -117,59 +117,76 @@ async def test_list_and_get_roundtrip(manager: SandboxManager) -> None:
     b = await manager.create("learner-2")
     listed = await manager.list()
     assert {h.id for h in listed} == {a.id, b.id}
-    assert (await manager.get(a.id)).id == a.id
+    assert {h.learner_id for h in listed} == {"learner-1", "learner-2"}
+    got = await manager.get(a.id)
+    assert got.id == a.id
+    assert got.learner_id == "learner-1"
     with pytest.raises(SandboxNotFoundError):
         await manager.get("sbx-nope")
 
 
 async def test_snapshot_copies_workspace(manager: SandboxManager) -> None:
-    handle = await manager.create("learner-1")
-    (handle.workdir / "workspace" / "solution.py").write_text("print(42)\n")
-    snap = await manager.snapshot(handle.id)
-    assert snap.parent == handle.workdir / "snapshots"
+    info = await manager.create("learner-1")
+    (info.workdir / "workspace" / "solution.py").write_text("print(42)\n")
+    snap = await manager.snapshot(info.id)
+    assert snap.parent == info.workdir / "snapshots"
     assert (snap / "solution.py").read_text() == "print(42)\n"
 
 
 async def test_destroy_keeps_workdir_with_snapshots(
     manager: SandboxManager, backend: FakeBackend
 ) -> None:
-    handle = await manager.create("learner-1")
-    (handle.workdir / "workspace" / "keep.txt").write_text("state")
-    await manager.snapshot(handle.id)
-    await manager.destroy(handle.id)
-    assert handle.id in backend.destroyed_ids
+    info = await manager.create("learner-1")
+    (info.workdir / "workspace" / "keep.txt").write_text("state")
+    await manager.snapshot(info.id)
+    await manager.destroy(info.id)
+    assert info.id in backend.destroyed_ids
     assert manager.active_count == 0
-    assert handle.workdir.is_dir()  # snapshots survive destroy (restore path)
-    assert list((handle.workdir / "snapshots").iterdir())
+    assert info.workdir.is_dir()  # snapshots survive destroy (restore path)
+    assert list((info.workdir / "snapshots").iterdir())
     with pytest.raises(SandboxNotFoundError):
-        await manager.get(handle.id)
+        await manager.get(info.id)
 
 
 async def test_destroy_purge_removes_workdir(
     manager: SandboxManager, backend: FakeBackend
 ) -> None:
-    handle = await manager.create("learner-1")
-    await manager.destroy(handle.id, purge_workdir=True)
-    assert handle.id in backend.destroyed_ids
-    assert not handle.workdir.exists()
+    info = await manager.create("learner-1")
+    await manager.destroy(info.id, purge_workdir=True)
+    assert info.id in backend.destroyed_ids
+    assert not info.workdir.exists()
 
 
 async def test_destroy_unknown_id_is_idempotent(manager: SandboxManager) -> None:
     await manager.destroy("sbx-ghost")  # must not raise
 
 
+async def test_destroy_all_empties_registry_keeps_workdirs(
+    manager: SandboxManager, backend: FakeBackend
+) -> None:
+    a = await manager.create("learner-1")
+    b = await manager.create("learner-2")
+    await manager.destroy_all()
+    assert manager.active_count == 0
+    assert backend.destroyed_ids == [a.id, b.id]
+    assert a.workdir.is_dir() and b.workdir.is_dir()  # snapshots survive
+    await manager.destroy_all()  # idempotent no-op on an empty registry
+    assert backend.destroyed_ids == [a.id, b.id]
+
+
 # -- reap_expired: wall-clock + G-2 sweep -------------------------------------
 
 
 async def test_reap_expired_destroys_timed_out_sandbox(
     manager: SandboxManager, backend: FakeBackend
 ) -> None:
-    handle = await manager.create("learner-1")
-    handle.created_at = datetime.now(UTC) - timedelta(seconds=901)
+    info = await manager.create("learner-1")
+    # created_at on the returned info is a snapshot; mutate the registry row.
+    manager._handles[info.id].created_at = datetime.now(UTC) - timedelta(seconds=901)
     fresh = await manager.create("learner-2")
     destroyed = await manager.reap_expired()
-    assert destroyed == [handle.id]
-    assert handle.id in backend.destroyed_ids
+    assert destroyed == [info.id]
+    assert info.id in backend.destroyed_ids
     assert (await manager.list())[0].id == fresh.id
 
 
@@ -187,30 +204,30 @@ async def test_workdir_size_sweep_destroys_over_cap_and_records_signal(
 ) -> None:
     settings.sandbox_max_workdir_mb = 1  # 1 MiB cap so the test stays tiny
     manager = SandboxManager(backend=backend, settings=settings)
-    handle = await manager.create("learner-greedy")
+    info = await manager.create("learner-greedy")
     # Over the cap, spread across many files (G-2 is the aggregate guard;
     # RLIMIT_FSIZE alone does not catch this).
     for i in range(9):
-        (handle.workdir / "workspace" / f"chunk-{i}.bin").write_bytes(b"x" * 256 * 1024)
+        (info.workdir / "workspace" / f"chunk-{i}.bin").write_bytes(b"x" * 256 * 1024)
     with caplog.at_level("WARNING"):
         destroyed = await manager.reap_expired()
-    assert destroyed == [handle.id]
-    assert handle.id in backend.destroyed_ids
+    assert destroyed == [info.id]
+    assert info.id in backend.destroyed_ids
     events = manager.integrity_events
     assert len(events) == 1
     assert events[0].kind == "workdir_size_cap"
-    assert events[0].sandbox_id == handle.id
+    assert events[0].sandbox_id == info.id
     assert events[0].learner_id == "learner-greedy"
     assert any("G-2" in rec.message for rec in caplog.records)
     # Snapshot-then-destroy: evidence preserved on disk after the reap.
-    snapshots = list((handle.workdir / "snapshots").iterdir())
+    snapshots = list((info.workdir / "snapshots").iterdir())
     assert len(snapshots) == 1
     assert (snapshots[0] / "chunk-0.bin").is_file()
 
 
 async def test_sweep_skips_under_cap_sandbox(manager: SandboxManager) -> None:
-    handle = await manager.create("learner-1")
-    (handle.workdir / "workspace" / "small.txt").write_text("ok")
+    info = await manager.create("learner-1")
+    (info.workdir / "workspace" / "small.txt").write_text("ok")
     assert await manager.reap_expired() == []
     assert manager.active_count == 1
 
@@ -283,9 +300,11 @@ async def test_real_unshare_backend_create_exec_destroy(tmp_path: Path) -> None:
     settings = Settings(provider="mock", sandbox_dir=sandbox_root)
     backend = UnshareBackend()
     manager = SandboxManager(backend=backend, settings=settings)
-    handle = await manager.create("learner-smoke")
+    info = await manager.create("learner-smoke")
+    # exec takes the raw backend handle; the registry holds it for info.id.
+    handle = manager._handles[info.id]
     result = await backend.exec(handle, ["echo", "hello-from-manager"])
     assert result.returncode == 0 and result.stdout.strip() == "hello-from-manager"
-    await manager.destroy(handle.id)
+    await manager.destroy(info.id)
     assert manager.active_count == 0
-    assert handle.workdir.is_dir()  # workdir kept for snapshot restore
+    assert info.workdir.is_dir()  # workdir kept for snapshot restore