Merge remote-tracking branch 'forgejo/main' into fix/ci-lane-shas-timeout-diagnostics

.gitignore

@@ -15,7 +15,6 @@ workbench_data/*
 core-rs/target/
 core-rs/Cargo.lock
 
-uv.lock
 
 # Environment secrets — never commit real keys
 .env

core/physics/goldtether.py

@@ -542,6 +542,28 @@ class GoldTetherMonitor:
 # ---------------------------------------------------------------------------
 
 
+def kappa_search_event(kappa: float, result: object) -> dict[str, Any]:
+    """ADR-0242 §5-P1: JSONL-ready execution-telemetry event for a κ search.
+
+    The memo requires the certificate be "written to the execution telemetry"
+    (audit trail) with failures falling back to baseline κ. Both typed results
+    carry ``as_dict()``; this wraps them in a stable event envelope. Pure
+    serialization — no state, no COHERENT standing, no truth status (§6
+    sovereignty invariant).
+    """
+    from core.physics.fibonacci_search import FibonacciSearchCertificate
+
+    outcome = (
+        "certificate" if isinstance(result, FibonacciSearchCertificate) else "failure"
+    )
+    return {
+        "kind": "fibonacci_kappa_search",
+        "outcome": outcome,
+        "kappa": float(kappa),
+        "result": result.as_dict(),  # type: ignore[attr-defined]
+    }
+
+
 def propose_kappa_line_search(
     residual_fn,
     *,
@@ -550,13 +572,18 @@ def propose_kappa_line_search(
     evaluation_budget: int = 16,
     objective_id: str = "goldtether_kappa",
     objective_version: str = "v1",
+    sink: Any = None,
 ) -> tuple[float, object]:
     """Optional κ search via Fibonacci section (ADR-0242 Phase 1 seam).
 
     Returns ``(kappa, cert_or_failure)``. On failure, kappa is baseline 1.0.
     Does **not** mutate GoldTetherMonitor state, COHERENT standing, or serve
-    autonomy — caller may record the result as telemetry only.
+    autonomy. When ``sink`` (any object with ``emit(line: str)``, e.g. a
+    :class:`chat.telemetry.TurnEventSink`) is provided, the §5-P1 execution-
+    telemetry event is emitted as one deterministic JSONL line.
     """
+    import json
+
     from core.physics.fibonacci_search import (
         BoundedUnimodalObjective,
         fibonacci_section_search,
@@ -571,4 +598,9 @@ def propose_kappa_line_search(
         objective_version=str(objective_version),
     )
     result = fibonacci_section_search(objective, residual_fn)
-    return propose_kappa_from_search(result)
+    kappa, outcome = propose_kappa_from_search(result)
+    if sink is not None:
+        sink.emit(
+            json.dumps(kappa_search_event(kappa, outcome), sort_keys=True)
+        )
+    return kappa, outcome

core/physics/multi_scale_energy.py

@@ -16,6 +16,7 @@ Reuses ``fibonacci_number`` / ``fibonacci_tau_schedule`` — no parallel Fibonac
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from math import exp, isfinite
 from typing import Sequence
 
@@ -169,8 +170,104 @@ def schedule_mid_span_fraction(taus: Sequence[float], *, index: int | None = Non
     return float(vals[i] / peak)
 
 
+# --- ADR-0242 §5-P2: F5–F7 cross-band surprise persistence gate ---------------
+#
+# "Emits a DiscoveryCandidate in the contemplation loop *only* when the
+# surprise signal persists across multiple Fibonacci-scaled temporal bands
+# (F5 to F7), preventing transient noise from triggering ungrounded updates."
+#
+# Pure verdict function for the contemplation loop — PROPOSAL-side only.
+# Lives here (Tier-2, serve-quarantined) so the gate can never touch serving.
+
+_DISCOVERY_BANDS: tuple[int, int, int] = (5, 8, 13)  # F_5, F_6, F_7
+
+
+@dataclass(frozen=True, slots=True)
+class CrossBandVerdict:
+    """Typed persistence verdict; never emits or promotes anything itself."""
+
+    eligible: bool
+    bands: tuple[int, int, int]
+    band_energies: tuple[float, float, float]
+    gamma: float
+    reason: str  # "eligible" | "insufficient_span" | "band_below_gamma"
+
+
+def cross_band_discovery_gate(
+    events: Sequence[tuple[float, float]],
+    *,
+    now: float,
+    tau0: float = _DEFAULT_TAU0,
+    gamma: float,
+) -> CrossBandVerdict:
+    """Persistence gate over a surprise-event history.
+
+    ``events`` are ``(t, energy)`` samples with ``t <= now`` and
+    ``energy >= 0``. Each band accumulates decay-weighted surprise
+
+        E_band(now) = Σ_i energy_i · exp(-(now - t_i) / (F_band · τ0))
+
+    Eligible ⇔ the history spans at least the shortest band (F_5·τ0 —
+    a single fresh spike carries full weight in every band but has zero
+    temporal persistence) AND every band's accumulation ≥ ``gamma``.
+    Deterministic and pure; the caller (contemplation loop) decides whether
+    to emit a DiscoveryCandidate.
+    """
+    if not events:
+        raise ValueError("cross_band_discovery_gate: empty event history")
+    tau0 = _validate_tau0(tau0)
+    if not isfinite(gamma) or gamma <= 0.0:
+        raise ValueError("gamma must be finite and > 0")
+    times = []
+    for t, e in events:
+        t = float(t)
+        e = float(e)
+        if not (isfinite(t) and isfinite(e)):
+            raise ValueError("event times/energies must be finite")
+        if e < 0.0:
+            raise ValueError(f"negative surprise energy {e} at t={t}")
+        if t > float(now):
+            raise ValueError(f"event at t={t} is after now={now}")
+        times.append(t)
+
+    band_energies = tuple(
+        sum(
+            float(e) * exp(-(float(now) - float(t)) / (band * tau0))
+            for t, e in events
+        )
+        for band in _DISCOVERY_BANDS
+    )
+
+    span = max(times) - min(times)
+    if span < _DISCOVERY_BANDS[0] * tau0:
+        return CrossBandVerdict(
+            eligible=False,
+            bands=_DISCOVERY_BANDS,
+            band_energies=band_energies,  # type: ignore[arg-type]
+            gamma=float(gamma),
+            reason="insufficient_span",
+        )
+    if any(e < gamma for e in band_energies):
+        return CrossBandVerdict(
+            eligible=False,
+            bands=_DISCOVERY_BANDS,
+            band_energies=band_energies,  # type: ignore[arg-type]
+            gamma=float(gamma),
+            reason="band_below_gamma",
+        )
+    return CrossBandVerdict(
+        eligible=True,
+        bands=_DISCOVERY_BANDS,
+        band_energies=band_energies,  # type: ignore[arg-type]
+        gamma=float(gamma),
+        reason="eligible",
+    )
+
+
 __all__ = [
+    "CrossBandVerdict",
     "comparative_residual_separation",
+    "cross_band_discovery_gate",
     "dyadic_tau_schedule",
     "multi_scale_energy_for_schedule",
     "multi_scale_energy_vector",

docs/analysis/full-suite-reds-ledger-2026-07.md

@@ -10,7 +10,7 @@ Base: `forgejo/main` @ `54228d45`
 |------|---------|--------|
 | Fast | `pytest -m "not quarantine and not slow" -n auto -q` | **5 failed**, 11660 passed, 106 skipped (pre-fix) |
 | Fast recheck | same after fix | (see verification below) |
-| Slow | `pytest -m "slow and not quarantine" -n auto -q` | running / recorded at PR time |
+| Slow | `pytest -m "slow and not quarantine" -n auto -q` | **1 failed, 909 passed, 1 skipped, 1 xfailed** (37m06s, 2026-07-15 sweep @ `8f67590d`+#47). The 1 red = `test_public_showcase.py::TestRuntimeBudget` — stale constant pin (30) vs the deliberate 640dbe8f re-budget (60); fixed alongside this row. Slow lane is otherwise clean → the historical "~31 reds" figure is fully retired. |
 | Full | `pytest -m "not quarantine"` | union of fast + slow |
 
 The brief’s “~31 reds” appears to reflect an older main tip. On current `forgejo/main`, the **non-quarantine fast lane** surfaces **exactly 5** dedicated failures. Smoke/pack/lane gates stay green because they never execute these nodes.

evals/gsm8k_math/train_sample/v1/runner.py

@@ -25,6 +25,7 @@ single scoring path.
 from __future__ import annotations
 
 import json
+import os
 import sys
 from pathlib import Path
 from typing import Any
@@ -33,7 +34,13 @@ from evals.gsm8k_math.runner import _score_one_candidate_graph
 
 _HERE = Path(__file__).resolve().parent
 _CASES_PATH = _HERE / "cases.jsonl"
-_REPORT_PATH = _HERE / "report.json"
+# Overridable so concurrent test invocations (xdist workers) can each point
+# the CLI entrypoint at an isolated tmp path instead of racing on the shared
+# committed report.json. Unset (the default) preserves the real operator
+# contract: `python -m ...runner` writes the tracked repo file.
+_REPORT_PATH = Path(
+    os.environ.get("CORE_GSM8K_TRAIN_SAMPLE_REPORT_PATH", str(_HERE / "report.json"))
+)
 _SAMPLE_REL = "evals/gsm8k_math/train_sample/v1/cases.jsonl"
 _EXPECTED_COUNT = 50
 _CORRECT_MIN = 10

tests/test_adr_0241_governance_p12.py

@@ -4,7 +4,8 @@ Does not re-execute the full physics suite. Asserts the load-bearing
 governance surfaces for ADR-0241/0242 cohesion remain present and honest:
   * runtime_contracts documents off-serve quarantine + epistemic standing
   * acceptance checklist maps C0–C8 to tests
-  * ADRs stay Proposed (ready for Joshua) — not self-Accepted
+  * ADR statuses carry recorded ruling provenance (ratified 2026-07-15;
+    a silent status flip in either direction fails)
   * cohesion suite still names I-01…I-05 pins
 """
 
@@ -82,21 +83,37 @@ def test_cohesion_suite_names_entity_invariants():
         assert name in text, f"missing entity pin {name}"
 
 
-def test_adrs_ready_for_acceptance_not_self_accepted():
-    """P12: implementation complete → Proposed + ready; Joshua alone Accepts."""
+def test_adrs_accepted_with_recorded_ruling_provenance():
+    """P12 (post-ratification): status flips are valid ONLY with provenance.
+
+    Joshua Shay ruled "Ratify" on 2026-07-15 (D10 acceptance packet §8), so
+    the anti-self-Accept guard evolves rather than dies: each ADR must say
+    Accepted WITH the ratification provenance inline, and the packet must
+    carry the ruling record. A silent status flip in either direction — an
+    Accept without provenance, or a quiet demotion — fails here.
+    """
+    packet = (
+        _ROOT / "docs/audit/adr-0241-0242-acceptance-packet-2026-07-15.md"
+    ).read_text(encoding="utf-8")
+    assert "## 8. RULING RECORD" in packet, "ruling record section missing"
+    assert "RATIFIED — Joshua Shay, 2026-07-15" in packet, "ruling attribution missing"
+
     for rel in (
         "docs/adr/ADR-0241-wave-field-driven-hyperbolic-atlas-and-resonant-cognition.md",
         "docs/adr/ADR-0242-atlas-packing-and-fibonacci.md",
     ):
         text = (_ROOT / rel).read_text(encoding="utf-8")
-        # First status line must remain Proposed until human Accept.
         status_line = next(
             (ln for ln in text.splitlines() if ln.startswith("**Status**")),
             "",
         )
-        assert "Proposed" in status_line, f"{rel} lost Proposed status"
-        assert "Accepted" not in status_line, f"{rel} must not self-Accept"
-        assert "Joshua" in status_line or "ready" in status_line.lower()
+        assert "Accepted" in status_line, f"{rel} lost Accepted status"
+        assert "ratified by Joshua Shay" in status_line, (
+            f"{rel} status lacks ratification provenance"
+        )
+        assert "acceptance-packet" in status_line, (
+            f"{rel} status must cite the ruling packet"
+        )
 
 
 def test_serve_quarantine_list_matches_cohesion_ast_pin():

tests/test_adr_0242_carry_seams.py

@@ -0,0 +1,187 @@
+"""ADR-0242 §5 carry seams — cert telemetry (P1) + F5–F7 cross-band gate (P2).
+
+RED until both seams land. These are the two staged items recorded in the
+acceptance packet §7 (ruled non-blocking at ratification, tracked work):
+
+P1 (memo §5 Phase 1): "The returned FibonacciSearchCertificate is written to
+the execution telemetry. If the certificate fails … defaults to the safe
+baseline kappa = 1.0 and logs an OptimizationFailure warning." The search
+returns typed results; nothing produced a telemetry-ready event.
+
+P2 (memo §5 Phase 2): "Emits a DiscoveryCandidate in the contemplation loop
+*only* when the surprise signal persists across multiple Fibonacci-scaled
+temporal bands (F5 to F7), preventing transient noise from triggering
+ungrounded updates." Band helpers existed; the persistence gate did not.
+
+Both stay PROPOSAL/telemetry-side: no serve import, no COHERENT promotion,
+no truth-status effect (memo §6 sovereignty invariant).
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from core.physics.goldtether import kappa_search_event, propose_kappa_line_search
+from core.physics.multi_scale_energy import (
+    CrossBandVerdict,
+    cross_band_discovery_gate,
+)
+
+# --- P1: κ-search certificate → execution telemetry event ---------------------
+
+
+def _good_objective(x: float) -> float:
+    return (x - 0.789) ** 2
+
+
+def _multimodal(x: float) -> float:
+    return x**4 - x**2
+
+
+class _CaptureSink:
+    def __init__(self) -> None:
+        self.lines: list[str] = []
+
+    def emit(self, line: str) -> None:
+        self.lines.append(line)
+
+
+def test_kappa_search_event_certificate_payload():
+    kappa, result = propose_kappa_line_search(_good_objective, evaluation_budget=16)
+    event = kappa_search_event(kappa, result)
+
+    assert event["kind"] == "fibonacci_kappa_search"
+    assert event["outcome"] == "certificate"
+    assert event["kappa"] == pytest.approx(kappa)
+    cert = event["result"]
+    assert cert["kind"] == "FibonacciSearchCertificate"
+    assert len(cert["cert_id"]) == 64  # content-addressed audit trail
+    assert cert["evaluations"] == 16
+    assert cert["objective_id"] == "goldtether_kappa"
+    # JSONL-ready: round-trips through json without custom encoders.
+    assert json.loads(json.dumps(event)) == event
+
+
+def test_kappa_search_event_failure_payload_falls_back_to_baseline():
+    kappa, result = propose_kappa_line_search(
+        _multimodal, lower=-2.0, upper=2.0, evaluation_budget=10
+    )
+    event = kappa_search_event(kappa, result)
+
+    assert event["outcome"] == "failure"
+    assert event["kappa"] == 1.0  # BASELINE_KAPPA — never a silent minimizer
+    assert event["result"]["kind"] == "OptimizationFailure"
+    assert "unimodality" in event["result"]["reason"]
+    assert json.loads(json.dumps(event)) == event
+
+
+def test_propose_kappa_line_search_emits_to_sink_when_provided():
+    sink = _CaptureSink()
+    kappa, result = propose_kappa_line_search(
+        _good_objective, evaluation_budget=12, sink=sink
+    )
+    assert len(sink.lines) == 1
+    event = json.loads(sink.lines[0])
+    assert event["kind"] == "fibonacci_kappa_search"
+    assert event["kappa"] == pytest.approx(kappa)
+    assert event["result"]["cert_id"]
+
+
+def test_kappa_search_event_deterministic():
+    _, result_a = propose_kappa_line_search(_good_objective, evaluation_budget=12)
+    _, result_b = propose_kappa_line_search(_good_objective, evaluation_budget=12)
+    kappa_a, _ = propose_kappa_line_search(_good_objective, evaluation_budget=12)
+    assert kappa_search_event(kappa_a, result_a) == kappa_search_event(
+        kappa_a, result_b
+    )
+
+
+# --- P2: F5–F7 cross-band surprise persistence gate ----------------------------
+#
+# Bands: tau_n = F_n · tau0 with (F5, F6, F7) = (5, 8, 13).
+# eligible ⇔ every band's decay-weighted accumulated surprise ≥ gamma AND the
+# event history spans at least the shortest band (temporal extent — a single
+# fresh spike has full weight in every band but zero persistence).
+
+
+def _sustained(tau0: float, *, per_event: float = 0.5, n: int = 14) -> list[tuple[float, float]]:
+    return [(float(i) * tau0, per_event) for i in range(n)]
+
+
+def test_sustained_signal_is_eligible_across_all_bands():
+    tau0 = 1.0
+    events = _sustained(tau0)
+    verdict = cross_band_discovery_gate(
+        events, now=13.0 * tau0, tau0=tau0, gamma=0.35
+    )
+    assert isinstance(verdict, CrossBandVerdict)
+    assert verdict.eligible is True
+    assert verdict.bands == (5, 8, 13)
+    assert all(e >= verdict.gamma for e in verdict.band_energies)
+
+
+def test_single_fresh_spike_is_not_eligible_no_temporal_extent():
+    tau0 = 1.0
+    verdict = cross_band_discovery_gate(
+        [(10.0, 5.0)], now=10.0, tau0=tau0, gamma=0.35
+    )
+    assert verdict.eligible is False
+    assert verdict.reason == "insufficient_span"
+
+
+def test_old_transient_is_not_eligible_short_band_decayed():
+    tau0 = 1.0
+    # Burst long ago: spans the extent requirement, but by `now` the F5 band
+    # (tau = 5·tau0) has decayed it below gamma while F7 may still hold mass.
+    events = [(0.0, 1.0), (2.0, 1.0), (4.0, 1.0), (6.0, 1.0)]
+    verdict = cross_band_discovery_gate(events, now=40.0, tau0=tau0, gamma=0.35)
+    assert verdict.eligible is False
+    assert verdict.reason == "band_below_gamma"
+    # The shortest band is the one that fails first.
+    assert verdict.band_energies[0] < verdict.gamma
+
+
+def test_gate_is_deterministic_and_pure():
+    tau0 = 2.0
+    events = _sustained(tau0, per_event=0.4)
+    a = cross_band_discovery_gate(events, now=30.0, tau0=tau0, gamma=0.2)
+    b = cross_band_discovery_gate(events, now=30.0, tau0=tau0, gamma=0.2)
+    assert a == b
+
+
+def test_gate_refuses_invalid_inputs():
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([], now=1.0, tau0=1.0, gamma=0.35)
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([(0.0, 1.0)], now=1.0, tau0=-1.0, gamma=0.35)
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([(2.0, 1.0)], now=1.0, tau0=1.0, gamma=0.35)  # event after now
+    with pytest.raises(ValueError):
+        cross_band_discovery_gate([(0.0, -1.0)], now=1.0, tau0=1.0, gamma=0.35)  # negative energy
+
+
+def test_gate_module_stays_off_serve():
+    """multi_scale_energy remains Tier-2: never imported by chat.runtime."""
+    import subprocess
+    import sys
+    from pathlib import Path
+
+    root = Path(__file__).resolve().parents[1]
+    probe = (
+        "import importlib, sys;"
+        "importlib.import_module('chat.runtime');"
+        "print('LEAK' if any(m == 'core.physics.multi_scale_energy' or "
+        "m.startswith('core.physics.multi_scale_energy.') for m in sys.modules)"
+        " else 'CLEAN')"
+    )
+    out = subprocess.run(
+        [sys.executable, "-c", probe],
+        cwd=str(root),
+        capture_output=True,
+        text=True,
+        env={"PYTHONPATH": str(root), "PATH": ""},
+    )
+    assert out.returncode == 0, out.stderr[-500:]
+    assert out.stdout.strip().splitlines()[-1] == "CLEAN"

tests/test_cli_demo.py

@@ -10,6 +10,24 @@ import pytest
 from core import cli
 
 
+@pytest.fixture(autouse=True)
+def _isolate_demo_results_dir(tmp_path, monkeypatch):
+    """Redirect the demo results dir to an isolated per-test tmp dir.
+
+    ``core.cli._DEMO_RESULTS_DIR`` is a module-level constant pointing at the
+    shared repo dir ``evals/forward_semantic_control/results/`` that every
+    ``core demo`` subcommand reads, writes, and glob-scans (including the
+    ``index.json`` rebuild in ``_write_results_index``). Multiple tests below
+    invoke ``cli.main(["demo", ...])``; left unpatched they race on that one
+    shared directory under ``-n auto`` (concurrent report writes plus a
+    scan-then-rewrite of ``index.json``). Follows the same monkeypatch-the-
+    module-attribute isolation idiom as the root-conftest engine-state fixture
+    (#782). ``TestResultsReadme`` reads the real checked-in README by a
+    hardcoded path, so it is unaffected by this redirection.
+    """
+    monkeypatch.setattr(cli, "_DEMO_RESULTS_DIR", tmp_path / "results")
+
+
 class TestADR0024SuiteAliases:
     """Layer 1: pin the new suite aliases so they don't drift."""
 
@@ -93,7 +111,7 @@ class TestDemoSubcommand:
         assert "ALL THREE CONDITIONS" in captured.out
         assert "PASS" in captured.out
         # Report file present and well-formed.
-        report = Path("evals/forward_semantic_control/results/phase6_demo_report.json")
+        report = cli._DEMO_RESULTS_DIR / "phase6_demo_report.json"
         assert report.exists()
         data = json.loads(report.read_text())
         assert data["metrics"]["all_three_conditions_pass"] is True
@@ -109,6 +127,12 @@ class TestDemoSubcommand:
         assert "all_three_conditions_pass" in payload["metrics"]
 
     def test_demo_list_results_indexes_reports(self, capsys) -> None:
+        # Self-contained: write a report before indexing rather than relying
+        # on a sibling test's shared-directory side effect (that implicit
+        # ordering coupling only "worked" by accident of file-declaration
+        # order and would break under `-n auto` reordering).
+        cli.main(["demo", "phase6"])
+        capsys.readouterr()
         rc = cli.main(["demo", "list-results"])
         assert rc == 0
         captured = capsys.readouterr()
@@ -116,6 +140,10 @@ class TestDemoSubcommand:
         assert "phase6_demo_report.json" in captured.out
 
     def test_demo_list_results_json_well_formed(self, capsys) -> None:
+        # Self-contained for the same reason as
+        # test_demo_list_results_indexes_reports above.
+        cli.main(["demo", "phase6"])
+        capsys.readouterr()
         rc = cli.main(["demo", "list-results", "--json"])
         assert rc == 0
         captured = capsys.readouterr()
@@ -127,7 +155,7 @@ class TestDemoSubcommand:
 
     def test_demo_index_file_refreshed_after_run(self) -> None:
         cli.main(["demo", "phase6"])
-        index_path = Path("evals/forward_semantic_control/results/index.json")
+        index_path = cli._DEMO_RESULTS_DIR / "index.json"
         assert index_path.exists()
         data = json.loads(index_path.read_text())
         names = [e["file"] for e in data["reports"]]

tests/test_public_showcase.py

@@ -162,5 +162,11 @@ class TestPureCompositionGate:
 
 
 class TestRuntimeBudget:
-    def test_budget_is_thirty_seconds(self) -> None:
-        assert MAX_RUNTIME_SECONDS == 30
+    def test_budget_is_sixty_seconds(self) -> None:
+        # 30 → 60 was a deliberate re-budget (commit 640dbe8f: post-CGA
+        # substrate work makes a cold RegisterTour alone ~30s+; see
+        # evals/public_demo/contract.md "Known Environment Caveat"). This slow
+        # pin was missed then because the slow lane hadn't been run since
+        # (nightly lives on the billing-locked GitHub side) — caught by the
+        # 2026-07 slow-lane sweep. Pin the deliberate value.
+        assert MAX_RUNTIME_SECONDS == 60

tests/test_rat1_end_to_end_admission.py

@@ -114,19 +114,19 @@ def test_canonical_pack_admits_cross_sentence_composition_when_seeded():
         )
 
 
-def test_wrong_zero_preserved_on_train_sample():
+def test_wrong_zero_preserved_on_train_sample(tmp_path):
     """The full train_sample eval must preserve wrong == 0 after RAT-1."""
+    import os
     import subprocess
     import sys
 
+    report_path = tmp_path / "report.json"
     result = subprocess.run(
         [sys.executable, "-m", "evals.gsm8k_math.train_sample.v1.runner"],
         cwd=_repo_root(),
         capture_output=True,
         text=True,
-    )
-    report_path = (
-        _repo_root() / "evals" / "gsm8k_math" / "train_sample" / "v1" / "report.json"
+        env={**os.environ, "CORE_GSM8K_TRAIN_SAMPLE_REPORT_PATH": str(report_path)},
     )
     assert report_path.exists(), "train_sample runner must emit report.json"
     report = json.loads(report_path.read_text())
@@ -134,18 +134,18 @@ def test_wrong_zero_preserved_on_train_sample():
     assert counts["wrong"] == 0, f"wrong-zero invariant violated: {counts}"
 
 
-def test_case_0050_remains_refused_after_rat1():
+def test_case_0050_remains_refused_after_rat1(tmp_path):
     """Hazard pin: case 0050 must not admit after any RAT-1 wiring."""
+    import os
     import subprocess
     import sys
 
+    report_path = tmp_path / "report.json"
     subprocess.run(
         [sys.executable, "-m", "evals.gsm8k_math.train_sample.v1.runner"],
         cwd=_repo_root(),
         capture_output=True,
-    )
-    report_path = (
-        _repo_root() / "evals" / "gsm8k_math" / "train_sample" / "v1" / "report.json"
+        env={**os.environ, "CORE_GSM8K_TRAIN_SAMPLE_REPORT_PATH": str(report_path)},
     )
     report = json.loads(report_path.read_text())
     case = next(

tests/test_teaching_queue.py

@@ -9,6 +9,7 @@ from unittest.mock import patch
 
 import pytest
 
+import engine_state
 from core.cli import main
 from teaching.discovery import DiscoveryCandidate, EvidencePointer
 from teaching.proposals import ProposalLog, ReplayEvidence, build_proposal
@@ -515,11 +516,21 @@ def snapshot_dir(directory: Path) -> dict[Path, bytes]:
 
 
 def test_read_only_invariant():
+    """Directories the hitl-queue commands must not mutate.
+
+    ``engine_state`` is read via ``engine_state._DEFAULT_DIR`` (not a
+    hardcoded ``project_root / "engine_state"``) because the root-conftest
+    ``_isolate_engine_state_default`` autouse fixture repoints it at a
+    per-test tmp dir; a hardcoded real path would instead snapshot the
+    shared repo dir and false-fail under ``-n auto`` whenever an unrelated
+    concurrent worker wrote to it (observed: AssertionError on this exact
+    directory under a full `-n auto` fast-lane run).
+    """
     project_root = Path(__file__).resolve().parent.parent
     dirs = [
         project_root / "teaching" / "proposals",
         project_root / "packs",
-        project_root / "engine_state",
+        Path(engine_state._DEFAULT_DIR),
         project_root / "contemplation" / "runs",
     ]
 

tests/test_wave_a_multiplicative_aggregation_injector.py

@@ -171,41 +171,43 @@ def _has_wave_a_seed() -> bool:
     )
 
 
-def test_wrong_zero_preserved():
+def test_wrong_zero_preserved(tmp_path):
     """The full train_sample eval keeps wrong == 0 after WAVE-A."""
+    import os
     import subprocess
     import sys
 
     here = Path(__file__).resolve()
     while here.parent != here and not (here / "pyproject.toml").exists():
         here = here.parent
+    report_path = tmp_path / "report.json"
     subprocess.run(
         [sys.executable, "-m", "evals.gsm8k_math.train_sample.v1.runner"],
         cwd=here,
         capture_output=True,
+        env={**os.environ, "CORE_GSM8K_TRAIN_SAMPLE_REPORT_PATH": str(report_path)},
     )
-    report = json.loads(
-        (here / "evals" / "gsm8k_math" / "train_sample" / "v1" / "report.json").read_text()
-    )
+    report = json.loads(report_path.read_text())
     assert report["counts"]["wrong"] == 0
 
 
-def test_case_0050_remains_refused():
+def test_case_0050_remains_refused(tmp_path):
     """Hazard pin."""
+    import os
     import subprocess
     import sys
 
     here = Path(__file__).resolve()
     while here.parent != here and not (here / "pyproject.toml").exists():
         here = here.parent
+    report_path = tmp_path / "report.json"
     subprocess.run(
         [sys.executable, "-m", "evals.gsm8k_math.train_sample.v1.runner"],
         cwd=here,
         capture_output=True,
+        env={**os.environ, "CORE_GSM8K_TRAIN_SAMPLE_REPORT_PATH": str(report_path)},
     )
-    report = json.loads(
-        (here / "evals" / "gsm8k_math" / "train_sample" / "v1" / "report.json").read_text()
-    )
+    report = json.loads(report_path.read_text())
     case_0050 = next(
         (c for c in report["per_case"] if c["case_id"].endswith("-0050")),
         None,

tests/test_workbench_replay.py

@@ -12,14 +12,13 @@ from pathlib import Path
 
 import pytest
 
+import engine_state
 from chat.runtime import ChatRuntime
 from workbench import api as workbench_api
 from workbench.api import WorkbenchApi, _run_sealed_chat_turn, _with_turn_cost_and_id
 from workbench.journal import TurnJournal, TurnJournalEntry
 from workbench.replay import CRITICAL_FIELDS, INFORMATIONAL_FIELDS, replay_turn
 
-_ENGINE_STATE_DIR = Path("engine_state")
-
 
 def _snapshot(root: Path) -> dict[str, bytes]:
     snap: dict[str, bytes] = {}
@@ -150,19 +149,26 @@ def test_hashless_legacy_turn_is_not_replayable(tmp_path, recorded, monkeypatch)
 
 
 def test_replay_leaves_no_trace(tmp_path, recorded) -> None:
-    """Obligation 4: GET /replay writes nothing — journal and engine state."""
+    """Obligation 4: GET /replay writes nothing — journal and engine state.
+
+    ``engine_state._DEFAULT_DIR`` is read fresh (not a module-level constant)
+    because the root-conftest ``_isolate_engine_state_default`` autouse
+    fixture repoints it at a per-test tmp dir; a hardcoded ``Path("engine_state")``
+    would instead snapshot the real shared repo dir and false-fail under
+    ``-n auto`` whenever an unrelated concurrent worker wrote to it.
+    """
     entry, _ = recorded
     journal = TurnJournal(journal_dir=tmp_path / "workbench_data")
     journal.append(entry)
     api = WorkbenchApi(journal=journal)
     journal_before = _snapshot(tmp_path / "workbench_data")
-    engine_state_before = _snapshot(_ENGINE_STATE_DIR)
+    engine_state_before = _snapshot(Path(engine_state._DEFAULT_DIR))
 
     response = api.handle("GET", "/replay/1", b"")
 
     assert response.status == 200
     assert _snapshot(tmp_path / "workbench_data") == journal_before
-    assert _snapshot(_ENGINE_STATE_DIR) == engine_state_before
+    assert _snapshot(Path(engine_state._DEFAULT_DIR)) == engine_state_before
 
 
 def test_wall_clock_divergence_does_not_break_equivalence(tmp_path, recorded) -> None: