From 93ff44693a1798aefb6cd0f7bb28a9217cd4d14a Mon Sep 17 00:00:00 2001 From: kewton Date: Sun, 11 Oct 2026 19:13:17 +0900 Subject: [PATCH] Acquire the machine-wide heavy slot in live eval entry points MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live Python evals now take the same mkdir slot as the with-heavy-lock wrapper, so two live runs serialize and exclude cargo tests and any other holder. scripts/eval_lib/heavy_lock.py provides the shared context manager: atomic mkdir, pid/cmd in the wrapper shape, HEAVY_LOCK_DIR/HEAVY_LOCK_HELD compatibility, stale reclaim of a confirmed-dead owner guarded by an atomic reclaim mutex, nested-call skip, child inheritance, release of only the lock this process owns, and append-only wait records in /heavy-lock-waits.jsonl with nested calls distinguished. Wired: goal_verify live campaign (v0/v3/v4 and contract smoke), blind review, recovery-v4 smoke, the non-dry-run eval-run matrix (wait linked through the existing extras_json, no TSV column or schema change), and preflight with an explicit --live-provider-smoke. Dry-run, plain preflight, deterministic evaluation, and model-list GETs take no slot. Evidence: - python3 -m pytest tests/eval/test_heavy_lock.py -q -> 26 passed, 2 subtests passed - ruff check --isolated --select E4,E7,E9,F,I --ignore E402 -> All checks passed - Two-point: mutating HeavyLock._release_own to ignore ownership makes test_release_leaves_a_slot_it_no_longer_owns fail (1 failed); restoring it returns to 26 passed. 判断: added an atomic sibling reclaim mutex and HEAVY_LOCK_POLL_SECONDS because a bare rmtree reclaim raced with a freshly acquired slot when several waiters detected the same dead owner (observed as a FileNotFoundError on pid write). 判断: wait time is recorded in a new append-only /heavy-lock-waits.jsonl and only linked to the existing extras_json so frozen manifest, raw records, TSV header, and schema_version stay unchanged. 読み替え: the acceptance criterion "every generation of live campaign" is covered by a parametrized test that drives the live CLI with v3 and v4 contracts. 本文に無い指摘: pre-existing failures in tests/eval/test_goal_verify_live_v2.py, test_goal_verify_recovery_a15.py, and test_goal_verify_v3.py reproduce on the unmodified baseline and are not part of this issue's gates. Co-authored-by: CommandCodeBot --- scripts/eval-goal-verify-blind.py | 16 +- scripts/eval-goal-verify-live.py | 4 +- scripts/eval-goal-verify-recovery-v4.py | 18 +- scripts/eval-preflight.py | 24 +- scripts/eval-run.py | 109 ++-- scripts/eval_lib/heavy_lock.py | 318 +++++++++++ tests/eval/test_heavy_lock.py | 669 ++++++++++++++++++++++++ 7 files changed, 1081 insertions(+), 77 deletions(-) create mode 100644 scripts/eval_lib/heavy_lock.py create mode 100644 tests/eval/test_heavy_lock.py diff --git a/scripts/eval-goal-verify-blind.py b/scripts/eval-goal-verify-blind.py index 3e2866020..951500af0 100755 --- a/scripts/eval-goal-verify-blind.py +++ b/scripts/eval-goal-verify-blind.py @@ -9,6 +9,7 @@ sys.path.insert(0, str(ROOT / "scripts")) from eval_lib.goal_verify_blind import run_blind_review +from eval_lib.heavy_lock import heavy_lock def main() -> int: @@ -26,13 +27,14 @@ def main() -> int: def absolute(path: Path) -> Path: return path if path.is_absolute() else ROOT / path - report = run_blind_review( - root=ROOT, - baseline_path=absolute(args.baseline), - candidate_draft_path=absolute(args.candidate_draft), - contract_path=absolute(args.contract), - run_dir=absolute(args.run_dir), - ) + with heavy_lock(run_dir=absolute(args.run_dir), invocation="goal_verify_blind_review"): + report = run_blind_review( + root=ROOT, + baseline_path=absolute(args.baseline), + candidate_draft_path=absolute(args.candidate_draft), + contract_path=absolute(args.contract), + run_dir=absolute(args.run_dir), + ) print( f"[done] reviewed_pairs={report['reviewed_pairs']}/{report['expected_pairs']} " f"complete={str(report['complete']).lower()}" diff --git a/scripts/eval-goal-verify-live.py b/scripts/eval-goal-verify-live.py index 4ba81ca63..a99125264 100755 --- a/scripts/eval-goal-verify-live.py +++ b/scripts/eval-goal-verify-live.py @@ -12,6 +12,7 @@ from eval_lib.goal_verify_live import run_campaign from eval_lib.goal_verify_live_v3 import run_campaign_v3 from eval_lib.goal_verify_live_v4 import run_campaign_v4 +from eval_lib.heavy_lock import heavy_lock def main() -> int: @@ -98,7 +99,8 @@ def main() -> int: parser.error("contract.smoke pair_ids or request_namespace is invalid") runner_args["pair_ids"] = pair_ids runner_args["request_namespace"] = request_namespace - summary = runner(**runner_args) + with heavy_lock(run_dir=paths["run_dir"], invocation="goal_verify_live_campaign"): + summary = runner(**runner_args) print( f"[done] completed_pairs={summary['completed_pairs']}/{summary['target_pairs']} " f"completed_proposals={summary.get('completed_proposals', 'n/a')}" diff --git a/scripts/eval-goal-verify-recovery-v4.py b/scripts/eval-goal-verify-recovery-v4.py index 5e98f392a..0faefddf5 100755 --- a/scripts/eval-goal-verify-recovery-v4.py +++ b/scripts/eval-goal-verify-recovery-v4.py @@ -5,6 +5,7 @@ from pathlib import Path from eval_lib.goal_verify_recovery_live_v4 import run_recovery_smoke +from eval_lib.heavy_lock import heavy_lock def main() -> int: @@ -18,14 +19,15 @@ def main() -> int: parser.add_argument("--commandagent-bin", type=Path, required=True) parser.add_argument("--limit", type=int) args = parser.parse_args() - summary = run_recovery_smoke( - root=args.root.resolve(), - contract_path=args.contract.resolve(), - run_dir=args.run_dir.resolve(), - execution_root=args.execution_root.resolve(), - commandagent_bin=args.commandagent_bin.resolve(), - limit=args.limit, - ) + with heavy_lock(run_dir=args.run_dir.resolve(), invocation="goal_verify_recovery_v4"): + summary = run_recovery_smoke( + root=args.root.resolve(), + contract_path=args.contract.resolve(), + run_dir=args.run_dir.resolve(), + execution_root=args.execution_root.resolve(), + commandagent_bin=args.commandagent_bin.resolve(), + limit=args.limit, + ) print(summary) return 0 diff --git a/scripts/eval-preflight.py b/scripts/eval-preflight.py index 5c45808d9..656086db8 100755 --- a/scripts/eval-preflight.py +++ b/scripts/eval-preflight.py @@ -8,6 +8,7 @@ from eval_lib.artifacts import create_run_root, write_json, write_jsonl from eval_lib.config import credential_status, env_value, load_dotenv +from eval_lib.heavy_lock import heavy_lock from eval_lib.matrix import expand_matrix, parse_modes from eval_lib.models import ( gemini_interactions_smoke, @@ -109,17 +110,18 @@ def main() -> int: ollama = {"error": str(err)} live_provider_smoke = {} if args.live_provider_smoke: - requested = ( - required_providers_for_profile(profile) - if args.live_provider_smoke == "all" - else {part.strip() for part in args.live_provider_smoke.split(",") if part.strip()} - ) - if "openai" in requested: - key = env_value("OPENAI_API_KEY", dotenv) - live_provider_smoke["openai"] = smoke_openai_models(profile, key) - if "gemini" in requested: - key = env_value("GEMINI_API_KEY", dotenv) - live_provider_smoke["gemini"] = smoke_gemini_models(profile, key) + with heavy_lock(run_dir=run_root, invocation="eval_preflight_live_provider_smoke"): + requested = ( + required_providers_for_profile(profile) + if args.live_provider_smoke == "all" + else {part.strip() for part in args.live_provider_smoke.split(",") if part.strip()} + ) + if "openai" in requested: + key = env_value("OPENAI_API_KEY", dotenv) + live_provider_smoke["openai"] = smoke_openai_models(profile, key) + if "gemini" in requested: + key = env_value("GEMINI_API_KEY", dotenv) + live_provider_smoke["gemini"] = smoke_gemini_models(profile, key) ok = True if any(status == "missing" for status in deps.values()): diff --git a/scripts/eval-run.py b/scripts/eval-run.py index 2fd9097c7..419bf8c85 100755 --- a/scripts/eval-run.py +++ b/scripts/eval-run.py @@ -13,8 +13,8 @@ from pathlib import Path from env_compat import getenv -from eval_lib.acceptance_outcome import evaluate_acceptance_outcome from eval_lib.acceptance_contract import contract_from_scenario +from eval_lib.acceptance_outcome import evaluate_acceptance_outcome from eval_lib.artifacts import create_run_root, write_json, write_jsonl from eval_lib.browser_oracle import evaluate_browser_oracle from eval_lib.config import merge_dotenv_into_env @@ -25,8 +25,10 @@ normalize_failure_kind, read_jsonl, ) +from eval_lib.heavy_lock import heavy_lock from eval_lib.matrix import expand_matrix, parse_modes from eval_lib.models import load_model_profiles +from eval_lib.plan_capability_contract import score_plan_capability_contract from eval_lib.plan_readiness import ( READINESS_FIELDS, aggregate_ultra_phase_readiness, @@ -34,12 +36,11 @@ empty_plan_readiness_scores, score_plan_readiness_file, ) -from eval_lib.plan_capability_contract import score_plan_capability_contract -from eval_lib.plan_verify_coverage import score_plan_verify_coverage from eval_lib.plan_scoring import score_plan_file +from eval_lib.plan_verify_coverage import score_plan_verify_coverage from eval_lib.postcheck import is_dependency_command, run_postcheck -from eval_lib.redaction import redact_json from eval_lib.process import run_capture +from eval_lib.redaction import redact_json from eval_lib.run_summary import ( calculate_overall, calculate_plan_run_predictive_score, @@ -160,52 +161,58 @@ def main() -> int: print(f"[write] {run_root / 'events.jsonl'}") print("[done] dry-run complete") return 0 - semaphores = { - "openai": threading.Semaphore(provider_limit), - "gemini": threading.Semaphore(provider_limit), - } - parallel_items = [item for item in prepared if not item[1].get("serial_lane") and not item[1].get("port_mutex")] - serial_items = [item for item in prepared if item not in parallel_items] - results: list[tuple[int, dict, list[dict]]] = [] - if parallel_items: - print(f"[scheduler] cloud_parallel={min(args.parallel, len(parallel_items))} serial={len(serial_items)}") - queued_at = time.monotonic() - for _, spec, _, _, _ in parallel_items: - spec["_queued_at"] = queued_at - spec["_scheduler_lane"] = "cloud" - spec["_effective_parallelism"] = min(args.parallel, len(parallel_items)) - with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, args.parallel)) as pool: - futures = [pool.submit(run_prepared, item, args.timeout_sec, semaphores, len(matrix)) for item in parallel_items] - for future in concurrent.futures.as_completed(futures): - results.append(future.result()) - for item in serial_items: - _, spec, _, _, _ = item - spec["_queued_at"] = time.monotonic() - spec["_scheduler_lane"] = "serial" - spec["_effective_parallelism"] = 1 - results.append(run_prepared(item, args.timeout_sec, semaphores, len(matrix))) - for _, row, run_events in sorted(results, key=lambda item: item[0]): - rows.append(row) - events.extend(run_events) - apply_time_scores(rows) - apply_stability_scores(rows) - write_summary(run_root / "summary.eval.tsv", rows) - write_jsonl(run_root / "events.jsonl", events) - print(f"[write] {run_root / 'summary.eval.tsv'}") - print(f"[write] {run_root / 'events.jsonl'}") - trace = write_trace_artifacts( - run_root, - subject=trace_subject(matrix, args.binary_kind), - binary_kind=trace_binary_kind(matrix, args.binary_kind), - binary_path=args.binary, - commit_sha=getenv("COMMANDAGENT_EVAL_COMMIT_SHA", "") or "", - ) - print(f"[write] {trace['report_path']}") - print(f"[write] {trace['manifest_path']}") - failed = [row for row in rows if row.get("success") not in {True, "true", "diagnostic_skipped"}] - skipped = [row for row in rows if row.get("success") == "diagnostic_skipped"] - print(f"[done] success={len(rows)-len(failed)-len(skipped)} failed={len(failed)} skipped={len(skipped)}") - return 1 if failed else 0 + with heavy_lock(run_dir=run_root, invocation="eval_run_matrix") as heavy: + heavy_wait_sec = round(heavy.record.wait_sec, 3) if heavy.record else 0.0 + heavy_nested = bool(heavy.record.nested) if heavy.record else False + for spec in matrix: + spec["_heavy_lock_wait_sec"] = heavy_wait_sec + spec["_heavy_lock_nested"] = heavy_nested + semaphores = { + "openai": threading.Semaphore(provider_limit), + "gemini": threading.Semaphore(provider_limit), + } + parallel_items = [item for item in prepared if not item[1].get("serial_lane") and not item[1].get("port_mutex")] + serial_items = [item for item in prepared if item not in parallel_items] + results: list[tuple[int, dict, list[dict]]] = [] + if parallel_items: + print(f"[scheduler] cloud_parallel={min(args.parallel, len(parallel_items))} serial={len(serial_items)}") + queued_at = time.monotonic() + for _, spec, _, _, _ in parallel_items: + spec["_queued_at"] = queued_at + spec["_scheduler_lane"] = "cloud" + spec["_effective_parallelism"] = min(args.parallel, len(parallel_items)) + with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, args.parallel)) as pool: + futures = [pool.submit(run_prepared, item, args.timeout_sec, semaphores, len(matrix)) for item in parallel_items] + for future in concurrent.futures.as_completed(futures): + results.append(future.result()) + for item in serial_items: + _, spec, _, _, _ = item + spec["_queued_at"] = time.monotonic() + spec["_scheduler_lane"] = "serial" + spec["_effective_parallelism"] = 1 + results.append(run_prepared(item, args.timeout_sec, semaphores, len(matrix))) + for _, row, run_events in sorted(results, key=lambda item: item[0]): + rows.append(row) + events.extend(run_events) + apply_time_scores(rows) + apply_stability_scores(rows) + write_summary(run_root / "summary.eval.tsv", rows) + write_jsonl(run_root / "events.jsonl", events) + print(f"[write] {run_root / 'summary.eval.tsv'}") + print(f"[write] {run_root / 'events.jsonl'}") + trace = write_trace_artifacts( + run_root, + subject=trace_subject(matrix, args.binary_kind), + binary_kind=trace_binary_kind(matrix, args.binary_kind), + binary_path=args.binary, + commit_sha=getenv("COMMANDAGENT_EVAL_COMMIT_SHA", "") or "", + ) + print(f"[write] {trace['report_path']}") + print(f"[write] {trace['manifest_path']}") + failed = [row for row in rows if row.get("success") not in {True, "true", "diagnostic_skipped"}] + skipped = [row for row in rows if row.get("success") == "diagnostic_skipped"] + print(f"[done] success={len(rows)-len(failed)-len(skipped)} failed={len(failed)} skipped={len(skipped)}") + return 1 if failed else 0 def trace_binary_kind(matrix: list[dict], requested: str) -> str: @@ -421,6 +428,8 @@ def run_prepared( "provider_wait_sec": row["provider_wait_sec"], "queue_wait_sec": row["queue_wait_sec"], "wall_clock_sec": row["wall_clock_sec"], + "heavy_lock_wait_sec": spec.get("_heavy_lock_wait_sec", ""), + "heavy_lock_nested": spec.get("_heavy_lock_nested", ""), } ) run_events.append( diff --git a/scripts/eval_lib/heavy_lock.py b/scripts/eval_lib/heavy_lock.py new file mode 100644 index 000000000..b7f617b57 --- /dev/null +++ b/scripts/eval_lib/heavy_lock.py @@ -0,0 +1,318 @@ +from __future__ import annotations + +import json +import os +import shutil +import sys +import time +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path + +HEAVY_LOCK_DIR_ENV = "HEAVY_LOCK_DIR" +HEAVY_LOCK_HELD_ENV = "HEAVY_LOCK_HELD" +HEAVY_LOCK_POLL_SECONDS_ENV = "HEAVY_LOCK_POLL_SECONDS" +DEFAULT_HEAVY_LOCK_DIR = "/Volumes/SSD_NX/tmp/heavy-lock/slot.d" +DEFAULT_POLL_SECONDS = 2.0 +WAIT_RECORD_FILE = "heavy-lock-waits.jsonl" +WAIT_RECORD_EVENT = "heavy_lock" + +OWNER_MISSING = "missing" +OWNER_UNKNOWN = "unknown" +OWNER_DEAD = "dead" +OWNER_ALIVE = "alive" + + +class HeavyLockTimeout(RuntimeError): + """The machine-wide heavy slot was not acquired within the wait budget.""" + + +@dataclass(frozen=True) +class HeavyLockRecord: + invocation: str + lock_dir: str + held: bool + nested: bool + wait_sec: float + pid: int + + def as_event(self) -> dict[str, object]: + return { + "event": WAIT_RECORD_EVENT, + "invocation": self.invocation, + "lock_dir": self.lock_dir, + "held": self.held, + "nested": self.nested, + "wait_sec": self.wait_sec, + "pid": self.pid, + } + + +def resolve_lock_dir( + env: dict[str, str] | None = None, + override: str | os.PathLike[str] | None = None, +) -> Path: + if override is not None: + return Path(override) + env = os.environ if env is None else env + value = env.get(HEAVY_LOCK_DIR_ENV) + return Path(value) if value else Path(DEFAULT_HEAVY_LOCK_DIR) + + +def resolve_poll_seconds(env: dict[str, str] | None = None) -> float: + env = os.environ if env is None else env + raw = env.get(HEAVY_LOCK_POLL_SECONDS_ENV) + if not raw: + return DEFAULT_POLL_SECONDS + try: + value = float(raw) + except ValueError: + return DEFAULT_POLL_SECONDS + return value if value > 0 else DEFAULT_POLL_SECONDS + + +def wait_record_path(run_dir: str | os.PathLike[str]) -> Path: + return Path(run_dir) / WAIT_RECORD_FILE + + +def owner_state(lock_dir: Path) -> tuple[str, int | None]: + """Classify the slot owner without guessing about a missing or broken pid. + + ``OWNER_DEAD`` is the only state that may be reclaimed: the pid file must + hold a positive pid whose process no longer exists. A pid file that has not + been written yet, holds a non-numeric value, or cannot be read is reported + as ``OWNER_MISSING`` or ``OWNER_UNKNOWN`` and left in place. + """ + try: + raw = (lock_dir / "pid").read_text(encoding="utf-8").strip() + except FileNotFoundError: + return OWNER_MISSING, None + except OSError: + return OWNER_UNKNOWN, None + if not raw: + return OWNER_MISSING, None + try: + pid = int(raw) + except ValueError: + return OWNER_UNKNOWN, None + if pid <= 0: + return OWNER_UNKNOWN, None + try: + os.kill(pid, 0) + except ProcessLookupError: + return OWNER_DEAD, pid + except PermissionError: + return OWNER_ALIVE, pid + except OSError: + return OWNER_UNKNOWN, None + return OWNER_ALIVE, pid + + +def _write_owner(lock_dir: Path, invocation: str) -> None: + (lock_dir / "pid").write_text(f"{os.getpid()}\n", encoding="utf-8") + label = invocation or Path(sys.argv[0]).name + (lock_dir / "cmd").write_text(f"{Path.cwd()} :: {label}\n", encoding="utf-8") + + +def _reclaim_path(lock_dir: Path) -> Path: + return lock_dir.with_name(f"{lock_dir.name}.reclaim") + + +def _clear_dead_reclaimer(path: Path) -> None: + try: + pid = int(path.read_text(encoding="utf-8").strip()) + except (OSError, ValueError): + return + if pid <= 0: + return + try: + os.kill(pid, 0) + except ProcessLookupError: + try: + path.unlink() + except OSError: + pass + except OSError: + pass + + +def _enter_reclaim(lock_dir: Path) -> bool: + path = _reclaim_path(lock_dir) + try: + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) + except FileExistsError: + _clear_dead_reclaimer(path) + return False + try: + os.write(fd, f"{os.getpid()}\n".encode("utf-8")) + finally: + os.close(fd) + return True + + +def _leave_reclaim(lock_dir: Path) -> None: + try: + _reclaim_path(lock_dir).unlink() + except OSError: + pass + + +def _reclaim_stale(lock_dir: Path, dead_pid: int) -> bool: + """Remove a slot whose recorded owner is gone, one reclaimer at a time. + + The re-check and the removal happen while holding an atomic sibling mutex, + so two waiters cannot both delete a freshly re-acquired slot. + """ + if not _enter_reclaim(lock_dir): + return False + try: + state, pid = owner_state(lock_dir) + if state != OWNER_DEAD or pid != dead_pid: + return False + shutil.rmtree(lock_dir, ignore_errors=True) + return True + finally: + _leave_reclaim(lock_dir) + + +def _acquire( + lock_dir: Path, + *, + poll_seconds: float, + timeout_seconds: float | None, + invocation: str, + monotonic=time.monotonic, + sleep=time.sleep, +) -> float: + lock_dir.parent.mkdir(parents=True, exist_ok=True) + start = monotonic() + while True: + try: + lock_dir.mkdir() + except FileExistsError: + pass + else: + _write_owner(lock_dir, invocation) + return max(0.0, monotonic() - start) + state, pid = owner_state(lock_dir) + if state == OWNER_DEAD and pid is not None and _reclaim_stale(lock_dir, pid): + continue + elapsed = monotonic() - start + if timeout_seconds is not None and elapsed >= timeout_seconds: + raise HeavyLockTimeout( + f"heavy lock at {lock_dir} still held by a {state} owner after {elapsed:.3f}s" + ) + sleep(poll_seconds) + + +class HeavyLock: + """Context manager for the machine-wide heavy slot. + + The slot is the same ``mkdir`` directory the ``with-heavy-lock`` wrapper + uses, so Python live evals exclude against cargo tests and any other holder. + Nested callers (``HEAVY_LOCK_HELD`` already set) neither acquire nor release. + """ + + def __init__( + self, + *, + run_dir: str | os.PathLike[str] | None = None, + invocation: str = "", + lock_dir: str | os.PathLike[str] | None = None, + poll_seconds: float | None = None, + timeout_seconds: float | None = None, + env: dict[str, str] | None = None, + ) -> None: + self._env = os.environ if env is None else env + self.run_dir = Path(run_dir) if run_dir is not None else None + self.invocation = invocation or "unknown" + self.lock_dir = resolve_lock_dir(self._env, lock_dir) + self.poll_seconds = ( + poll_seconds if poll_seconds is not None else resolve_poll_seconds(self._env) + ) + self.timeout_seconds = timeout_seconds + self.record: HeavyLockRecord | None = None + self._acquired = False + self._had_previous_held = False + self._previous_held: str | None = None + + @property + def held(self) -> bool: + return self._acquired + + def __enter__(self) -> HeavyLock: + if self._env.get(HEAVY_LOCK_HELD_ENV): + self.record = HeavyLockRecord( + self.invocation, str(self.lock_dir), False, True, 0.0, os.getpid() + ) + self._write_record() + return self + wait_sec = _acquire( + self.lock_dir, + poll_seconds=self.poll_seconds, + timeout_seconds=self.timeout_seconds, + invocation=self.invocation, + ) + self._had_previous_held = HEAVY_LOCK_HELD_ENV in self._env + self._previous_held = self._env.get(HEAVY_LOCK_HELD_ENV) + self._env[HEAVY_LOCK_HELD_ENV] = str(os.getpid()) + self._acquired = True + self.record = HeavyLockRecord( + self.invocation, str(self.lock_dir), True, False, round(wait_sec, 6), os.getpid() + ) + self._write_record() + return self + + def __exit__(self, exc_type, exc, tb) -> bool: + self.release() + return False + + def release(self) -> None: + if not self._acquired: + return + self._release_own() + self._restore_env() + self._acquired = False + + def _release_own(self) -> None: + state, pid = owner_state(self.lock_dir) + if state == OWNER_ALIVE and pid == os.getpid(): + shutil.rmtree(self.lock_dir, ignore_errors=True) + + def _restore_env(self) -> None: + if self._had_previous_held: + self._env[HEAVY_LOCK_HELD_ENV] = self._previous_held or "" + else: + self._env.pop(HEAVY_LOCK_HELD_ENV, None) + + def _write_record(self) -> None: + if self.record is None or self.run_dir is None: + return + path = wait_record_path(self.run_dir) + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a", encoding="utf-8") as handle: + handle.write( + json.dumps(self.record.as_event(), ensure_ascii=False, sort_keys=True) + "\n" + ) + + +@contextmanager +def heavy_lock( + run_dir: str | os.PathLike[str] | None = None, + invocation: str = "", + *, + lock_dir: str | os.PathLike[str] | None = None, + poll_seconds: float | None = None, + timeout_seconds: float | None = None, + env: dict[str, str] | None = None, +): + lock = HeavyLock( + run_dir=run_dir, + invocation=invocation, + lock_dir=lock_dir, + poll_seconds=poll_seconds, + timeout_seconds=timeout_seconds, + env=env, + ) + with lock as entered: + yield entered diff --git a/tests/eval/test_heavy_lock.py b/tests/eval/test_heavy_lock.py new file mode 100644 index 000000000..15cf560a5 --- /dev/null +++ b/tests/eval/test_heavy_lock.py @@ -0,0 +1,669 @@ +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import threading +import time +import unittest +from contextlib import contextmanager +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT / "scripts")) + +from eval_lib.heavy_lock import ( + DEFAULT_HEAVY_LOCK_DIR, + HEAVY_LOCK_DIR_ENV, + HEAVY_LOCK_HELD_ENV, + HEAVY_LOCK_POLL_SECONDS_ENV, + HeavyLockTimeout, + heavy_lock, + owner_state, + resolve_lock_dir, + resolve_poll_seconds, + wait_record_path, +) +from eval_lib.run_summary import read_summary + + +def load_script(name: str, relative: str): + module_path = ROOT / relative + spec = importlib.util.spec_from_file_location(name, module_path) + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +@contextmanager +def heavy_env(lock_dir: Path): + keys = (HEAVY_LOCK_DIR_ENV, HEAVY_LOCK_HELD_ENV, HEAVY_LOCK_POLL_SECONDS_ENV) + previous = {key: os.environ.get(key) for key in keys} + os.environ[HEAVY_LOCK_DIR_ENV] = str(lock_dir) + os.environ.pop(HEAVY_LOCK_HELD_ENV, None) + os.environ[HEAVY_LOCK_POLL_SECONDS_ENV] = "0.01" + try: + yield + finally: + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + +def read_records(run_dir: Path) -> list[dict]: + path = wait_record_path(run_dir) + if not path.exists(): + return [] + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] + + +def dead_pid() -> int | None: + proc = subprocess.Popen([sys.executable, "-c", "pass"]) + proc.wait() + try: + os.kill(proc.pid, 0) + except ProcessLookupError: + return proc.pid + except OSError: + return None + return None + + +class HeavyLockCoreTest(unittest.TestCase): + def test_acquire_records_owner_shape_and_releases(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_dir = Path(td) / "run" + env: dict[str, str] = {} + with heavy_lock( + run_dir=run_dir, + invocation="unit", + lock_dir=lock_dir, + poll_seconds=0.01, + timeout_seconds=5, + env=env, + ) as lock: + self.assertTrue(lock.held) + self.assertFalse(lock.record.nested) + self.assertTrue(lock_dir.is_dir()) + self.assertEqual((lock_dir / "pid").read_text().strip(), str(os.getpid())) + self.assertIn("unit", (lock_dir / "cmd").read_text()) + self.assertEqual(env[HEAVY_LOCK_HELD_ENV], str(os.getpid())) + self.assertFalse(lock_dir.exists()) + self.assertNotIn(HEAVY_LOCK_HELD_ENV, env) + records = read_records(run_dir) + self.assertEqual(len(records), 1) + self.assertTrue(records[0]["held"]) + self.assertFalse(records[0]["nested"]) + self.assertGreaterEqual(records[0]["wait_sec"], 0.0) + self.assertEqual(records[0]["invocation"], "unit") + + def test_nested_holder_skips_acquire_and_release(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_dir = Path(td) / "run" + env = {HEAVY_LOCK_HELD_ENV: "4242"} + with heavy_lock( + run_dir=run_dir, + invocation="nested", + lock_dir=lock_dir, + env=env, + ) as lock: + self.assertFalse(lock.held) + self.assertTrue(lock.record.nested) + self.assertEqual(lock.record.wait_sec, 0.0) + self.assertFalse(lock_dir.exists()) + self.assertFalse(lock_dir.exists()) + self.assertEqual(env[HEAVY_LOCK_HELD_ENV], "4242") + records = read_records(run_dir) + self.assertEqual(len(records), 1) + self.assertFalse(records[0]["held"]) + self.assertTrue(records[0]["nested"]) + + def test_second_holder_waits_for_first_to_release(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + order: list[str] = [] + waits: dict[str, float] = {} + first_entered = threading.Event() + + def worker(name: str, hold_sec: float): + with heavy_lock( + invocation=name, + lock_dir=lock_dir, + poll_seconds=0.01, + timeout_seconds=10, + env={}, + ) as lock: + waits[name] = lock.record.wait_sec + order.append(name) + if name == "first": + first_entered.set() + time.sleep(hold_sec) + + first = threading.Thread(target=worker, args=("first", 0.3)) + first.start() + self.assertTrue(first_entered.wait(timeout=5)) + second = threading.Thread(target=worker, args=("second", 0.0)) + second.start() + first.join(timeout=10) + second.join(timeout=10) + self.assertEqual(order, ["first", "second"]) + self.assertGreaterEqual(waits["second"], 0.15) + self.assertFalse(lock_dir.exists()) + + def test_child_process_inherits_the_held_slot(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + with heavy_env(lock_dir): + with heavy_lock(invocation="parent"): + child = subprocess.run( + [ + sys.executable, + "-c", + "import os; print(os.environ.get('HEAVY_LOCK_HELD', ''))", + ], + text=True, + capture_output=True, + ) + self.assertEqual(child.returncode, 0, child.stderr) + self.assertEqual(child.stdout.strip(), str(os.getpid())) + + def test_child_nested_call_does_not_touch_the_slot(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + code = ( + "import sys; sys.path.insert(0, sys.argv[1]);" + "from eval_lib.heavy_lock import heavy_lock;" + "ctx = heavy_lock(lock_dir=sys.argv[2], invocation='child');" + "lock = ctx.__enter__();" + "print(lock.record.nested, lock.held);" + "ctx.__exit__(None, None, None)" + ) + proc = subprocess.run( + [sys.executable, "-c", code, str(ROOT / "scripts"), str(lock_dir)], + env={**os.environ, HEAVY_LOCK_HELD_ENV: "777"}, + text=True, + capture_output=True, + ) + self.assertEqual(proc.returncode, 0, proc.stderr) + self.assertEqual(proc.stdout.strip(), "True False") + self.assertFalse(lock_dir.exists()) + + def test_stale_owner_is_reclaimed(self): + pid = dead_pid() + if pid is None: + self.skipTest("could not obtain a dead pid") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + lock_dir.mkdir(parents=True) + (lock_dir / "pid").write_text(f"{pid}\n", encoding="utf-8") + (lock_dir / "cmd").write_text("dead holder\n", encoding="utf-8") + with heavy_lock( + invocation="reclaim", + lock_dir=lock_dir, + poll_seconds=0.01, + timeout_seconds=5, + env={}, + ) as lock: + self.assertTrue(lock.held) + self.assertEqual((lock_dir / "pid").read_text().strip(), str(os.getpid())) + self.assertFalse(lock_dir.exists()) + + def test_alive_owner_is_protected_until_timeout(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + lock_dir.mkdir(parents=True) + (lock_dir / "pid").write_text(f"{os.getpid()}\n", encoding="utf-8") + with self.assertRaises(HeavyLockTimeout): + with heavy_lock( + lock_dir=lock_dir, + poll_seconds=0.01, + timeout_seconds=0.1, + env={}, + ): + self.fail("must not enter while the slot is held") + self.assertTrue(lock_dir.exists()) + self.assertEqual((lock_dir / "pid").read_text().strip(), str(os.getpid())) + + def test_missing_pid_before_recording_is_protected(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + lock_dir.mkdir(parents=True) + with self.assertRaises(HeavyLockTimeout): + with heavy_lock(lock_dir=lock_dir, poll_seconds=0.01, timeout_seconds=0.1, env={}): + self.fail("must not enter on an unrecorded owner") + self.assertTrue(lock_dir.exists()) + + def test_corrupt_pid_is_protected(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + lock_dir.mkdir(parents=True) + (lock_dir / "pid").write_text("not-a-pid\n", encoding="utf-8") + state, pid = owner_state(lock_dir) + self.assertEqual(state, "unknown") + self.assertIsNone(pid) + with self.assertRaises(HeavyLockTimeout): + with heavy_lock(lock_dir=lock_dir, poll_seconds=0.01, timeout_seconds=0.1, env={}): + self.fail("must not enter on a corrupt owner") + self.assertTrue(lock_dir.exists()) + + def test_non_positive_pid_is_protected(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + lock_dir.mkdir(parents=True) + (lock_dir / "pid").write_text("0\n", encoding="utf-8") + with self.assertRaises(HeavyLockTimeout): + with heavy_lock(lock_dir=lock_dir, poll_seconds=0.01, timeout_seconds=0.1, env={}): + self.fail("must not enter on an invalid owner") + self.assertTrue(lock_dir.exists()) + + def test_release_leaves_a_slot_it_no_longer_owns(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + env: dict[str, str] = {} + with heavy_lock(lock_dir=lock_dir, env=env) as lock: + (lock_dir / "pid").write_text("1\n", encoding="utf-8") + lock.release() + self.assertTrue(lock_dir.exists()) + self.assertTrue(lock_dir.exists()) + self.assertNotIn(HEAVY_LOCK_HELD_ENV, env) + + def test_exception_releases_the_slot(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + env: dict[str, str] = {} + with self.assertRaises(ValueError): + with heavy_lock(lock_dir=lock_dir, env=env): + raise ValueError("boom") + self.assertFalse(lock_dir.exists()) + self.assertNotIn(HEAVY_LOCK_HELD_ENV, env) + + def test_keyboard_interrupt_releases_the_slot(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + env: dict[str, str] = {} + with self.assertRaises(KeyboardInterrupt): + with heavy_lock(lock_dir=lock_dir, env=env): + raise KeyboardInterrupt + self.assertFalse(lock_dir.exists()) + self.assertNotIn(HEAVY_LOCK_HELD_ENV, env) + + def test_wait_records_append_and_do_not_overwrite_on_resume(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_dir = Path(td) / "run" + run_dir.mkdir(parents=True) + wait_record_path(run_dir).write_text( + json.dumps({"event": "heavy_lock", "invocation": "previous"}) + "\n", + encoding="utf-8", + ) + with heavy_lock(run_dir=run_dir, invocation="one", lock_dir=lock_dir, env={}): + pass + with heavy_lock(run_dir=run_dir, invocation="two", lock_dir=lock_dir, env={}): + pass + records = read_records(run_dir) + self.assertEqual( + [record["invocation"] for record in records], + ["previous", "one", "two"], + ) + + def test_many_waiters_keep_single_holder_and_all_enter(self): + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + guard = threading.Lock() + active = 0 + max_active = 0 + entered = 0 + + def worker(): + nonlocal active, max_active, entered + with heavy_lock( + invocation="w", + lock_dir=lock_dir, + poll_seconds=0.005, + timeout_seconds=15, + env={}, + ): + with guard: + active += 1 + max_active = max(max_active, active) + entered += 1 + time.sleep(0.02) + with guard: + active -= 1 + + threads = [threading.Thread(target=worker) for _ in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=20) + self.assertEqual(entered, 4) + self.assertEqual(max_active, 1) + self.assertFalse(lock_dir.exists()) + + def test_dead_holder_race_yields_one_holder(self): + pid = dead_pid() + if pid is None: + self.skipTest("could not obtain a dead pid") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + lock_dir.mkdir(parents=True) + (lock_dir / "pid").write_text(f"{pid}\n", encoding="utf-8") + guard = threading.Lock() + active = 0 + max_active = 0 + entered = 0 + + def worker(): + nonlocal active, max_active, entered + with heavy_lock( + invocation="r", + lock_dir=lock_dir, + poll_seconds=0.005, + timeout_seconds=15, + env={}, + ): + with guard: + active += 1 + max_active = max(max_active, active) + entered += 1 + time.sleep(0.02) + with guard: + active -= 1 + + threads = [threading.Thread(target=worker) for _ in range(3)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=20) + self.assertEqual(entered, 3) + self.assertEqual(max_active, 1) + self.assertFalse(lock_dir.exists()) + + def test_default_slot_matches_the_wrapper(self): + self.assertEqual(DEFAULT_HEAVY_LOCK_DIR, "/Volumes/SSD_NX/tmp/heavy-lock/slot.d") + self.assertEqual(resolve_lock_dir({}), Path(DEFAULT_HEAVY_LOCK_DIR)) + self.assertEqual(resolve_lock_dir({}, "/tmp/other.d"), Path("/tmp/other.d")) + self.assertEqual( + resolve_lock_dir({HEAVY_LOCK_DIR_ENV: "/tmp/env.d"}), + Path("/tmp/env.d"), + ) + + def test_poll_seconds_env_is_honoured(self): + self.assertEqual(resolve_poll_seconds({}), 2.0) + self.assertEqual(resolve_poll_seconds({HEAVY_LOCK_POLL_SECONDS_ENV: "0.25"}), 0.25) + self.assertEqual(resolve_poll_seconds({HEAVY_LOCK_POLL_SECONDS_ENV: "bad"}), 2.0) + + +class HeavyLockEntrypointTest(unittest.TestCase): + def _assert_slot_was_released(self, lock_dir: Path, run_dir: Path): + self.assertFalse(lock_dir.exists()) + records = read_records(run_dir) + self.assertEqual(len(records), 1) + self.assertTrue(records[0]["held"]) + self.assertFalse(records[0]["nested"]) + + def test_goal_verify_live_entrypoint_takes_the_slot(self): + module = load_script("cli_goal_verify_live", "scripts/eval-goal-verify-live.py") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_dir = Path(td) / "run" + observed = {} + + def fake_campaign(**kwargs): + observed["held"] = os.environ.get(HEAVY_LOCK_HELD_ENV) + observed["slot_exists"] = lock_dir.exists() + return {"completed_pairs": 0, "target_pairs": 0} + + module.run_campaign = fake_campaign + with heavy_env(lock_dir), mock.patch.object( + sys, "argv", ["eval-goal-verify-live.py", "--run-dir", str(run_dir)] + ): + self.assertEqual(module.main(), 0) + self.assertEqual(observed["held"], str(os.getpid())) + self.assertTrue(observed["slot_exists"]) + self._assert_slot_was_released(lock_dir, run_dir) + + def test_goal_verify_live_acquires_for_every_generation(self): + module = load_script("cli_goal_verify_live_gen", "scripts/eval-goal-verify-live.py") + generations = { + "v3": ("commandagent.goal_verify.phase6_preflight_contract.v3", "run_campaign_v3"), + "v4": ("commandagent.goal_verify.phase6_preflight_contract.v4", "run_campaign_v4"), + } + for name, (schema_version, runner_name) in generations.items(): + with self.subTest(generation=name), tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_dir = Path(td) / "run" + contract = Path(td) / "contract.json" + contract.write_text( + json.dumps({"schema_version": schema_version}), encoding="utf-8" + ) + called = {} + + def fake_campaign(**kwargs): + called["held"] = os.environ.get(HEAVY_LOCK_HELD_ENV) + called["slot_exists"] = lock_dir.exists() + return {"completed_pairs": 0, "target_pairs": 0} + + setattr(module, runner_name, fake_campaign) + argv = [ + "eval-goal-verify-live.py", + "--contract", + str(contract), + "--run-dir", + str(run_dir), + "--execution-root", + str(Path(td) / "exec"), + ] + with heavy_env(lock_dir), mock.patch.object(sys, "argv", argv): + self.assertEqual(module.main(), 0) + self.assertEqual(called["held"], str(os.getpid())) + self.assertTrue(called["slot_exists"]) + self._assert_slot_was_released(lock_dir, run_dir) + + def test_goal_verify_blind_entrypoint_takes_the_slot(self): + module = load_script("cli_goal_verify_blind", "scripts/eval-goal-verify-blind.py") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_dir = Path(td) / "run" + observed = {} + + def fake_review(**kwargs): + observed["held"] = os.environ.get(HEAVY_LOCK_HELD_ENV) + observed["slot_exists"] = lock_dir.exists() + return {"reviewed_pairs": 0, "expected_pairs": 0, "complete": True} + + module.run_blind_review = fake_review + argv = [ + "eval-goal-verify-blind.py", + "--baseline", + str(Path(td) / "baseline"), + "--candidate-draft", + str(Path(td) / "draft"), + "--run-dir", + str(run_dir), + ] + with heavy_env(lock_dir), mock.patch.object(sys, "argv", argv): + self.assertEqual(module.main(), 0) + self.assertEqual(observed["held"], str(os.getpid())) + self.assertTrue(observed["slot_exists"]) + self._assert_slot_was_released(lock_dir, run_dir) + + def test_goal_verify_recovery_entrypoint_takes_the_slot(self): + module = load_script("cli_goal_verify_recovery", "scripts/eval-goal-verify-recovery-v4.py") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_dir = Path(td) / "run" + observed = {} + + def fake_recovery(**kwargs): + observed["held"] = os.environ.get(HEAVY_LOCK_HELD_ENV) + observed["slot_exists"] = lock_dir.exists() + return "recovery-summary" + + module.run_recovery_smoke = fake_recovery + argv = [ + "eval-goal-verify-recovery-v4.py", + "--contract", + str(Path(td) / "contract.json"), + "--run-dir", + str(run_dir), + "--execution-root", + str(Path(td) / "exec"), + "--commandagent-bin", + str(Path(td) / "commandagent"), + ] + with heavy_env(lock_dir), mock.patch.object(sys, "argv", argv): + self.assertEqual(module.main(), 0) + self.assertEqual(observed["held"], str(os.getpid())) + self.assertTrue(observed["slot_exists"]) + self._assert_slot_was_released(lock_dir, run_dir) + + def test_eval_run_matrix_takes_the_slot_and_records_wait_in_extras(self): + module = load_script("cli_eval_run", "scripts/eval-run.py") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_root = Path(td) / "run" + observed = {} + original_run_prepared = module.run_prepared + + def fake_run_prepared(item, timeout_sec, semaphores, total): + index, spec, _, run_dir, workdir = item + observed["held"] = os.environ.get(HEAVY_LOCK_HELD_ENV) + observed["slot_exists"] = lock_dir.exists() + return original_run_prepared( + (index, spec, ["true"], run_dir, workdir), + timeout_sec, + semaphores, + total, + ) + + module.run_prepared = fake_run_prepared + module.write_trace_artifacts = lambda *a, **k: { + "report_path": "report", + "manifest_path": "manifest", + } + argv = [ + "eval-run.py", + "--suite", + "eval/suites/mvp-smoke.yaml", + "--model-profile", + "openai-only", + "--modes", + "minimal-loop", + "--runs", + "1", + "--scenario", + "docs-heading-update-small", + "--run-root", + str(run_root), + ] + with heavy_env(lock_dir), mock.patch.object(sys, "argv", argv): + self.assertIn(module.main(), (0, 1)) + self.assertEqual(observed["held"], str(os.getpid())) + self.assertTrue(observed["slot_exists"]) + self._assert_slot_was_released(lock_dir, run_root) + rows = read_summary(run_root / "summary.eval.tsv") + extras = json.loads(rows[0]["extras_json"]) + self.assertIn("heavy_lock_wait_sec", extras) + self.assertIn("heavy_lock_nested", extras) + self.assertFalse(extras["heavy_lock_nested"]) + + def test_eval_run_dry_run_does_not_take_the_slot(self): + module = load_script("cli_eval_run_dry", "scripts/eval-run.py") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_root = Path(td) / "run" + argv = [ + "eval-run.py", + "--suite", + "eval/suites/mvp-smoke.yaml", + "--model-profile", + "openai-only", + "--modes", + "minimal-loop", + "--runs", + "1", + "--run-root", + str(run_root), + "--dry-run", + ] + with heavy_env(lock_dir), mock.patch.object(sys, "argv", argv): + self.assertEqual(module.main(), 0) + self.assertFalse(lock_dir.exists()) + self.assertEqual(read_records(run_root), []) + + def test_eval_preflight_live_provider_smoke_takes_the_slot(self): + module = load_script("cli_eval_preflight", "scripts/eval-preflight.py") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_root = Path(td) / "run" + observed = {} + + def fake_smoke(profile, api_key): + observed["held"] = os.environ.get(HEAVY_LOCK_HELD_ENV) + observed["slot_exists"] = lock_dir.exists() + return {"model": {"ok": True}} + + module.smoke_openai_models = fake_smoke + argv = [ + "eval-preflight.py", + "--suite", + "eval/suites/mvp-smoke.yaml", + "--model-profile", + "openai-only", + "--modes", + "minimal-loop", + "--runs", + "1", + "--offline-ok", + "--binary", + "python3", + "--live-provider-smoke", + "openai", + "--run-root", + str(run_root), + ] + with heavy_env(lock_dir), mock.patch.object(sys, "argv", argv): + self.assertEqual(module.main(), 0) + self.assertEqual(observed["held"], str(os.getpid())) + self.assertTrue(observed["slot_exists"]) + self._assert_slot_was_released(lock_dir, run_root) + payload = json.loads((run_root / "preflight.json").read_text(encoding="utf-8")) + self.assertIn("openai", payload["live_provider_smoke"]) + + def test_eval_preflight_without_smoke_does_not_take_the_slot(self): + module = load_script("cli_eval_preflight_plain", "scripts/eval-preflight.py") + with tempfile.TemporaryDirectory() as td: + lock_dir = Path(td) / "slot.d" + run_root = Path(td) / "run" + argv = [ + "eval-preflight.py", + "--suite", + "eval/suites/mvp-smoke.yaml", + "--model-profile", + "openai-only", + "--modes", + "minimal-loop", + "--runs", + "1", + "--offline-ok", + "--binary", + "python3", + "--run-root", + str(run_root), + ] + with heavy_env(lock_dir), mock.patch.object(sys, "argv", argv): + self.assertEqual(module.main(), 0) + self.assertFalse(lock_dir.exists()) + self.assertEqual(read_records(run_root), []) + + +if __name__ == "__main__": + unittest.main()