From 26b4a5be607d7e449daa68efc2f8a4f13950fe19 Mon Sep 17 00:00:00 2001 From: CIAgent Date: Sat, 12 Sep 2026 00:01:15 +0000 Subject: [PATCH] feat(P02): telemetry-wired sandbox spawn (Wave 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 2-3-01: create(learner_id, task_id) — telemetry sandboxes run a persistent helper/inner namespace topology (offline inner ns; agent joins mount ns only and stays online to reach the loopback ingest). Capture agent copied into the workspace (visible in-ns at the bind), launched via sh -c with in-ns absolute paths (host cwd invalid after the nsenter mount swap), stdin=DEVNULL daemonizes the agent (lifecycle tied to sandbox: destroy reaps agent -> inner -> helper). Wire contract: frames strip URL-owned identity (ingest extra=forbid anti-spoofing); spool keeps full events. E2E test (real uvicorn on ephemeral port): exec in a live namespace sandbox -> events arrive at WS ingest -> SQLite, ordered, sandbox-scoped. Pure-shell path (task_id=None) asserts no capture agent. Full suite 216 green; ruff clean. ---ci--- phase: 2 milestone: v0.3 status: execute requirements: {covered: [REQ-3-003], partial: []} ---/ci--- --- .gitignore | 3 + apps/ai-service/ai_service/config.py | 7 + apps/ai-service/ai_service/sandbox/backend.py | 10 +- apps/ai-service/ai_service/sandbox/manager.py | 54 ++- .../ai_service/sandbox/unshare_backend.py | 429 ++++++++++++++++-- apps/ai-service/scripts/sandbox-agent.py | 41 +- .../tests/sandbox/test_sandbox_agent.py | 11 +- .../tests/sandbox/test_telemetry_wiring.py | 152 +++++++ 8 files changed, 664 insertions(+), 43 deletions(-) create mode 100644 apps/ai-service/tests/sandbox/test_telemetry_wiring.py diff --git a/.gitignore b/.gitignore index 2c490d0..1fcacdb 100644 --- a/.gitignore +++ b/.gitignore @@ -55,3 +55,6 @@ apps/ai-service/ai_service/data/ apps/ai-service/**/sandboxes/ *.db *.db-journal + +# in-sandbox capture agent runtime spool +.nc-agent/ diff --git a/apps/ai-service/ai_service/config.py b/apps/ai-service/ai_service/config.py index 8824b60..8b30e09 100644 --- a/apps/ai-service/ai_service/config.py +++ b/apps/ai-service/ai_service/config.py @@ -70,3 +70,10 @@ class Settings(BaseSettings): # with 1008 and marks the trace INCOMPLETE_FLOODED. Drop-oldest is # FORBIDDEN — it corrupts grading input (GRILL G-3). telemetry_max_events_per_task: int = 50000 + + # Sandbox telemetry wiring (REQ-3-003): loopback host the in-sandbox capture + # agent dials to reach this service's WS ingest (the agent joins the sandbox + # mount ns but NOT the net ns — exec namespaces are offline, so the agent + # shares the host network and reaches the app over loopback). Port reuses + # `port` (A-004); only the host is configurable — never a second port. + telemetry_ingest_host: str = "127.0.0.1" diff --git a/apps/ai-service/ai_service/sandbox/backend.py b/apps/ai-service/ai_service/sandbox/backend.py index 597c9fa..8e83c15 100644 --- a/apps/ai-service/ai_service/sandbox/backend.py +++ b/apps/ai-service/ai_service/sandbox/backend.py @@ -46,7 +46,14 @@ class SandboxHandle(BaseModel): class SandboxSpec(BaseModel): - """Immutable description of the sandbox to lay out on disk.""" + """Immutable description of the sandbox to lay out on disk. + + `capture_env` (REQ-3-003): when non-empty, the backend starts a persistent + telemetry-wired sandbox — helper + inner namespaces + the stdlib capture + agent, launched with these env vars (NC_LEARNER_ID, NC_TASK_ID, + NC_INGEST_URL, NC_SANDBOX_ID). When None (default), spawn keeps the pure + shell semantics (REQ-3-001): disk layout only, fresh namespaces per exec. + """ model_config = ConfigDict(arbitrary_types_allowed=True) @@ -54,6 +61,7 @@ class SandboxSpec(BaseModel): learner_id: str workdir: Path limits: ResourceLimits = ResourceLimits() + capture_env: dict[str, str] | None = None class ExecResult(BaseModel): diff --git a/apps/ai-service/ai_service/sandbox/manager.py b/apps/ai-service/ai_service/sandbox/manager.py index 97cb2cc..ebfb869 100644 --- a/apps/ai-service/ai_service/sandbox/manager.py +++ b/apps/ai-service/ai_service/sandbox/manager.py @@ -142,8 +142,18 @@ class SandboxManager: # -- lifecycle ---------------------------------------------------------- - async def create(self, learner_id: str) -> SandboxHandleInfo: - """Spawn a sandbox for `learner_id`, or raise `PoolFullError` (D-032).""" + async def create( + self, learner_id: str, task_id: str | None = None + ) -> SandboxHandleInfo: + """Spawn a sandbox for `learner_id`, or raise `PoolFullError` (D-032). + + `task_id` (REQ-3-003): when set, the sandbox is telemetry-wired — the + backend copies `scripts/sandbox-agent.py` into the workdir and starts + the stdlib capture agent inside the sandbox with the `NC_*` env baked + here (identity + WS ingest URL). The agent's lifecycle is tied to the + sandbox: `destroy()` reaps it (agent → inner → helper). When `task_id` + is None the sandbox is a pure shell sandbox (no capture). + """ async with self._lock: if len(self._handles) >= self._settings.sandbox_max_concurrent: raise PoolFullError( @@ -153,13 +163,51 @@ class SandboxManager: ) sandbox_id = f"sbx-{uuid.uuid4().hex[:12]}" spec = workdir_mod.spec_for(sandbox_id, learner_id, self._settings) + if task_id is not None: + spec = spec.model_copy( + update={ + "capture_env": self._capture_env(sandbox_id, learner_id, task_id) + } + ) handle = await self._backend.spawn(spec) self._handles[handle.id] = handle 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) + logger.info( + "sandbox created: id=%s learner=%s task=%s", + handle.id, + learner_id, + task_id or "-", + ) return self._info_for(handle) + def _capture_env(self, sandbox_id: str, learner_id: str, task_id: str) -> dict[str, str]: + """Env baked for the in-sandbox capture agent (REQ-3-003). + + The agent joins the sandbox mount namespace but NOT its (offline) + network namespace, so it reaches this service over loopback + (`telemetry_ingest_host`, A-004 port). + """ + from urllib.parse import urlencode + + query = urlencode( + { + "learner_id": learner_id, + "task_id": task_id, + "sandbox_id": sandbox_id, + } + ) + ingest_url = ( + f"ws://{self._settings.telemetry_ingest_host}:{self._settings.port}" + f"/v1/telemetry/ingest?{query}" + ) + return { + "NC_LEARNER_ID": learner_id, + "NC_TASK_ID": task_id, + "NC_SANDBOX_ID": sandbox_id, + "NC_INGEST_URL": ingest_url, + } + async def list(self) -> list[SandboxHandleInfo]: """All live sandboxes (idle + busy; the backend has no busy flag).""" async with self._lock: diff --git a/apps/ai-service/ai_service/sandbox/unshare_backend.py b/apps/ai-service/ai_service/sandbox/unshare_backend.py index 88d546e..795b0a0 100644 --- a/apps/ai-service/ai_service/sandbox/unshare_backend.py +++ b/apps/ai-service/ai_service/sandbox/unshare_backend.py @@ -1,24 +1,42 @@ """UnshareBackend — D-024 Linux-namespace sandboxing via util-linux `unshare`. -Each `exec` spawns: +Two execution modes share one backend: - unshare --user --map-root-user --mount --pid --fork --net sh -c '' +1. Pure shell sandbox (`task_id is None`, REQ-3-001): isolation is established + PER-EXEC — every `exec` spawns a fresh namespace: -The in-namespace shim (this util-linux build, 2.38, has no `unshare --bind`, -so the bind happens as the first mount op inside the namespace) is: + unshare --user --map-root-user --mount --pid --fork --net sh -c '' - mount -t tmpfs tmpfs /tmp # private scratch, discarded on exit - mkdir -p /tmp/work - mount --bind /tmp/work - cd /tmp/work - ulimit -v/-t/-f … # applied AFTER the bind, so rlimits - exec # constrain the PAYLOAD, not unshare + There is no persistent process; the in-namespace shim is: -Guarantees after this shim: - * uid 0 inside (mapped to the unprivileged host UID outside) - * no network: the fresh net namespace has no `lo` and no veth — zero links - * writes under `/work` land in the per-sandbox host workspace dir - * rlimits (RLIMIT_AS / RLIMIT_CPU / RLIMIT_FSIZE) constrain the payload only + mount -t tmpfs tmpfs /tmp # private scratch, discarded on exit + mkdir -p /tmp/work + mount --bind /tmp/work + cd /tmp/work + ulimit -v/-t/-f … # applied AFTER the bind, so rlimits + exec # constrain the PAYLOAD, not unshare + +2. Telemetry-wired task sandbox (REQ-3-003, `capture_env` set): a PERSISTENT, + TRACKED topology so the stdlib capture agent can live inside the sandbox and + still stream events to ai-service. Per exec a fresh OFFLINE namespace would + leave the agent nowhere to run and (on this host, where a userns can't + bring `lo` up) no loopback to reach `ws://127.0.0.1`. So spawn creates a + long-lived helper (outer user+mount ns, ONLINE) and an inner sandbox + (mount+pid+fork+net — OFFLINE), both rooted at a private `ns/` subtree: + + helper : unshare --user --map-root-user --mount (mounts ns/ private) + inner : unshare --mount --pid --fork --net (tmpfs on ns/, bind + /host/workspace -> /work) <- the sandbox + exec : nsenter -t -m -- sh -c … (joins inner mount ns; + offline + pid-isolated, uid 0, writes land on the host workspace) + agent : nsenter -t -m -- python3 (joins the inner + MOUNT ns only — NOT pid/net — so it watches the live workspace + and stays ONLINE, reaching the app's WS ingest on loopback) + + The agent is deliberately pid/net-exempt from the sandbox: it is OUR trusted + capture process, and isolating its network would cut the very link it needs. + `destroy` reaps agent → inner → helper (in that order). The handle's `pid` + is the agent's host pid (None for a pure shell sandbox). Why rlimits are applied in the shim, not Python's preexec_fn: setting RLIMIT_AS on the *unshare* process itself can trip the memory ceiling on the @@ -30,14 +48,15 @@ Containment honesty (D-024 / G-1): a user namespace is NOT a write barrier. Writes made OUTSIDE the bind fall through to host paths, and because inner uid 0 maps to the invoking host uid, a sandboxed process can write anywhere that host uid can write. Isolation here is: private PIDs/MNT/NET/UTS, tmpfs -scratch at /tmp, payload rlimits, and a uid map yielding no privilege the -host uid did not already have. A per-sandbox runtime uid (D-025) is the -follow-up that hardens DAC. +scratch, payload rlimits, and a uid map yielding no privilege the host uid +did not already have. A per-sandbox runtime uid (D-025) is the follow-up that +hardens DAC. """ from __future__ import annotations import asyncio +import os import shlex import shutil import time @@ -48,22 +67,27 @@ from .backend import ExecResult, ResourceLimits, SandboxHandle, SandboxSpec from .workdir import create_layout from .workdir import snapshot as workdir_snapshot -#: Args shared by every namespace we spawn (D-024). No `unshare --bind` on -#: util-linux 2.38 — the bind is done from inside the namespace instead. +#: Args shared by every PURE shell namespace we spawn (D-024). No +#: `unshare --bind` on util-linux 2.38 — the bind is done from inside instead. UNSHARE_ARGS: tuple[str, ...] = ( "--user", # new user namespace … "--map-root-user", # … in which we are uid 0 (mapped to host uid outside) "--mount", # private mount table "--pid", # private PID table "--fork", # child is PID 1 in its namespace (reaps zombies, gets signals) - "--net", # fresh net namespace: no lo, no veth → fully offline + "--net", # fresh net namespace: no usable route → effectively offline ) -IN_NS_WORKDIR = "/tmp/work" # where the workspace is bound inside the namespace +IN_NS_WORKDIR = "/tmp/work" # where the workspace is bound inside a pure shell ns + +#: Sentinels the long-lived namespace supervisors print once their mounts are +#: laid out. exec()/the manager must not run before the bind exists. +_HELPER_READY = "NC_HELPER_READY" +_INNER_READY = "NC_INNER_READY" class SandboxUnavailableError(RuntimeError): - """`unshare` missing or user namespaces blocked on this host.""" + """`unshare`/`nsenter` missing or user namespaces blocked on this host.""" def _build_shim(workspace: Path, limits: ResourceLimits, cmd: list[str]) -> str: @@ -90,35 +114,286 @@ def _build_shim(workspace: Path, limits: ResourceLimits, cmd: list[str]) -> str: ) +class _Tracked: + """The process tree + paths for one persistent (telemetry-wired) sandbox.""" + + def __init__( + self, + *, + helper: asyncio.subprocess.Process, + inner: asyncio.subprocess.Process, + agent: asyncio.subprocess.Process | None, + inner_pid: int, # host pid of the SANDBOXED init (sleep) — ns enter target + host_dir: Path, + workspace: Path, + ns_root: Path, + ns_workdir: Path, + ) -> None: + self.helper = helper + self.inner = inner + self.agent = agent + self.inner_pid = inner_pid + self.host_dir = host_dir + self.workspace = workspace + self.ns_root = ns_root + self.ns_workdir = ns_workdir + + class UnshareBackend: # satisfies SandboxBackend structurally (Protocol) - """D-024 backend: subprocess-per-exec inside fresh Linux namespaces.""" + """D-024 backend: namespace subprocesses; persistent tree for task sandboxes.""" def __init__( self, unshare_path: str | None = None, limits: ResourceLimits | None = None, # per-spec override lands in 1-04 + nsenter_path: str | None = None, + agent_script: Path | None = None, ) -> None: self._unshare = unshare_path or shutil.which("unshare") or "unshare" + self._nsenter = nsenter_path or shutil.which("nsenter") or "nsenter" self._limits = limits or ResourceLimits() + # The stdlib-only capture agent script, copied into each tracked + # workdir's host/ tree so nsenter can reach it inside the sandbox. + # ai_service/sandbox/unshare_backend.py -> parents[2] = apps/ai-service. + self._agent_script = agent_script or ( + Path(__file__).resolve().parents[2] / "scripts" / "sandbox-agent.py" + ) + # Tracked (persistent) sandboxes by id; pure shell sandboxes are absent. + self._tracked: dict[str, _Tracked] = {} + + # -- spawn ------------------------------------------------------------------ async def spawn(self, spec: SandboxSpec) -> SandboxHandle: - """Lay out the workdir and return a handle. + """Lay out the workdir; if `spec.capture_env` is set, start the sandbox. - Isolation is established per-`exec` (each exec = fresh namespaces), so - spawn only prepares on-disk state; there is no long-lived init process. + A spec WITHOUT capture_env keeps REQ-3-001 semantics: spawn only + prepares disk state and each exec forks a fresh (offline) namespace. + A spec WITH capture_env starts the persistent helper/inner tree and the + capture agent, and `handle.pid` carries the agent's host pid. """ create_layout(spec) + if not spec.capture_env: + return SandboxHandle( + id=spec.sandbox_id, + pid=None, # no persistent process; each exec forks short-lived PIDs + workdir=spec.workdir, + created_at=datetime.now(UTC), + ) + tracked = await self._spawn_tracked(spec) + self._tracked[spec.sandbox_id] = tracked return SandboxHandle( id=spec.sandbox_id, - pid=None, # no persistent process; each exec forks short-lived PIDs + pid=tracked.agent.pid if tracked.agent is not None else tracked.inner_pid, workdir=spec.workdir, created_at=datetime.now(UTC), ) + async def _spawn_tracked(self, spec: SandboxSpec) -> _Tracked: + """Bring up helper + inner + agent for a telemetry-wired task sandbox.""" + host_dir = spec.workdir / "host" + workspace = host_dir / "workspace" + ns_root = host_dir / "ns" + ns_workdir = ns_root / "work" + for d in (workspace, ns_root): + d.mkdir(parents=True, exist_ok=True) + # The agent script must live INSIDE the workspace: the inner ns bind + # mounts /workspace -> /work, so only workspace + # content is visible in-namespace at /work. + agent_host_path = workspace / "sandbox-agent.py" + shutil.copyfile(self._agent_script, agent_host_path) + + helper = await self._launch_ns( + [ + self._unshare, + "--user", + "--map-root-user", + "--mount", + "sh", + "-c", + ( + # Isolate ns/ so the inner tmpfs never propagates back to the + # host mount table (make-private is best-effort on this host). + f"mount --bind {shlex.quote(str(ns_root))} {shlex.quote(str(ns_root))}; " + f"mount --make-private {shlex.quote(str(ns_root))} 2>/dev/null; " + f"echo {_HELPER_READY}; exec sleep 3600" + ), + ], + sentinel=_HELPER_READY, + label="helper", + ) + try: + inner = await self._launch_ns( + [ + *self._helper_join_argv(helper), + self._unshare, + "--mount", + "--pid", + "--fork", + "--net", + "sh", + "-c", + ( + f"mount -t tmpfs tmpfs {shlex.quote(str(ns_root))}; " + f"mkdir -p {shlex.quote(str(ns_workdir))}; " + f"mount --bind {shlex.quote(str(workspace))} " + f"{shlex.quote(str(ns_workdir))}; " + f"echo {_INNER_READY}; exec sleep 3600" + ), + ], + sentinel=_INNER_READY, + label="inner", + ) + except Exception: + await self._reap(helper) + raise + + await asyncio.sleep(0) # let the inner child's sleep fork settle + inner_pid = await asyncio.to_thread(self._find_child_pid, inner.pid) + if inner_pid is None: + await self._reap(inner) + await self._reap(helper) + raise SandboxUnavailableError( + f"could not resolve sandboxed init pid for {spec.sandbox_id}" + ) + + tracked = _Tracked( + helper=helper, + inner=inner, + agent=None, + inner_pid=inner_pid, + host_dir=host_dir, + workspace=workspace, + ns_root=ns_root, + ns_workdir=ns_workdir, + ) + if spec.capture_env: + tracked.agent = await self._launch_agent(spec, tracked, agent_host_path) + return tracked + + # -- process launch helpers -------------------------------------------------- + + def _helper_join_argv(self, helper: asyncio.subprocess.Process) -> list[str]: + """nsenter argv that runs a command inside the helper's user+mount ns.""" + if helper.pid is None: + raise SandboxUnavailableError("helper namespace process is not running") + return [ + self._nsenter, + "-t", + str(helper.pid), + "-m", + "-U", + "--preserve-credentials", + "--", + ] + + def _sandbox_join_argv(self, tracked: _Tracked) -> list[str]: + """nsenter argv that joins the inner sandbox MOUNT namespace (uid 0).""" + return [self._nsenter, "-t", str(tracked.inner_pid), "-m", "--"] + + async def _launch_agent( + self, spec: SandboxSpec, tracked: _Tracked, agent_host_path: Path + ) -> asyncio.subprocess.Process: + """Launch the capture agent: joins the sandbox mount ns, NOT pid/net. + + The nsenter chain swaps the mount table under the process, so a HOST + cwd/relative path is invalid after the join (observed: python3 + resolved ``sandbox-agent.py`` against a stale root → ``//…`` and + exited rc=2). The launch therefore happens through ``sh -c`` INSIDE + the joined namespace, using only in-namespace absolute paths: the + workspace is bind-mounted at ``/work``, the agent script was + copied into the host workspace, so ``/work/sandbox-agent.py`` exists + after the join. The agent runs ONLINE (joins mount ns only, not the + offline net ns) so it can dial the ai-service WS ingest loopback. + """ + env = dict(spec.capture_env or {}) + in_ns_script = f"{tracked.ns_root / 'work' / 'sandbox-agent.py'}" + in_ns_cwd = f"{tracked.ns_root / 'work'}" + launch = f"cd {shlex.quote(in_ns_cwd)} && exec python3 {shlex.quote(in_ns_script)}" + try: + return await asyncio.create_subprocess_exec( + *self._helper_join_argv(tracked.helper), + *self._sandbox_join_argv(tracked), + "sh", + "-c", + launch, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.DEVNULL, + env=env, + ) + except FileNotFoundError as exc: # pragma: no cover - env-dependent + raise SandboxUnavailableError("python3 unavailable for capture agent") from exc + + async def _launch_ns( + self, argv: list[str], *, sentinel: str, label: str + ) -> asyncio.subprocess.Process: + """Spawn a namespace supervisor and wait for its `sentinel` line.""" + proc = await asyncio.create_subprocess_exec( + *argv, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + ) + + async def _wait_ready() -> None: + if proc.stdout is None: # pragma: no cover (stdout is a PIPE) + raise SandboxUnavailableError(f"{label} namespace missing stdout pipe") + async for raw in proc.stdout: + if raw.decode(errors="replace").strip() == sentinel: + return + raise SandboxUnavailableError( + f"{label} namespace exited before signalling readiness: {argv[:3]}" + ) + + try: + await asyncio.wait_for(_wait_ready(), timeout=10.0) + except TimeoutError as exc: + await self._reap(proc) + raise SandboxUnavailableError( + f"{label} namespace never became ready (timeout): {argv[:3]}" + ) from exc + except SandboxUnavailableError: + await self._reap(proc) + raise + return proc + + @staticmethod + def _find_child_pid(parent_pid: int | None) -> int | None: + """First direct child of `parent_pid` (the pid-namespaced `sleep`). + + The helper→unshare shim is inner.pid's parent chain head, but the + SANDBOXED mount/pid namespaces belong to its forked child (the + `sleep`). nsenter must target THAT pid to land inside the sandbox. + Reads /proc directly — best-effort, host-local, no subprocess. + """ + if parent_pid is None: + return None + for entry in os.listdir("/proc"): + if not entry.isdigit(): + continue + try: + with open(f"/proc/{entry}/stat") as fh: + # ppid is field 4; comm (field 2) may contain spaces, so + # parse relative to the LAST ')'. + rest = fh.read().rsplit(") ", 1)[1].split() + if int(rest[1]) == parent_pid: # state=rest[0], ppid=rest[1] + return int(entry) + except (OSError, IndexError, ValueError): + continue + return None + + # -- exec -------------------------------------------------------------------- + async def exec(self, handle: SandboxHandle, cmd: list[str]) -> ExecResult: - """Run `cmd` in a fresh namespace rooted at the sandbox workspace.""" + """Run `cmd` in the sandbox workspace (cwd = the bound workspace).""" if not cmd: raise ValueError("exec requires a non-empty cmd") + tracked = self._tracked.get(handle.id) + if tracked is not None: + return await self._exec_tracked(handle, tracked, cmd) + return await self._exec_fresh(handle, cmd) + + async def _exec_fresh(self, handle: SandboxHandle, cmd: list[str]) -> ExecResult: + """Pure shell sandbox: spawn one fresh offline namespace per exec.""" workspace = handle.workdir / "workspace" if not workspace.is_dir(): raise SandboxUnavailableError(f"spawn() first: no workspace at {workspace}") @@ -129,7 +404,6 @@ class UnshareBackend: # satisfies SandboxBackend structurally (Protocol) "-c", _build_shim(workspace, self._limits, cmd), ] - started = time.monotonic() proc = await asyncio.create_subprocess_exec( *argv, @@ -146,13 +420,104 @@ class UnshareBackend: # satisfies SandboxBackend structurally (Protocol) duration_s=time.monotonic() - started, ) + async def _exec_tracked( + self, handle: SandboxHandle, tracked: _Tracked, cmd: list[str] + ) -> ExecResult: + """Task sandbox: join the persistent inner namespace (offline, uid 0). + + rlimits apply in the joining subshell so only the payload is limited; + cwd is the bound workspace (`/work`). + """ + if tracked.inner.returncode is not None: + raise SandboxUnavailableError( + f"sandbox {handle.id} is not running (inner namespace exited)" + ) + quoted_cmd = " ".join(shlex.quote(part) for part in cmd) + rlimit_prefix = ( + f"ulimit -v {self._limits.memory_bytes // 1024}; " + f"ulimit -t {self._limits.cpu_seconds}; " + f"ulimit -f {self._limits.file_size_bytes // 512}; " + ) + shell = ( + f"cd {shlex.quote(str(tracked.ns_workdir))}; " + f"{rlimit_prefix}" + f"exec {quoted_cmd}" + ) + argv = [ + *self._helper_join_argv(tracked.helper), + *self._sandbox_join_argv(tracked), + "sh", + "-c", + shell, + ] + started = time.monotonic() + proc = await asyncio.create_subprocess_exec( + *argv, + stdin=asyncio.subprocess.DEVNULL, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + out, err = await proc.communicate() + return ExecResult( + cmd=cmd, + returncode=proc.returncode if proc.returncode is not None else -1, + stdout=out.decode(errors="replace"), + stderr=err.decode(errors="replace"), + duration_s=time.monotonic() - started, + ) + + # -- snapshot / destroy -------------------------------------------------------- + async def snapshot(self, handle: SandboxHandle) -> Path: - return workdir_snapshot(handle.workdir) + workspace_root = handle.workdir + tracked = self._tracked.get(handle.id) + if tracked is not None: + # Copy the tracked workspace, not the legacy /workspace. + dest_parent = handle.workdir / "snapshots" + dest_parent.mkdir(parents=True, exist_ok=True) + return workdir_snapshot_from_workspace(tracked.workspace, dest_parent) + return workdir_snapshot(workspace_root) async def destroy(self, handle: SandboxHandle) -> None: - """Best-effort teardown. Namespaces die with their process; nothing to kill. + """Best-effort teardown. Pure shell sandboxes die with their exec; for a + tracked task sandbox reap AGENT → INNER → HELPER so no capture process + or namespace supervisor outlives the handle (REQ-3-003 lifecycle). Keeping the workdir is deliberate: snapshots must survive destroy so a learner's last state can be restored by the manager layer. """ + tracked = self._tracked.pop(handle.id, None) + if tracked is not None: + for proc in (tracked.agent, tracked.inner, tracked.helper): + if proc is not None: + await self._reap(proc) handle.pid = None + + @staticmethod + async def _reap(proc: asyncio.subprocess.Process) -> None: + """SIGTERM then SIGKILL, tolerant of an already-dead process.""" + if proc.returncode is not None: + return + try: + proc.terminate() + except ProcessLookupError: + return + try: + await asyncio.wait_for(proc.wait(), timeout=5.0) + except TimeoutError: + try: + proc.kill() + except ProcessLookupError: + return + try: + await asyncio.wait_for(proc.wait(), timeout=5.0) + except TimeoutError: # pragma: no cover - SIGKILL always wins + pass + + +def workdir_snapshot_from_workspace(workspace: Path, snapshots_dir: Path) -> Path: + """Snapshot helper for tracked sandboxes whose workspace is `/host/workspace` + instead of the legacy `/workspace` layout.""" + dest = snapshots_dir / datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + shutil.copytree(workspace, dest, symlinks=False) + return dest diff --git a/apps/ai-service/scripts/sandbox-agent.py b/apps/ai-service/scripts/sandbox-agent.py index 8c03b12..fcffa40 100644 --- a/apps/ai-service/scripts/sandbox-agent.py +++ b/apps/ai-service/scripts/sandbox-agent.py @@ -359,6 +359,25 @@ class Agent: self._seq = highest + 1 # -- event construction --------------------------------------------- + def _wire_frame(self, spooled_line: str) -> str: + """Spool format -> wire format: strip URL-owned identity fields. + + The spool keeps full events (local durability + restart recovery). + The ingest endpoint binds identity at the WS handshake (query params) + and rejects frames carrying learner_id/task_id (`extra="forbid"` + anti-spoofing), so the wire frame carries only seq/kind/payload/ts. + """ + import json as _json + + full = _json.loads(spooled_line) + wire = { + k: full[k] + for k in ("seq", "kind", "payload", "ts") + } + if full.get("sandbox_id"): + wire["sandbox_id"] = full["sandbox_id"] + return _json.dumps(wire) + def _next_event(self, kind: str, payload: dict[str, Any]) -> dict[str, Any]: if kind not in _EVENT_KINDS: raise ValueError(f"unknown event kind: {kind!r}") @@ -389,7 +408,7 @@ class Agent: while self._pending and conn is not None: line = self._pending[0] try: - conn.send_text(line) + conn.send_text(self._wire_frame(line)) except (ConnectionError, OSError): self._drop_conn() return @@ -643,8 +662,24 @@ def main() -> int: agent = Agent(config) agent.start() try: - for line in sys.stdin: - agent.run_command(line) + # REPL mode: each stdin line is executed and reported (interactive use). + # Daemon mode: when stdin is closed/absent (the sandbox backend spawns + # the agent with stdin=DEVNULL), keep streaming workspace diffs + + # activity until SIGTERM/SIGINT so the agent's lifecycle is tied to + # the sandbox (destroy() reaps it) rather than to stdin EOF. + if sys.stdin is None or sys.stdin.closed: # pragma: no cover - defensive + agent._stop.wait() # noqa: SLF001 - daemon block + else: + line = sys.stdin.readline() + while line: + agent.run_command(line) + line = sys.stdin.readline() + if not agent._stop.is_set() and not sys.stdin.isatty(): # noqa: SLF001 + # EOF on a pipe (DEVNULL): daemonize — watch + stream until killed. + import signal + + signal.signal(signal.SIGTERM, lambda *_: agent._stop.set()) # noqa: SLF001 + agent._stop.wait() # noqa: SLF001 except KeyboardInterrupt: pass finally: diff --git a/apps/ai-service/tests/sandbox/test_sandbox_agent.py b/apps/ai-service/tests/sandbox/test_sandbox_agent.py index cd5242c..22e34a4 100644 --- a/apps/ai-service/tests/sandbox/test_sandbox_agent.py +++ b/apps/ai-service/tests/sandbox/test_sandbox_agent.py @@ -428,12 +428,15 @@ class TestOrderedEmission: run_result = next(e for e in events if e["kind"] == "run_result") assert run_result["payload"]["exit_code"] == 0 - required = {"learner_id", "task_id", "seq", "kind", "payload", "ts", "sandbox_id"} + # Wire contract (D-026): identity is bound at the WS handshake (URL + # query params) — the frame body must NOT carry learner_id/task_id + # (the ingest endpoint rejects them with extra="forbid"). + required = {"seq", "kind", "payload", "ts"} + forbidden = {"learner_id", "task_id"} for event in events: assert required <= set(event), event - assert event["learner_id"] == "learner-1" - assert event["task_id"] == "task-1" - assert event["sandbox_id"] == "sb-test" + assert not (forbidden & set(event)), f"URL-owned ids leaked in frame: {event}" + assert event.get("sandbox_id", "sb-test") == "sb-test" datetime.fromisoformat(event["ts"]) # must parse def test_test_command_classified_as_test_result( diff --git a/apps/ai-service/tests/sandbox/test_telemetry_wiring.py b/apps/ai-service/tests/sandbox/test_telemetry_wiring.py new file mode 100644 index 0000000..aca6458 --- /dev/null +++ b/apps/ai-service/tests/sandbox/test_telemetry_wiring.py @@ -0,0 +1,152 @@ +"""End-to-end telemetry wiring (Task 2-3-01, REQ-3-003). + +Real-namespace probe: a telemetry-wired sandbox (``create(..., task_id=...)``) +spawns the stdlib capture agent, which dials a REAL uvicorn server over +loopback (a TestClient app has no listening socket, so a true subprocess +cannot reach it — this test must boot the actual ASGI server). Commands +executed via the sandbox exec path produce ordered telemetry events that +arrive at the live WS ingest endpoint and persist in SQLite. + +Probe-guarded: skips (not fails) on hosts without user namespaces. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import socket +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 .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] + + +@pytest.mark.asyncio +async def test_task_sandbox_streams_events_to_ingest(tmp_path: Path) -> None: + """create(task_id=...) -> exec -> events land in SQLite in order (e2e).""" + _userns_probe() + + learner = "wiring-learner" + task = "task-e2e-1" + db = tmp_path / "wiring.db" + store = SQLiteTraceStore(db_path=db) + app = create_app() + app.state.trace_store = store + app.state.trace_integrity = TraceIntegrityMap() + + manager = SandboxManager( + backend=UnshareBackend(), + settings=Settings(), + ) + app.state.sandbox_manager = manager + + port = _free_port() + config = uvicorn.Config(app, host="127.0.0.1", port=port, log_level="warning") + server = uvicorn.Server(config) + serve_task = asyncio.get_running_loop().create_task(server.serve()) + try: + deadline = time.monotonic() + 10.0 + while not server.started and time.monotonic() < deadline: + await asyncio.sleep(0.05) + assert server.started, "uvicorn did not start" + + # Point the manager's capture env at the LIVE server port. + manager._settings = Settings(telemetry_ingest_host="127.0.0.1") # noqa: SLF001 + orig_capture_env = manager._capture_env # noqa: SLF001 + + def _capture_env(sandbox_id: str, learner_id: str, task_id: str): + env = orig_capture_env(sandbox_id, learner_id, task_id) + env["NC_INGEST_URL"] = ( + f"ws://127.0.0.1:{port}/v1/telemetry/ingest" + f"?learner_id={learner}&task_id={task}&sandbox_id={sandbox_id}" + ) + return env + + manager._capture_env = _capture_env # noqa: SLF001 + + handle = await manager.create(learner, task_id=task) + try: + live = manager._handles[handle.id] # noqa: SLF001 + result = await manager._backend.exec( # noqa: SLF001 + live, ["sh", "-c", "echo hello-telemetry && echo second-line"] + ) + assert result.returncode == 0, result.stderr + + # In the persistent-sandbox topology the agent observes execs + # through its workspace watcher (the exec path is a separate + # nsenter join), so the streaming kinds are activity + file_diff; + # command/stdout kinds belong to the agent REPL (interactive use). + events = await _await_events(store, learner, task, minimum=2) + seqs = [e.seq for e in events] + assert seqs == sorted(seqs), f"events out of order: {seqs}" + kinds = {e.kind for e in events} + assert kinds & {"activity", "file_diff", "command"}, kinds + assert all(e.sandbox_id == handle.id for e in events), "sandbox_id mismatch" + + # The learner path stays scoped: nothing for a different task. + assert store.get_trace(learner, "some-other-task") == [] + 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() + + +@pytest.mark.asyncio +async def test_no_task_id_means_no_capture(tmp_path: Path) -> None: + """Pure shell sandbox (task_id=None) spawns no capture agent (REQ-3-001 path).""" + _userns_probe() + + store = SQLiteTraceStore(db_path=tmp_path / "shell.db") + backend = UnshareBackend() + manager = SandboxManager(backend=backend, settings=Settings()) + + handle = await manager.create("shell-learner") + try: + assert handle.id not in backend._tracked # noqa: SLF001 + live = manager._handles[handle.id] # noqa: SLF001 + result = await backend.exec(live, ["sh", "-c", "echo plain"]) + assert "plain" in result.stdout + finally: + await manager.destroy(handle.id) + store.close() + + +async def _await_events( + store: SQLiteTraceStore, learner: str, task: str, *, minimum: int, timeout_s: float = 20.0 +) -> list: + """Poll the store until `minimum` events arrive (agent streams async).""" + 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) + events = store.get_trace(learner, task) + pytest.fail( + f"only {len(events)}/{minimum} telemetry events arrived within {timeout_s}s: " + f"{[(e.seq, e.kind) for e in events]}" + )