codex: contain and reap workflow-run sandbox descendants (opt-in cgroup)

docs/config.md

@@ -1045,8 +1045,24 @@ codex:                     # active when a codex backend is selected
     max_concurrency: 2          # hard cap, even if a workflow asks for more
     default_token_budget: 250000 # default and maximum per workflow
     max_agents: 8               # lifetime worker cap per workflow
+  lifecycle:
+    mode: disabled              # disabled | strict (workflow-run cgroup containment)
 ```
 
+`codex.lifecycle.mode: strict` launches each Codex workflow run's app-server
+inside a transient systemd user scope (`systemd-run --user --scope`). Sandboxed
+commands that `setsid` or double-fork out of the app-server's process group stay
+in that cgroup. Closing the run's client stops the scope, which kills them as
+well. Interactive and cron Codex sessions are not affected.
+
+`strict` needs Linux, cgroup v2, and a systemd user manager that the daemon can
+reach (`systemctl --user show-environment` succeeds as the daemon's user).
+Docker containers and the system-level unit in [setup](setup.md) usually have
+no user manager. There `strict` fails every Codex workflow run at launch rather
+than run it uncontained, so leave it `disabled` on those hosts. Each contained
+run writes `lifecycle.json` in its run directory. On startup, Nerve stops any
+scope still recorded there in either mode, which cleans up after a crash.
+
 `codex.ultracode.dashboard` exposes run journals in Nerve's authenticated UI.
 It does not start Ultracode's detached dashboard process. Keep
 `codex.ultracode.ui: false`: the upstream process serves unauthenticated

nerve/agent/backends/codex/appserver.py

@@ -28,6 +28,7 @@
 from typing import Any, Awaitable, Callable
 
 from nerve.agent.backends.base import TransportDiedError
+from nerve.agent.backends.codex import lifecycle
 
 logger = logging.getLogger(__name__)
 
@@ -139,6 +140,7 @@ def __init__(
         client_name: str = "nerve",
         client_version: str = "1.0.0",
         request_timeout: float = 60.0,
+        containment: lifecycle.WorkflowContainment | None = None,
     ) -> None:
         self._bin_path = bin_path
         self._cwd = cwd
@@ -148,6 +150,9 @@ def __init__(
         self._client_version = client_version
         self._request_timeout = request_timeout
         self._on_server_request = server_request_handler
+        # Set only for contained Codex workflow runs (codex/lifecycle.py).
+        self._containment = containment
+        self._lifecycle_record: lifecycle.LifecycleRecord | None = None
 
         self._proc: asyncio.subprocess.Process | None = None
         self._reader_task: asyncio.Task | None = None
@@ -178,6 +183,20 @@ async def start(self) -> dict:
             args.extend(["--config", kv])
         args.extend(["app-server", "--listen", "stdio://"])
 
+        # Wrap the argv in a systemd scope; raises before exec on failure.
+        containment = self._containment
+        if containment is not None:
+            try:
+                args, self._lifecycle_record = await asyncio.to_thread(
+                    lifecycle.prepare_launch, containment, args,
+                )
+            except lifecycle.ContainmentUnavailable as e:
+                raise TransportDiedError(str(e)) from e
+            except (lifecycle.LifecycleError, OSError) as e:
+                raise TransportDiedError(
+                    f"codex workflow containment could not be persisted: {e}"
+                ) from e
+
         try:
             self._proc = await asyncio.create_subprocess_exec(
                 *args,
@@ -194,6 +213,19 @@ async def start(self) -> dict:
                 f"Failed to spawn codex app-server ({self._bin_path}): {e}"
             ) from e
 
+        # Record the scope's InvocationID + cgroup.
+        if containment is not None and self._lifecycle_record is not None:
+            try:
+                self._lifecycle_record = await asyncio.to_thread(
+                    lifecycle.record_launched,
+                    containment.run_dir, self._lifecycle_record,
+                )
+            except (lifecycle.LifecycleError, OSError) as e:
+                await self.close()
+                raise TransportDiedError(
+                    f"codex workflow scope did not register: {e}"
+                ) from e
+
         self._reader_task = asyncio.create_task(
             self._reader_loop(), name=f"codex-reader:{self._proc.pid}",
         )
@@ -215,11 +247,27 @@ async def start(self) -> dict:
                 },
             })
             await self.notify("initialized", None)
-            return response
         except BaseException:
             await self.close()
             raise
 
+        # Verify the app-server is inside its scope cgroup before returning.
+        if (
+            containment is not None
+            and self._lifecycle_record is not None
+            and self._proc is not None and self._proc.pid is not None
+        ):
+            member = await asyncio.to_thread(
+                lifecycle.verify_membership,
+                self._lifecycle_record.control_group, self._proc.pid,
+            )
+            if not member:
+                await self.close()
+                raise TransportDiedError(
+                    "codex app-server is not contained in its cgroup scope"
+                )
+        return response
+
     def is_alive(self) -> bool:
         return (
             not self._closed
@@ -228,7 +276,8 @@ def is_alive(self) -> bool:
         )
 
     async def close(self) -> None:
-        """Terminate the subprocess and fail everything pending."""
+        """Terminate the subprocess, fail everything pending, then stop the
+        run's scope if the launch was contained."""
         self._closed = True
         tasks = [
             task for task in (self._reader_task, self._stderr_task)
@@ -261,6 +310,22 @@ async def close(self) -> None:
         self._stderr_task = None
         self._server_request_tasks.clear()
 
+        # Stop the scope too: it holds descendants outside the process group.
+        if self._containment is not None and self._lifecycle_record is not None:
+            try:
+                receipt = await asyncio.to_thread(
+                    lifecycle.reap, self._containment.run_dir,
+                    unit=self._lifecycle_record.unit,
+                )
+            except Exception:  # noqa: BLE001
+                logger.exception("codex scope reap failed on close")
+            else:
+                if receipt.outcome not in ("complete", "no_scope"):
+                    logger.warning(
+                        "codex scope %s not reaped on close (%s: %s)",
+                        self._lifecycle_record.unit, receipt.outcome, receipt.error,
+                    )
+
     @staticmethod
     def _signal_process_tree(
         proc: asyncio.subprocess.Process, sig: signal.Signals,

nerve/agent/backends/codex/backend.py

@@ -44,6 +44,7 @@
     TurnInput,
     config_excluded_tools,
 )
+from nerve.agent.backends.codex import lifecycle
 from nerve.agent.backends.codex.appserver import (
     CodexAppServerClient,
     CodexRpcError,
@@ -665,6 +666,30 @@ def resolve_api_key(self) -> str | None:
         return None
 
 
+_WORKFLOW_SESSION_PREFIX = "workflow:"
+
+
+def _workflow_containment(
+    backend: "CodexBackend", spec: SessionSpec,
+) -> lifecycle.WorkflowContainment | None:
+    """Containment for a ``workflow:`` session in strict mode. None for other
+    sessions, disabled mode, or a run id that isn't a run dir under
+    ``runs_dir``."""
+    sid = spec.session_id or ""
+    if not sid.startswith(_WORKFLOW_SESSION_PREFIX):
+        return None
+    if backend.codex.lifecycle.mode != lifecycle.MODE_STRICT:
+        return None
+    run_id = sid[len(_WORKFLOW_SESSION_PREFIX):]
+    if not lifecycle._RUN_ID_RE.match(run_id):
+        return None
+    runs_dir = Path(backend.config.workflows.runs_dir).expanduser().resolve()
+    run_dir = (runs_dir / run_id).resolve()
+    if run_dir.parent != runs_dir:
+        return None
+    return lifecycle.WorkflowContainment(run_dir=run_dir, run_id=run_id)
+
+
 class CodexClient(AgentClient):
     """One live ``codex app-server`` subprocess for one nerve session."""
 
@@ -708,6 +733,7 @@ def __init__(self, backend: CodexBackend, spec: SessionSpec):
             env=env,
             server_request_handler=self._handle_server_request,
             config_overrides=config_overrides,
+            containment=_workflow_containment(backend, spec),
         )
         self._thread_id: str | None = None
         self._turn_id: str | None = None