Fix virtualenv installation, lock recovery, and platform test ownership

.github/workflows/ci.yml

@@ -1,7 +1,7 @@
 # Ouroboros CI — Five-tier cross-platform testing and release pipeline
 #
 # Tier 1 (Quick):       Push to ouroboros or PR targeting ouroboros                   → Ubuntu-only tests (~1 min)
-# Tier 2 (Full):        Push to ouroboros-stable / manual / tag                       → Full 3-OS matrix          (~5 min)
+# Tier 2 (Full):        Every PR → Windows/macOS; stable / manual / tag → full 3-OS matrix (~5 min)
 # Tier 2.5 (Integration): Push to main / ouroboros / ouroboros-stable / manual / tag  → Real-provider tests       (~2 min)
 # Tier 2.6 (Skill smoke): Push to ouroboros-stable / manual / tag                     → LIVE OuroborosHub official-skill install smoke (3-OS, ~5 min) + review→grant→enable-persistence flow on one cheap reviewer slot (ubuntu-only step, OPENROUTER_API_KEY, ~$1.2/run)
 # Tier 3 (Build+Release): Tag v*                                                       → PyInstaller + GitHub Release (~15 min)
@@ -53,10 +53,9 @@ on:
   # Only fork-safe jobs match pull_request refs, including narrow Publish UI proof.
   pull_request:
     branches: [ouroboros]
-  # A plain dispatch (the pre-tag 3-OS matrix: `gh workflow run CI --ref <branch>`)
-  # never spends money: the paid e2e-live lane runs only when its opt-in input is
-  # set (`-f e2e_live=true`). Inputs do not change `github.event_name`, so every
-  # job gated on the bare `workflow_dispatch` event is unaffected by this block.
+  # Plain dispatch retains the provider and skill-review jobs and can spend
+  # money. The separate paid e2e-live stand additionally requires its opt-in
+  # input (`-f e2e_live=true`); all other dispatch gates stay unchanged.
   workflow_dispatch:
     inputs:
       e2e_live:
@@ -197,17 +196,18 @@ jobs:
         run: python -m pytest tests/test_devtools_benchmarks.py devtools/benchmarks -m "not integration and not browser and not ui_browser and not ui_browser_docker and not portable_detail and not skill_smoke and not size_ratchet" --timeout=300 --timeout-method=thread -q --tb=short
 
   # ──────────────────────────────────────────────────────────────────
-  # Tier 2: Full matrix (stable branch, manual, or tag push)
+  # Tier 2: Ordinary tests on PR Windows/macOS; full 3-OS stable/manual/tag matrix
   # ──────────────────────────────────────────────────────────────────
   full-test:
     if: |
       github.ref == 'refs/heads/ouroboros-stable'
       || github.event_name == 'workflow_dispatch'
       || startsWith(github.ref, 'refs/tags/v')
+      || (github.event_name == 'pull_request' && github.base_ref == 'ouroboros')
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest, windows-latest, macos-latest]
+        os: ${{ fromJSON(github.event_name == 'pull_request' && '["windows-latest","macos-latest"]' || '["ubuntu-latest","windows-latest","macos-latest"]') }}
     runs-on: ${{ matrix.os }}
     env:
       # Same contract as quick-test: this job provisions the parallel-pass
@@ -243,7 +243,7 @@ jobs:
       - name: Run tests (serial — real subprocess/port/global-state suites that flake under -n)
         run: python -m pytest tests/ -m "serial and not integration and not browser and not ui_browser and not ui_browser_docker and not portable_detail and not skill_smoke and not size_ratchet" -q --tb=short
       # Same size-ratchet enforcement contract as quick-test (see its comment).
-      # full-test triggers: stable pushes carry event.before; dispatch runs leave
+      # PRs carry their base SHA; stable pushes carry event.before; dispatch runs leave
       # the env empty and tag pushes carry an all-zeros before — both degrade to
       # the tip's parent manifest (never a skip: a skip would let a recreated
       # branch grandfather debt in one green run).

ouroboros/platform_layer.py

@@ -296,9 +296,9 @@ def acquire_exclusive_file_lock(
                     for field in os.read(probe, 512).decode("utf-8", "replace").split():
                         if field.startswith("pid=") and field[4:].isdigit():
                             owner_pid = int(field[4:])
-                    stale = bool(judged) and (time.time() - judged[2] / 1e9) > stale_sec and not (
-                        owner_aware_stale and owner_pid > 0 and pid_is_alive(owner_pid)
-                    )
+                    stale = bool(judged) and (time.time() - judged[2] / 1e9) > stale_sec
+                    if judged and owner_aware_stale and owner_pid > 0:
+                        stale = not pid_is_alive(owner_pid)  # Proven death needs no age grace.
                     # Judge and evict the same inode under a kernel hold.
                     if stale and enforced:
                         try:
@@ -1078,6 +1078,11 @@ def pip_install_target_args(interpreter: str) -> List[str]:
     """Use the embedded interpreter's userbase; never add --user for a dev venv.
 
     Bundle-signature and target policy: ARCHITECTURE §1 CLI / Headless Boundary."""
+    invocation = pathlib.Path(interpreter)
+    if invocation.parent.name.lower() in {"bin", "scripts"} and (
+        invocation.parent.parent / "pyvenv.cfg"
+    ).is_file():
+        return []  # Resolve only after Python's lexical venv selection is known.
     return ["--user"] if interpreter_is_embedded(interpreter) else []
 
 

tests/test_platform_ci_events.py

@@ -0,0 +1,89 @@
+"""Ordinary desktop PR coverage at the existing CI matrix/event seam."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import yaml
+
+
+ROOT = Path(__file__).resolve().parents[1]
+WORKFLOW = yaml.safe_load((ROOT / ".github/workflows/ci.yml").read_text(encoding="utf-8"))
+
+
+def _value(expression, *, event, ref, base="", schedule=""):
+    """Evaluate the workflow's small expression vocabulary against real event shapes."""
+    expression = expression.strip()
+    if expression.startswith("$" + "{{"):
+        expression = expression[3:-2]
+    expression = " ".join(expression.split()).replace("&&", " and ").replace("||", " or ")
+    github = SimpleNamespace(
+        event_name=event, ref=ref, base_ref=base,
+        event=SimpleNamespace(
+            schedule=schedule, inputs=SimpleNamespace(e2e_live="false"), before="previous-tip",
+            pull_request=SimpleNamespace(base=SimpleNamespace(sha="pr-base")),
+        ),
+    )
+    return eval(expression, {"__builtins__": {}}, {
+        "github": github, "fromJSON": json.loads,
+        "startsWith": lambda value, prefix: value.startswith(prefix),
+    })
+
+
+@pytest.mark.parametrize("event,ref,base,quick,platforms", [
+    ("pull_request", "refs/pull/42/merge", "ouroboros", True, ["windows-latest", "macos-latest"]),
+    ("pull_request", "refs/pull/42/merge", "main", False, []),
+    ("push", "refs/heads/ouroboros", "", True, []),
+    ("push", "refs/heads/main", "", False, []),
+    ("push", "refs/heads/ouroboros-stable", "", False, ["ubuntu-latest", "windows-latest", "macos-latest"]),
+    ("push", "refs/tags/v7.0.0", "", False, ["ubuntu-latest", "windows-latest", "macos-latest"]),
+    ("workflow_dispatch", "refs/heads/candidate", "", True, ["ubuntu-latest", "windows-latest", "macos-latest"]),
+    ("schedule", "refs/heads/main", "", False, []),
+])
+def test_ordinary_matrix_expands_for_prs_without_changing_existing_events(event, ref, base, quick, platforms):
+    facts = {"event": event, "ref": ref, "base": base}
+    assert bool(_value(WORKFLOW["jobs"]["quick-test"]["if"], **facts)) is quick
+    job = WORKFLOW["jobs"]["full-test"]
+    admitted = _value(job["if"], **facts)
+    actual = _value(job["strategy"]["matrix"]["os"], **facts) if admitted else []
+    assert actual == platforms
+    assert job["strategy"]["fail-fast"] is False
+    assert job["runs-on"] == "$" + "{{ matrix.os }}"
+
+
+def test_desktop_pr_matrix_keeps_merge_checkout_and_pr_base_evidence_secret_free():
+    triggers = WORKFLOW.get("on") or WORKFLOW[True]  # PyYAML's YAML 1.1 spelling of "on".
+    assert triggers["pull_request"]["branches"] == ["ouroboros"]
+    assert "pull_request_target" not in triggers
+    assert WORKFLOW["permissions"] == {"contents": "read"}
+    job = WORKFLOW["jobs"]["full-test"]
+    assert "secrets." not in json.dumps(job)
+    assert not job.get("permissions")
+    checkout = job["steps"][0]
+    assert checkout["uses"] == "actions/checkout@v4"
+    assert checkout["with"] == {"fetch-depth": 0}  # Default PR checkout tests the merge ref.
+    base = next(step["env"]["OURO_SIZE_RATCHET_BASE_REF"] for step in job["steps"]
+                if "OURO_SIZE_RATCHET_BASE_REF" in step.get("env", {}))
+    assert _value(base, event="pull_request", ref="refs/pull/42/merge", base="ouroboros") == "pr-base"
+    assert _value(base, event="push", ref="refs/heads/ouroboros-stable") == "previous-tip"
+
+
+@pytest.mark.parametrize("cron", ["37 4 * * *", "17 3 * * *"])
+def test_scheduled_main_runs_do_not_enter_the_ordinary_matrix(cron):
+    for name in ("quick-test", "full-test"):
+        assert not _value(
+            WORKFLOW["jobs"][name]["if"], event="schedule", ref="refs/heads/main", schedule=cron,
+        )
+
+
+@pytest.mark.parametrize("name", [
+    "integration-test", "skill-smoke", "system-e2e-mock", "e2e-live",
+    "marker-guards", "docker-ui-smoke", "docker-portable-test",
+    "release-preflight", "build", "release", "vendor-package-smoke",
+])
+def test_desktop_pr_coverage_does_not_admit_provider_or_release_jobs(name):
+    job = WORKFLOW["jobs"][name]
+    assert not _value(job["if"], event="pull_request", ref="refs/pull/42/merge", base="ouroboros")

tests/test_platform_install_targets.py

@@ -0,0 +1,90 @@
+"""Pip targets the invoked venv even when its binary belongs to the bundle."""
+
+import json
+import os
+import pathlib
+import shutil
+import subprocess
+import sys
+import zipfile
+
+import pytest
+
+from ouroboros.launcher_bootstrap import embedded_python_env
+from ouroboros.platform_layer import pip_install_target_args
+
+
+@pytest.mark.parametrize("layout", ["bin/python3", "Scripts/python.exe"])
+@pytest.mark.parametrize("linked", [False, True], ids=["copied", "symlink"])
+def test_confirmed_venv_precedes_embedded_binary_resolution(tmp_path, layout, linked):
+    bundled = tmp_path / "python-standalone" / "bin" / "python3"
+    bundled.parent.mkdir(parents=True)
+    bundled.write_bytes(b"interpreter fixture")
+    # A copied venv under this ancestor also defeats ancestry-only classification.
+    venv = tmp_path / "python-standalone" / "work-environment"
+    invocation = venv / pathlib.Path(layout)
+    invocation.parent.mkdir(parents=True)
+    (venv / "pyvenv.cfg").write_text("home = bundled\n", encoding="utf-8")
+    if linked:
+        try:
+            invocation.symlink_to(bundled)
+        except OSError as exc:
+            pytest.skip(f"symlink creation unavailable: {exc}")
+    else:
+        shutil.copyfile(bundled, invocation)
+    assert pip_install_target_args(str(invocation)) == []
+
+
+def test_venv_name_and_ambient_env_do_not_override_direct_bundle(tmp_path, monkeypatch):
+    bundled = tmp_path / "python-standalone" / ".venv" / "bin" / "python3"
+    bundled.parent.mkdir(parents=True)
+    bundled.write_bytes(b"interpreter fixture")
+    monkeypatch.setenv("VIRTUAL_ENV", str(tmp_path / "unrelated-env"))
+    assert pip_install_target_args(str(bundled)) == ["--user"]
+    alias = tmp_path / "python-alias"
+    try:
+        alias.symlink_to(bundled)
+    except OSError as exc:
+        pytest.skip(f"symlink creation unavailable: {exc}")
+    assert pip_install_target_args(str(alias)) == ["--user"]
+
+
+def test_plain_python_and_nonstandard_cfg_neighbor(tmp_path):
+    plain = tmp_path / "usr" / "bin" / "python3"
+    assert pip_install_target_args(str(plain)) == []
+    bundled = tmp_path / "python-standalone" / "python.exe"
+    bundled.parent.mkdir()
+    bundled.write_bytes(b"interpreter fixture")
+    (bundled.parent / "pyvenv.cfg").write_text("home = elsewhere\n", encoding="utf-8")
+    assert pip_install_target_args(str(bundled)) == ["--user"]
+
+
+@pytest.mark.serial
+def test_real_venv_installs_and_imports_local_wheel(tmp_path):
+    venv = tmp_path / "work-environment"
+    env = embedded_python_env(tmp_path / "data")
+    env["PYTHONDONTWRITEBYTECODE"] = "1"
+    subprocess.run([sys.executable, "-m", "venv", str(venv)], env=env, check=True, timeout=90)
+    interpreter = venv / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
+    wheel = tmp_path / "ouroboros_venv_probe-1.0-py3-none-any.whl"
+    with zipfile.ZipFile(wheel, "w") as archive:
+        archive.writestr("ouroboros_venv_probe.py", "VALUE = 'local wheel imported'\n")
+        archive.writestr("ouroboros_venv_probe-1.0.dist-info/METADATA",
+                         "Metadata-Version: 2.1\nName: ouroboros-venv-probe\nVersion: 1.0\n")
+        archive.writestr("ouroboros_venv_probe-1.0.dist-info/WHEEL",
+                         "Wheel-Version: 1.0\nRoot-Is-Purelib: true\nTag: py3-none-any\n")
+        archive.writestr("ouroboros_venv_probe-1.0.dist-info/RECORD", "")
+    subprocess.run(
+        [str(interpreter), "-m", "pip", "install", *pip_install_target_args(str(interpreter)),
+         "--no-index", "--no-deps", "--no-cache-dir", str(wheel)],
+        env=env, check=True, capture_output=True, text=True, timeout=60,
+    )
+    result = subprocess.run(
+        [str(interpreter), "-c", "import json,sys,ouroboros_venv_probe as p; "
+         "print(json.dumps([sys.prefix,p.__file__,p.VALUE]))"],
+        env=env, check=True, capture_output=True, text=True, timeout=10,
+    )
+    prefix, module, value = json.loads(result.stdout)
+    assert pathlib.Path(prefix).resolve() == venv.resolve()
+    assert venv.resolve() in pathlib.Path(module).resolve().parents
+    assert value == "local wheel imported"

tests/test_platform_lock_recovery.py

@@ -0,0 +1,62 @@
+"""Owner death, age and kernel ownership are separate lock observations."""
+
+import os
+import subprocess
+import sys
+import time
+
+import pytest
+
+from ouroboros import platform_layer as platform
+
+
+@pytest.mark.parametrize("age", [0, 120], ids=["fresh", "old"])
+def test_dead_owner_recovery_keeps_kernel_exclusion(tmp_path, monkeypatch, age):
+    lock = tmp_path / "state.lock"
+    fd = platform.acquire_exclusive_file_lock(lock, metadata="pid=424242\n")
+    assert fd is not None
+    try:
+        os.utime(lock, (time.time() - age,) * 2)
+        monkeypatch.setattr(platform, "pid_is_alive", lambda pid: False)
+        assert platform.acquire_exclusive_file_lock(
+            lock, timeout_sec=0.1, stale_sec=90, owner_aware_stale=True,
+        ) is None
+        assert lock.read_text() == "pid=424242\n"
+    finally:
+        platform.release_exclusive_file_lock(lock, fd)
+
+
+@pytest.mark.parametrize("metadata,owner_aware", [("unknown\n", True), ("pid=424242\n", False)])
+def test_unproven_or_unrequested_owner_recovery_keeps_age_grace(tmp_path, monkeypatch, metadata, owner_aware):
+    lock = tmp_path / "state.lock"
+    lock.write_text(metadata)
+    monkeypatch.setattr(platform, "pid_is_alive", lambda pid: False)
+    assert platform.acquire_exclusive_file_lock(
+        lock, timeout_sec=0.1, stale_sec=90, owner_aware_stale=owner_aware,
+    ) is None
+    assert lock.read_text() == metadata
+    os.utime(lock, (0, 0))
+    fd = platform.acquire_exclusive_file_lock(
+        lock, timeout_sec=1, stale_sec=90, owner_aware_stale=owner_aware,
+    )
+    assert fd is not None
+    platform.release_exclusive_file_lock(lock, fd)
+
+
+@pytest.mark.serial
+def test_fresh_lock_of_reaped_process_is_recovered_within_caller_budget(tmp_path):
+    child = subprocess.Popen([sys.executable, "-c", "pass"])
+    child.wait(timeout=10)
+    assert not platform.pid_is_alive(child.pid)
+    lock = tmp_path / "usage_attempts.lock"
+    lock.write_text(f"pid={child.pid}\n")
+    started = time.monotonic()
+    fd = platform.acquire_exclusive_file_lock(
+        lock, timeout_sec=2, stale_sec=90, owner_aware_stale=True,
+    )
+    assert fd is not None
+    try:
+        assert time.monotonic() - started < 2
+        assert f"pid={os.getpid()}" in lock.read_text()
+    finally:
+        platform.release_exclusive_file_lock(lock, fd)

tests/test_platform_paid_response_recovery.py

@@ -0,0 +1,68 @@
+"""A ledger lock failure must preserve the response already paid for."""
+
+import asyncio
+import errno
+import json
+
+import pytest
+
+from ouroboros import platform_layer, usage_accounting, usage_ledger
+
+
+@pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"])
+def test_paid_response_survives_kernel_lock_refusal(
+    tmp_path, monkeypatch, caplog, asynchronous,
+):
+    root = tmp_path / "data"
+    (root / "state").mkdir(parents=True)
+    monkeypatch.setenv("OUROBOROS_DATA_DIR", str(root))
+    monkeypatch.setenv("OUROBOROS_SETTINGS_PATH", str(root / "settings.json"))
+    monkeypatch.setenv("TOTAL_BUDGET", "100")
+    usage_accounting._reset_task_cache_splits()
+    request = usage_accounting.AttemptRequest(
+        model="openai/gpt-5.2", provider="openai", reservation_usd=1.0,
+        drive_root=root, task_id="child", root_task_id="root", source="test",
+    )
+    assert platform_layer.kernel_file_locks_enforced(root / usage_ledger.LOCK_REL)
+    response = {"content": "useful result", "usage": {
+        "prompt_tokens": 3, "completion_tokens": 2,
+    }}
+    sends = 0
+    refused = []
+    actual_kernel_lock = platform_layer.file_lock_exclusive_nb
+
+    def refuse_after_response(fd):
+        if sends:
+            refused.append(fd)
+            raise OSError(errno.EIO, "ledger kernel lock unavailable after response")
+        return actual_kernel_lock(fd)
+
+    def send():
+        nonlocal sends
+        sends += 1
+        return response
+
+    async def send_async():
+        return send()
+
+    with monkeypatch.context() as failure:
+        failure.setattr(platform_layer, "file_lock_exclusive_nb", refuse_after_response)
+        if asynchronous:
+            actual = asyncio.run(usage_accounting.execute_physical_attempt_async(
+                request, send_async,
+            ))
+        else:
+            actual = usage_accounting.execute_physical_attempt(request, send)
+
+    assert actual is response
+    assert sends == 1
+    assert len(refused) >= 2  # Settlement and the fallback unresolved write both failed.
+    assert "Failed to mark post-response accounting failure unresolved" in caplog.text
+    rows = [json.loads(line) for line in (root / usage_ledger.LEDGER_REL).read_text().splitlines()]
+    assert [row["state"] for row in rows] == ["reserved", "dispatched"]
+    projection = usage_accounting.usage_projection(root)
+    assert projection["unresolved_upper_bound_usd"] == 1.0
+    assert projection["cost_final"] is False
+    assert not (root / usage_ledger.LOCK_REL).exists()
+    assert len({row["attempt_id"] for row in rows}) == 1
+    usage_accounting._reset_task_cache_splits()

tests/test_ui_fixture_lifecycle.py

@@ -0,0 +1,312 @@
+"""The shared UI fixture owns each spawned tree through readiness, restart and teardown.
+
+The sleeper payload keeps process containment real without starting a server or
+requiring browser binaries. Real UI consumers run in their existing marker lane.
+"""
+
+from __future__ import annotations
+
+from contextlib import nullcontext, suppress
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from ouroboros import platform_layer as pl, process_containment as pc
+from tests import test_ui_smoke_playwright as ui
+
+
+pytestmark = pytest.mark.serial
+
+_TREE_SCRIPT = r"""
+import json, os, pathlib, subprocess, sys, time
+receipt, entered, ready = map(pathlib.Path, sys.argv[1:4])
+entered.write_text("entered", encoding="utf-8")
+child = subprocess.Popen(
+    [sys.executable, "-c",
+     "import pathlib,sys,time; pathlib.Path(sys.argv[1]).write_text('ready', encoding='utf-8'); time.sleep(90)",
+     str(ready)],
+    stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    start_new_session=os.name != "nt",
+)
+pending = receipt.with_suffix(".pending")
+pending.write_text(json.dumps({"parent": os.getpid(), "child": child.pid}), encoding="utf-8")
+pending.replace(receipt)
+if sys.argv[4] == "exited":
+    sys.exit(0)
+time.sleep(90)
+"""
+
+
+def _wait(predicate, detail, timeout=15):
+    deadline = time.monotonic() + timeout
+    while not predicate():
+        if time.monotonic() >= deadline:
+            raise AssertionError(detail)
+        time.sleep(0.05)
+
+
+def _gone(pid):
+    # A reparented POSIX zombie cannot execute, even before init has waited it.
+    return not pl.pid_is_alive(pid) or pc.pid_is_zombie(pid)
+
+
+def _assert_stopped(probe, run):
+    _wait(lambda: _gone(run.proc.pid) and _gone(run.child), "fixture-owned tree survived")
+    assert run.proc.poll() is not None
+    assert ("reap", run.index) in probe.events
+    assert ("close", run.index) in probe.events
+    assert probe.sentinel.poll() is None, "cleanup killed an unrelated process"
+
+
+@pytest.fixture
+def fixture_probe(tmp_path, monkeypatch):
+    """Use a sleeper tree at the fixture's one spawn seam; keep containment real."""
+    original_container = pc.ProcessContainer
+    probe = SimpleNamespace(
+        mode="alive", fail_at="", reap_error="", reap_exception=False,
+        runs=[], containers=[], generators=[], events=[], prior_gone=[],
+    )
+    sentinel_ready = tmp_path / "sentinel-ready"
+    probe.sentinel = subprocess.Popen(
+        [sys.executable, "-c",
+         "import pathlib,sys; pathlib.Path(sys.argv[1]).write_text('ready', encoding='utf-8'); sys.stdin.read()",
+         str(sentinel_ready)],
+        stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        **pl.subprocess_new_group_kwargs(),
+    )
+
+    class ObservedContainer(original_container):
+        def __init__(self):
+            super().__init__()
+            self.index = len(probe.containers)
+            self.initial_token = dict(self.containment_env())
+            probe.containers.append(self)
+
+        def spawn(self, argv, **kwargs):
+            assert Path(argv[1]).name == "server.py", argv
+            if probe.runs:
+                previous = probe.runs[-1]
+                probe.prior_gone.append(
+                    _gone(previous.proc.pid) and _gone(previous.child)
+                )
+            run = SimpleNamespace(
+                index=self.index, container=self, proc=None, child=0,
+                receipt=tmp_path / f"tree-{len(probe.runs)}.json",
+                entered=tmp_path / f"entered-{len(probe.runs)}",
+                ready=tmp_path / f"ready-{len(probe.runs)}",
+            )
+            probe.runs.append(run)
+            probe.events.append(("spawn", self.index))
+            run.proc = super().spawn(
+                [sys.executable, "-c", _TREE_SCRIPT,
+                 str(run.receipt), str(run.entered), str(run.ready), probe.mode],
+                **kwargs,
+            )
+            return run.proc
+
+        def reap(self):
+            probe.events.append(("reap", self.index))
+            actual = super().reap()
+            if actual:
+                return actual
+            if probe.reap_exception:
+                raise RuntimeError("injected reap exception")
+            return probe.reap_error
+
+        def close(self):
+            probe.events.append(("close", self.index))
+            return super().close()
+
+    def health(_url):
+        # The OLD bare-Popen fixture must fail without accidentally starting a
+        # real server. Its direct Popen is refused below; only this container's
+        # script may start.
+        assert probe.runs, "the UI fixture did not use ProcessContainer.spawn"
+        run = probe.runs[-1]
+        _wait(lambda: run.receipt.exists() and run.ready.exists(), "sleeper tree did not start")
+        info = json.loads(run.receipt.read_text(encoding="utf-8"))
+        assert info["parent"] == run.proc.pid
+        run.child = info["child"]
+        assert not _gone(run.child), "child must be live before teardown"
+        if probe.mode == "exited":
+            assert run.proc.wait(timeout=15) == 0
+        else:
+            assert run.proc.poll() is None
+        if probe.fail_at == "health":
+            raise RuntimeError("injected health failure")
+
+    def supervisor(_url):
+        if probe.fail_at == "supervisor":
+            raise RuntimeError("injected supervisor failure")
+
+    original_popen = subprocess.Popen
+
+    def no_real_server(argv, *args, **kwargs):
+        if isinstance(argv, (list, tuple)) and len(argv) > 1:
+            assert Path(str(argv[1])).name != "server.py", "uncontained real server spawn"
+        proc = original_popen(argv, *args, **kwargs)
+        if isinstance(argv, list) and len(argv) > 2 and argv[2] == _TREE_SCRIPT:
+            # Retain custody even if a later native assignment/resume step raises
+            # before ProcessContainer.spawn has returned its Popen to the caller.
+            probe.runs[-1].proc = proc
+        return proc
+
+    def open_fixture():
+        gen = ui.direct_server_with_data.__wrapped__(tmp_path / f"fixture-{len(probe.generators)}")
+        probe.generators.append(gen)
+        return gen, next(gen)
+
+    monkeypatch.setenv("OUROBOROS_RUN_UI_SMOKE", "1")
+    monkeypatch.setattr(pc, "ProcessContainer", ObservedContainer)
+    monkeypatch.setattr(ui, "MockLLMServer", lambda: nullcontext(
+        SimpleNamespace(base_url="http://127.0.0.1:9/v1")
+    ))
+    monkeypatch.setattr(ui, "_free_port", lambda: 27991)  # no port is actually bound
+    monkeypatch.setattr(ui, "_wait_health", health)
+    monkeypatch.setattr(ui, "_wait_supervisor_ready", supervisor)
+    monkeypatch.setattr(subprocess, "Popen", no_real_server)
+    probe.open = open_fixture
+    try:
+        _wait(sentinel_ready.exists, "unrelated sentinel did not start")
+        yield probe
+    finally:
+        # This runs only after test assertions. It cleans our own exact children
+        # if a regression made the generator's cleanup fail; never sweeps argv,
+        # filenames, global process lists, or another test's historical runs.
+        try:
+            for gen in probe.generators:
+                with suppress(RuntimeError, pytest.fail.Exception):
+                    gen.close()
+            for run in probe.runs:
+                if run.receipt.exists():
+                    run.child = json.loads(run.receipt.read_text(encoding="utf-8"))["child"]
+                try:
+                    original_container.reap(run.container)
+                finally:
+                    original_container.close(run.container)
+                    if run.child and not _gone(run.child):
+                        pl.force_kill_pid(run.child)
+                    if run.proc is not None:
+                        if run.proc.poll() is None:
+                            run.proc.kill()
+                        run.proc.wait(timeout=15)
+        finally:
+            if probe.sentinel.stdin is not None:
+                probe.sentinel.stdin.close()
+            if probe.sentinel.poll() is None:
+                probe.sentinel.terminate()
+            probe.sentinel.wait(timeout=15)
+
+
+@pytest.mark.parametrize("mode", ["exited", "alive"])
+def test_ui_fixture_reaps_descendants_after_parent_exit_or_normal_stop(fixture_probe, mode):
+    probe = fixture_probe
+    probe.mode = mode
+    gen, _ = probe.open()
+    run = probe.runs[0]
+    assert (run.proc.poll() is not None) == (mode == "exited")
+    gen.close()
+    _assert_stopped(probe, run)
+
+
+@pytest.mark.parametrize("stage", ["health", "supervisor"])
+def test_ui_fixture_reaps_after_readiness_failure(fixture_probe, stage):
+    probe = fixture_probe
+    probe.fail_at = stage
+    with pytest.raises(RuntimeError, match=f"injected {stage} failure"):
+        probe.open()
+    assert len(probe.runs) == 1
+    _assert_stopped(probe, probe.runs[0])
+
+
+def test_ui_fixture_restart_retires_one_container_before_spawning_another(fixture_probe):
+    probe = fixture_probe
+    gen, fixture = probe.open()
+    first = probe.runs[0]
+    fixture["restart_server"]()
+    assert len(probe.runs) == len(probe.containers) == 2
+    second = probe.runs[1]
+    assert first.container.initial_token != second.container.initial_token
+    assert probe.prior_gone == [True]
+    assert probe.events.index(("close", 0)) < probe.events.index(("spawn", 1))
+    _assert_stopped(probe, first)
+    assert second.proc.poll() is None and not _gone(second.child)
+    gen.close()
+    _assert_stopped(probe, second)
+
+
+@pytest.mark.parametrize("raises", [False, True], ids=["returned-error", "raised-error"])
+@pytest.mark.parametrize("operation", ["teardown", "restart"])
+def test_ui_fixture_surfaces_reap_failure_and_still_closes(fixture_probe, raises, operation):
+    probe = fixture_probe
+    gen, fixture = probe.open()
+    probe.reap_exception = raises
+    probe.reap_error = "injected reap failure"
+    with pytest.raises((RuntimeError, pytest.fail.Exception), match="injected reap"):
+        if operation == "teardown":
+            gen.close()
+        else:
+            fixture["restart_server"]()
+    # A failed retirement cannot start a second generation or erase the error.
+    assert len(probe.runs) == 1
+    _assert_stopped(probe, probe.runs[0])
+    # The generator still reaches its outer finally after restart failed.
+    # Root may keep or clear active references; either way this cannot spawn.
+    with suppress(RuntimeError, pytest.fail.Exception):
+        gen.close()
+    assert len(probe.runs) == 1
+
+
+@pytest.mark.skipif(os.name != "nt", reason="requires actual Windows Job and suspended-start APIs")
+def test_ui_fixture_native_windows_assigns_suspended_root_then_reaps_orphan(fixture_probe, monkeypatch):
+    probe = fixture_probe
+    probe.mode = "exited"
+    native_popen = subprocess.Popen
+    native_assign = pl.assign_pid_to_job
+    native_resume = pl.resume_process
+    ordering = []
+    assigned = set()
+    observations = []
+
+    def popen(argv, **kwargs):
+        if isinstance(argv, list) and len(argv) > 2 and argv[2] == _TREE_SCRIPT:
+            assert int(kwargs.get("creationflags", 0)) & 0x4, "root was not CREATE_SUSPENDED"
+            ordering.append("popen")
+        return native_popen(argv, **kwargs)
+
+    def assign(job, pid):
+        observations.append(("before_assign", probe.runs[-1].entered.exists()))
+        result = native_assign(job, pid)
+        observations.append(("assigned", result))
+        if result:
+            assigned.add(pid)
+        ordering.append("assign")
+        return result
+
+    def resume(pid):
+        observations.append(("before_resume", pid in assigned, probe.runs[-1].entered.exists()))
+        result = native_resume(pid)
+        observations.append(("resumed", result))
+        ordering.append("resume")
+        return result
+
+    monkeypatch.setattr(subprocess, "Popen", popen)
+    monkeypatch.setattr(pl, "assign_pid_to_job", assign)
+    monkeypatch.setattr(pl, "resume_process", resume)
+    gen, _ = probe.open()
+    run = probe.runs[0]
+    assert ordering == ["popen", "assign", "resume"]
+    assert observations == [
+        ("before_assign", False), ("assigned", True),
+        ("before_resume", True, False), ("resumed", True),
+    ]
+    assert run.entered.read_text(encoding="utf-8") == "entered"
+    assert run.proc.poll() == 0 and not _gone(run.child)
+    gen.close()
+    _assert_stopped(probe, run)

tests/test_ui_smoke_playwright.py

@@ -267,47 +267,44 @@ def direct_server_with_data(tmp_path):
             "OUROBOROS_NETWORK_PASSWORD": "ui-smoke-password",
         }
         url = f"http://127.0.0.1:{port}"
-        active_proc = None
+        active_proc = active_container = None
 
         def stop_server() -> None:
-            nonlocal active_proc
-            if active_proc is None or active_proc.poll() is not None:
+            nonlocal active_proc, active_container
+            proc, container = active_proc, active_container
+            active_proc = active_container = None
+            if container is None:
                 return
-            from ouroboros.platform_layer import IS_WINDOWS, kill_process_tree
-
-            # Windows terminate() is an immediate TerminateProcess, so the parent
-            # can disappear before its worker tree and bypass the timeout cleanup.
-            # taskkill /T must own that path from the start.
-            if IS_WINDOWS:
-                kill_process_tree(active_proc)
-                active_proc.wait(timeout=5)
-                active_proc = None
-                return
-            active_proc.terminate()
             try:
-                active_proc.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                # A timed-out UI-smoke server still owns its worker pool. Killing
-                # only the parent leaks ten orphan workers into later smoke tests,
-                # producing suite-order history/card timeouts. The server starts in
-                # its own process group below, so the shared cross-platform helper
-                # can close the complete tree without touching pytest.
-                kill_process_tree(active_proc)
-                active_proc.wait(timeout=5)
+                if proc is not None and proc.poll() is None:
+                    proc.terminate()
+                    try:
+                        proc.wait(timeout=10)
+                    except subprocess.TimeoutExpired:
+                        pass  # The container below also owns surviving descendants.
             finally:
-                active_proc = None
+                try:
+                    # Parent exit never proves the entire incarnation is gone.
+                    error = container.reap()
+                    if error:
+                        raise RuntimeError(f"UI fixture process cleanup failed: {error}")
+                finally:
+                    container.close()
+                    if proc is not None:
+                        proc.wait(timeout=5)
 
         def start_server() -> None:
-            nonlocal active_proc
-            from ouroboros.platform_layer import subprocess_new_group_kwargs
+            nonlocal active_proc, active_container
+            from ouroboros.process_containment import ProcessContainer
 
-            active_proc = subprocess.Popen(
+            # Reap consumes the token/Job: every restart needs fresh containment.
+            active_container = ProcessContainer()
+            active_proc = active_container.spawn(
                 [sys.executable, "server.py"],
                 cwd=REPO_ROOT,
                 env=env,
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
-                **subprocess_new_group_kwargs(),
             )
             _wait_health(url)
             _wait_supervisor_ready(url)