diff --git a/docs/config.md b/docs/config.md
index 59e2d5ebb..e45fa3d0a 100644
--- a/docs/config.md
+++ b/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
diff --git a/nerve/agent/backends/codex/appserver.py b/nerve/agent/backends/codex/appserver.py
index 5291db730..c7ca57ebf 100644
--- a/nerve/agent/backends/codex/appserver.py
+++ b/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,
diff --git a/nerve/agent/backends/codex/backend.py b/nerve/agent/backends/codex/backend.py
index e131f62a2..892396b15 100644
--- a/nerve/agent/backends/codex/backend.py
+++ b/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
diff --git a/nerve/agent/backends/codex/lifecycle.py b/nerve/agent/backends/codex/lifecycle.py
new file mode 100644
index 000000000..8b12f4517
--- /dev/null
+++ b/nerve/agent/backends/codex/lifecycle.py
@@ -0,0 +1,443 @@
+"""Cgroup containment + reaping for Codex workflow-run descendants.
+
+In ``strict`` mode a workflow run's ``codex app-server`` is launched inside a
+delegated systemd user scope, so its whole descendant tree — including
+``codex-linux-sandbox`` children that ``setsid`` into their own process groups —
+stays in one cgroup and a single ``systemctl --user stop`` reaps it. A durable
+per-run record lets the reap run at terminal cleanup or from startup
+reconciliation after a daemon crash. Requires Linux + a reachable systemd
+``--user`` manager + cgroup v2; strict fails before exec where that is absent.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+import shutil
+import subprocess
+import time
+import uuid
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+from nerve.utils.fs import atomic_write_text
+
+logger = logging.getLogger(__name__)
+
+MODE_STRICT = "strict"
+
+RECORD_NAME = "lifecycle.json"
+SCOPE_DESCRIPTION = "Nerve Codex workflow run containment"
+
+# systemd-run sends TERM, then SIGKILL after this grace, over the whole cgroup.
+TERM_GRACE_SECONDS = 5
+STOP_TIMEOUT_SECONDS = 20
+EMPTY_WAIT_SECONDS = 3
+
+_UNIT_RE = re.compile(r"^nerve-wf-[A-Za-z0-9:_.-]+\.scope$")
+_RUN_ID_RE = re.compile(r"^wfr-[A-Za-z0-9_]+$")
+_CGROUP_ROOT = "/sys/fs/cgroup"
+_USER_SLICE_PREFIX = "/user.slice/"
+
+# `complete` and `refused` are final; `pending_retry` is re-attempted by the
+# next startup reconciliation.
+_TERMINAL_OUTCOMES = frozenset({"complete", "refused"})
+
+
+class LifecycleError(Exception):
+    """A containment/reap operation failed."""
+
+
+class ContainmentUnavailable(LifecycleError):
+    """Strict mode requested but the host cannot create a scope (raised before exec)."""
+
+
+# -- capabilities ------------------------------------------------------------ #
+
+
+@dataclass(frozen=True)
+class Capabilities:
+    ok: bool
+    reason: str
+
+
+_caps_cache: Capabilities | None = None
+
+
+def detect_capabilities(*, refresh: bool = False) -> Capabilities:
+    """Whether a delegated user scope can be created here. Cached per process."""
+    global _caps_cache
+    if _caps_cache is not None and not refresh:
+        return _caps_cache
+    reasons = []
+    if shutil.which("systemd-run") is None:
+        reasons.append("systemd-run not found")
+    if shutil.which("systemctl") is None:
+        reasons.append("systemctl not found")
+    if not Path(_CGROUP_ROOT, "cgroup.controllers").exists():
+        reasons.append("cgroup v2 not mounted")
+    else:
+        try:
+            proc = subprocess.run(["systemctl", "--user", "show-environment"],
+                                  capture_output=True, timeout=5, check=False)
+            if proc.returncode != 0:
+                reasons.append("systemd --user manager not reachable")
+        except (OSError, subprocess.SubprocessError):
+            reasons.append("systemd --user manager not reachable")
+    _caps_cache = Capabilities(ok=not reasons, reason="; ".join(reasons) or "ok")
+    return _caps_cache
+
+
+def _boot_id() -> str:
+    try:
+        return Path("/proc/sys/kernel/random/boot_id").read_text().strip()
+    except OSError:
+        return ""
+
+
+# -- durable record ---------------------------------------------------------- #
+
+
+@dataclass
+class Receipt:
+    outcome: str          # complete | pending_retry | refused | no_scope
+    error: str = ""
+    ts: str = ""
+
+    def is_complete(self) -> bool:
+        return self.outcome == "complete"
+
+    def is_terminal(self) -> bool:
+        return self.outcome in _TERMINAL_OUTCOMES
+
+
+@dataclass
+class LifecycleRecord:
+    run_id: str
+    unit: str
+    boot_id: str
+    invocation_id: str = ""
+    control_group: str = ""
+    receipt: dict | None = None
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, d: dict) -> "LifecycleRecord":
+        if not isinstance(d, dict):
+            raise LifecycleError("record is not an object")
+        try:
+            rec = cls(
+                run_id=str(d["run_id"]), unit=str(d["unit"]),
+                boot_id=str(d.get("boot_id", "")),
+                invocation_id=str(d.get("invocation_id", "")),
+                control_group=str(d.get("control_group", "")),
+                receipt=d.get("receipt") if isinstance(d.get("receipt"), dict) else None,
+            )
+        except (KeyError, TypeError) as e:
+            raise LifecycleError(f"malformed record: {e}") from e
+        rec.validate()
+        return rec
+
+    def validate(self) -> None:
+        if not _UNIT_RE.match(self.unit):
+            raise LifecycleError(f"unsafe unit name {self.unit!r}")
+        if self.control_group:
+            _validate_cgroup_path(self.control_group, self.unit)
+
+
+def _validate_cgroup_path(control_group: str, unit: str) -> None:
+    """A resolved scope cgroup must be a leaf under the user slice named for its
+    unit — reject a root/parent, a ``..`` traversal, or a foreign path."""
+    if not control_group.startswith(_USER_SLICE_PREFIX):
+        raise LifecycleError(f"cgroup not under user slice: {control_group!r}")
+    if ".." in control_group.split("/"):
+        raise LifecycleError(f"cgroup traverses parent: {control_group!r}")
+    if control_group.rstrip("/").split("/")[-1] != unit:
+        raise LifecycleError(f"cgroup leaf != unit: {control_group!r} / {unit!r}")
+
+
+def record_path(run_dir: Path) -> Path:
+    return Path(run_dir) / RECORD_NAME
+
+
+def read_record(run_dir: Path) -> LifecycleRecord | None:
+    try:
+        raw = record_path(run_dir).read_text(encoding="utf-8")
+    except FileNotFoundError:
+        return None
+    except OSError as e:
+        raise LifecycleError(f"cannot read record: {e}") from e
+    try:
+        return LifecycleRecord.from_dict(json.loads(raw))
+    except ValueError as e:
+        raise LifecycleError(f"corrupt record JSON: {e}") from e
+
+
+def write_record(run_dir: Path, record: LifecycleRecord) -> None:
+    record.validate()
+    atomic_write_text(record_path(run_dir),
+                      json.dumps(record.to_dict(), indent=2, sort_keys=True), mode=0o600)
+
+
+# -- unit naming + launch wrapper -------------------------------------------- #
+
+
+def scope_unit_name(run_id: str, nonce: str) -> str:
+    safe_run = re.sub(r"[^A-Za-z0-9_.-]", "_", run_id)[:80]
+    safe_nonce = re.sub(r"[^A-Za-z0-9]", "", nonce)[:16]
+    return f"nerve-wf-{safe_run}-{safe_nonce}.scope"
+
+
+def build_scope_argv(unit: str, inner_argv: list[str]) -> list[str]:
+    """Wrap ``inner_argv`` to run inside a delegated transient user scope.
+    systemd puts the process in the scope's cgroup before it execs, so the whole
+    fork/setsid subtree stays contained."""
+    if not _UNIT_RE.match(unit):
+        raise LifecycleError(f"unsafe unit {unit!r}")
+    return [
+        "systemd-run", "--user", "--scope", "--quiet", "--collect",
+        "--expand-environment=no",
+        "-p", "Delegate=yes",
+        "-p", "KillMode=control-group",
+        "-p", "SendSIGKILL=yes",
+        "-p", f"TimeoutStopSec={TERM_GRACE_SECONDS}",
+        "--description", SCOPE_DESCRIPTION,
+        "--unit", unit,
+        "--", *inner_argv,
+    ]
+
+
+# -- systemd / cgroup queries ------------------------------------------------ #
+
+
+def _unit_identity(unit: str) -> dict[str, str] | None:
+    """``systemctl --user show`` the unit's identity props, or None if the query
+    failed."""
+    try:
+        proc = subprocess.run(
+            ["systemctl", "--user", "show", unit,
+             "--property", "LoadState,InvocationID,ControlGroup,Description"],
+            capture_output=True, text=True, timeout=10, check=False,
+        )
+    except (OSError, subprocess.SubprocessError) as e:
+        logger.debug("systemctl show %s failed: %s", unit, e)
+        return None
+    if proc.returncode != 0:
+        return None
+    out: dict[str, str] = {}
+    for line in proc.stdout.splitlines():
+        key, sep, value = line.partition("=")
+        if sep:
+            out[key.strip()] = value.strip()
+    return out
+
+
+def _await_unit_identity(unit: str, *, timeout: float = 4.0) -> tuple[str, str]:
+    """Poll until the scope reports its InvocationID + ControlGroup."""
+    deadline = time.monotonic() + timeout
+    inv = cg = ""
+    while time.monotonic() < deadline:
+        ident = _unit_identity(unit) or {}
+        inv, cg = ident.get("InvocationID", ""), ident.get("ControlGroup", "")
+        if inv and cg:
+            return inv, cg
+        time.sleep(0.1)
+    return inv, cg
+
+
+def _cgroup_gone_or_empty(control_group: str) -> bool | None:
+    """True if the scope cgroup is gone or recursively unpopulated, False if it
+    still holds a process, None if it can't be read. ``cgroup.events``
+    ``populated`` is recursive, so it covers delegated child cgroups."""
+    if not control_group:
+        return None
+    path = Path(_CGROUP_ROOT + control_group) / "cgroup.events"
+    try:
+        text = path.read_text()
+    except FileNotFoundError:
+        return True
+    except OSError:
+        return None
+    for line in text.splitlines():
+        if line.startswith("populated"):
+            return line.split()[-1].strip() == "0"
+    return None
+
+
+def _scope_stop(unit: str) -> tuple[bool, str]:
+    try:
+        proc = subprocess.run(["systemctl", "--user", "stop", unit],
+                              capture_output=True, text=True,
+                              timeout=STOP_TIMEOUT_SECONDS, check=False)
+    except subprocess.TimeoutExpired:
+        return False, "stop timed out"
+    except (OSError, subprocess.SubprocessError) as e:
+        return False, f"stop failed: {e}"
+    if proc.returncode != 0:
+        return False, (proc.stderr or f"rc={proc.returncode}").strip()[:200]
+    return True, ""
+
+
+def proc_cgroup(pid: int) -> str | None:
+    try:
+        text = Path(f"/proc/{pid}/cgroup").read_text()
+    except OSError:
+        return None
+    for line in text.splitlines():
+        if line.startswith("0::"):
+            return line[3:].strip()
+    return None
+
+
+def verify_membership(control_group: str, pid: int) -> bool:
+    """True if ``pid`` is in ``control_group`` (or a descendant of it)."""
+    if not control_group:
+        return False
+    member = proc_cgroup(pid)
+    if member is None:
+        return False
+    return member == control_group or member.startswith(control_group.rstrip("/") + "/")
+
+
+# -- launch config handed to the app-server client -------------------------- #
+
+
+@dataclass(frozen=True)
+class WorkflowContainment:
+    run_dir: Path
+    run_id: str
+
+
+def prepare_launch(
+    containment: WorkflowContainment, inner_argv: list[str],
+) -> tuple[list[str], LifecycleRecord]:
+    """Record the owner, then return the scope-wrapped argv. A record left by an
+    earlier launch of the same run is reaped first. Raises before exec if the
+    host can't contain the run, the record can't be written, or the prior scope
+    isn't confirmed reaped."""
+    caps = detect_capabilities()
+    if not caps.ok:
+        raise ContainmentUnavailable(f"strict cgroup containment unavailable: {caps.reason}")
+    if record_path(containment.run_dir).exists():
+        prior = reap(containment.run_dir)
+        if not prior.is_complete():
+            raise LifecycleError(
+                f"previous scope not confirmed reaped ({prior.outcome}: {prior.error}); "
+                "refusing to relaunch"
+            )
+    unit = scope_unit_name(containment.run_id, uuid.uuid4().hex)
+    record = LifecycleRecord(run_id=containment.run_id, unit=unit, boot_id=_boot_id())
+    write_record(containment.run_dir, record)
+    return build_scope_argv(unit, inner_argv), record
+
+
+def record_launched(run_dir: Path, record: LifecycleRecord) -> LifecycleRecord:
+    """Record the scope's InvocationID + ControlGroup once systemd-run created
+    it. Raises if they don't resolve — the caller then fails the launch."""
+    inv, cg = _await_unit_identity(record.unit)
+    if not inv or not cg:
+        raise LifecycleError(f"scope {record.unit} reported no InvocationID/ControlGroup")
+    record.invocation_id, record.control_group = inv, cg
+    write_record(Path(run_dir), record)
+    return record
+
+
+# -- reap + reconcile -------------------------------------------------------- #
+
+
+def reap(run_dir: Path, *, unit: str | None = None) -> Receipt:
+    """Reap a run's contained descendants from its durable record. Idempotent:
+    a record already carrying a terminal receipt is returned unchanged. With
+    ``unit``, a record naming a different unit is left alone (``no_scope``)."""
+    run_dir = Path(run_dir)
+    if not record_path(run_dir).exists():
+        return Receipt(outcome="no_scope")
+    try:
+        record = read_record(run_dir)
+    except LifecycleError as e:
+        logger.warning("refusing to reap from corrupt record in %s: %s", run_dir, e)
+        return Receipt(outcome="pending_retry", error="corrupt record")
+    if record is None or (unit is not None and record.unit != unit):
+        return Receipt(outcome="no_scope")
+    prev = _receipt_from(record)
+    if prev is not None and prev.is_terminal():
+        return prev
+    receipt = _do_reap(record)
+    record.receipt = asdict(receipt)
+    try:
+        write_record(run_dir, record)
+    except OSError as e:
+        return Receipt(outcome="pending_retry", error=f"receipt not persisted: {e}")
+    return receipt
+
+
+def _do_reap(record: LifecycleRecord) -> Receipt:
+    # Recorded on a previous boot: nothing from it is alive; don't signal.
+    if record.boot_id and _boot_id() and record.boot_id != _boot_id():
+        return Receipt(outcome="complete")
+
+    ident = _unit_identity(record.unit)
+    if ident is None:
+        return Receipt(outcome="pending_retry", error="unit query failed")
+    if ident.get("LoadState", "") == "not-found":
+        # A transient scope is unloaded only once its cgroup is empty.
+        return Receipt(outcome="complete")
+
+    # Verify identity before signalling: an exact InvocationID when recorded, or
+    # the exact scope description for a pre-registration record.
+    live_inv = ident.get("InvocationID", "")
+    if record.invocation_id:
+        if not live_inv:
+            return Receipt(outcome="pending_retry", error="unit InvocationID unavailable")
+        if live_inv != record.invocation_id:
+            return Receipt(outcome="refused", error="unit replaced by a foreign instance")
+    elif ident.get("Description", "") != SCOPE_DESCRIPTION:
+        return Receipt(outcome="refused", error="unit description mismatch")
+
+    ok, err = _scope_stop(record.unit)
+    if not ok:
+        return Receipt(outcome="pending_retry", error=err)
+    # The cgroup can report populated briefly after the stop returns. A record
+    # written before registration has no cgroup path; the unit unloading then
+    # confirms the scope is empty.
+    deadline = time.monotonic() + EMPTY_WAIT_SECONDS
+    while True:
+        if _cgroup_gone_or_empty(record.control_group) is True:
+            return Receipt(outcome="complete")
+        if (_unit_identity(record.unit) or {}).get("LoadState") == "not-found":
+            return Receipt(outcome="complete")
+        if time.monotonic() >= deadline:
+            return Receipt(outcome="pending_retry", error="scope still populated after stop")
+        time.sleep(0.1)
+
+
+def reconcile(runs_dir: Path) -> list[tuple[str, Receipt]]:
+    """Reap every run's scope from its record. Called at startup, after active
+    runs have been marked failed."""
+    root = Path(runs_dir)
+    if not root.is_dir():
+        return []
+    results: list[tuple[str, Receipt]] = []
+    for child in sorted(p for p in root.iterdir() if p.is_dir()):
+        try:
+            record = read_record(child)
+        except LifecycleError:
+            logger.warning("skipping corrupt record in %s", child.name)
+            continue
+        if record is None:
+            continue
+        results.append((child.name, reap(child)))
+    return results
+
+
+def _receipt_from(record: LifecycleRecord) -> Receipt | None:
+    if not record.receipt:
+        return None
+    try:
+        return Receipt(**{k: v for k, v in record.receipt.items()
+                          if k in Receipt.__dataclass_fields__})
+    except TypeError:
+        return None
diff --git a/nerve/config.py b/nerve/config.py
index 5acacf5e6..34dce5a2e 100644
--- a/nerve/config.py
+++ b/nerve/config.py
@@ -3076,6 +3076,29 @@ def validate(self) -> list[str]:
         return problems
 
 
+@dataclass
+class CodexLifecycleConfig:
+    """Cgroup containment for Codex workflow runs.
+
+    ``mode``: ``disabled`` (default; launch unchanged) or ``strict`` (launch the
+    workflow app-server inside a systemd user scope so its descendants can be
+    reaped; fails before exec where the host can't contain). Codex workflow runs
+    only. See ``nerve/agent/backends/codex/lifecycle.py``.
+    """
+
+    mode: str = "disabled"
+
+    @classmethod
+    @_coerced
+    def from_dict(cls, raw: dict | None) -> "CodexLifecycleConfig":
+        return cls(mode=str((raw or {}).get("mode") or "disabled").strip().lower())
+
+    def validate(self) -> list[str]:
+        if self.mode not in ("disabled", "strict"):
+            return [f"codex.lifecycle.mode must be 'disabled' or 'strict', got {self.mode!r}"]
+        return []
+
+
 @dataclass
 class CodexConfig:
     """OpenAI Codex backend (``codex app-server``) settings.
@@ -3116,6 +3139,7 @@ class CodexConfig:
     # Arbitrary codex config-override passthrough (-c key=value at spawn)
     extra_config: dict[str, Any] = field(default_factory=dict)
     ultracode: UltracodeConfig = field(default_factory=UltracodeConfig)
+    lifecycle: CodexLifecycleConfig = field(default_factory=CodexLifecycleConfig)
 
     @classmethod
     @_coerced
@@ -3168,6 +3192,7 @@ def from_dict(cls, d: dict) -> "CodexConfig":
             pricing=pricing,
             extra_config=dict(d.get("extra_config") or {}),
             ultracode=UltracodeConfig.from_dict(d.get("ultracode")),
+            lifecycle=CodexLifecycleConfig.from_dict(d.get("lifecycle")),
         )
 
     def validate(self) -> list[str]:
@@ -3189,6 +3214,7 @@ def validate(self) -> list[str]:
                 f"got {self.sandbox!r}"
             )
         problems.extend(self.ultracode.validate())
+        problems.extend(self.lifecycle.validate())
         return problems
 
 
diff --git a/nerve/workflows/service.py b/nerve/workflows/service.py
index 4ef1b4d7a..2b4316ad5 100644
--- a/nerve/workflows/service.py
+++ b/nerve/workflows/service.py
@@ -55,6 +55,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from nerve.agent.backends.codex import lifecycle as codex_lifecycle
 from nerve.agent.streaming import broadcaster
 from nerve.db.usage import estimate_turn_cost
 from nerve.db.workflow_runs import ACTIVE_STATUSES
@@ -162,6 +163,8 @@ async def start(self) -> None:
                 "survive the restart; restart them explicitly if still needed.",
                 priority="high",
             )
+        # Reap Codex workflow-run scopes left by the previous daemon.
+        await self._reconcile_codex_lifecycle()
         interval = max(5, int(self.config.workflows.poll_interval_seconds))
         self._monitor_task = asyncio.create_task(self._monitor_loop(interval))
         logger.info(
@@ -759,6 +762,41 @@ def _is_quiet(run: dict | None) -> bool:
         notifications (a 3-iteration loop would otherwise ping 6-8 times)."""
         return str((run or {}).get("created_by") or "").startswith("review-loop:")
 
+    # ------------------------------------------------------------------ #
+    #  Codex workflow-run descendant containment (cgroup lifecycle)       #
+    # ------------------------------------------------------------------ #
+
+    async def _reconcile_codex_lifecycle(self) -> None:
+        """Reap scopes recorded by earlier runs, in every mode. Errors are
+        logged."""
+        try:
+            results = await asyncio.to_thread(
+                codex_lifecycle.reconcile, self.runs_dir(),
+            )
+        except Exception:  # noqa: BLE001
+            logger.exception("codex lifecycle reconciliation failed")
+            return
+        stuck = [rid for rid, rc in results if not rc.is_complete()]
+        if stuck:
+            logger.warning(
+                "codex lifecycle reconciliation could not confirm cleanup for: %s",
+                ", ".join(stuck),
+            )
+
+    async def _reap_codex_lifecycle(self, run_id: str) -> None:
+        """Reap one terminal run's contained descendants from its record."""
+        try:
+            receipt = await asyncio.to_thread(codex_lifecycle.reap, self.runs_dir() / run_id)
+        except Exception:  # noqa: BLE001
+            logger.exception("codex lifecycle reap failed for %s", run_id)
+            return
+        if receipt.outcome not in ("complete", "no_scope"):
+            logger.warning(
+                "codex lifecycle reap for %s did not complete (%s: %s); "
+                "the next startup reconciliation will retry",
+                run_id, receipt.outcome, receipt.error,
+            )
+
     # ------------------------------------------------------------------ #
     #  Budget monitor                                                     #
     # ------------------------------------------------------------------ #
@@ -985,6 +1023,9 @@ async def _teardown_client(self, run_id: str) -> None:
             )
         except Exception:  # noqa: BLE001
             logger.exception("client teardown failed for %s", session_id)
+        # Closing a contained client stops its scope; this reaps one the close
+        # left behind (no-op without a record).
+        await self._reap_codex_lifecycle(run_id)
 
     async def _stop_session_hard(self, run_id: str) -> None:
         """Graceful interrupt → grace window → force client discard.
diff --git a/tests/fixtures/fake_codex_appserver.py b/tests/fixtures/fake_codex_appserver.py
index b25d1d288..61da90ce8 100755
--- a/tests/fixtures/fake_codex_appserver.py
+++ b/tests/fixtures/fake_codex_appserver.py
@@ -27,6 +27,8 @@
                  transport)
   big_stderr   — writes one ~2.5 MiB stderr line mid-turn, then
                  completes normally (stderr loop must keep draining)
+  stall        — starts the turn, then emits nothing and ignores
+                 turn/interrupt
   close_stdout_mid_turn — closes stdout mid-turn but KEEPS RUNNING
                  (reader-death liveness: is_alive() must flip False even
                  though the process is still up)
@@ -257,6 +259,12 @@ def run_turn(thread_id: str, turn_id: str) -> None:
         sys.stdout.flush()
         os._exit(1)
 
+    if MODE == "stall":
+        notify("turn/started", {"threadId": thread_id,
+                                "turn": {"id": turn_id, "status": "inProgress", "items": []}})
+        time.sleep(3600)
+        return
+
     if MODE == "interrupt":
         notify("turn/started", {"threadId": thread_id,
                                 "turn": {"id": turn_id, "status": "inProgress", "items": []}})
diff --git a/tests/fixtures/lifecycle_escaper.py b/tests/fixtures/lifecycle_escaper.py
new file mode 100755
index 000000000..cb5b75355
--- /dev/null
+++ b/tests/fixtures/lifecycle_escaper.py
@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""Escaping-descendant fixture for Codex lifecycle tests (no model, no network).
+
+Run as the main command of a systemd ``--scope``, it setsid+double-forks a
+sleeper grandchild that outlives the scope's main process — the escape a
+single ``killpg`` cannot reach. Every process carries a unique ``ESCAPER_TOKEN``
+so cleanup only ever signals this fixture's own processes, by pidfd.
+
+Commands: ``spawn-escapee <outdir>`` (plain), ``spawn-escapee-nested <outdir>``
+(migrate the grandchild into a child cgroup so the scope root empties but
+recursive ``cgroup.events populated`` stays 1), ``sleeper``, ``sweep <token>``.
+"""
+
+from __future__ import annotations
+
+import os
+import signal
+import sys
+import time
+
+SELF = os.path.abspath(__file__)
+_SLEEP = 300
+
+
+# --------------------------- identity + cleanup --------------------------- #
+
+
+def _environ_token(pid: int) -> str | None:
+    try:
+        with open(f"/proc/{pid}/environ", "rb") as fh:
+            raw = fh.read()
+    except OSError:
+        return None
+    for entry in raw.split(b"\0"):
+        if entry.startswith(b"ESCAPER_TOKEN="):
+            return entry[len(b"ESCAPER_TOKEN="):].decode("utf-8", "replace")
+    return None
+
+
+def _cmdline(pid: int) -> str:
+    try:
+        with open(f"/proc/{pid}/cmdline", "rb") as fh:
+            return fh.read().replace(b"\0", b" ").decode("utf-8", "replace")
+    except OSError:
+        return ""
+
+
+def is_alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+        return True
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+
+
+def owned_by_me(pid: int, token: str) -> bool:
+    if pid <= 1 or pid == os.getpid():
+        return False
+    if _environ_token(pid) != token:
+        return False
+    return SELF in _cmdline(pid)
+
+
+def identity_kill(pid: int, token: str) -> str:
+    if not owned_by_me(pid, token):
+        return "skipped(not-owned)"
+    try:
+        pidfd = os.pidfd_open(pid)
+    except (ProcessLookupError, OSError) as e:
+        return f"gone({e.__class__.__name__})"
+    try:
+        if not owned_by_me(pid, token):
+            return "skipped(reuse)"
+        try:
+            signal.pidfd_send_signal(pidfd, signal.SIGKILL)
+            return "killed"
+        except ProcessLookupError:
+            return "already-gone"
+    finally:
+        os.close(pidfd)
+
+
+def sweep(token: str) -> list[tuple[int, str]]:
+    out = []
+    for name in os.listdir("/proc"):
+        if not name.isdigit():
+            continue
+        pid = int(name)
+        if owned_by_me(pid, token):
+            out.append((pid, identity_kill(pid, token)))
+    return out
+
+
+# ------------------------------ the fork tree ----------------------------- #
+
+
+def _become_sleeper(token: str, role: str, outdir: str) -> None:
+    os.execv(sys.executable, [sys.executable, SELF, "sleeper", token, role, outdir])
+
+
+def _write(outdir: str, name: str, value: str) -> None:
+    with open(os.path.join(outdir, name), "w") as fh:
+        fh.write(value)
+
+
+def _own_cgroup() -> str | None:
+    try:
+        with open("/proc/self/cgroup") as fh:
+            for line in fh:
+                if line.startswith("0::"):
+                    return line[3:].strip()
+    except OSError:
+        return None
+    return None
+
+
+def _double_fork_grandchild(token: str, outdir: str, variant: str) -> None:
+    """In a child: setsid → NEW group, close stdio, double-fork the survivor."""
+    os.setsid()
+    devnull = os.open(os.devnull, os.O_RDWR)
+    for fd in (0, 1, 2):
+        os.dup2(devnull, fd)
+    g = os.fork()
+    if g != 0:
+        os._exit(0)  # sandbox exits → grandchild is double-fork orphaned
+
+    # grandchild (the survivor)
+    if variant == "nested":
+        # Migrate into a child cgroup of the delegated scope: the scope root's
+        # cgroup.procs empties, but recursive cgroup.events populated stays 1.
+        cg = _own_cgroup()
+        if cg:
+            leaf = "/sys/fs/cgroup" + cg + "/nested"
+            try:
+                os.mkdir(leaf)
+                with open(leaf + "/cgroup.procs", "w") as fh:
+                    fh.write(str(os.getpid()))
+            except OSError:
+                pass  # non-fatal: fall back to plain containment
+
+    _write(outdir, "grandchild.pid", str(os.getpid()))
+    _become_sleeper(token, "grandchild", outdir)
+
+
+def _spawn(variant: str, argv: list[str]) -> int:
+    token = os.environ["ESCAPER_TOKEN"]
+    outdir = argv[0]
+    child = os.fork()
+    if child == 0:
+        _double_fork_grandchild(token, outdir, variant)
+        os._exit(0)  # unreachable
+    # scope main: wait for the grandchild to materialize, reap the middle
+    # child (so it is not a zombie), then exit — leaving the escapee behind.
+    deadline = time.monotonic() + 5
+    gc = os.path.join(outdir, "grandchild.pid")
+    while time.monotonic() < deadline and not os.path.exists(gc):
+        time.sleep(0.05)
+    try:
+        os.waitpid(child, 0)
+    except ChildProcessError:
+        pass
+    return 0
+
+
+def main() -> int:
+    if len(sys.argv) < 2:
+        print("usage: lifecycle_escaper.py <cmd> ...", file=sys.stderr)
+        return 2
+    cmd = sys.argv[1]
+    if cmd == "sleeper":
+        time.sleep(_SLEEP)
+        return 0
+    if cmd == "spawn-escapee":
+        return _spawn("plain", sys.argv[2:])
+    if cmd == "spawn-escapee-nested":
+        return _spawn("nested", sys.argv[2:])
+    if cmd == "sweep":
+        token = os.environ.get("ESCAPER_TOKEN") or (sys.argv[2] if len(sys.argv) > 2 else "")
+        print(sweep(token))
+        return 0
+    print(f"unknown cmd {cmd}", file=sys.stderr)
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_codex_lifecycle.py b/tests/test_codex_lifecycle.py
new file mode 100644
index 000000000..60ade7403
--- /dev/null
+++ b/tests/test_codex_lifecycle.py
@@ -0,0 +1,484 @@
+"""Tests for Codex workflow-run cgroup containment (codex/lifecycle.py).
+
+Containment/reap cases use real systemd user scopes running real setsid escapers
+(tests/fixtures/lifecycle_escaper.py) and the real fake app-server — no model,
+no network. Every scope/process is torn down by the ``scopes`` fixture. Scope
+cases skip where a systemd ``--user`` manager isn't reachable.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import uuid
+from pathlib import Path
+
+import pytest
+
+from nerve.agent.backends import BackendDeps, SessionSpec
+from nerve.agent.backends.codex import CodexBackend, lifecycle
+from nerve.agent.backends.codex.appserver import CodexAppServerClient
+from nerve.agent.backends.codex.backend import _workflow_containment
+from nerve.agent.interactive import InteractiveToolHandler
+from nerve.config import NerveConfig
+
+FIX = Path(__file__).parent / "fixtures"
+ESCAPER = str(FIX / "lifecycle_escaper.py")
+FAKE = str(FIX / "fake_codex_appserver.py")
+
+CAPS = lifecycle.detect_capabilities()
+requires_scope = pytest.mark.skipif(
+    not CAPS.ok, reason=f"systemd user scope unavailable: {CAPS.reason}",
+)
+
+
+def _alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+        return True
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+
+
+@pytest.fixture
+def scopes():
+    reg: dict[str, set] = {"units": set(), "tokens": set(), "run_dirs": set()}
+    try:
+        yield reg
+    finally:
+        for rd in reg["run_dirs"]:
+            try:
+                rec = lifecycle.read_record(Path(rd))
+                if rec is not None:
+                    reg["units"].add(rec.unit)
+            except Exception:  # noqa: BLE001
+                pass
+        for unit in reg["units"]:
+            subprocess.run(["systemctl", "--user", "stop", unit], capture_output=True, timeout=15)
+        for tok in reg["tokens"]:
+            subprocess.run([sys.executable, ESCAPER, "sweep", tok],
+                           env={**os.environ, "ESCAPER_TOKEN": tok}, capture_output=True, timeout=15)
+
+
+def _launch_escaper(scopes, run_dir: Path, run_id: str, *, variant: str = "",
+                    invocation_id: str | None = None):
+    """Launch a real escaper in a real scope; return (record, gc_pid)."""
+    run_dir = Path(run_dir)
+    run_dir.mkdir(parents=True, exist_ok=True)
+    token = "lc" + uuid.uuid4().hex[:12]
+    scopes["tokens"].add(token)
+    unit = lifecycle.scope_unit_name(run_id, uuid.uuid4().hex)
+    scopes["units"].add(unit)
+    cmd = "spawn-escapee" + (f"-{variant}" if variant else "")
+    argv = lifecycle.build_scope_argv(unit, [sys.executable, ESCAPER, cmd, str(run_dir)])
+    subprocess.run(argv, env={**os.environ, "ESCAPER_TOKEN": token},
+                   capture_output=True, timeout=30, check=True)
+    gc_pid = int((run_dir / "grandchild.pid").read_text())
+    rec = lifecycle.LifecycleRecord(run_id=run_id, unit=unit, boot_id=lifecycle._boot_id())
+    lifecycle.write_record(run_dir, rec)
+    if invocation_id is not None:
+        _inv, cg = lifecycle._await_unit_identity(unit)
+        rec.invocation_id, rec.control_group = invocation_id, cg
+        lifecycle.write_record(run_dir, rec)
+    else:
+        rec = lifecycle.record_launched(run_dir, rec)
+    return rec, gc_pid
+
+
+def _deps(cfg: NerveConfig) -> BackendDeps:
+    return BackendDeps(
+        config=lambda: cfg, db=None, registry=None, tool_ctx_factory=lambda sid: None,
+        external_mcp_servers=lambda: [], gateway_port=lambda: 8900,
+        mint_session_token=lambda sid: f"tok-{sid}",
+    )
+
+
+def _spec(cfg: NerveConfig, *, session_id: str, source: str) -> SessionSpec:
+    hub = InteractiveToolHandler(session_id=session_id, broadcast_fn=_noop, interactive_capable=False)
+    return SessionSpec(
+        session_id=session_id, source=source, model=None, effort="high",
+        system_prompt="You are Nerve.", cwd=str(cfg.workspace), resume_native_id=None,
+        fork=False, interactive=hub, snapshot=None, record_wakeup=None, idle_timeout=15.0,
+    )
+
+
+async def _noop(session_id: str, message: dict) -> None:
+    return None
+
+
+def _wf_config(tmp_path: Path, *, mode: str = "strict") -> NerveConfig:
+    cfg = NerveConfig.from_dict({
+        "workspace": str(tmp_path / "ws"),
+        "workflows": {"runs_dir": str(tmp_path / "workflow-runs")},
+        "codex": {"bin_path": FAKE, "home_dir": str(tmp_path / "home"),
+                  "model": "gpt-5.6-sol", "lifecycle": {"mode": mode}},
+    })
+    (tmp_path / "ws").mkdir(parents=True, exist_ok=True)
+    (tmp_path / "workflow-runs").mkdir(parents=True, exist_ok=True)
+    return cfg
+
+
+# -- launch path ------------------------------------------------------------ #
+
+
+@requires_scope
+@pytest.mark.asyncio
+async def test_strict_launch_contains_and_roundtrips(tmp_path, scopes):
+    run_dir = tmp_path / "workflow-runs" / "wfr-l1"
+    scopes["run_dirs"].add(run_dir)
+    cwd = tmp_path / "ws"
+    cwd.mkdir(parents=True, exist_ok=True)
+    containment = lifecycle.WorkflowContainment(run_dir=run_dir, run_id="wfr-l1")
+
+    async def _handler(method, params):
+        return {}
+
+    client = CodexAppServerClient(
+        bin_path=FAKE, cwd=str(cwd),
+        env={**os.environ, "CODEX_HOME": str(tmp_path / "home"), "FAKE_CODEX_MODE": "basic"},
+        server_request_handler=_handler, config_overrides=[], containment=containment,
+    )
+    resp = await client.start()
+    try:
+        assert resp.get("userAgent")                       # clean JSON through the wrapper
+        rec = lifecycle.read_record(run_dir)
+        assert rec.invocation_id and rec.control_group     # scope identity recorded
+        assert lifecycle._cgroup_gone_or_empty(rec.control_group) is False  # contained + populated
+    finally:
+        await client.close()
+
+
+def test_scope_argv_has_bounded_kill_directives():
+    argv = lifecycle.build_scope_argv("nerve-wf-x-n.scope", ["/bin/true"])
+    assert "Delegate=yes" in argv
+    assert "KillMode=control-group" in argv
+    assert "SendSIGKILL=yes" in argv
+    assert any(a.startswith("TimeoutStopSec=") for a in argv)
+    assert "--expand-environment=no" in argv
+
+
+def test_strict_unavailable_fails_before_exec(tmp_path, monkeypatch):
+    run_dir = tmp_path / "u"
+    run_dir.mkdir()
+    monkeypatch.setattr(lifecycle, "detect_capabilities",
+                        lambda **k: lifecycle.Capabilities(ok=False, reason="no systemd-run"))
+    c = lifecycle.WorkflowContainment(run_dir=run_dir, run_id="wfr-u")
+    with pytest.raises(lifecycle.ContainmentUnavailable):
+        lifecycle.prepare_launch(c, ["/usr/bin/codex", "app-server"])
+    assert not lifecycle.record_path(run_dir).exists()   # nothing launched/recorded
+
+
+def test_containment_gating_rejects_disabled_nonworkflow_and_traversal(tmp_path):
+    cfg = _wf_config(tmp_path, mode="strict")
+    backend = CodexBackend(_deps(cfg))
+    # real workflow session in strict mode → contained
+    c = _workflow_containment(backend, _spec(cfg, session_id="workflow:wfr-abc123", source="workflow"))
+    assert c is not None and c.run_id == "wfr-abc123"
+    # interactive session → never contained, even in strict mode
+    assert _workflow_containment(backend, _spec(cfg, session_id="s1", source="web")) is None
+    # path-traversal / spoofed run id → refused (no containment, no escape)
+    assert _workflow_containment(backend, _spec(cfg, session_id="workflow:../../etc", source="workflow")) is None
+    assert _workflow_containment(backend, _spec(cfg, session_id="workflow:wfr-a/b", source="workflow")) is None
+    # disabled mode → not contained
+    cfg2 = _wf_config(tmp_path, mode="disabled")
+    b2 = CodexBackend(_deps(cfg2))
+    assert _workflow_containment(b2, _spec(cfg2, session_id="workflow:wfr-x", source="workflow")) is None
+
+
+# -- reap -------------------------------------------------------------------- #
+
+
+@requires_scope
+def test_reap_kills_contained_spares_foreign_and_is_idempotent(tmp_path, scopes, monkeypatch):
+    target, gc_t = _launch_escaper(scopes, tmp_path / "t", "wfr-target")
+    foreign, gc_f = _launch_escaper(scopes, tmp_path / "f", "wfr-foreign")
+    assert _alive(gc_t) and _alive(gc_f)
+    assert lifecycle.proc_cgroup(gc_t) == target.control_group
+
+    r = lifecycle.reap(tmp_path / "t")
+    assert r.outcome == "complete"
+    assert not _alive(gc_t)          # target reaped
+    assert _alive(gc_f)              # foreign run untouched
+
+    # Idempotent: a second reap returns the stored terminal receipt, no re-stop.
+    monkeypatch.setattr(lifecycle, "_scope_stop", _fail_if_called)
+    assert lifecycle.reap(tmp_path / "t").outcome == "complete"
+
+
+@requires_scope
+def test_reap_nested_cgroup_uses_recursive_populated(tmp_path, scopes):
+    rec, gc = _launch_escaper(scopes, tmp_path / "n", "wfr-nested", variant="nested")
+    # grandchild migrated into a child cgroup: scope root cgroup.procs is empty,
+    # but recursive cgroup.events populated is 1 (what reap must check).
+    assert Path("/sys/fs/cgroup" + rec.control_group + "/cgroup.procs").read_text().split() == []
+    assert lifecycle._cgroup_gone_or_empty(rec.control_group) is False
+    assert lifecycle.proc_cgroup(gc) == rec.control_group + "/nested"
+    assert lifecycle.reap(tmp_path / "n").outcome == "complete"
+    assert not _alive(gc)
+
+
+@requires_scope
+def test_reap_refuses_replaced_unit(tmp_path, scopes):
+    # Record a wrong InvocationID: the live unit is a foreign instance of the
+    # same name; reap must refuse to signal it.
+    rec, gc = _launch_escaper(scopes, tmp_path / "r", "wfr-rep",
+                              invocation_id="00000000000000000000000000000000")
+    r = lifecycle.reap(tmp_path / "r")
+    assert r.outcome == "refused" and not r.is_complete()
+    assert _alive(gc)
+
+
+@requires_scope
+def test_relaunch_reaps_prior_scope_before_overwriting_record(tmp_path, scopes):
+    # A crashed client is recreated for the same run: prepare_launch must reap
+    # the prior scope before overwriting the sole record, or attempt 1's escapee
+    # would become unreachable.
+    run_dir = tmp_path / "workflow-runs" / "wfr-relaunch"
+    scopes["run_dirs"].add(run_dir)
+    rec1, gc1 = _launch_escaper(scopes, run_dir, "wfr-relaunch")
+    assert _alive(gc1)
+    c = lifecycle.WorkflowContainment(run_dir=run_dir, run_id="wfr-relaunch")
+    _argv, rec2 = lifecycle.prepare_launch(c, ["/bin/true"])
+    assert not _alive(gc1)                       # prior scope reaped
+    assert rec2.unit != rec1.unit                # new attempt gets a new unit
+    assert lifecycle.read_record(run_dir).unit == rec2.unit
+
+
+def test_relaunch_refused_when_prior_scope_not_confirmed(tmp_path, monkeypatch):
+    run_dir = tmp_path / "workflow-runs" / "wfr-stuck"
+    run_dir.mkdir(parents=True)
+    lifecycle.write_record(run_dir, lifecycle.LifecycleRecord(
+        run_id="wfr-stuck", unit="nerve-wf-wfr-stuck-n.scope", boot_id=lifecycle._boot_id()))
+    monkeypatch.setattr(lifecycle, "reap", lambda rd: lifecycle.Receipt(outcome="pending_retry", error="x"))
+    c = lifecycle.WorkflowContainment(run_dir=run_dir, run_id="wfr-stuck")
+    with pytest.raises(lifecycle.LifecycleError):
+        lifecycle.prepare_launch(c, ["/bin/true"])
+
+
+def test_reap_cross_boot_record_never_signals(tmp_path, monkeypatch):
+    run_dir = tmp_path / "b"
+    run_dir.mkdir()
+    lifecycle.write_record(run_dir, lifecycle.LifecycleRecord(
+        run_id="wfr-b", unit="nerve-wf-wfr-b-n.scope", boot_id="prior-boot",
+        invocation_id="i", control_group="/user.slice/nerve-wf-wfr-b-n.scope"))
+    monkeypatch.setattr(lifecycle, "_unit_identity", _fail_if_called)
+    monkeypatch.setattr(lifecycle, "_scope_stop", _fail_if_called)
+    assert lifecycle.reap(run_dir).outcome == "complete"   # nothing survives a reboot
+
+
+def test_reap_query_failure_is_pending_not_complete(tmp_path, monkeypatch):
+    run_dir = tmp_path / "q"
+    run_dir.mkdir()
+    lifecycle.write_record(run_dir, lifecycle.LifecycleRecord(
+        run_id="wfr-q", unit="nerve-wf-wfr-q-n.scope", boot_id=lifecycle._boot_id(),
+        invocation_id="inv", control_group="/user.slice/nerve-wf-wfr-q-n.scope"))
+    # systemctl query failed → must NOT be read as "scope gone"
+    monkeypatch.setattr(lifecycle, "_unit_identity", lambda u: None)
+    monkeypatch.setattr(lifecycle, "_scope_stop", _fail_if_called)
+    assert lifecycle.reap(run_dir).outcome == "pending_retry"
+    # unit present but InvocationID blank (recorded one exists) → don't signal
+    monkeypatch.setattr(lifecycle, "_unit_identity",
+                        lambda u: {"LoadState": "loaded", "InvocationID": "", "ControlGroup": "", "Description": ""})
+    # reset the stored receipt so reap re-evaluates
+    lifecycle.write_record(run_dir, lifecycle.LifecycleRecord(
+        run_id="wfr-q", unit="nerve-wf-wfr-q-n.scope", boot_id=lifecycle._boot_id(),
+        invocation_id="inv", control_group="/user.slice/nerve-wf-wfr-q-n.scope"))
+    assert lifecycle.reap(run_dir).outcome == "pending_retry"
+
+
+@pytest.mark.parametrize("body", ["{ not valid json ", '{"run_id":"r","unit":"evil; rm.scope","boot_id":"b"}'])
+def test_reap_corrupt_or_invalid_record_never_signals(tmp_path, monkeypatch, body):
+    run_dir = tmp_path / "c"
+    run_dir.mkdir()
+    lifecycle.record_path(run_dir).write_text(body)
+    monkeypatch.setattr(lifecycle, "_scope_stop", _fail_if_called)
+    assert lifecycle.reap(run_dir).outcome == "pending_retry"
+
+
+def test_record_rejects_foreign_cgroup_path():
+    with pytest.raises(lifecycle.LifecycleError):
+        lifecycle.LifecycleRecord(run_id="r", unit="nerve-wf-r-n.scope", boot_id="b",
+                                  control_group="/system.slice/other.scope").validate()
+
+
+def _fail_if_called(*a, **k):
+    raise AssertionError("must not be called for this case")
+
+
+# -- reconcile --------------------------------------------------------------- #
+
+
+@requires_scope
+def test_reconcile_reaps_recorded_scope(tmp_path, scopes):
+    rec, gc = _launch_escaper(scopes, tmp_path / "wf" / "wfr-old", "wfr-old")
+    results = lifecycle.reconcile(tmp_path / "wf")
+    assert ("wfr-old", ) == tuple(rid for rid, _ in results)
+    assert results[0][1].outcome == "complete"
+    assert not _alive(gc)
+
+
+def test_reconcile_skips_dirs_without_a_record(tmp_path):
+    (tmp_path / "wf" / "plain-run").mkdir(parents=True)   # e.g. a Claude run dir
+    assert lifecycle.reconcile(tmp_path / "wf") == []
+
+
+# -- client close + workflow stop paths ------------------------------------- #
+
+
+def _escaping_launcher(tmp_path: Path) -> tuple[Path, Path]:
+    """An executable that starts a setsid escapee (in its own launch dir), then
+    execs the fake app-server. Returns (launcher, launches_dir)."""
+    launches = tmp_path / "launches"
+    launches.mkdir()
+    launcher = tmp_path / "codex-launcher"
+    launcher.write_text(
+        "#!/bin/sh\n"
+        'case "$*" in *--version*) exec "$NERVE_PY" "$NERVE_FAKE" "$@";; esac\n'
+        'out=$(mktemp -d "$NERVE_LAUNCHES/l-XXXXXX")\n'
+        '"$NERVE_PY" "$NERVE_ESCAPER" spawn-escapee "$out" >/dev/null 2>&1\n'
+        'exec "$NERVE_PY" "$NERVE_FAKE" "$@"\n'
+    )
+    launcher.chmod(0o755)
+    return launcher, launches
+
+
+def _launcher_env(launches: Path, token: str) -> dict[str, str]:
+    return {"NERVE_PY": sys.executable, "NERVE_FAKE": FAKE, "NERVE_ESCAPER": ESCAPER,
+            "NERVE_LAUNCHES": str(launches), "ESCAPER_TOKEN": token}
+
+
+def _escapees(launches: Path) -> list[int]:
+    return [int((d / "grandchild.pid").read_text())
+            for d in sorted(launches.iterdir()) if (d / "grandchild.pid").exists()]
+
+
+def test_reap_with_unit_leaves_a_newer_record_alone(tmp_path, monkeypatch):
+    run_dir = tmp_path / "w"
+    run_dir.mkdir()
+    lifecycle.write_record(run_dir, lifecycle.LifecycleRecord(
+        run_id="wfr-w", unit="nerve-wf-wfr-w-new.scope", boot_id=lifecycle._boot_id()))
+    monkeypatch.setattr(lifecycle, "_unit_identity", _fail_if_called)
+    monkeypatch.setattr(lifecycle, "_scope_stop", _fail_if_called)
+    assert lifecycle.reap(run_dir, unit="nerve-wf-wfr-w-old.scope").outcome == "no_scope"
+
+
+def test_reap_before_registration_confirms_by_unit_unload(tmp_path, monkeypatch):
+    run_dir = tmp_path / "p"
+    run_dir.mkdir()
+    lifecycle.write_record(run_dir, lifecycle.LifecycleRecord(
+        run_id="wfr-p", unit="nerve-wf-wfr-p-n.scope", boot_id=lifecycle._boot_id()))
+    states = iter([
+        {"LoadState": "loaded", "InvocationID": "i", "ControlGroup": "",
+         "Description": lifecycle.SCOPE_DESCRIPTION},
+        {"LoadState": "not-found"},
+    ])
+    stops = []
+    monkeypatch.setattr(lifecycle, "_unit_identity", lambda u: next(states))
+    monkeypatch.setattr(lifecycle, "_scope_stop", lambda u: (stops.append(u), (True, ""))[1])
+    assert lifecycle.reap(run_dir).outcome == "complete"
+    assert stops == ["nerve-wf-wfr-p-n.scope"]
+
+
+@requires_scope
+@pytest.mark.asyncio
+async def test_close_stops_own_scope(tmp_path, scopes):
+    run_dir = tmp_path / "workflow-runs" / "wfr-close"
+    scopes["run_dirs"].add(run_dir)
+    launcher, launches = _escaping_launcher(tmp_path)
+    token = "lc" + uuid.uuid4().hex[:12]
+    scopes["tokens"].add(token)
+    cwd = tmp_path / "ws"
+    cwd.mkdir()
+
+    async def _handler(method, params):
+        return {}
+
+    client = CodexAppServerClient(
+        bin_path=str(launcher), cwd=str(cwd),
+        env={**os.environ, **_launcher_env(launches, token),
+             "CODEX_HOME": str(tmp_path / "home"), "FAKE_CODEX_MODE": "basic"},
+        server_request_handler=_handler, config_overrides=[],
+        containment=lifecycle.WorkflowContainment(run_dir=run_dir, run_id="wfr-close"),
+    )
+    await client.start()
+    [gc] = _escapees(launches)
+    rec = lifecycle.read_record(run_dir)
+    assert _alive(gc) and lifecycle.proc_cgroup(gc) == rec.control_group
+    await client.close()
+    assert not _alive(gc)                          # escaped the process group, not the scope
+    assert lifecycle.read_record(run_dir).receipt["outcome"] == "complete"
+
+
+@requires_scope
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stop_via", ["kill_run", "budget_kill", "stop_session"])
+async def test_workflow_stop_before_content_reaps_and_never_relaunches(
+    db, tmp_path, scopes, monkeypatch, stop_via,
+):
+    """A stop or budget kill while the turn has produced no content must not
+    let the engine's transport retry launch a second scope, and must leave no
+    descendant alive. Real AgentEngine, WorkflowRunService and CodexBackend."""
+    import asyncio
+
+    from nerve.agent.engine import AgentEngine
+    from nerve.workflows.service import ENGINE_CODEX, WorkflowRunService
+    from tests.actor_rows import ensure_system_principal
+
+    await ensure_system_principal(db)
+    launcher, launches = _escaping_launcher(tmp_path)
+    token = "lc" + uuid.uuid4().hex[:12]
+    scopes["tokens"].add(token)
+    for k, v in _launcher_env(launches, token).items():
+        monkeypatch.setenv(k, v)
+    monkeypatch.setenv("FAKE_CODEX_MODE", "stall")
+    runs_dir = tmp_path / "workflow-runs"
+    cfg = NerveConfig.from_dict({
+        "workspace": str(tmp_path / "ws"),
+        "workflows": {"runs_dir": str(runs_dir), "kill_grace_seconds": 1},
+        "codex": {"bin_path": str(launcher), "home_dir": str(tmp_path / "home"),
+                  "model": "gpt-5.6-sol", "lifecycle": {"mode": "strict"}},
+    })
+    (tmp_path / "ws").mkdir()
+    runs_dir.mkdir()
+    engine = AgentEngine(cfg, db)
+    svc = WorkflowRunService(cfg, db, engine)
+    engine.add_stop_listener(svc._on_session_stop)   # registered by svc.start()
+    run_id = "wfr-race" + stop_via.replace("_", "")
+    run_dir = runs_dir / run_id
+    scopes["run_dirs"].add(run_dir)
+    await db.create_workflow_run(run_id, ENGINE_CODEX, {"prompt": "p"}, 5.0)
+    await db.transition_workflow_run(run_id, "running", expect=("pending",))
+    task = asyncio.create_task(svc._execute(run_id))
+    svc._exec_tasks[run_id] = task
+    try:
+        deadline = asyncio.get_running_loop().time() + 30
+        while not (_escapees(launches) and engine.is_session_running(f"workflow:{run_id}")
+                   and (lifecycle.read_record(run_dir) or lifecycle.LifecycleRecord(
+                       run_id=run_id, unit="nerve-wf-x.scope", boot_id="")).control_group):
+            assert asyncio.get_running_loop().time() < deadline, "turn never started"
+            await asyncio.sleep(0.1)
+        [gc] = _escapees(launches)
+        assert _alive(gc)
+
+        if stop_via == "kill_run":
+            await svc.kill_run(run_id, "test")
+        elif stop_via == "budget_kill":
+            await svc._budget_kill(await db.get_workflow_run(run_id), 6.0)
+        else:
+            await engine.stop_session(f"workflow:{run_id}")
+
+        deadline = asyncio.get_running_loop().time() + 60
+        while svc._exec_tasks or svc._kill_tasks:
+            assert asyncio.get_running_loop().time() < deadline, "stop did not settle"
+            await asyncio.gather(*svc._exec_tasks.values(), *svc._kill_tasks,
+                                 return_exceptions=True)
+    finally:
+        await engine.shutdown()
+
+    assert len(list(launches.iterdir())) == 1      # no relaunch after terminal
+    assert not any(_alive(pid) for pid in _escapees(launches))
+    assert lifecycle.read_record(run_dir).receipt["outcome"] == "complete"
+    assert (await db.get_workflow_run(run_id))["status"] in ("killed", "budget_exhausted")