Close completed test harness Bead

devtools/incident_coverage_ledger.py

@@ -398,10 +398,22 @@ def _committed_paths() -> set[str]:
 
 
 def _validate_graph_provenance(graph: JsonObject, *, beads_path: Path) -> None:
-    source_commit = _string(graph.get("source_commit"), context="campaign graph source_commit")
     source_path = _string(graph.get("source_path"), context="campaign graph source_path")
     if source_path != ".beads/issues.jsonl":
         _fail("graph_source_path_invalid", f"campaign graph source path must be .beads/issues.jsonl, got {source_path}")
+    snapshot_digest = graph.get("source_snapshot_sha256")
+    if isinstance(snapshot_digest, str) and snapshot_digest:
+        actual_digest = hashlib.sha256(beads_path.read_bytes()).hexdigest()
+        if actual_digest != snapshot_digest:
+            _fail(
+                "graph_source_snapshot_mismatch",
+                "campaign graph source snapshot digest does not match current Beads export",
+                source_path=source_path,
+                expected_digest=snapshot_digest,
+                actual_digest=actual_digest,
+            )
+        return
+    source_commit = _string(graph.get("source_commit"), context="campaign graph source_commit")
     try:
         subprocess.run(
             ["git", "cat-file", "-e", f"{source_commit}^{{commit}}"],

devtools/pytest_progress_plugin.py

@@ -120,6 +120,10 @@ def pytest_sessionstart(session: Any) -> None:
     _SLOWEST_REPORTS.clear()
     _COLLECTION_STARTED_AT = None
     _COLLECTION_DURATION_S = None
+    # The worker environment is assigned after process exec, so it is not
+    # reliably visible through /proc/<pid>/environ.  Emit the identity from
+    # inside the worker for the supervisor's process-state sampler.
+    _write_event({"event": "session_started"})
 
 
 @pytest.hookimpl

devtools/verify.py

@@ -1493,7 +1493,19 @@ def _run(
         except PytestResourceError as exc:
             elapsed = time.monotonic() - t0
             sys.stderr.write(f"FAILED ({elapsed:.1f}s)\nverify: {exc}\n")
-            return 125, elapsed, {"diagnosis": "pytest_resource_preflight_failed", "error": str(exc)}
+            refusal_metadata: dict[str, Any] = {
+                "diagnosis": "pytest_resource_preflight_failed",
+                "error": str(exc),
+                "termination_reason": "pytest resource preflight refused basetemp admission",
+                "verification_scope": "narrow-terminal",
+                "release_baseline_allowed": False,
+            }
+            if run is not None and artifacts is not None:
+                run.finish_step(
+                    step_id=artifacts.step_id,
+                    result={"duration_s": round(elapsed, 2), "exit": 125, **refusal_metadata},
+                )
+            return 125, elapsed, refusal_metadata
         pytest_tmpfs = env.get("POLYLOGUE_PYTEST_TMPFS") == "1"
         budget_kb = pytest_tmpfs_budget_kb(env)
         pytest_tmpfs_budget_mb = budget_kb / 1024 if budget_kb is not None else None

devtools/verify_runs.py

@@ -875,6 +875,29 @@ def cleanup_managed_pytest_basetemp(*, root: Path, run_id: str, env: dict[str, s
     return None
 
 
+def _pytest_event_worker_ids(events_dir: Path | None) -> dict[int, str]:
+    """Recover xdist worker identities emitted after process exec.
+
+    ``PYTEST_XDIST_WORKER`` is not guaranteed to appear in ``/proc``'s
+    exec-time environment.  The progress plugin emits a session-start event
+    from inside each worker, which is the authoritative identity for the
+    supervisor sampler.
+    """
+    if events_dir is None or not events_dir.is_dir():
+        return {}
+    identities: dict[int, str] = {}
+    for path in events_dir.glob("*.jsonl"):
+        with contextlib.suppress(OSError, UnicodeDecodeError):
+            for line in path.read_text(encoding="utf-8").splitlines():
+                with contextlib.suppress(json.JSONDecodeError):
+                    payload = json.loads(line)
+                    pid = payload.get("pid")
+                    worker_id = payload.get("worker_id")
+                    if isinstance(pid, int) and isinstance(worker_id, str) and worker_id != "controller":
+                        identities[pid] = worker_id
+    return identities
+
+
 class ResourceSampler:
     """Samples host and process-tree resources for one subprocess tree."""
 
@@ -884,6 +907,7 @@ def __init__(self, *, root_pid: int, run_id: str, root: Path, env: dict[str, str
         self.root = root
         self.env = env
         self.output_path = output_path
+        self.events_dir = Path(env["POLYLOGUE_PYTEST_EVENTS_DIR"]) if env.get("POLYLOGUE_PYTEST_EVENTS_DIR") else None
         self.sample_count = 0
         self.peak_rss_kb = 0
         self.peak_pss_kb: int | None = None
@@ -930,6 +954,7 @@ def sample(self, *, event: str) -> dict[str, Any]:
         total_cpu = 0.0
         xdist_worker_count = 0
         xdist_uninterruptible_count = 0
+        event_worker_ids = _pytest_event_worker_ids(self.events_dir)
         for pid in pids:
             status = _status_values(pid)
             rss = int(status.get("rss_kb") or 0)
@@ -940,7 +965,7 @@ def sample(self, *, event: str) -> dict[str, Any]:
             swap_pss = smaps.get("SwapPss")
             cpu = _cpu_seconds(pid)
             process_identity = _process_identity(pid)
-            worker_id = _process_environ_value(pid, "PYTEST_XDIST_WORKER")
+            worker_id = _process_environ_value(pid, "PYTEST_XDIST_WORKER") or event_worker_ids.get(pid)
             if worker_id is not None:
                 xdist_worker_count += 1
                 if str(status.get("state") or "").startswith("D"):

tests/fixtures/reindex_incident_coverage/campaign_graph.json

@@ -1,6 +1,7 @@
 {
   "schema_version": 1,
-  "source_commit": "dfb4854b67076f8c8708eedafc31c513c6873526",
+  "source_commit": "12618c007a987b227e73cfc9e4f462dcaa669914",
+  "source_snapshot_sha256": "767d60ce405c65d6fa19a8302c6a6920f054ca8be6adbfaca1d0c88444b9b424",
   "source_path": ".beads/issues.jsonl",
   "target_bead_id": "polylogue-818fy",
   "forcing_dependencies": [

tests/infra/workload_artifacts.py

@@ -345,6 +345,26 @@ def _remove_tree(path: Path) -> None:
     shutil.rmtree(path)
 
 
+def _recover_stale_staging(*, staging_root: Path, artifact_name: str) -> tuple[str, ...]:
+    """Remove only crash-left staging trees for the currently owned build.
+
+    The per-key flock is held by the caller, so no live builder for this
+    artifact can be using these paths while this sweep runs.  A completed
+    artifact is published by ``os.replace``; anything left under the matching
+    staging prefix is therefore an incomplete build from a process that died
+    before publication.  Keeping those trees made a SIGKILL leak large SQLite
+    databases indefinitely and allowed a later cache inspection to mistake a
+    partial build for reusable state.
+    """
+    removed: list[str] = []
+    for candidate in sorted(staging_root.glob(f"{artifact_name}.*")):
+        if not candidate.is_dir():
+            continue
+        _remove_tree(candidate)
+        removed.append(candidate.name)
+    return tuple(removed)
+
+
 def _validate_facts(root: Path, facts: tuple[SyntheticArtifactFacts, ...]) -> None:
     with contextlib.closing(sqlite3.connect(root / "index.db")) as conn:
         session_ids = {str(row[0]) for row in conn.execute("SELECT session_id FROM sessions")}
@@ -423,6 +443,7 @@ def build_seeded_archive(
 
     with lock_path.open("a+") as handle:
         fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
+        _recover_stale_staging(staging_root=staging_root, artifact_name=final_root.name)
         cached = _validate_artifact(final_root, key)
         if cached is not None:
             return cached

tests/unit/devtools/test_incident_coverage_ledger.py

@@ -12,6 +12,7 @@
     CAMPAIGN_GRAPH_PATH,
     LEDGER_PATH,
     IncidentCoverageLedgerError,
+    _validate_graph_provenance,
     load_beads_jsonl,
     load_campaign_graph,
     load_ledger,
@@ -55,6 +56,19 @@ def test_real_campaign_graph_resolves_the_current_forcing_set() -> None:
     assert CAMPAIGN_GRAPH_PATH.is_file()
 
 
+def test_campaign_graph_snapshot_digest_survives_missing_feature_commit(tmp_path: Path) -> None:
+    graph = _graph()
+    graph.pop("source_commit", None)
+    graph["source_snapshot_sha256"] = "not-the-current-export"
+    beads_path = tmp_path / "issues.jsonl"
+    beads_path.write_bytes((Path.cwd() / ".beads" / "issues.jsonl").read_bytes())
+
+    with pytest.raises(IncidentCoverageLedgerError) as error:
+        _validate_graph_provenance(graph, beads_path=beads_path)
+
+    assert error.value.diagnostic["error"] == "graph_source_snapshot_mismatch"
+
+
 def test_deleting_a_ledger_row_emits_machine_readable_missing_id() -> None:
     ledger = _ledger()
     _rows(ledger)[:] = [row for row in _rows(ledger) if row["bead_id"] != "polylogue-xselt"]

tests/unit/devtools/test_verify.py

@@ -1384,6 +1384,41 @@ def test_xdist_uninterruptible_stall_ignores_partial_or_moving_workers() -> None
     )
 
 
+def test_resource_sampler_resolves_worker_identity_from_in_process_events(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    events = tmp_path / "events"
+    events.mkdir()
+    (events / "worker.jsonl").write_text(
+        json.dumps({"event": "session_started", "pid": 101, "worker_id": "gw0"}) + "\n",
+        encoding="utf-8",
+    )
+    monkeypatch.setattr("devtools.verify_runs.process_tree", lambda _root_pid: [101])
+    monkeypatch.setattr("devtools.verify_runs._status_values", lambda _pid: {"state": "D", "rss_kb": 30})
+    monkeypatch.setattr("devtools.verify_runs._smaps_rollup_kb", lambda _pid: {})
+    monkeypatch.setattr("devtools.verify_runs._process_io_bytes", lambda _pid: {})
+    monkeypatch.setattr("devtools.verify_runs._process_identity", lambda _pid: "101:1")
+    monkeypatch.setattr("devtools.verify_runs._cpu_seconds", lambda _pid: 1.0)
+    monkeypatch.setattr("devtools.verify_runs._process_environ_value", lambda _pid, _key: None)
+
+    sampler = ResourceSampler(
+        root_pid=101,
+        run_id="worker-events",
+        root=tmp_path,
+        env={
+            "POLYLOGUE_PYTEST_BASETEMP_ROOT": str(tmp_path),
+            "POLYLOGUE_PYTEST_EVENTS_DIR": str(events),
+        },
+        output_path=tmp_path / "resources.jsonl",
+    )
+
+    sample = sampler.sample(event="sample")
+
+    assert sample["xdist_worker_count"] == 1
+    assert sample["xdist_uninterruptible_count"] == 1
+
+
 def test_resource_sampler_accounts_memory_swap_and_io_deltas(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -1604,6 +1639,26 @@ def fake_fs_usage(path: Path) -> dict[str, int] | None:
     assert label == "scratch"
 
 
+def test_resolve_basetemp_reroutes_known_demand_before_tmpfs_run(
+    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_BASETEMP_REQUIRED_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:

tests/unit/infra/test_workload_artifacts.py

@@ -120,6 +120,25 @@ async def fail_parse(*args: object, **kwargs: object) -> None:
     assert not list((cache_root / ".staging").iterdir())
 
 
+def test_seeded_archive_recovers_crash_left_staging_before_rebuild(tmp_path: Path) -> None:
+    import tests.infra.workload_artifacts as artifacts
+
+    cache_root = tmp_path / "cache"
+    cache_root.joinpath("artifacts").mkdir(parents=True)
+    cache_root.joinpath(".locks").mkdir()
+    staging_root = cache_root / ".staging"
+    staging_root.mkdir()
+    stale = staging_root / "dead-build.123"
+    stale.mkdir()
+    stale.joinpath("index.db").write_bytes(b"partial sqlite")
+    stale.joinpath(".build.done").write_text("written before the crash", encoding="utf-8")
+
+    removed = artifacts._recover_stale_staging(staging_root=staging_root, artifact_name="dead-build")
+
+    assert removed == ("dead-build.123",)
+    assert not stale.exists()
+
+
 class _FlakyLockConnection:
     """Fakes ``PRAGMA journal_mode=DELETE`` raising a transient same-process lock.