diff --git a/CLAUDE.md b/CLAUDE.md index 2f412e8de8..949e1089e8 100644 --- a/CLAUDE.md +++ b/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 diff --git a/TESTING.md b/TESTING.md index 4b457351d2..1d5eda8c40 100644 --- a/TESTING.md +++ b/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. diff --git a/devtools/run_tests.py b/devtools/run_tests.py index ef2228bf55..59948062c7 100644 --- a/devtools/run_tests.py +++ b/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) diff --git a/devtools/verify.py b/devtools/verify.py index 73f153536d..b5ccff19df 100644 --- a/devtools/verify.py +++ b/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,6 +1495,9 @@ 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 @@ -1486,10 +1505,24 @@ def _run( 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 = { diff --git a/devtools/verify_runs.py b/devtools/verify_runs.py index c3798d7b28..1e1b4655f0 100644 --- a/devtools/verify_runs.py +++ b/devtools/verify_runs.py @@ -23,6 +23,8 @@ from pathlib import Path from typing import Any +from polylogue.core.metrics import read_cgroup_memory_headroom_bytes + VERIFY_CACHE = Path(".cache/verify") VERIFY_RUNS_DIR = VERIFY_CACHE / "runs" CURRENT_RUN_PATH = VERIFY_CACHE / "current-run.json" @@ -37,12 +39,29 @@ DEFAULT_PYTEST_TMPFS_MAX_MB = 512 MAX_PYTEST_TMPFS_MAX_MB = 2048 MIN_PYTEST_AVAILABLE_KB = 1024 * 1024 -PYTEST_WORKER_MEMORY_KB = 768 * 1024 PYTEST_HOST_RESERVE_KB = 512 * 1024 +MIN_PYTEST_TMPFS_BUDGET_KB = 64 * 1024 MAX_ADAPTIVE_PYTEST_WORKERS = 12 PYTEST_BASETEMP_MIN_FREE_MB_ENV = "POLYLOGUE_PYTEST_BASETEMP_MIN_FREE_MB" DEFAULT_PYTEST_BASETEMP_MIN_FREE_MB = 1024 PYTEST_BASETEMP_REQUIRED_MB_ENV = "POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB" +PYTEST_MEMORY_ENVELOPE_WORKERS = 4 +PYTEST_MEMORY_ENVELOPE_PSS_KB = 4_353_168 +PYTEST_MEMORY_ENVELOPE_TMPFS_KB = 1_472_636 +PYTEST_MEMORY_ENVELOPE_CGROUP_BYTES = 6_278_623_232 +PYTEST_BASETEMP_PEAK_KB = 1_522 * 1024 +PYTEST_PROCESS_MEMORY_PER_WORKER_KB = 768 * 1024 +PYTEST_FOCUSED_CONTROLLER_MEMORY_KB = 256 * 1024 +PYTEST_PROCESS_MEMORY_FIXED_KB = max( + 0, + PYTEST_MEMORY_ENVELOPE_PSS_KB - PYTEST_MEMORY_ENVELOPE_WORKERS * PYTEST_PROCESS_MEMORY_PER_WORKER_KB, +) +PYTEST_CGROUP_OVERHEAD_FLOOR_KB = max( + 0, + (PYTEST_MEMORY_ENVELOPE_CGROUP_BYTES + 1023) // 1024 + - PYTEST_MEMORY_ENVELOPE_PSS_KB + - PYTEST_MEMORY_ENVELOPE_TMPFS_KB, +) class PytestResourceError(RuntimeError): @@ -61,6 +80,7 @@ class PytestRuntimePolicy: basetemp_label: str | None = None basetemp_required_mb: int | None = None basetemp_free_mb: int | None = None + tmpfs_predicted_mb: int | None = None def to_dict(self) -> dict[str, int | float | str | None]: return { @@ -72,6 +92,7 @@ def to_dict(self) -> dict[str, int | float | str | None]: "basetemp_label": self.basetemp_label, "basetemp_required_mb": self.basetemp_required_mb, "basetemp_free_mb": self.basetemp_free_mb, + "tmpfs_predicted_mb": self.tmpfs_predicted_mb, } @@ -550,6 +571,15 @@ def checkout_hash(root: Path) -> str: PYTEST_TMPFS_ROOT = Path("/dev/shm") +def _is_beneath(path: Path, root: Path) -> bool: + """Return whether *path* resolves within *root*, including *root* itself.""" + try: + path.resolve().relative_to(root.resolve()) + except (OSError, ValueError): + return False + return True + + def normalize_pytest_basetemp_env(env: Mapping[str, str]) -> dict[str, str]: """Keep cloud pytest defaults from escaping a workstation scratch volume. @@ -567,16 +597,110 @@ def normalize_pytest_basetemp_env(env: Mapping[str, str]) -> dict[str, str]: return normalized +def force_managed_pytest_scratch(env: Mapping[str, str]) -> dict[str, str]: + """Route an unsupervised nested pytest away from tmpfs. + + Preserve a genuine disk-backed operator root. An inherited root beneath + ``/dev/shm`` is not a safe override for a subprocess whose parent does not + sample or terminate it at the managed tmpfs cap. + """ + normalized = normalize_pytest_basetemp_env(env) + configured_root = normalized.get("POLYLOGUE_PYTEST_BASETEMP_ROOT") + if configured_root is not None and _is_beneath(Path(configured_root), PYTEST_TMPFS_ROOT): + normalized.pop("POLYLOGUE_PYTEST_BASETEMP_ROOT", None) + normalized["POLYLOGUE_PYTEST_TMPFS"] = "0" + return normalized + + +def _pytest_process_memory_reserve_kb(workers: int) -> int: + """Keep the measured controller floor while scaling worker processes.""" + return PYTEST_PROCESS_MEMORY_FIXED_KB + workers * PYTEST_PROCESS_MEMORY_PER_WORKER_KB + + +def _pytest_cgroup_overhead_reserve_kb(workers: int) -> int: + """Keep the measured residual floor, then scale it above four workers.""" + scale_workers = max(workers, PYTEST_MEMORY_ENVELOPE_WORKERS) + return ( + PYTEST_CGROUP_OVERHEAD_FLOOR_KB * scale_workers + PYTEST_MEMORY_ENVELOPE_WORKERS - 1 + ) // PYTEST_MEMORY_ENVELOPE_WORKERS + + +def _pytest_non_tmpfs_memory_reserve_kb(workers: int, *, full_suite: bool = True) -> int: + if not full_suite: + focused_process_kb = PYTEST_FOCUSED_CONTROLLER_MEMORY_KB + workers * PYTEST_PROCESS_MEMORY_PER_WORKER_KB + focused_cgroup_overhead_kb = ( + PYTEST_CGROUP_OVERHEAD_FLOOR_KB * max(1, workers) + PYTEST_MEMORY_ENVELOPE_WORKERS - 1 + ) // PYTEST_MEMORY_ENVELOPE_WORKERS + return focused_process_kb + focused_cgroup_overhead_kb + PYTEST_HOST_RESERVE_KB + return ( + _pytest_process_memory_reserve_kb(workers) + + _pytest_cgroup_overhead_reserve_kb(workers) + + PYTEST_HOST_RESERVE_KB + ) + + +def _default_pytest_workers(*, available_kb: int, cpu_cap: int, memory_full_avg10: float) -> int: + """Choose a pool that the same launch-time envelope can admit. + + Prefer the largest pool that preserves the measured aggregate basetemp in + memory. If even one worker cannot do that, retain the largest pool that can + start with the minimum tmpfs allowance so placement can reroute the shared + basetemp to scratch. + """ + + maximum = min(MAX_ADAPTIVE_PYTEST_WORKERS, cpu_cap) + + def largest_with(tmpfs_reserve_kb: int) -> int | None: + return next( + ( + workers + for workers in range(maximum, 0, -1) + if _pytest_non_tmpfs_memory_reserve_kb(workers) + tmpfs_reserve_kb <= available_kb + ), + None, + ) + + workers = largest_with(PYTEST_BASETEMP_PEAK_KB) + if workers is None: + workers = largest_with(MIN_PYTEST_TMPFS_BUDGET_KB) or 1 + if memory_full_avg10 >= 2.0: + return max(1, workers // 4) + if memory_full_avg10 >= 0.5: + return max(1, workers // 2) + return workers + + def adaptive_pytest_runtime_policy( *, available_kb: int | None = None, memory_full_avg10: float | None = None, cpu_count: int | None = None, shm_free_kb: int | None = None, + worker_count: int | None = None, + full_suite: bool = True, ) -> PytestRuntimePolicy: - """Size tmpfs and xdist from current headroom without a disk fallback.""" + """Size tmpfs and xdist from one measured resource envelope. + + The tmpfs cap is an admission limit, so it must be derived from capacity + the host can actually reserve. A percentage of ``MemAvailable`` made a + well-provisioned host advertise a 1.5 GiB cap to a full pytest run that + had 15 GiB free, then the supervisor correctly killed the run at that + artificial limit. Reserve the command's known workers and host headroom + from the measured four-worker process/cgroup/tmpfs envelope first. + + The four-worker PSS measurement contains a controller and supervisor that + do not disappear at one worker. Preserve the remainder after the existing + 768 MiB marginal worker allowance as a fixed process component. Keep the + measured residual cgroup charge as a floor through four workers and scale + it above the observed concurrency. Reserve the separately observed 1,521.6 + MiB basetemp peak at its next whole-MiB ceiling because it is one aggregate + run tree, independent of how xdist partitions the tests. + """ if available_kb is None: available_kb = _meminfo().get("MemAvailable") + cgroup_headroom_bytes = read_cgroup_memory_headroom_bytes() + if available_kb is not None and cgroup_headroom_bytes is not None: + available_kb = min(available_kb, cgroup_headroom_bytes // 1024) if available_kb is None: raise PytestResourceError("cannot read MemAvailable; refusing an unbudgeted pytest run") if available_kb < MIN_PYTEST_AVAILABLE_KB: @@ -587,46 +711,49 @@ def adaptive_pytest_runtime_policy( if memory_full_avg10 is None: memory_full_avg10 = _pressure("memory").get("full_avg10", 0.0) if shm_free_kb is None: - shm = _fs_usage(Path("/dev/shm")) - shm_free_kb = shm.get("free_kb") if shm is not None else None - if shm_free_kb is None: - raise PytestResourceError("/dev/shm is unavailable; refusing to fall back to disk-backed pytest") - - adaptive_budget_mb = max( - DEFAULT_PYTEST_TMPFS_MAX_MB, - min(MAX_PYTEST_TMPFS_MAX_MB, int(available_kb / 1024 * 0.10)), - ) - safe_shm_budget_mb = int((shm_free_kb / 1024) * 0.80) - tmpfs_budget_mb = min(adaptive_budget_mb, safe_shm_budget_mb) - if tmpfs_budget_mb < 64: - raise PytestResourceError(f"only {shm_free_kb / 1024:.0f} MiB free in /dev/shm; refusing disk-backed pytest") + shm = _fs_usage(PYTEST_TMPFS_ROOT) + shm_free_kb = shm.get("free_kb", 0) if shm is not None else 0 logical_cpus = cpu_count if cpu_count is not None else (os.cpu_count() or 1) cpu_cap = max(1, logical_cpus // 2) - worker_pool_kb = max(0, available_kb - tmpfs_budget_mb * 1024 - PYTEST_HOST_RESERVE_KB) - workers = max( - 1, - min( - MAX_ADAPTIVE_PYTEST_WORKERS, - cpu_cap, - max(1, worker_pool_kb // PYTEST_WORKER_MEMORY_KB), - ), - ) - if memory_full_avg10 >= 2.0: - workers = max(1, workers // 4) - elif memory_full_avg10 >= 0.5: - workers = max(1, workers // 2) + if worker_count is not None: + if worker_count < 0: + raise PytestResourceError(f"invalid pytest worker count {worker_count}") + reserved_workers = worker_count + else: + reserved_workers = _default_pytest_workers( + available_kb=available_kb, + cpu_cap=cpu_cap, + memory_full_avg10=memory_full_avg10, + ) + + fixed_reserve_kb = _pytest_non_tmpfs_memory_reserve_kb(reserved_workers, full_suite=full_suite) + tmpfs_predicted_kb = PYTEST_BASETEMP_PEAK_KB if full_suite else None + tmpfs_floor_kb = MIN_PYTEST_TMPFS_BUDGET_KB if shm_free_kb >= MIN_PYTEST_TMPFS_BUDGET_KB else 0 + if fixed_reserve_kb + tmpfs_floor_kb > available_kb: + raise PytestResourceError( + "cannot reserve measured pytest cgroup memory and host headroom " + f"(available={available_kb / 1024:.0f} MiB, workers={reserved_workers}, " + f"required={(fixed_reserve_kb + tmpfs_floor_kb) / 1024:.0f} MiB)" + ) + memory_safe_tmpfs_kb = available_kb - fixed_reserve_kb + safe_shm_budget_kb = int(shm_free_kb * 0.80) + tmpfs_budget_kb = min(MAX_PYTEST_TMPFS_MAX_MB * 1024, safe_shm_budget_kb, memory_safe_tmpfs_kb) + tmpfs_budget_mb = int(tmpfs_budget_kb / 1024) return PytestRuntimePolicy( available_kb=available_kb, tmpfs_budget_mb=tmpfs_budget_mb, - workers=workers, + workers=reserved_workers, memory_full_avg10=memory_full_avg10, + tmpfs_predicted_mb=(tmpfs_predicted_kb + 1023) // 1024 if tmpfs_predicted_kb is not None else None, ) -def apply_managed_pytest_runtime_policy(env: Mapping[str, str]) -> tuple[dict[str, str], PytestRuntimePolicy | None]: - """Enable bounded tmpfs by default; preserve explicit storage choices. +def apply_managed_pytest_runtime_policy( + env: Mapping[str, str], *, worker_count: int | None = None, full_suite: bool = True +) -> tuple[dict[str, str], PytestRuntimePolicy | None]: + """Place broad runs on scratch and focused runs on bounded tmpfs by default. Also runs the basetemp disk-headroom preflight (:func:`resolve_pytest_basetemp_root`) so a starved basetemp location is @@ -635,25 +762,55 @@ def apply_managed_pytest_runtime_policy(env: Mapping[str, str]) -> tuple[dict[st unrelated command minutes or hours later. """ normalized = normalize_pytest_basetemp_env(env) - policy: PytestRuntimePolicy | None = None - if not normalized.get("POLYLOGUE_PYTEST_BASETEMP_ROOT") and normalized.get("POLYLOGUE_PYTEST_TMPFS") != "0": - policy = adaptive_pytest_runtime_policy() + default_full_suite_scratch = ( + full_suite + and not normalized.get("POLYLOGUE_PYTEST_BASETEMP_ROOT") + and "POLYLOGUE_PYTEST_TMPFS" not in normalized + ) + configured_root = normalized.get("POLYLOGUE_PYTEST_BASETEMP_ROOT") + configured_tmpfs = configured_root is not None and _is_beneath(Path(configured_root), PYTEST_TMPFS_ROOT) + manages_tmpfs = configured_tmpfs or (configured_root is None and normalized.get("POLYLOGUE_PYTEST_TMPFS") != "0") + policy = adaptive_pytest_runtime_policy( + worker_count=worker_count, + shm_free_kb=None if manages_tmpfs else 0, + full_suite=full_suite, + ) + if full_suite and policy.tmpfs_predicted_mb is not None: + normalized.setdefault(PYTEST_BASETEMP_REQUIRED_MB_ENV, str(policy.tmpfs_predicted_mb)) + if manages_tmpfs: normalized["POLYLOGUE_PYTEST_TMPFS"] = "1" normalized.setdefault(PYTEST_TMPFS_MAX_MB_ENV, str(policy.tmpfs_budget_mb)) + effective_tmpfs_budget_kb = pytest_tmpfs_budget_kb(normalized) + policy_tmpfs_budget_kb = policy.tmpfs_budget_mb * 1024 + if effective_tmpfs_budget_kb is not None and effective_tmpfs_budget_kb > policy_tmpfs_budget_kb: + normalized[PYTEST_TMPFS_MAX_MB_ENV] = str(policy.tmpfs_budget_mb) + effective_tmpfs_budget_kb = pytest_tmpfs_budget_kb(normalized) + required_basetemp_kb = pytest_basetemp_required_kb(normalized) + if ( + required_basetemp_kb is not None + and effective_tmpfs_budget_kb is not None + and effective_tmpfs_budget_kb < required_basetemp_kb + ): + normalized["POLYLOGUE_PYTEST_TMPFS"] = "0" + if default_full_suite_scratch: + # Broad-suite demand grows with the fixture universe and has exceeded + # the supervised 2 GiB ceiling while tests were still progressing. + # Keep that ceiling for explicit tmpfs runs; use NVMe for the default + # broad route instead of guessing the next aggregate peak. + normalized["POLYLOGUE_PYTEST_TMPFS"] = "0" selected_root, selected_label = resolve_pytest_basetemp_root(normalized) if not normalized.get("POLYLOGUE_PYTEST_BASETEMP_ROOT") and selected_root != PYTEST_TMPFS_ROOT: normalized["POLYLOGUE_PYTEST_BASETEMP_ROOT"] = str(selected_root) normalized["POLYLOGUE_PYTEST_TMPFS"] = "0" - if policy is not None: - free_kb = _headroom_kb(selected_root) - required_kb = pytest_basetemp_required_kb(normalized) - policy = replace( - policy, - basetemp_root=str(selected_root), - basetemp_label=selected_label, - basetemp_required_mb=round(required_kb / 1024) if required_kb is not None else None, - basetemp_free_mb=round(free_kb / 1024) if free_kb is not None else None, - ) + free_kb = _headroom_kb(selected_root) + required_kb = pytest_basetemp_required_kb(normalized) + policy = replace( + policy, + basetemp_root=str(selected_root), + basetemp_label=selected_label, + basetemp_required_mb=round(required_kb / 1024) if required_kb is not None else None, + basetemp_free_mb=round(free_kb / 1024) if free_kb is not None else None, + ) return normalized, policy @@ -781,10 +938,15 @@ def resolve_pytest_basetemp_root(env: Mapping[str, str]) -> tuple[Path, str]: if configured: root = Path(configured) free_kb = _headroom_kb(root) - if free_kb is not None and free_kb >= min_free_kb: + configured_required_kb = min_free_kb + if _is_beneath(root, PYTEST_TMPFS_ROOT) and normalized.get("POLYLOGUE_PYTEST_TMPFS") == "1": + budget_kb = pytest_tmpfs_budget_kb(normalized) + headroom_kb = pytest_basetemp_min_free_kb(normalized) + configured_required_kb = headroom_kb + max(required_kb or 0, budget_kb or 0) + if free_kb is not None and free_kb >= configured_required_kb: return root, "configured" - checked.append(_describe_candidate(root, "configured", free_kb, min_free_kb)) - raise _basetemp_refusal(checked, min_free_kb) + checked.append(_describe_candidate(root, "configured", free_kb, configured_required_kb)) + raise _basetemp_refusal(checked, configured_required_kb) if normalized.get("POLYLOGUE_PYTEST_TMPFS", "1") != "0": shm = PYTEST_TMPFS_ROOT @@ -796,7 +958,7 @@ def resolve_pytest_basetemp_root(env: Mapping[str, str]) -> tuple[Path, str]: # cap so a bounded run cannot fill /dev/shm and strand unrelated # processes before the supervisor notices. headroom_kb = pytest_basetemp_min_free_kb(normalized) - declared_demand_kb = required_kb if required_kb is not None else (budget_kb or 0) + declared_demand_kb = max(required_kb or 0, budget_kb or 0) tmpfs_required_kb = headroom_kb + declared_demand_kb demand_fits_budget = free_kb is not None and free_kb >= tmpfs_required_kb if demand_fits_budget: @@ -845,7 +1007,9 @@ def pytest_basetemp_path(*, root: Path, run_id: str, env: dict[str, str]) -> Pat def pytest_tmpfs_budget_kb(env: Mapping[str, str]) -> int | None: """Return the bounded per-run tmpfs budget shared by all pytest workers.""" - if env.get("POLYLOGUE_PYTEST_TMPFS") != "1" or env.get("POLYLOGUE_PYTEST_BASETEMP_ROOT"): + configured_root = env.get("POLYLOGUE_PYTEST_BASETEMP_ROOT") + configured_tmpfs = configured_root is not None and _is_beneath(Path(configured_root), PYTEST_TMPFS_ROOT) + if env.get("POLYLOGUE_PYTEST_TMPFS") != "1" or (configured_root is not None and not configured_tmpfs): return None raw = env.get(PYTEST_TMPFS_MAX_MB_ENV, str(DEFAULT_PYTEST_TMPFS_MAX_MB)) with contextlib.suppress(ValueError): diff --git a/devtools/verify_slos.py b/devtools/verify_slos.py index 1dff6bc5c4..9cb949c7ba 100644 --- a/devtools/verify_slos.py +++ b/devtools/verify_slos.py @@ -15,6 +15,7 @@ import argparse import hashlib import json +import os import subprocess import sys import tempfile @@ -22,7 +23,7 @@ from devtools import repo_root as _get_root from devtools.benchmark_results import parse_pytest_benchmark_stats -from devtools.verify_runs import git_head +from devtools.verify_runs import apply_managed_pytest_runtime_policy, force_managed_pytest_scratch, git_head from polylogue.scenarios.workload import ( BudgetMeasure, BudgetSemantics, @@ -126,7 +127,12 @@ def _run_benchmarks(test_ids: set[str]) -> dict[str, dict[str, float]]: *sorted(test_ids), ] - result = subprocess.run(cmd, cwd=ROOT, capture_output=True, text=True) + env, _policy = apply_managed_pytest_runtime_policy( + force_managed_pytest_scratch(os.environ), + worker_count=0, + full_suite=False, + ) + result = subprocess.run(cmd, cwd=ROOT, capture_output=True, text=True, env=env) if not json_path.exists() or json_path.stat().st_size == 0: print("verify-slos: no benchmark results produced", file=sys.stderr) diff --git a/polylogue/core/metrics.py b/polylogue/core/metrics.py index 43d4ac605c..aa85f1a257 100644 --- a/polylogue/core/metrics.py +++ b/polylogue/core/metrics.py @@ -146,6 +146,27 @@ def read_cgroup_memory_max_bytes() -> int | None: return _read_cgroup_limit_int("memory.max") +def read_cgroup_memory_headroom_bytes() -> int | None: + """Return the smallest remaining ``memory.max`` allowance in the hierarchy. + + A child can have ample local allowance while a finite ancestor is nearly + full because of sibling workloads. Pair every finite limit with that same + cgroup's ``memory.current`` value instead of subtracting the leaf usage from + an unrelated ancestor limit. A finite limit with unreadable usage returns + zero so callers do not admit work against unmeasured capacity. + """ + headrooms: list[int] = [] + for limit_path in _cgroup_limit_files("memory.max"): + limit = _read_cgroup_int_from_path(limit_path) + if limit is None: + continue + current = _read_cgroup_int_from_path(limit_path.with_name("memory.current")) + if current is None: + return 0 + headrooms.append(max(0, limit - current)) + return min(headrooms) if headrooms else None + + def read_cgroup_memory_high_bytes() -> int | None: """Return the effective cgroup hierarchy ``memory.high`` threshold in bytes. @@ -322,6 +343,7 @@ def to_summary(self) -> JSONDocument: "SlowItemTracker", "StageMetrics", "read_cgroup_memory_current_mb", + "read_cgroup_memory_headroom_bytes", "read_cgroup_memory_high_bytes", "read_cgroup_memory_max_bytes", "read_cgroup_memory_peak_mb", diff --git a/tests/conftest.py b/tests/conftest.py index 3aaff9d0e7..6a8fdf4a36 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,6 +6,7 @@ import os import shutil import sqlite3 +import stat import subprocess import sys import threading @@ -35,7 +36,8 @@ assert_polylogue_matches_checkout, resolved_polylogue_path, ) -from devtools.verify_runs import PytestResourceError, resolve_pytest_basetemp_root +from devtools.pytest_supervisor import _process_start_ticks +from devtools.verify_runs import PytestResourceError, normalize_pytest_basetemp_env, resolve_pytest_basetemp_root # Resolve (but don't yet raise on) the polylogue-vs-checkout mismatch check # before the first `from polylogue...` import below: a shared/editable venv's @@ -102,6 +104,16 @@ def pytest_configure(config: pytest.Config) -> None: ) if config.option.basetemp is None: + normalized_basetemp_env = normalize_pytest_basetemp_env(os.environ) + if ( + "POLYLOGUE_VERIFY_RUN_ID" not in os.environ + and "POLYLOGUE_PYTEST_BASETEMP_ROOT" not in normalized_basetemp_env + ): + # Bare pytest has no devtools supervisor to enforce a tmpfs cap. + # Keep its basetemp on scratch; managed devtools runs carry the + # verify-run id and may opt into bounded tmpfs safely. + os.environ.pop("POLYLOGUE_PYTEST_BASETEMP_ROOT", None) + os.environ["POLYLOGUE_PYTEST_TMPFS"] = "0" checkout = hashlib.sha1(str(config.rootpath).encode("utf-8"), usedforsecurity=False).hexdigest()[:8] os.environ["POLYLOGUE_PYTEST_CHECKOUT"] = checkout run_id = os.environ.get("POLYLOGUE_PYTEST_RUN_ID") @@ -153,20 +165,53 @@ def _managed_pytest_temp_root() -> tuple[Path, str]: def _mark_basetemp_owner(basetemp: Path) -> None: - """Record the owning process pid so a stale-directory sweep never races a live run.""" + """Record the owning process identity so a sweep never races a live run.""" with contextlib.suppress(OSError): basetemp.mkdir(parents=True, exist_ok=True) - (basetemp / _OWNER_PID_MARKER).write_text(str(os.getpid()), encoding="utf-8") + pid = os.getpid() + start_ticks = _process_start_ticks(pid) + identity = f"{pid}:{start_ticks}" if start_ticks is not None else str(pid) + (basetemp / _OWNER_PID_MARKER).write_text(identity, encoding="utf-8") def _basetemp_owner_alive(entry: Path) -> bool | None: - """True/False when the owner pid marker resolves a live/dead process, else None (unknown).""" + """True/False when the owner marker resolves a live/dead process, else None.""" marker = entry / _OWNER_PID_MARKER try: - pid = int(marker.read_text(encoding="utf-8").strip()) + raw_identity = marker.read_text(encoding="utf-8").strip() + raw_pid, separator, raw_start_ticks = raw_identity.partition(":") + pid = int(raw_pid) + start_ticks = int(raw_start_ticks) if separator else None except (OSError, ValueError): return None - return Path(f"/proc/{pid}").exists() + if not Path(f"/proc/{pid}").exists(): + return False + return start_ticks is None or _process_start_ticks(pid) == start_ticks + + +def _remove_stale_basetemp(entry: Path) -> None: + """Remove an already-adjudicated stale tree, including read-only fixtures.""" + if entry.is_symlink(): + return + owner_directory_mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR + owner_file_mode = stat.S_IRUSR | stat.S_IWUSR + for current_root, directories, files in os.walk(entry): + root = Path(current_root) + with contextlib.suppress(OSError): + root.chmod(root.stat().st_mode | owner_directory_mode) + for name in directories: + child = root / name + if child.is_symlink(): + continue + with contextlib.suppress(OSError): + child.chmod(child.stat().st_mode | owner_directory_mode) + for name in files: + child = root / name + if child.is_symlink(): + continue + with contextlib.suppress(OSError): + child.chmod(child.stat().st_mode | owner_file_mode) + shutil.rmtree(entry) def _mark_btrfs_nocow(path: Path) -> None: @@ -239,9 +284,9 @@ def _sweep_stale_polylogue_basetemps( mtime = entry.stat().st_mtime if owner_alive is False: if mtime < cutoff: - shutil.rmtree(entry, ignore_errors=True) + _remove_stale_basetemp(entry) elif mtime < unknown_owner_cutoff: - shutil.rmtree(entry, ignore_errors=True) + _remove_stale_basetemp(entry) except OSError: pass diff --git a/tests/unit/core/test_metrics_cgroup_memory_limits.py b/tests/unit/core/test_metrics_cgroup_memory_limits.py index c240cd9546..6b372c53cc 100644 --- a/tests/unit/core/test_metrics_cgroup_memory_limits.py +++ b/tests/unit/core/test_metrics_cgroup_memory_limits.py @@ -1,10 +1,10 @@ """polylogue-e98k: cgroup memory.max/memory.high readers used by the mmap budget check. -Production dependency exercised: the real ``read_cgroup_memory_max_bytes``/ -``read_cgroup_memory_high_bytes`` -> ancestor-aware limit reader, not a -reimplemented parser. Reverting either reader to always return ``None``, to -read only the leaf cgroup, or to mis-treat the literal ``"max"`` value as a -real 0-byte limit would make these tests fail. +Production dependency exercised: the real max, high, and remaining-headroom +readers use the ancestor-aware cgroup hierarchy, not a reimplemented parser. +Returning ``None`` unconditionally, reading only the leaf cgroup, pairing an +ancestor limit with leaf usage, or treating literal ``"max"`` as zero makes +these tests fail. """ from __future__ import annotations @@ -14,7 +14,11 @@ import pytest from polylogue.core import metrics as metrics_module -from polylogue.core.metrics import read_cgroup_memory_high_bytes, read_cgroup_memory_max_bytes +from polylogue.core.metrics import ( + read_cgroup_memory_headroom_bytes, + read_cgroup_memory_high_bytes, + read_cgroup_memory_max_bytes, +) def _patch_cgroup_hierarchy(monkeypatch: pytest.MonkeyPatch, cgroup_root: Path, cgroup_path: str = "/") -> None: @@ -70,6 +74,33 @@ def test_read_cgroup_memory_limits_use_minimum_finite_nested_ancestor( assert read_cgroup_memory_high_bytes() == 10737418240 +def test_read_cgroup_memory_headroom_pairs_each_limit_with_its_own_usage( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "cgroup" + parent = root / "parent" + leaf = parent / "leaf" + leaf.mkdir(parents=True) + (root / "memory.max").write_text("max\n") + (root / "memory.current").write_text("1073741824\n") + (parent / "memory.max").write_text("12884901888\n") + (parent / "memory.current").write_text("10737418240\n") + (leaf / "memory.max").write_text("17179869184\n") + (leaf / "memory.current").write_text("5368709120\n") + _patch_cgroup_hierarchy(monkeypatch, root, "/parent/leaf") + + assert read_cgroup_memory_headroom_bytes() == 2147483648 + + +def test_read_cgroup_memory_headroom_fails_closed_when_finite_usage_is_unreadable( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + (tmp_path / "memory.max").write_text("4294967296\n") + _patch_cgroup_hierarchy(monkeypatch, tmp_path) + + assert read_cgroup_memory_headroom_bytes() == 0 + + def test_read_cgroup_memory_limits_return_none_when_nested_hierarchy_is_unlimited( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/unit/devtools/test_run_tests.py b/tests/unit/devtools/test_run_tests.py index 974a6958aa..bef2f53356 100644 --- a/tests/unit/devtools/test_run_tests.py +++ b/tests/unit/devtools/test_run_tests.py @@ -2,7 +2,6 @@ from __future__ import annotations -import os import subprocess import sys from pathlib import Path @@ -10,8 +9,7 @@ import pytest -from devtools import run_tests -from devtools.verify import PYTEST_EVENTS_PATH, PYTEST_SELECTION_PATH, PYTEST_SUMMARY_PATH +from devtools import run_tests, verify from devtools.verify_runs import git_head @@ -35,12 +33,39 @@ def test_build_pytest_cmd_respects_explicit_worker_flag() -> None: assert cmd[-2:] == ["-n", "4"] +@pytest.mark.parametrize( + "selection", + [ + ["tests/unit", "-n4"], + ["tests/unit", "-n=4"], + ["tests/unit", "--numprocesses", "8"], + ["tests/unit", "--numprocesses=8"], + ], +) +def test_build_pytest_cmd_forwards_all_xdist_worker_spellings(selection: list[str]) -> None: + command = run_tests.build_pytest_cmd(selection) + + for arg in selection: + assert arg in command + + def test_build_pytest_cmd_honors_workers_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("POLYLOGUE_PYTEST_WORKERS", "8") cmd = run_tests.build_pytest_cmd(["tests/unit"]) assert cmd[-2:] == ["-n", "8"] +def test_subprocess_env_anchors_pytest_artifacts_to_checkout(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.chdir(run_tests.ROOT / "tests") + + env = verify._subprocess_env() + + assert env["POLYLOGUE_PYTEST_EVENTS_DIR"] == str(run_tests.ROOT / verify.PYTEST_EVENTS_DIR) + assert env["POLYLOGUE_PYTEST_EVENTS_PATH"] == str(run_tests.ROOT / verify.PYTEST_EVENTS_PATH) + assert env["POLYLOGUE_PYTEST_SELECTION_PATH"] == str(run_tests.ROOT / verify.PYTEST_SELECTION_PATH) + assert env["POLYLOGUE_PYTEST_SUMMARY_PATH"] == str(run_tests.ROOT / verify.PYTEST_SUMMARY_PATH) + + def test_main_requires_a_selection(capsys: pytest.CaptureFixture[str]) -> None: assert run_tests.main([]) == 2 err = capsys.readouterr().err @@ -107,33 +132,3 @@ def _fake_run(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str] monkeypatch.setattr("devtools.verify_runs.subprocess.run", _fake_run) assert git_head(tmp_path) is None - - -def test_managed_env_sets_repo_roots() -> None: - env = run_tests._managed_env() - assert env["POLYLOGUE_ROOT"] == str(run_tests.ROOT) - assert env["POLYLOGUE_REPO_ROOT"] == str(run_tests.ROOT) - assert env["PYTHONPYCACHEPREFIX"] == str(run_tests.ROOT / ".cache" / "pycache") - assert env["PYTHONPATH"].split(os.pathsep)[0] == str(run_tests.ROOT) - assert env["POLYLOGUE_PYTEST_EVENTS_PATH"] == str(run_tests.ROOT / PYTEST_EVENTS_PATH) - assert env["POLYLOGUE_PYTEST_SELECTION_PATH"] == str(run_tests.ROOT / PYTEST_SELECTION_PATH) - assert env["POLYLOGUE_PYTEST_SUMMARY_PATH"] == str(run_tests.ROOT / PYTEST_SUMMARY_PATH) - assert Path(env["POLYLOGUE_ROOT"]).is_dir() - - -def test_managed_env_replaces_cloud_basetemp_with_local_tmpfs(monkeypatch: pytest.MonkeyPatch) -> None: - """Exercise devtools test's child environment, not conftest in isolation.""" - monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", "/tmp/polylogue-pytest") - - env = run_tests._managed_env() - - assert "POLYLOGUE_PYTEST_BASETEMP_ROOT" not in env - assert env["POLYLOGUE_PYTEST_TMPFS"] == "1" - - -def test_managed_env_preserves_explicit_custom_basetemp(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: - monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(tmp_path)) - - env = run_tests._managed_env() - - assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(tmp_path) diff --git a/tests/unit/devtools/test_slo_catalog.py b/tests/unit/devtools/test_slo_catalog.py index 1876576fb8..71fe4717a5 100644 --- a/tests/unit/devtools/test_slo_catalog.py +++ b/tests/unit/devtools/test_slo_catalog.py @@ -20,12 +20,16 @@ import io import json +import os +import subprocess +from collections.abc import Mapping from contextlib import redirect_stdout from pathlib import Path import pytest from devtools import verify_slos +from devtools.verify_runs import PytestRuntimePolicy REPO_ROOT = Path(__file__).resolve().parents[3] CATALOG_PATH = REPO_ROOT / "docs" / "plans" / "slo-catalog.yaml" @@ -34,6 +38,47 @@ REQUIRED_SURFACES = ("query", "reader", "facets", "context", "cost") +def test_benchmark_runner_routes_inherited_tmpfs_root_to_managed_scratch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + inherited_env = { + "POLYLOGUE_VERIFY_RUN_ID": "verify-run-123", + "POLYLOGUE_PYTEST_RUN_ID": "verify-run-123", + "POLYLOGUE_PYTEST_TMPFS": "1", + "POLYLOGUE_PYTEST_TMPFS_MAX_MB": "512", + "POLYLOGUE_PYTEST_BASETEMP_ROOT": "/dev/shm/inherited-benchmark", + } + monkeypatch.setattr(os, "environ", inherited_env) + captured: dict[str, object] = {} + + def fake_policy( + env: Mapping[str, str], *, worker_count: int | None, full_suite: bool + ) -> tuple[dict[str, str], PytestRuntimePolicy | None]: + captured["policy_env"] = dict(env) + captured["worker_count"] = worker_count + captured["full_suite"] = full_suite + return dict(env), None + + def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + captured["command"] = command + captured["env"] = kwargs["env"] + return subprocess.CompletedProcess(command, 0, stdout="", stderr="") + + monkeypatch.setattr(verify_slos, "apply_managed_pytest_runtime_policy", fake_policy) + monkeypatch.setattr(subprocess, "run", fake_run) + + assert verify_slos._run_benchmarks({"tests/benchmarks/test_reader_api.py::test_bench_reader_status"}) == {} + assert captured["policy_env"] == { + "POLYLOGUE_VERIFY_RUN_ID": "verify-run-123", + "POLYLOGUE_PYTEST_RUN_ID": "verify-run-123", + "POLYLOGUE_PYTEST_TMPFS": "0", + "POLYLOGUE_PYTEST_TMPFS_MAX_MB": "512", + } + assert captured["worker_count"] == 0 + assert captured["full_suite"] is False + assert captured["env"] == captured["policy_env"] + + def test_catalog_exists_and_covers_required_surfaces() -> None: """The committed SLO catalog must cover the surfaces enumerated in #872.""" assert CATALOG_PATH.exists(), f"missing SLO catalog at {CATALOG_PATH}" diff --git a/tests/unit/devtools/test_verify.py b/tests/unit/devtools/test_verify.py index 634bc653f1..094fe0db9d 100644 --- a/tests/unit/devtools/test_verify.py +++ b/tests/unit/devtools/test_verify.py @@ -11,7 +11,7 @@ import pytest -from devtools import verify, verify_runs +from devtools import run_tests, verify, verify_runs from devtools.testmon_state import ( BaselineStatus, BindingMode, @@ -73,6 +73,7 @@ ResourceSampler, VerifyRun, adaptive_pytest_runtime_policy, + adaptive_pytest_worker_count, apply_managed_pytest_runtime_policy, classify_pytest_result, cleanup_managed_pytest_basetemp, @@ -320,7 +321,7 @@ def test_seed_testmon_worker_count_can_be_overridden(monkeypatch: pytest.MonkeyP assert command[command.index("-n") + 1] == "4" -def test_seed_auto_enables_bounded_tmpfs(monkeypatch: pytest.MonkeyPatch) -> None: +def test_seed_defaults_to_managed_scratch(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("POLYLOGUE_PYTEST_TMPFS", raising=False) completed = subprocess.CompletedProcess(args=["pytest"], returncode=0, stdout="1 passed in 0.1s\n", stderr="") @@ -331,8 +332,11 @@ def test_seed_auto_enables_bounded_tmpfs(monkeypatch: pytest.MonkeyPatch) -> Non rc, _elapsed, metadata = _run("pytest seed-testmon", ["pytest", "--testmon", "--testmon-noselect"]) assert rc == 0 - assert metadata["pytest_tmpfs"] is True - assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_TMPFS"] == "1" + assert metadata["pytest_tmpfs"] is False + assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str( + verify_runs.DEFAULT_PYTEST_BASETEMP_ROOT + ) assert run.call_args.kwargs["env"]["POLYLOGUE_PYTEST_SELECTION_NODEID_LIMIT"] == "50000" @@ -1567,18 +1571,148 @@ def test_pytest_tmpfs_budget_is_shared_and_bounded() -> None: ) is None ) + assert ( + pytest_tmpfs_budget_kb( + { + "POLYLOGUE_PYTEST_TMPFS": "1", + "POLYLOGUE_PYTEST_BASETEMP_ROOT": "/dev/shm/polylogue-explicit", + } + ) + == 512 * 1024 + ) -def test_adaptive_pytest_policy_scales_workers_and_tmpfs() -> None: +def test_adaptive_pytest_policy_uses_host_capacity_not_ten_percent_cap() -> None: + """Reproduce ce4dd629's 15,190 MiB host headroom without under-capping tmpfs. + + The full parallel suite used four workers and reached 1,521.6 MiB in its + basetemp. Before this regression, the production policy returned 1,519 + MiB solely because it used ten percent of ``MemAvailable`` as the cap. + """ policy = adaptive_pytest_runtime_policy( - available_kb=16 * 1024 * 1024, + available_kb=15_190 * 1024, + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=15_190 * 1024, + worker_count=4, + ) + + assert policy.workers == 4 + assert policy.tmpfs_budget_mb == 2048 + assert policy.tmpfs_predicted_mb == 1522 + + +def test_adaptive_pytest_policy_refuses_four_workers_at_four_gib_from_measured_envelope() -> None: + with pytest.raises(PytestResourceError, match="cannot reserve measured pytest cgroup memory"): + adaptive_pytest_runtime_policy( + available_kb=4 * 1024 * 1024, + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=16 * 1024 * 1024, + worker_count=4, + ) + + +def test_adaptive_pytest_policy_caps_near_threshold_from_full_cgroup_peak() -> None: + policy = adaptive_pytest_runtime_policy( + available_kb=6400 * 1024, + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=16 * 1024 * 1024, + worker_count=4, + ) + + assert policy.tmpfs_budget_mb == 1338 + assert policy.tmpfs_predicted_mb is not None + assert policy.tmpfs_budget_mb < policy.tmpfs_predicted_mb + + +def test_adaptive_pytest_policy_preserves_controller_memory_at_one_worker() -> None: + with pytest.raises(PytestResourceError, match="cannot reserve measured pytest cgroup memory"): + adaptive_pytest_runtime_policy( + available_kb=2800 * 1024, + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=16 * 1024 * 1024, + worker_count=1, + ) + + +def test_adaptive_pytest_policy_does_not_charge_a_serial_run_for_an_xdist_worker() -> None: + policy = adaptive_pytest_runtime_policy( + available_kb=2500 * 1024, memory_full_avg10=0.0, cpu_count=24, shm_free_kb=16 * 1024 * 1024, + worker_count=0, ) - assert policy.workers == 12 - assert policy.tmpfs_budget_mb == 1638 + assert policy.workers == 0 + + +def test_focused_serial_policy_does_not_inherit_full_suite_memory_residuals() -> None: + policy = adaptive_pytest_runtime_policy( + available_kb=1500 * 1024, + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=0, + worker_count=0, + full_suite=False, + ) + + assert policy.workers == 0 + assert policy.tmpfs_predicted_mb is None + + with pytest.raises(PytestResourceError, match="cannot reserve measured pytest cgroup memory"): + adaptive_pytest_runtime_policy( + available_kb=1500 * 1024, + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=0, + worker_count=0, + full_suite=True, + ) + + +def test_adaptive_pytest_policy_treats_full_run_basetemp_as_aggregate_demand() -> None: + predictions = { + adaptive_pytest_runtime_policy( + available_kb=16 * 1024 * 1024, + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=16 * 1024 * 1024, + worker_count=workers, + ).tmpfs_predicted_mb + for workers in (1, 4, 8) + } + + assert predictions == {1522} + + +@pytest.mark.parametrize( + ("worker_args", "expected"), + [ + (["-n", "4"], 4), + (["-n", "0"], 0), + (["-n4"], 4), + (["-n=4"], 4), + (["--numprocesses", "8"], 8), + (["--numprocesses=8"], 8), + (["-n", "auto"], max(1, os.cpu_count() or 1)), + (["-nauto"], max(1, os.cpu_count() or 1)), + (["--numprocesses=auto"], max(1, os.cpu_count() or 1)), + ], +) +def test_production_pytest_commands_reserve_every_xdist_spelling(worker_args: list[str], expected: int) -> None: + command = run_tests.build_pytest_cmd(["tests/unit/devtools", *worker_args]) + + assert verify._pytest_command_concurrency(command) == expected + + +def test_pytest_auto_workers_reserve_environment_override() -> None: + command = run_tests.build_pytest_cmd(["tests/unit/devtools", "-n", "auto"]) + + assert verify._pytest_command_concurrency(command, env={"PYTEST_XDIST_AUTO_NUM_WORKERS": "32"}) == 32 def test_adaptive_pytest_policy_reduces_workers_under_pressure() -> None: @@ -1602,11 +1736,101 @@ def test_managed_pytest_policy_refuses_low_memory() -> None: ) -def test_managed_pytest_policy_preserves_explicit_custom_root(tmp_path: Path) -> None: - env, policy = apply_managed_pytest_runtime_policy({"POLYLOGUE_PYTEST_BASETEMP_ROOT": str(tmp_path)}) +def test_adaptive_pytest_policy_caps_host_capacity_to_cgroup_headroom(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 16 * 1024 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: 4 * 1024 * 1024 * 1024) + + with pytest.raises(PytestResourceError, match="cannot reserve measured pytest cgroup memory"): + adaptive_pytest_runtime_policy( + memory_full_avg10=0.0, + cpu_count=24, + shm_free_kb=16 * 1024 * 1024, + worker_count=4, + ) + + +def test_managed_pytest_policy_preserves_explicit_custom_root_and_memory_admission( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 8 * 1024 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + monkeypatch.setattr(verify_runs, "_fs_usage", lambda _path: {"used_kb": 0, "free_kb": 16 * 1024 * 1024}) + env, policy = apply_managed_pytest_runtime_policy({"POLYLOGUE_PYTEST_BASETEMP_ROOT": str(tmp_path)}, worker_count=4) assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(tmp_path) - assert policy is None + assert policy is not None + assert policy.workers == 4 + assert policy.basetemp_label == "configured" + + +def test_managed_pytest_policy_bounds_explicit_tmpfs_root(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 8 * 1024 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + monkeypatch.setattr(verify_runs, "_fs_usage", lambda _path: {"used_kb": 0, "free_kb": 16 * 1024 * 1024}) + + env, policy = apply_managed_pytest_runtime_policy( + { + "POLYLOGUE_PYTEST_BASETEMP_ROOT": "/dev/shm/polylogue-explicit", + "POLYLOGUE_PYTEST_TMPFS": "0", + }, + worker_count=4, + full_suite=False, + ) + + assert policy is not None + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == "/dev/shm/polylogue-explicit" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "1" + assert pytest_tmpfs_budget_kb(env) == policy.tmpfs_budget_mb * 1024 + assert policy.basetemp_label == "configured" + + +def test_managed_pytest_policy_preserves_headroom_for_explicit_tmpfs_root( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 8 * 1024 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + monkeypatch.setattr(verify_runs, "_fs_usage", lambda _path: {"used_kb": 0, "free_kb": 2500 * 1024}) + + with pytest.raises(PytestResourceError, match="need >= 3024 MiB"): + apply_managed_pytest_runtime_policy( + { + "POLYLOGUE_PYTEST_BASETEMP_ROOT": "/dev/shm/polylogue-explicit", + "POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB": "1522", + "POLYLOGUE_PYTEST_TMPFS_MAX_MB": "2048", + }, + worker_count=4, + full_suite=True, + ) + + +def test_full_suite_explicit_root_requires_measured_basetemp_space( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 8 * 1024 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + monkeypatch.setattr(verify_runs, "_fs_usage", lambda _path: {"used_kb": 0, "free_kb": 1200 * 1024}) + + with pytest.raises(PytestResourceError, match="no pytest basetemp location has enough free space"): + apply_managed_pytest_runtime_policy( + {"POLYLOGUE_PYTEST_BASETEMP_ROOT": str(tmp_path)}, + worker_count=4, + full_suite=True, + ) + + +def test_managed_pytest_policy_rejects_explicit_root_when_workers_exceed_memory( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 4 * 1024 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + + with pytest.raises(PytestResourceError, match="cannot reserve measured pytest cgroup memory"): + apply_managed_pytest_runtime_policy({"POLYLOGUE_PYTEST_BASETEMP_ROOT": str(tmp_path)}, worker_count=12) # ── basetemp placement: one resolution order, disk-headroom preflight ────── @@ -1647,6 +1871,169 @@ def _patch_basetemp_roots(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, *, re return shm, scratch +def _patch_resource_capacity( + monkeypatch: pytest.MonkeyPatch, + *, + shm: Path, + scratch: Path, + available_mb: int, +) -> None: + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": available_mb * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + monkeypatch.setattr(os, "cpu_count", lambda: 24) + + def fake_fs_usage(path: Path) -> dict[str, int] | None: + if path in {shm, scratch.parent}: + return {"used_kb": 0, "free_kb": 16 * 1024 * 1024} + return None + + monkeypatch.setattr(verify_runs, "_fs_usage", fake_fs_usage) + + +def test_default_full_suite_workers_use_scratch_through_placement( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=8192) + + workers = adaptive_pytest_worker_count({}) + env, policy = apply_managed_pytest_runtime_policy({}, worker_count=workers) + + assert workers == 5 + assert policy is not None + assert policy.workers == workers + assert policy.basetemp_label == "scratch" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) + + +def test_focused_selection_keeps_bounded_tmpfs_default(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=8192) + + env, policy = apply_managed_pytest_runtime_policy({}, worker_count=1, full_suite=False) + + assert policy is not None + assert policy.basetemp_label == "tmpfs opt-in" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "1" + + +def test_explicit_full_suite_tmpfs_choice_remains_bounded(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=15_190) + + env, policy = apply_managed_pytest_runtime_policy({"POLYLOGUE_PYTEST_TMPFS": "1"}, worker_count=4, full_suite=True) + + assert policy is not None + assert policy.basetemp_label == "tmpfs opt-in" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "1" + assert env["POLYLOGUE_PYTEST_TMPFS_MAX_MB"] == "2048" + + +def test_inherited_512_mib_tmpfs_cap_reroutes_measured_demand_to_scratch( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=15_190) + + env, policy = apply_managed_pytest_runtime_policy( + {"POLYLOGUE_PYTEST_TMPFS_MAX_MB": "512"}, + worker_count=4, + ) + + assert policy is not None + assert policy.tmpfs_budget_mb == 2048 + assert policy.tmpfs_predicted_mb == 1522 + assert policy.basetemp_label == "scratch" + assert env["POLYLOGUE_PYTEST_TMPFS_MAX_MB"] == "512" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) + + +def test_focused_policy_keeps_full_suite_basetemp_demand_out_of_scratch_preflight( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=8192) + + def fake_fs_usage(path: Path) -> dict[str, int] | None: + if path == shm: + return None + if path == scratch.parent: + return {"used_kb": 0, "free_kb": 1200 * 1024} + return None + + monkeypatch.setattr(verify_runs, "_fs_usage", fake_fs_usage) + env, policy = apply_managed_pytest_runtime_policy({}, worker_count=0, full_suite=False) + + assert policy is not None + assert policy.basetemp_required_mb is None + assert policy.basetemp_label == "scratch" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) + + +def test_inherited_tmpfs_cap_is_clamped_to_the_measured_host_budget( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=6400) + + env, policy = apply_managed_pytest_runtime_policy( + {"POLYLOGUE_PYTEST_TMPFS_MAX_MB": "2048"}, + worker_count=4, + ) + + assert policy is not None + assert policy.tmpfs_budget_mb == 1338 + assert env["POLYLOGUE_PYTEST_TMPFS_MAX_MB"] == "1338" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) + + +def test_managed_policy_uses_scratch_when_tmpfs_is_unavailable(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + monkeypatch.setattr(verify_runs, "_meminfo", lambda: {"MemAvailable": 15_190 * 1024}) + monkeypatch.setattr(verify_runs, "read_cgroup_memory_headroom_bytes", lambda: None) + monkeypatch.setattr(verify_runs, "_pressure", lambda _kind: {"full_avg10": 0.0}) + monkeypatch.setattr(os, "cpu_count", lambda: 24) + + def fake_fs_usage(path: Path) -> dict[str, int] | None: + if path == shm: + return None + if path == scratch.parent: + return {"used_kb": 0, "free_kb": 16 * 1024 * 1024} + return None + + monkeypatch.setattr(verify_runs, "_fs_usage", fake_fs_usage) + + env, policy = apply_managed_pytest_runtime_policy({}, worker_count=4) + + assert policy is not None + assert policy.tmpfs_budget_mb == 0 + assert policy.basetemp_label == "scratch" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) + + +def test_declared_basetemp_demand_above_tmpfs_cap_uses_scratch(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + _patch_resource_capacity(monkeypatch, shm=shm, scratch=scratch, available_mb=15_190) + + env, policy = apply_managed_pytest_runtime_policy( + {"POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB": "4096"}, + worker_count=4, + ) + + assert policy is not None + assert policy.tmpfs_budget_mb == 2048 + assert policy.basetemp_required_mb == 4096 + assert policy.basetemp_label == "scratch" + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == str(scratch) + + def test_resolve_basetemp_prefers_tmpfs_when_it_has_headroom(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: shm, _scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) monkeypatch.setattr( @@ -1699,6 +2086,32 @@ def fake_fs_usage(path: Path) -> dict[str, int] | None: assert label == "scratch" +def test_resolve_basetemp_reserves_the_allowed_cap_not_only_the_prediction( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + shm, scratch = _patch_basetemp_roots(monkeypatch, tmp_path, realm_mounted=True) + + def fake_fs_usage(path: Path) -> dict[str, int] | None: + if path == shm: + return {"used_kb": 0, "free_kb": 2500 * 1024} + if path == scratch.parent: + return {"used_kb": 0, "free_kb": 4096 * 1024} + return None + + monkeypatch.setattr(verify_runs, "_fs_usage", fake_fs_usage) + + root, label = resolve_pytest_basetemp_root( + { + "POLYLOGUE_PYTEST_TMPFS": "1", + "POLYLOGUE_PYTEST_BASETEMP_REQUIRED_MB": "1522", + "POLYLOGUE_PYTEST_TMPFS_MAX_MB": "2048", + } + ) + + assert root == scratch + assert label == "scratch" + + def test_resolve_basetemp_refuses_loudly_when_every_candidate_is_full( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: @@ -1901,6 +2314,26 @@ def test_run_records_managed_basetemp_cleanup_metadata(tmp_path: Path) -> None: assert metadata["basetemp_cleanup"] == str(cleaned) +def test_explicit_basetemp_root_retains_managed_resource_monitoring( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + monkeypatch.setattr(verify_runs, "PYTEST_TMPFS_ROOT", tmp_path / "unselected-tmpfs") + nvme_root = tmp_path / "realm-tmp" / "polylogue-pytest" + monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(nvme_root)) + run = VerifyRun(tier="configured-nvme", argv=[], git_head=None, root=tmp_path) + + rc, _elapsed, metadata = _run( + "pytest configured NVMe root", + [sys.executable, "-c", "print('managed resource sampler remains active')"], + run=run, + ) + + assert rc == 0 + assert metadata["pytest_tmpfs"] is False + assert metadata["pytest_tmpfs_budget_mb"] is None + assert metadata["resource_sample_count"] >= 1 + + def test_run_receipt_uses_capped_pytest_command_concurrency() -> None: completed = subprocess.CompletedProcess(args=["pytest"], returncode=0, stdout="1 passed in 0.1s\n", stderr="") @@ -1911,17 +2344,86 @@ def to_dict(self) -> dict[str, int]: return {"workers": self.workers} with ( - patch("devtools.verify.apply_managed_pytest_runtime_policy", return_value=({}, UncappedPolicy())), + patch( + "devtools.verify.apply_managed_pytest_runtime_policy", return_value=({}, UncappedPolicy()) + ) as apply_policy, patch("devtools.verify._run_pytest_with_heartbeat", return_value=completed), patch("devtools.verify._read_pytest_report", return_value=None), ): rc, _elapsed, metadata = _run("pytest seed-testmon", ["pytest", "--testmon", "-n", "4"]) assert rc == 0 + assert apply_policy.call_args.kwargs["worker_count"] == 4 + assert apply_policy.call_args.kwargs["full_suite"] is True assert metadata["pytest_runtime_policy"] == {"workers": 12} assert metadata["workload_receipt"]["spec"]["concurrency"] == 4 +@pytest.mark.parametrize( + ("label", "full_suite"), + [ + ("pytest focused", False), + ("pytest testmon", False), + ("pytest testmon (broad)", True), + ("pytest seed-testmon", True), + ("pytest full (parallel)", True), + ("pytest load-sensitive (isolated)", True), + ], +) +def test_run_scopes_measured_full_suite_basetemp_demand(tmp_path: Path, label: str, full_suite: bool) -> None: + completed = subprocess.CompletedProcess(args=["pytest"], returncode=0, stdout="1 passed in 0.1s\n", stderr="") + + class FocusedPolicy: + workers = 0 + + def to_dict(self) -> dict[str, int]: + return {"workers": self.workers} + + run = VerifyRun(tier="focused-test", argv=["tests/unit/example.py"], git_head="head", root=tmp_path) + with ( + patch( + "devtools.verify.apply_managed_pytest_runtime_policy", return_value=({}, FocusedPolicy()) + ) as apply_policy, + patch("devtools.verify._run_pytest_with_heartbeat", return_value=completed), + patch("devtools.verify._read_pytest_report", return_value=None), + ): + rc, _elapsed, _metadata = _run(label, ["pytest", "tests/unit/example.py", "-n", "0"], run=run) + + assert rc == 0 + assert apply_policy.call_args.kwargs["worker_count"] == 0 + assert apply_policy.call_args.kwargs["full_suite"] is full_suite + + +def test_bench_slo_forces_nested_pytest_to_managed_scratch( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + run = VerifyRun(tier="lab", argv=[], git_head=None, root=tmp_path) + monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", "/dev/shm/inherited-benchmark") + managed_env = { + "POLYLOGUE_PYTEST_TMPFS": "0", + "POLYLOGUE_PYTEST_BASETEMP_ROOT": "/realm/tmp/polylogue-pytest", + } + completed = subprocess.CompletedProcess(args=["devtools", "bench", "slo"], returncode=0, stdout="", stderr="") + + with ( + patch("devtools.verify.apply_managed_pytest_runtime_policy", return_value=(managed_env, None)) as apply_policy, + patch("devtools.verify.subprocess.run", return_value=completed) as subprocess_run, + ): + rc, _elapsed, _metadata = _run("bench slo", ["devtools", "bench", "slo"], run=run) + + assert rc == 0 + assert apply_policy.call_args.kwargs == {"worker_count": 0, "full_suite": False} + policy_input = apply_policy.call_args.args[0] + assert policy_input["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert "POLYLOGUE_PYTEST_BASETEMP_ROOT" not in policy_input + env = subprocess_run.call_args.kwargs["env"] + assert env["POLYLOGUE_VERIFY_RUN_ID"] == run.run_id + assert env["POLYLOGUE_PYTEST_RUN_ID"] == run.run_id + assert env["POLYLOGUE_PYTEST_TMPFS"] == "0" + assert env["POLYLOGUE_PYTEST_BASETEMP_ROOT"] == "/realm/tmp/polylogue-pytest" + + def test_run_forces_subprocesses_to_current_checkout(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("POLYLOGUE_ROOT", "/stale/main") monkeypatch.setenv("POLYLOGUE_REPO_ROOT", "/stale/main") @@ -1938,7 +2440,7 @@ def test_run_forces_subprocesses_to_current_checkout(monkeypatch: pytest.MonkeyP assert env["POLYLOGUE_REPO_ROOT"] == str(ROOT) assert env["PYTHONPYCACHEPREFIX"] == str(ROOT / ".cache" / "pycache") assert env["PYTHONPATH"].split(os.pathsep)[0] == str(ROOT) - assert env["POLYLOGUE_PYTEST_EVENTS_PATH"] == str(Path.cwd() / PYTEST_EVENTS_PATH) + assert env["POLYLOGUE_PYTEST_EVENTS_PATH"] == str(ROOT / PYTEST_EVENTS_PATH) def test_verify_subprocess_env_removes_cloud_basetemp_in_local_worktree(monkeypatch: pytest.MonkeyPatch) -> None: @@ -2166,15 +2668,19 @@ def test_pytest_run_terminates_with_heartbeat_disabled( def test_pytest_run_terminates_after_output_stall( - monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], ) -> None: monkeypatch.setenv("POLYLOGUE_VERIFY_HEARTBEAT_S", "0.05") monkeypatch.setenv("POLYLOGUE_VERIFY_PYTEST_TIMEOUT_S", "0") monkeypatch.setenv("POLYLOGUE_VERIFY_PYTEST_STALL_TIMEOUT_S", "0.15") + run = VerifyRun(tier="output-stall", argv=[], git_head=None, root=tmp_path) rc, _elapsed, metadata = _run( "pytest stall", [sys.executable, "-c", "import time; print('progress', flush=True); time.sleep(5)"], + run=run, ) captured = capsys.readouterr() @@ -2187,7 +2693,9 @@ def test_pytest_run_terminates_after_output_stall( def test_pytest_run_terminates_on_progress_stall_despite_flowing_output( - monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], ) -> None: """polylogue-27rb: an xdist-master-keeps-emitting D-state deadlock. @@ -2200,15 +2708,17 @@ def test_pytest_run_terminates_on_progress_stall_despite_flowing_output( monkeypatch.setenv("POLYLOGUE_VERIFY_PYTEST_TIMEOUT_S", "0") monkeypatch.setenv("POLYLOGUE_VERIFY_PYTEST_STALL_TIMEOUT_S", "0.15") - # _run's _clear_pytest_report wipes PYTEST_EVENTS_PATH before the child + # _run's _clear_pytest_report wipes the event artifacts before the child # starts, so the child itself must write the one-and-only progress # event (matching a real pytest worker reporting "test started" then - # wedging mid-test) -- it reads the path devtools/verify.py's - # _subprocess_env() injects for it. + # wedging mid-test). Real xdist workers write their own live JSONL files + # under POLYLOGUE_PYTEST_EVENTS_DIR; the merged path appears only after + # pytest exits and therefore cannot drive an in-process stall detector. child_script = ( "import json, os, time\n" - "path = os.environ['POLYLOGUE_PYTEST_EVENTS_PATH']\n" - "os.makedirs(os.path.dirname(path), exist_ok=True)\n" + "directory = os.environ['POLYLOGUE_PYTEST_EVENTS_DIR']\n" + "os.makedirs(directory, exist_ok=True)\n" + "path = os.path.join(directory, 'gw0.jsonl')\n" "with open(path, 'w') as f:\n" " f.write(json.dumps({'event': 'test_started', 'nodeid': 'wedged::test', " "'updated_at': '2026-01-01T00:00:00Z'}) + '\\n')\n" @@ -2216,7 +2726,12 @@ def test_pytest_run_terminates_on_progress_stall_despite_flowing_output( " print('progress', flush=True)\n" " time.sleep(0.02)\n" ) - rc, _elapsed, metadata = _run("pytest stall", [sys.executable, "-c", child_script]) + run = VerifyRun(tier="progress-stall", argv=[], git_head=None, root=tmp_path) + rc, _elapsed, metadata = _run( + "pytest stall", + [sys.executable, "-c", child_script], + run=run, + ) captured = capsys.readouterr() assert rc == 124 diff --git a/tests/unit/operations/test_mutation_actuators.py b/tests/unit/operations/test_mutation_actuators.py index b67897cf6d..4bc357f06c 100644 --- a/tests/unit/operations/test_mutation_actuators.py +++ b/tests/unit/operations/test_mutation_actuators.py @@ -84,8 +84,7 @@ ) from polylogue.operations.mutation_transaction import ConfirmationRequiredError, OperationExecutor, PlanStaleError from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore -from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database -from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier +from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root from polylogue.storage.sqlite.archive_tiers.user_write import ( assertion_id_for_saved_view, assertion_id_for_workspace, @@ -95,8 +94,7 @@ def _seed_archive_session(archive_root: Path, *, native_id: str) -> str: source_db = archive_root / "source.db" index_db = archive_root / "index.db" - initialize_archive_database(source_db, ArchiveTier.SOURCE) - initialize_archive_database(index_db, ArchiveTier.INDEX) + initialize_active_archive_root(archive_root) session_id = f"codex-session:{native_id}" raw_id = f"raw-{native_id}" with sqlite3.connect(source_db) as conn: diff --git a/tests/unit/test_pytest_temp_policy.py b/tests/unit/test_pytest_temp_policy.py index 20f6403f9d..94bdb03eab 100644 --- a/tests/unit/test_pytest_temp_policy.py +++ b/tests/unit/test_pytest_temp_policy.py @@ -9,6 +9,7 @@ import tests.conftest as conftest from devtools import verify_runs +from tests.infra.frozen_clock import FrozenClock def _make_real_candidates( @@ -166,8 +167,57 @@ def test_pytest_configure_reports_low_space_as_usage_error( conftest.pytest_configure(cast("pytest.Config", config)) +def test_bare_pytest_configure_defaults_to_scratch_without_a_supervisor( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _shm, scratch = _make_real_candidates(monkeypatch, tmp_path) + for name in ( + "POLYLOGUE_VERIFY_RUN_ID", + "POLYLOGUE_PYTEST_BASETEMP_ROOT", + "POLYLOGUE_PYTEST_TMPFS", + "POLYLOGUE_PYTEST_RUN_ID", + "POLYLOGUE_PYTEST_CHECKOUT", + ): + monkeypatch.delenv(name, raising=False) + config = SimpleNamespace( + option=SimpleNamespace(basetemp=None), + addinivalue_line=lambda *args, **kwargs: None, + rootpath=tmp_path, + ) + + conftest.pytest_configure(cast("pytest.Config", config)) + + assert Path(str(config.option.basetemp)).parent == scratch + assert os.environ["POLYLOGUE_PYTEST_TMPFS"] == "0" + + +def test_bare_pytest_ignores_leaked_cloud_basetemp_on_workstation( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + _shm, scratch = _make_real_candidates(monkeypatch, tmp_path) + monkeypatch.setenv("POLYLOGUE_PYTEST_BASETEMP_ROOT", str(verify_runs._CLOUD_PYTEST_BASETEMP_ROOT)) + monkeypatch.delenv("POLYLOGUE_VERIFY_RUN_ID", raising=False) + monkeypatch.delenv("POLYLOGUE_PYTEST_TMPFS", raising=False) + monkeypatch.delenv("POLYLOGUE_PYTEST_RUN_ID", raising=False) + monkeypatch.delenv("POLYLOGUE_PYTEST_CHECKOUT", raising=False) + config = SimpleNamespace( + option=SimpleNamespace(basetemp=None), + addinivalue_line=lambda *args, **kwargs: None, + rootpath=tmp_path, + ) + + conftest.pytest_configure(cast("pytest.Config", config)) + + assert Path(str(config.option.basetemp)).parent == scratch + assert "POLYLOGUE_PYTEST_BASETEMP_ROOT" not in os.environ + assert os.environ["POLYLOGUE_PYTEST_TMPFS"] == "0" + + def test_sweep_stale_polylogue_basetemps_preserves_seeded_and_recent( tmp_path: Path, + frozen_clock: FrozenClock, ) -> None: stale = tmp_path / "pytest-polylogue-dead-123" seeded = tmp_path / "pytest-polylogue-seeded-dead" @@ -176,7 +226,7 @@ def test_sweep_stale_polylogue_basetemps_preserves_seeded_and_recent( for path in (stale, seeded, recent, unrelated): path.mkdir() - old = 1.0 + old = frozen_clock.time() - conftest._STALE_BASETEMP_UNKNOWN_OWNER_MAX_AGE_S - 1 os.utime(stale, (old, old)) os.utime(seeded, (old, old)) @@ -190,6 +240,7 @@ def test_sweep_stale_polylogue_basetemps_preserves_seeded_and_recent( def test_sweep_stale_polylogue_basetemps_never_deletes_a_live_owner( tmp_path: Path, + frozen_clock: FrozenClock, ) -> None: """Critical safety invariant: age alone must never justify deletion — a long-running lane's basetemp must survive the sweep even once it is @@ -198,7 +249,7 @@ def test_sweep_stale_polylogue_basetemps_never_deletes_a_live_owner( live = tmp_path / "pytest-polylogue-live-owner-123" live.mkdir() conftest._mark_basetemp_owner(live) - old = 1.0 + old = frozen_clock.time() - 120 os.utime(live, (old, old)) conftest._sweep_stale_polylogue_basetemps(max_age_s=60, roots=(tmp_path,)) @@ -208,13 +259,14 @@ def test_sweep_stale_polylogue_basetemps_never_deletes_a_live_owner( def test_sweep_stale_polylogue_basetemps_reclaims_a_confirmed_dead_owner( tmp_path: Path, + frozen_clock: FrozenClock, ) -> None: dead = tmp_path / "pytest-polylogue-dead-owner-123" dead.mkdir() # A pid that is guaranteed not to be alive right now (max pid + 1 territory # would flake on hosts near pid rollover; /proc simply never has this one). (dead / conftest._OWNER_PID_MARKER).write_text("999999999", encoding="utf-8") - old = 1.0 + old = frozen_clock.time() - 120 os.utime(dead, (old, old)) conftest._sweep_stale_polylogue_basetemps(max_age_s=60, roots=(tmp_path,)) @@ -222,6 +274,70 @@ def test_sweep_stale_polylogue_basetemps_reclaims_a_confirmed_dead_owner( assert not dead.exists() +def test_sweep_stale_polylogue_basetemps_reclaims_reused_pid_identity( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + frozen_clock: FrozenClock, +) -> None: + stale = tmp_path / "pytest-polylogue-reused-pid-123" + stale.mkdir() + (stale / conftest._OWNER_PID_MARKER).write_text(f"{os.getpid()}:100", encoding="utf-8") + monkeypatch.setattr(conftest, "_process_start_ticks", lambda _pid: 200) + old = frozen_clock.time() - 120 + os.utime(stale, (old, old)) + + conftest._sweep_stale_polylogue_basetemps(max_age_s=60, roots=(tmp_path,)) + + assert not stale.exists() + + +def test_sweep_stale_polylogue_basetemps_reclaims_read_only_fixture_tree( + tmp_path: Path, + frozen_clock: FrozenClock, +) -> None: + stale = tmp_path / "pytest-polylogue-read-only-123" + nested = stale / "published" / "artifact" + nested.mkdir(parents=True) + payload = nested / "payload.json" + payload.write_text("{}", encoding="utf-8") + payload.chmod(0o400) + nested.chmod(0o500) + (stale / "published").chmod(0o500) + old = frozen_clock.time() - conftest._STALE_BASETEMP_UNKNOWN_OWNER_MAX_AGE_S - 1 + os.utime(stale, (old, old)) + + conftest._sweep_stale_polylogue_basetemps(max_age_s=60, roots=(tmp_path,)) + + assert not stale.exists() + + +def test_sweep_stale_polylogue_basetemps_does_not_follow_top_level_symlink( + tmp_path: Path, + frozen_clock: FrozenClock, +) -> None: + target = tmp_path / "external-fixture" + nested = target / "published" + nested.mkdir(parents=True) + payload = nested / "payload.json" + payload.write_text("{}", encoding="utf-8") + payload.chmod(0o400) + nested.chmod(0o500) + target.chmod(0o500) + old = frozen_clock.time() - conftest._STALE_BASETEMP_UNKNOWN_OWNER_MAX_AGE_S - 1 + os.utime(target, (old, old)) + + link = tmp_path / "pytest-polylogue-stale-symlink-123" + link.symlink_to(target, target_is_directory=True) + before_modes = (target.stat().st_mode, nested.stat().st_mode, payload.stat().st_mode) + + conftest._sweep_stale_polylogue_basetemps(max_age_s=60, roots=(tmp_path,)) + + assert link.is_symlink() + assert target.is_dir() + assert payload.read_text(encoding="utf-8") == "{}" + assert (target.stat().st_mode, nested.stat().st_mode, payload.stat().st_mode) == before_modes + + def test_sweep_stale_polylogue_basetemps_gives_unknown_owner_a_long_grace_period( tmp_path: Path, ) -> None: