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

"""/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 pathlib import Path

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


# -- workspace files + exec (Phase 6, REQ-3-008; CUT-2) -------------------------
#
# The build surface reads/writes/list workspace files and runs Run/Test
# commands through the manager's backend. NO interactive shell relay (CUT-2:
# keystroke-level stdin/stdout is v0.4) — each exec is a bounded command with
# captured output. Paths are WORKSPACE-RELATIVE; traversal outside the
# workspace is rejected (the workdir bind is the boundary, but the API adds
# its own containment check — defense in depth).


class FileWriteRequest(BaseModel):
    path: str = Field(min_length=1)
    content: str


class ExecRequest(BaseModel):
    cmd: list[str] = Field(min_length=1)


class ExecResponse(BaseModel):
    cmd: list[str]
    returncode: int
    stdout: str
    stderr: str
    duration_s: float


async def _workspace_dir(manager: SandboxManager, sandbox_id: str):
    """Resolve the sandbox workspace (tracked layout or shell layout)."""
    info = await manager.get(sandbox_id)  # raises SandboxNotFoundError -> 404
    backend = manager._backend  # noqa: SLF001 - API owns the composition seam
    tracked = getattr(backend, "_tracked", {}).get(sandbox_id)
    if tracked is not None:
        return tracked.workspace, info
    return info.workdir / "workspace", info


def _safe_rel_path(raw: str) -> Path:
    """Workspace-relative path; reject absolute/traversal paths."""
    candidate = Path(raw)
    if candidate.is_absolute() or ".." in candidate.parts:
        raise HTTPException(status_code=422, detail=f"invalid workspace path {raw!r}")
    return candidate


@router.get("/{sandbox_id}/files")
async def list_files(
    sandbox_id: str,
    manager: SandboxManager = Depends(get_sandbox_manager),
) -> dict:
    try:
        workspace, _ = await _workspace_dir(manager, sandbox_id)
    except SandboxNotFoundError:
        raise HTTPException(status_code=404, detail=f"no sandbox {sandbox_id!r}") from None
    return {"files": sorted(p.name for p in workspace.iterdir()) if workspace.is_dir() else []}


@router.get("/{sandbox_id}/files/{path:path}")
async def read_file(
    sandbox_id: str,
    path: str,
    manager: SandboxManager = Depends(get_sandbox_manager),
) -> dict:
    workspace, _ = await _workspace_dir(manager, sandbox_id)
    rel = _safe_rel_path(path)
    target = workspace / rel
    if not target.is_file():
        raise HTTPException(status_code=404, detail=f"no file {path!r}")
    return {"path": path, "content": target.read_text(errors="replace")}


@router.put("/{sandbox_id}/files/{path:path}")
async def write_file(
    sandbox_id: str,
    path: str,
    body: FileWriteRequest,
    manager: SandboxManager = Depends(get_sandbox_manager),
) -> dict:
    workspace, _ = await _workspace_dir(manager, sandbox_id)
    rel = _safe_rel_path(body.path)
    target = workspace / rel
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(body.content)
    return {"path": body.path, "written": True}


@router.post("/{sandbox_id}/exec", response_model=ExecResponse)
async def exec_command(
    sandbox_id: str,
    body: ExecRequest,
    manager: SandboxManager = Depends(get_sandbox_manager),
) -> ExecResponse:
    try:
        await manager.get(sandbox_id)
    except SandboxNotFoundError:
        raise HTTPException(status_code=404, detail=f"no sandbox {sandbox_id!r}") from None
    backend = manager._backend  # noqa: SLF001 - API owns the composition seam
    handle = manager._handles.get(sandbox_id)  # noqa: SLF001
    if handle is None:
        raise HTTPException(status_code=404, detail=f"no live handle {sandbox_id!r}")
    result = await backend.exec(handle, body.cmd)
    return ExecResponse(**result.model_dump())