fix: budget pytest tmpfs from measured cgroup overhead

CLAUDE.md

@@ -509,9 +509,15 @@ isolated XDG paths + archive root.
   exceptions opt out inline via `@pytest.mark.uses_real_clock("reason")`.
 - Pytest temp DBs pick ONE basetemp root via
   `devtools.verify_runs.resolve_pytest_basetemp_root` (shared by
-  `tests/conftest.py` and the `devtools test`/`verify` preflight): `/dev/shm`
-  tmpfs by default when it has ≥1 GiB free (`POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB`
-  to override), else `/realm/tmp/polylogue-pytest` (NVMe), else `/tmp/polylogue-pytest`
+  `tests/conftest.py` and the `devtools test`/`verify` preflight): focused runs
+  use bounded `/dev/shm` tmpfs when it has ≥1 GiB free, while full-suite and
+  seed-testmon runs default to `/realm/tmp/polylogue-pytest` (NVMe) because
+  their aggregate fixture tree can exceed the supervised tmpfs ceiling.
+  `POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB` overrides required headroom; an
+  explicit `POLYLOGUE_PYTEST_TMPFS=1` requests bounded tmpfs, but the request
+  is honored only when the effective budget satisfies the declared basetemp
+  requirement.
+  `/tmp/polylogue-pytest` is used
   only when `/realm/tmp` genuinely isn't mounted (cloud sandbox). If nothing
   clears the headroom requirement the run refuses immediately with every
   candidate's free space named, instead of an unrelated command crashing on

TESTING.md

@@ -94,11 +94,14 @@ independent placement policy that can silently disagree with this one:
 
 1. `POLYLOGUE_PYTEST_BASETEMP_ROOT=/path` — an explicit operator override,
    still headroom-checked (see below), never silently downgraded.
-2. `/dev/shm` (tmpfs) — the default, because measured SQLite fsync traffic
-   made the disk-backed lane more than 20 times slower — used when it clears
-   the free-space requirement.
-3. `/realm/tmp/polylogue-pytest` (NVMe scratch) — used when `/dev/shm` lacks
-   headroom but `/realm/tmp` is mounted and has room.
+2. `/dev/shm` (tmpfs) — the focused-run default, because measured SQLite fsync
+   traffic makes it substantially faster when it clears the free-space
+   requirement. Full-suite and seed-testmon runs use it only when
+   `POLYLOGUE_PYTEST_TMPFS=1` is explicit.
+3. `/realm/tmp/polylogue-pytest` (NVMe scratch) — the broad-run default, and
+   the fallback when `/dev/shm` lacks headroom. Broad fixture trees have
+   exceeded the supervised 2 GiB tmpfs ceiling while still making progress,
+   so their normal route does not guess a future aggregate peak.
 4. `/tmp/polylogue-pytest` — reachable **only** when `/realm/tmp` is not
    mounted at all (a genuine cloud sandbox, where `.claude/settings.json`
    sets this as `POLYLOGUE_PYTEST_BASETEMP_ROOT`). On a workstation with
@@ -120,10 +123,12 @@ MiB to 2 GiB) once a tmpfs root is chosen. Per-run `pytest-polylogue-*`
 basetemps are removed at normal pytest shutdown, and pytest startup sweeps
 stale per-run dirs from every known root (`/dev/shm`, `/realm/tmp/polylogue-pytest`,
 `/tmp/polylogue-pytest`, plus any explicit configured root) — never based on
-age alone: each managed basetemp carries an owner-pid marker, and a
-directory whose owner process is still alive is never removed regardless of
-age; an owner that cannot be confirmed dead (no marker) gets a multi-hour
-grace period rather than the normal ~30-minute one. Shared
+age alone: each managed basetemp carries a PID plus process-start identity,
+and a directory whose exact owner process is still alive is never removed
+regardless of age. An owner that cannot be confirmed dead (no marker) gets a
+multi-hour grace period rather than the normal ~30-minute one. The sweeper
+restores owner-write permission only after a tree is adjudicated stale, so
+published read-only fixture copies cannot leak tmpfs indefinitely. Shared
 `pytest-polylogue-*-seeded-*` caches are never touched by the sweep — they
 are shared, reused, and built once behind their own `.build.done` guard.
 

devtools/run_tests.py

@@ -47,26 +47,12 @@
     _clear_pytest_report,
     _run,
 )
-from devtools.verify_runs import VerifyRun, apply_managed_pytest_runtime_policy, git_head
+from devtools.verify_runs import VerifyRun, git_head
 
 ROOT = Path(__file__).resolve().parent.parent
 _LOCK_PATH = ROOT / ".cache" / "test-run.lock"
 
 
-def _managed_env() -> dict[str, str]:
-    """Mirror devtools.verify's subprocess environment for parity."""
-    env, _policy = apply_managed_pytest_runtime_policy(os.environ)
-    env["POLYLOGUE_ROOT"] = str(ROOT)
-    env["POLYLOGUE_REPO_ROOT"] = str(ROOT)
-    inherited_pythonpath = env.get("PYTHONPATH", "")
-    env["PYTHONPATH"] = str(ROOT) if not inherited_pythonpath else f"{ROOT}{os.pathsep}{inherited_pythonpath}"
-    env["PYTHONPYCACHEPREFIX"] = str(ROOT / ".cache" / "pycache")
-    env["POLYLOGUE_PYTEST_EVENTS_PATH"] = str(ROOT / PYTEST_EVENTS_PATH)
-    env["POLYLOGUE_PYTEST_SELECTION_PATH"] = str(ROOT / PYTEST_SELECTION_PATH)
-    env["POLYLOGUE_PYTEST_SUMMARY_PATH"] = str(ROOT / PYTEST_SUMMARY_PATH)
-    return env
-
-
 def _has_worker_flag(selection: list[str]) -> bool:
     """True when the caller already chose an xdist worker count."""
     return any(arg.startswith(("-n", "--numprocesses")) for arg in selection)

devtools/verify.py

@@ -87,6 +87,7 @@
     cleanup_managed_pytest_basetemp,
     copy_current_pytest_artifacts,
     env_for_pytest_step,
+    force_managed_pytest_scratch,
     latest_event_from_paths,
     merge_worker_events,
     normalize_pytest_basetemp_env,
@@ -318,10 +319,16 @@ def _read_json_artifact(path: Path) -> dict[str, Any] | None:
     return raw if isinstance(raw, dict) else None
 
 
-def _read_latest_pytest_event(path: Path = PYTEST_EVENTS_PATH) -> dict[str, Any] | None:
+def _read_latest_pytest_event(
+    path: Path = PYTEST_EVENTS_PATH,
+    *,
+    events_dir: Path | None = None,
+) -> dict[str, Any] | None:
     """Return the latest valid pytest event from the live JSONL ledger."""
+    if events_dir is not None:
+        return latest_event_from_paths(events_dir, path)
     if path == PYTEST_EVENTS_PATH:
-        return latest_event_from_paths(PYTEST_EVENTS_DIR, PYTEST_EVENTS_PATH)
+        return latest_event_from_paths(PYTEST_EVENTS_DIR, path)
     try:
         with path.open("rb") as handle:
             handle.seek(0, os.SEEK_END)
@@ -364,12 +371,7 @@ def _pytest_metadata_from_report(report: dict[str, Any], *, report_path: Path) -
 def _pytest_command_metadata(cmd: list[str]) -> dict[str, Any]:
     """Return verify metadata that explains the pytest worker policy."""
     metadata: dict[str, Any] = {}
-    if "-n" in cmd:
-        index = cmd.index("-n")
-        if index + 1 < len(cmd):
-            metadata["pytest_workers"] = cmd[index + 1]
-    else:
-        metadata["pytest_workers"] = "unset"
+    metadata["pytest_workers"] = _pytest_command_worker_request(cmd) or "unset"
     if "--testmon" in cmd:
         metadata["pytest_selection"] = "testmon-noselect" if "--testmon-noselect" in cmd else "testmon"
     else:
@@ -611,6 +613,8 @@ def _write_pytest_progress(
     artifact_dir: str | None = None,
     resources: Mapping[str, Any] | None = None,
     containment: Mapping[str, Any] | None = None,
+    events_path: Path = PYTEST_EVENTS_PATH,
+    events_dir: Path | None = None,
 ) -> None:
     """Write a live pytest progress artifact for long verify runs."""
     if elapsed_s is None:
@@ -644,7 +648,7 @@ def _write_pytest_progress(
         payload["resources"] = dict(resources)
     if containment is not None:
         payload["containment"] = dict(containment)
-    latest_event = _read_latest_pytest_event()
+    latest_event = _read_latest_pytest_event(events_path, events_dir=events_dir)
     if latest_event is not None:
         payload["latest_test_event"] = {
             key: latest_event[key]
@@ -908,6 +912,8 @@ def _run_pytest_with_heartbeat(
     term_grace_s = _pytest_term_grace_s()
     resource_interval_s = _pytest_resource_interval_s()
     tmpfs_budget_kb = pytest_tmpfs_budget_kb(env)
+    events_path = Path(env.get("POLYLOGUE_PYTEST_EVENTS_PATH", str(PYTEST_EVENTS_PATH)))
+    events_dir = Path(env.get("POLYLOGUE_PYTEST_EVENTS_DIR", str(PYTEST_EVENTS_DIR)))
     runner_subreaper_enabled = enable_child_subreaper()
     preserved_runner_descendants = tuple(descendant_process_identities(os.getpid()))
     receipt_path = (
@@ -1130,6 +1136,8 @@ def _stop_startup_attempt(
         run_id=run.run_id if run is not None else None,
         artifact_dir=str(artifacts.step_dir) if artifacts is not None else None,
         containment=_containment_summary(launch, startup_receipt),
+        events_path=events_path,
+        events_dir=events_dir,
     )
     selector = selectors.DefaultSelector()
     selector.register(stdout_pipe, selectors.EVENT_READ, "stdout")
@@ -1156,7 +1164,7 @@ def _stop_startup_attempt(
     # latest test event's own updated_at timestamp across all workers
     # (devtools/pytest_progress_plugin.py); last_progress_at is the local
     # monotonic time that marker was last seen to change.
-    initial_event = _read_latest_pytest_event()
+    initial_event = _read_latest_pytest_event(events_path, events_dir=events_dir)
     last_progress_marker: str | None = initial_event.get("updated_at") if initial_event is not None else None
     last_progress_at = last_sample
     seen_any_progress_event = initial_event is not None
@@ -1165,7 +1173,7 @@ def _stop_startup_attempt(
     def _refresh_progress_marker(at: float, latest: dict[str, Any] | None = None) -> None:
         nonlocal last_progress_marker, last_progress_at, seen_any_progress_event
         if latest is None:
-            latest = _read_latest_pytest_event()
+            latest = _read_latest_pytest_event(events_path, events_dir=events_dir)
         if latest is None:
             return
         marker = latest.get("updated_at")
@@ -1309,6 +1317,8 @@ def _refresh_progress_marker(at: float, latest: dict[str, Any] | None = None) ->
                             run_id=run.run_id if run is not None else None,
                             artifact_dir=str(artifacts.step_dir) if artifacts is not None else None,
                             containment=_containment_summary(launch, receipt),
+                            events_path=events_path,
+                            events_dir=events_dir,
                         )
                     else:
                         selector.unregister(selector_key.fileobj)
@@ -1325,7 +1335,7 @@ def _refresh_progress_marker(at: float, latest: dict[str, Any] | None = None) ->
                 rss_text = f", rss={int(rss) // 1024} MiB" if isinstance(rss, int) else ""
                 cpu_text = f", cpu={cpu_pct:.0f}%" if cpu_pct is not None else ""
                 state_text = f", state={status['state']}" if status["state"] is not None else ""
-                latest_event = _read_latest_pytest_event()
+                latest_event = _read_latest_pytest_event(events_path, events_dir=events_dir)
                 _refresh_progress_marker(sample_now, latest_event)
                 if latest_event is not None:
                     event = latest_event.get("event")
@@ -1359,6 +1369,8 @@ def _refresh_progress_marker(at: float, latest: dict[str, Any] | None = None) ->
                     run_id=run.run_id if run is not None else None,
                     artifact_dir=str(artifacts.step_dir) if artifacts is not None else None,
                     containment=_containment_summary(launch, receipt),
+                    events_path=events_path,
+                    events_dir=events_dir,
                 )
             sample_now = time.monotonic()
             if (
@@ -1446,6 +1458,8 @@ def _refresh_progress_marker(at: float, latest: dict[str, Any] | None = None) ->
             artifact_dir=str(artifacts.step_dir) if artifacts is not None else None,
             resources=resource_summary,
             containment=containment,
+            events_path=events_path,
+            events_dir=events_dir,
         )
     else:
         _write_pytest_progress(
@@ -1459,6 +1473,8 @@ def _refresh_progress_marker(at: float, latest: dict[str, Any] | None = None) ->
             artifact_dir=str(artifacts.step_dir) if artifacts is not None else None,
             resources=resource_summary,
             containment=containment,
+            events_path=events_path,
+            events_dir=events_dir,
         )
     _write_pytest_output(stdout, stderr)
     if artifacts is not None:
@@ -1479,17 +1495,34 @@ def _run(
     sys.stderr.write(f"  {label} ... ")
     sys.stderr.flush()
     is_pytest = label.startswith("pytest")
+    # ``bench slo`` starts pytest-benchmark itself, so it needs the same
+    # bounded temp policy and run marker as a direct pytest step.
+    has_managed_pytest_child = label == "bench slo"
     if is_pytest:
         _clear_pytest_report(cmd)
     artifacts = run.start_step(label=label, cmd=cmd) if run is not None else None
     env = _subprocess_env()
     pytest_tmpfs = False
     pytest_tmpfs_budget_mb: float | None = None
     runtime_policy = None
+    pytest_concurrency = 0
     basetemp_cleanup: Path | None = None
-    if is_pytest:
+    if is_pytest or has_managed_pytest_child:
         try:
-            env, runtime_policy = apply_managed_pytest_runtime_policy(env)
+            if has_managed_pytest_child:
+                # The outer benchmark process is not supervised as pytest, so
+                # its nested pytest cannot safely consume a bounded tmpfs run:
+                # nobody samples or terminates it at the tmpfs cap. Preserve a
+                # custom disk root, but replace inherited /dev/shm placement
+                # with the managed scratch candidate before admission.
+                env = force_managed_pytest_scratch(env)
+            if is_pytest:
+                pytest_concurrency = _pytest_command_concurrency(cmd, env=env)
+            env, runtime_policy = apply_managed_pytest_runtime_policy(
+                env,
+                worker_count=pytest_concurrency,
+                full_suite=_pytest_uses_full_suite_basetemp(label),
+            )
         except PytestResourceError as exc:
             elapsed = time.monotonic() - t0
             sys.stderr.write(f"FAILED ({elapsed:.1f}s)\nverify: {exc}\n")
@@ -1516,6 +1549,7 @@ def _run(
             env["POLYLOGUE_PYTEST_SELECTION_NODEID_LIMIT"] = "50000"
         if run is not None and artifacts is not None:
             env = env_for_pytest_step(env, run=run, artifacts=artifacts)
+    if is_pytest:
         try:
             result = _run_pytest_with_heartbeat(cmd, cwd=cwd, env=env, t0=t0, run=run, artifacts=artifacts)
         finally:
@@ -1723,7 +1757,7 @@ def _run(
             last_resource_sample=last_resource_row,
             tmpfs_budget_mb=pytest_tmpfs_budget_mb,
             basetemp_cleanup=basetemp_cleanup,
-            concurrency=_pytest_command_concurrency(cmd),
+            concurrency=max(1, pytest_concurrency),
         )
         metadata["workload_receipt"] = workload_receipt
         if artifacts is not None:
@@ -1780,9 +1814,10 @@ def _subprocess_env() -> dict[str, str]:
     env["PYTHONPYCACHEPREFIX"] = str(ROOT / ".cache" / "pycache")
     TESTMON_DATA.parent.mkdir(parents=True, exist_ok=True)
     env["TESTMON_DATAFILE"] = str(TESTMON_DATA)
-    env["POLYLOGUE_PYTEST_EVENTS_PATH"] = str(Path.cwd() / PYTEST_EVENTS_PATH)
-    env["POLYLOGUE_PYTEST_SELECTION_PATH"] = str(Path.cwd() / PYTEST_SELECTION_PATH)
-    env["POLYLOGUE_PYTEST_SUMMARY_PATH"] = str(Path.cwd() / PYTEST_SUMMARY_PATH)
+    env["POLYLOGUE_PYTEST_EVENTS_DIR"] = str(ROOT / PYTEST_EVENTS_DIR)
+    env["POLYLOGUE_PYTEST_EVENTS_PATH"] = str(ROOT / PYTEST_EVENTS_PATH)
+    env["POLYLOGUE_PYTEST_SELECTION_PATH"] = str(ROOT / PYTEST_SELECTION_PATH)
+    env["POLYLOGUE_PYTEST_SUMMARY_PATH"] = str(ROOT / PYTEST_SUMMARY_PATH)
     return env
 
 
@@ -2117,16 +2152,60 @@ def _pytest_worker_args(*, maximum: int | None = None) -> list[str]:
     return ["-n", str(workers)]
 
 
-def _pytest_command_concurrency(cmd: Sequence[str]) -> int:
-    """Return the worker count actually requested by the final pytest command."""
-    for index in range(len(cmd) - 2, -1, -1):
-        if cmd[index] != "-n":
-            continue
-        try:
-            return max(1, int(cmd[index + 1]))
-        except ValueError:
-            return 1
-    return 1
+def _pytest_command_worker_request(cmd: Sequence[str]) -> str | None:
+    """Return the last xdist worker request from a final pytest command.
+
+    ``devtools test`` forwards pytest arguments unchanged, so this accepts
+    both xdist spellings and their compact forms.  The final occurrence wins,
+    matching pytest's normal option precedence.
+    """
+    request: str | None = None
+    for index, arg in enumerate(cmd):
+        if arg in {"-n", "--numprocesses"}:
+            if index + 1 < len(cmd):
+                request = cmd[index + 1]
+        elif arg.startswith("--numprocesses="):
+            request = arg.removeprefix("--numprocesses=")
+        elif arg.startswith("-n") and len(arg) > 2:
+            request = arg[2:].removeprefix("=")
+    return request
+
+
+def _pytest_command_concurrency(cmd: Sequence[str], *, env: Mapping[str, str] | None = None) -> int:
+    """Return a fail-closed reservation for the final pytest command.
+
+    ``-n auto`` can launch one worker per logical CPU.  Reserve that maximum
+    instead of guessing one worker; an unrecognised xdist value is treated the
+    same way so malformed or future values cannot weaken admission.
+    """
+    request = _pytest_command_worker_request(cmd)
+    if request is None:
+        return 0
+    if request == "auto":
+        auto_workers = (env if env is not None else os.environ).get("PYTEST_XDIST_AUTO_NUM_WORKERS", "").strip()
+        if auto_workers:
+            try:
+                configured = int(auto_workers)
+            except ValueError:
+                configured = 0
+            if configured > 0:
+                return configured
+    try:
+        return max(0, int(request))
+    except ValueError:
+        return max(1, os.cpu_count() or 1)
+
+
+def _pytest_uses_full_suite_basetemp(label: str) -> bool:
+    """Whether this pytest step can materialize the measured full-suite tree."""
+    return label.startswith(
+        (
+            "pytest seed-testmon",
+            "pytest full",
+            "pytest load-sensitive",
+            "pytest testmon (broad)",
+        )
+    )
 
 
 _BROAD_TESTMON_CHANGED_PATHS = {