merge(P02): phase/02 live build telemetry → milestone/v0.3-credential-engines

---ci---
phase: 2
milestone: v0.3
status: ship
---/ci---

.ciagent/CHECKPOINT.json

@@ -1,8 +1,8 @@
 {
-  "phase": 1,
-  "stage": "complete",
+  "phase": 2,
+  "stage": "verify",
   "milestone": "v0.3",
   "phase_role": "execution",
   "attempts": 0,
-  "updated_at": "2026-09-11T18:43:22Z"
+  "updated_at": "2026-09-12T01:41:08Z"
 }

.ciagent/REQUIREMENTS.md

@@ -8,7 +8,7 @@
 |----|-------------|----------|-------|--------|
 | REQ-3-001 | Sandbox fabric: isolated per-learner execution environments (sandboxed IDE, design tool, simulation) with lifecycle management | critical | 1 | complete |
 | REQ-3-002 | Sandbox isolation + resource limits: per-learner isolation boundary, CPU/memory quotas (rlimits), wall-clock time quota, disk-quota via per-sandbox workdir usage sweep (best-effort, not kernel-enforced), no cross-tenant access, snapshot support. **Known gap (v0.3): per-sandbox pids and hard disk caps are NOT kernel-enforceable without cgroup delegation/sudo — documented as accepted risk** | critical | 1 | complete |
-| REQ-3-003 | Live build telemetry: in-environment capture of process events (commands, file diffs, run/test results, activity) streamed reliably to ai-service with per-learner trace persistence | critical | 2 | pending |
+| REQ-3-003 | Live build telemetry: in-environment capture of process events (commands, file diffs, run/test results, activity) streamed reliably to ai-service with per-learner trace persistence | critical | 2 | complete |
 
 ### Credential Engines
 
@@ -176,7 +176,7 @@
 |-------------|-------|--------|
 | REQ-3-001 | 1 | complete |
 | REQ-3-002 | 1 | complete |
-| REQ-3-003 | 2 | pending |
+| REQ-3-003 | 2 | complete |
 | REQ-3-004 | 3 | pending |
 | REQ-3-005 | 4 | pending |
 | REQ-3-006 | 5 | pending |

.ciagent/ROADMAP.md

@@ -20,7 +20,7 @@
 |---|------|--------|------------|--------------|------------------|
 | 0 | Pre-execution | in-progress | — | — | Specification, clarify, research, plan complete; .ciagent/ files updated for v0.3 |
 | 1 | Sandbox fabric | complete | 0 | REQ-3-001, REQ-3-002 | Isolated per-learner sandbox environments provisioned (IDE / design / simulation); lifecycle API (create/destroy/snapshot); resource limits enforced; no cross-tenant access |
-| 2 | Live build telemetry | pending | 1 | REQ-3-003 | In-environment capture of process events (commands, file diffs, run/test results, keystroke-level activity) streamed to ai-service; reliable transport; per-learner trace persistence |
+| 2 | Live build telemetry | complete | 1 | REQ-3-003 | In-environment capture of process events (commands, file diffs, run/test results, keystroke-level activity) streamed to ai-service; reliable transport; per-learner trace persistence |
 | 3 | Process-trace grading engine | pending | 2 | REQ-3-004 | Grades artifacts from their full process traces (not just final output); emits structured rubric-aligned scores; feeds Assessor real inputs |
 | 4 | Variant task generation | pending | 1 | REQ-3-005 | Per-learner task variants generated so no two learners receive identical prompts; variant seed recorded for grading fairness |
 | 5 | Oral / voice defense | pending | 3 | REQ-3-006 | AI examiner conducts spoken defense of submitted work; STT → dialogue → TTS; transcript + integrity signals captured; feeds Proctor/Mentor |

apps/ai-service/README.md

@@ -194,6 +194,28 @@ Unprivileged user namespaces are on the box's kernel and need neither a
 daemon, nor suid helpers, nor network access — they are the only isolation
 primitive that works here, so that's what v0.3 uses.
 
+## Telemetry delivery semantics (v0.3, REQ-3-003)
+
+Delivery is **at-least-once**; storage is **exactly-once** — the two compose:
+
+- The in-sandbox capture agent (stdlib-only, `scripts/sandbox-agent.py`)
+  spools every event to a durable JSONL file (fsync per append) BEFORE any
+  send attempt, so no event can be lost to a dead socket or a SIGKILL.
+- The WS ingest endpoint (`WS /v1/telemetry/ingest?learner_id&task_id`,
+  D-026) dedups server-side on the `(learner_id, task_id, seq)` primary key:
+  re-sends (reconnect flushes, replay margin) are collapsed, never upserted.
+- On disconnect the agent reconnects with exponential backoff and flushes
+  the spool in `seq` order; a transient outage therefore loses nothing and
+  stores each event exactly once (`tests/telemetry/test_durability.py`
+  proves this end-to-end against a real namespace sandbox + live server).
+- Replay/read path: `GET /v1/telemetry/traces/{learner}/{task}` returns the
+  complete ordered trace; `GET /v1/telemetry/gaps/{learner}/{task}` returns
+  missing seqs for gap detection.
+- Flood boundary (G-3): a connection exceeding `AI_TELEMETRY_MAX_EVENTS_PER_TASK`
+  (default 50,000) is closed with WS code 1008 and its trace is marked
+  `INCOMPLETE_FLOODED` — a terminal integrity flag the grader refuses to
+  grade. Silent event dropping is forbidden: it would corrupt grading input.
+
 ## Layout
 
 ```

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

@@ -59,6 +59,7 @@ import asyncio
 import os
 import shlex
 import shutil
+import signal
 import time
 from datetime import UTC, datetime
 from pathlib import Path
@@ -488,11 +489,35 @@ class UnshareBackend:  # satisfies SandboxBackend structurally (Protocol)
         """
         tracked = self._tracked.pop(handle.id, None)
         if tracked is not None:
+            # Agent first (it must not flush a "stopped" event into a dead
+            # sandbox), then the namespace tree. The inner `unshare --fork`
+            # shim is NOT the namespace init: killing it orphans its child
+            # (the `sleep` that is PID 1 of the sandbox pid+mnt+net ns),
+            # which reparents to host init and holds the tmpfs + bind for
+            # a full hour (observed: ~30 leaked `sleep 3600` after a test
+            # run). `--kill-child` does not reach it either (util-linux
+            # 2.38 leaks the same child under this flag combo — the child
+            # is reparented before unshare's signal handler runs). The
+            # deterministic kill is SIGKILL on the ns-init's HOST pid,
+            # which we already track as `tracked.inner_pid` (nsenter uses
+            # it for exec); the kernel then tears down the namespace with
+            # its init (no processes remain).
             for proc in (tracked.agent, tracked.inner, tracked.helper):
                 if proc is not None:
                     await self._reap(proc)
+            self._kill_pid(tracked.inner_pid)
         handle.pid = None
 
+    @staticmethod
+    def _kill_pid(pid: int | None, sig: int = signal.SIGKILL) -> None:
+        """Best-effort host-side signal; pid recycled or gone is not an error."""
+        if pid is None:
+            return
+        try:
+            os.kill(pid, sig)
+        except (ProcessLookupError, PermissionError):
+            pass  # already dead, or not ours — nothing to do
+
     @staticmethod
     async def _reap(proc: asyncio.subprocess.Process) -> None:
         """SIGTERM then SIGKILL, tolerant of an already-dead process."""

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

@@ -135,6 +135,57 @@ async def test_no_task_id_means_no_capture(tmp_path: Path) -> None:
     store.close()
 
 
+@pytest.mark.asyncio
+async def test_destroy_kills_inner_namespace_not_just_the_shim() -> None:
+    """Destroy must reap the ns-init, not only the `unshare --fork` shim.
+
+    Regression: `_reap(inner)` kills the unshare PARENT, but its forked child
+    (PID 1 of the sandbox pid/mnt/net namespace, the `sleep`) reparents to
+    host init and holds the tmpfs + workspace bind for the full sleep
+    duration — a leaked sandbox per destroy (observed ~30 orphaned
+    `sleep 3600` processes after one suite run; `--kill-child` did not reach
+    it under this flag combo). The ns-init's host pid is `tracked.inner_pid`.
+    """
+    _userns_probe()
+
+    backend = UnshareBackend()
+    manager = SandboxManager(backend=backend, settings=Settings())
+
+    handle = await manager.create("lifecycle-learner", task_id="lifecycle-task")
+    try:
+        tracked = backend._tracked[handle.id]  # noqa: SLF001
+        assert tracked.agent is not None
+        assert _proc_alive(tracked.agent.pid), "agent should live with the sandbox"
+        assert _proc_alive(tracked.inner_pid), "ns-init should live with the sandbox"
+    finally:
+        await manager.destroy(handle.id)
+
+    # Reaping the unshare shim is asynchronous up to _reap's timeouts; poll
+    # until both the agent AND the ns-init host pid are gone.
+    deadline = time.monotonic() + 20.0
+    while time.monotonic() < deadline and (
+        _proc_alive(tracked.agent.pid) or _proc_alive(tracked.inner_pid)
+    ):
+        await asyncio.sleep(0.25)
+    assert not _proc_alive(tracked.agent.pid), "agent pid survived destroy (lifecycle)"
+    assert not _proc_alive(tracked.inner_pid), (
+        "inner namespace init survived destroy — leaked sandbox (tmpfs + bind "
+        "held for a full hour; every destroy leaked one namespace process)"
+    )
+
+
+def _proc_alive(pid: int | None) -> bool:
+    if pid is None:
+        return False
+    try:
+        with open(f"/proc/{pid}/stat") as fh:
+            # state is the first field after comm: Z (zombie) counts as dead
+            # for our purpose — it holds no namespace and is reaped next.
+            return fh.read().rsplit(") ", 1)[1].split()[0] != "Z"
+    except OSError:
+        return False
+
+
 async def _await_events(
     store: SQLiteTraceStore, learner: str, task: str, *, minimum: int, timeout_s: float = 20.0
 ) -> list:

apps/ai-service/tests/telemetry/test_durability.py

@@ -0,0 +1,239 @@
+"""Dropped-connection durability probe (Task 2-4-01, REQ-3-003).
+
+At-least-once delivery + server-side (learner,task,seq) dedup = every event
+stored exactly once, in order — proven against a REAL uvicorn ingest and a
+REAL namespace sandbox, with the connection severed mid-stream by a killable
+TCP proxy (deterministic "network down" window).
+
+Probe-guarded: skips (not fails) on hosts without user namespaces.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import socket
+import struct
+import time
+from pathlib import Path
+
+import pytest
+import uvicorn
+
+from ai_service.config import Settings
+from ai_service.main import create_app
+from ai_service.sandbox import SandboxManager
+from ai_service.sandbox.unshare_backend import UnshareBackend
+from ai_service.telemetry.ingest import TraceIntegrityMap
+from ai_service.telemetry.store import SQLiteTraceStore
+
+from ..sandbox.test_isolation import USERSNS_AVAILABLE
+
+
+def _userns_probe() -> None:
+    if not USERSNS_AVAILABLE:
+        pytest.skip("user namespaces unavailable on this host (probe)")
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+class KillableProxy:
+    """TCP forwarder whose data path dies on ``kill()`` — the listener stays.
+
+    Accepts are preserved so the agent's reconnects fail fast (connection
+    reset) instead of hanging on a black-holed socket, keeping the outage
+    window deterministic. ``revive()`` restores forwarding.
+    """
+
+    def __init__(self, target_port: int) -> None:
+        self.target_port = target_port
+        self.listen_port = _free_port()
+        self._listener: socket.socket | None = None
+        self._killed = False
+        self._conns: list[socket.socket] = []
+
+    async def start(self) -> None:
+        self._listener = socket.socket()
+        self._listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._listener.bind(("127.0.0.1", self.listen_port))
+        self._listener.listen(16)
+        self._listener.setblocking(False)
+        loop = asyncio.get_running_loop()
+        loop.create_task(self._accept_loop())
+
+    async def _accept_loop(self) -> None:
+        loop = asyncio.get_running_loop()
+        assert self._listener is not None
+        while True:
+            try:
+                client, _ = await loop.sock_accept(self._listener)
+            except OSError:
+                return  # listener closed
+            client.setblocking(False)  # accepted sockets default to blocking
+            if self._killed:
+                with contextlib.suppress(OSError):
+                    # RST on accept: reconnects fail fast while "down"
+                    client.setsockopt(
+                        socket.SOL_SOCKET, socket.SO_LINGER,
+                        struct.pack("<ii", 1, 0),
+                    )
+                    client.close()
+                continue
+            upstream = socket.socket()
+            upstream.setblocking(False)
+            try:
+                await loop.sock_connect(upstream, ("127.0.0.1", self.target_port))
+            except OSError:
+                with contextlib.suppress(OSError):
+                    client.close()
+                continue
+            self._conns.extend((client, upstream))
+            loop.create_task(self._pump(client, upstream))
+            loop.create_task(self._pump(upstream, client))
+
+    async def _pump(self, src: socket.socket, dst: socket.socket) -> None:
+        """Forward until EOF, error, or the killed flag is observed.
+
+        Only ``dst`` is closed here — the reverse-direction pump owns ``src``
+        (closing the peer's socket from this task causes EBADF cascades).
+        """
+        loop = asyncio.get_running_loop()
+        try:
+            while True:
+                data = await loop.sock_recv(src, 65536)
+                if not data:
+                    return
+                if self._killed:
+                    return  # outage: drop the relay, peer sees EOF on dst close
+                await loop.sock_sendall(dst, data)
+        except OSError:
+            return
+        finally:
+            with contextlib.suppress(OSError):
+                dst.close()
+
+    def kill(self) -> None:
+        """Sever the data path — flag only.
+
+        The pumps and the accept loop observe ``_killed`` themselves and close
+        THEIR OWN sockets from inside the event loop. Closing sockets that the
+        loop is currently awaiting from the outside deadlocks the loop's
+        child-watcher in this environment (observed: post-kill subprocess
+        spawns hung forever) — so no external socket surgery, ever.
+        """
+
+    def revive(self) -> None:
+        self._killed = False
+
+
+async def _await_events(
+    store: SQLiteTraceStore, learner: str, task: str, *, minimum: int, timeout_s: float = 30.0
+) -> list:
+    deadline = time.monotonic() + timeout_s
+    while time.monotonic() < deadline:
+        events = store.get_trace(learner, task)
+        if len(events) >= minimum:
+            return events
+        await asyncio.sleep(0.25)
+    return store.get_trace(learner, task)
+
+
+@pytest.mark.asyncio
+async def test_disconnect_reconnect_loses_nothing(tmp_path: Path) -> None:
+    """Sever the agent's WS mid-stream; every event lands exactly once, ordered."""
+    _userns_probe()
+
+    learner = "durability-learner"
+    task = "task-durability"
+
+    store = SQLiteTraceStore(db_path=tmp_path / "durability.db")
+    app = create_app()
+    app.state.trace_store = store
+    app.state.trace_integrity = TraceIntegrityMap()
+
+    server_port = _free_port()
+    server = uvicorn.Server(
+        uvicorn.Config(app, host="127.0.0.1", port=server_port, log_level="warning")
+    )
+    serve_task = asyncio.get_running_loop().create_task(server.serve())
+
+    proxy = KillableProxy(target_port=server_port)
+    await proxy.start()
+
+    manager = SandboxManager(backend=UnshareBackend(), settings=Settings())
+    app.state.sandbox_manager = manager
+
+    def capture_env(sandbox_id: str, learner_id: str, task_id: str) -> dict[str, str]:
+        return {
+            "NC_LEARNER_ID": learner_id,
+            "NC_TASK_ID": task_id,
+            "NC_SANDBOX_ID": sandbox_id,
+            # Agent dials the PROXY; the proxy forwards to the real server.
+            "NC_INGEST_URL": (
+                f"ws://127.0.0.1:{proxy.listen_port}/v1/telemetry/ingest"
+                f"?learner_id={learner_id}&task_id={task_id}&sandbox_id={sandbox_id}"
+            ),
+            # Fast reconnect so the probe stays bounded.
+            "NC_BACKOFF_BASE_S": "0.1",
+            "NC_BACKOFF_MAX_S": "0.5",
+        }
+
+    manager._capture_env = capture_env  # noqa: SLF001 - test seam
+
+    try:
+        for _ in range(100):
+            if server.started:
+                break
+            await asyncio.sleep(0.1)
+        assert server.started, "uvicorn did not start"
+
+        handle = await manager.create(learner, task_id=task)
+        try:
+            live = manager._handles[handle.id]  # noqa: SLF001 - test seam
+            backend: UnshareBackend = manager._backend  # noqa: SLF001 - test seam
+
+            # Phase A — connected: a write streams through.
+            result = await backend.exec(live, ["sh", "-c", "echo one > a.txt"])
+            assert result.returncode == 0, result.stderr
+            events = await _await_events(store, learner, task, minimum=1)
+            assert events, "no events arrived before the outage"
+
+            # Phase B — network down: sever mid-stream, keep generating.
+            proxy.kill()
+            for n in ("two", "three", "four"):
+                result = await backend.exec(live, ["sh", "-c", f"echo {n} > {n}.txt"])
+                assert result.returncode == 0, result.stderr
+                await asyncio.sleep(0.3)
+
+            # Phase C — revive: the agent reconnects (fast backoff) and
+            # flushes the spool. Server dedups on (learner,task,seq).
+            proxy.revive()
+            await asyncio.sleep(1.5)
+            for n in ("five", "six"):
+                result = await backend.exec(live, ["sh", "-c", f"echo {n} > {n}.txt"])
+                assert result.returncode == 0, result.stderr
+
+            events = await _await_events(
+                store, learner, task, minimum=6, timeout_s=30.0
+            )
+            seqs = [e.seq for e in events]
+            assert seqs == sorted(seqs), f"out of order after reconnect: {seqs}"
+            assert len(set(seqs)) == len(seqs), f"duplicates stored: {seqs}"
+            assert len(events) >= 6, f"lost events across the outage: {seqs}"
+            # Exactly-once storage despite at-least-once delivery: every spool
+            # flush may re-send the replay-margin line; dedup collapses it.
+            stored_files = {
+                e.payload.get("path") for e in events if e.kind == "file_diff"
+            }
+            assert stored_files, "file_diff events missing from the stored trace"
+        finally:
+            await manager.destroy(handle.id)
+    finally:
+        server.should_exit = True
+        with contextlib.suppress(Exception):
+            await asyncio.wait_for(serve_task, timeout=10.0)
+        store.close()