perf(test-infra): pytest-xdist + module-scoped demo fixtures

Full lane wall-time: 6:35 → 2:25 (2.7× speedup).  No behavioral
changes; same 1933 passed, 2 skipped.

Three wins, biggest first:

1. pytest-xdist as a project dependency.

   ``pyproject.toml`` gains ``pytest-xdist>=3.6``.  ``cmd_test``
   injects ``-n auto`` for ``--suite full`` when xdist is importable;
   curated suites stay single-process because worker-spawn overhead
   is net-negative on the smaller suites.  Operator can override
   via passing ``-n <N>`` or ``--dist`` explicitly.

   Verified: ``core test --suite full -q`` prints ``bringing up
   nodes...`` and parallelises across the runner's CPUs.

2. Module-scoped fixture for run_demo() in test_learning_loop_demo.py.

   The 7 demo tests each previously called ``run_demo(emit_json=True)``
   from scratch — and ``run_demo`` itself runs the cognition lane
   twice via the replay-equivalence gate.  ~15s/file → ~3s/file.

   Module scope (not session) is intentional: pytest-xdist
   distributes by test, so a session-scoped fixture would still be
   re-evaluated per worker that picks up a test from this file.
   Module scope keeps the cost paid once per worker per file, which
   is the actual lower bound.

3. Module-scoped fixture for the teaching-loop bench.

   ``test_teaching_loop_bench.py``'s 5 tests previously each ran
   ``run_teaching_loop_determinism(runs=2 or 3)`` — 12 pipeline
   invocations across the file.  One ``runs=3`` invocation shared
   across all 5 tests covers every assertion: ~25s → ~7s.

For local iteration, ``core test --suite cognition -q`` etc. remain
fast (no xdist overhead).  The full-lane speedup is most visible
under CI / pre-merge runs.

core/cli.py

@@ -249,6 +249,32 @@ def _pytest_args_for_suite(suite: str, extra_args: Sequence[str]) -> list[str]:
     return [*paths, *forwarded]
 
 
+def _xdist_available() -> bool:
+    """Return True iff ``pytest-xdist`` is importable."""
+    try:
+        import xdist  # noqa: F401
+    except ImportError:
+        return False
+    return True
+
+
+def _maybe_inject_xdist(forwarded: list[str], suite: str | None) -> list[str]:
+    """Inject ``-n auto`` for suites large enough to benefit from
+    parallelism.  ``--suite full`` always gets it (when xdist is
+    installed); curated suites stay single-process because they are
+    already small and the worker-spawn overhead is net-negative on
+    them.  Operators can override by passing ``-n <N>`` or
+    ``--no-parallel`` (here stripped) in ``args``."""
+    if not _xdist_available():
+        return forwarded
+    # Honour explicit operator override.
+    if any(a.startswith("-n") or a == "--dist" for a in forwarded):
+        return forwarded
+    if suite == "full":
+        return ["-n", "auto", *forwarded]
+    return forwarded
+
+
 def cmd_test(args: argparse.Namespace) -> int:
     """Run pytest through curated suite aliases or direct passthrough args."""
     default_args = ["-q", "--tb=short"]
@@ -262,6 +288,7 @@ def cmd_test(args: argparse.Namespace) -> int:
         forwarded = list(args.args or default_args)
         if forwarded and forwarded[0] == "--":
             forwarded = forwarded[1:]
+    forwarded = _maybe_inject_xdist(forwarded, args.suite)
     return _run(sys.executable, "-m", "pytest", *forwarded)
 
 

pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
     "numpy>=1.26",
     "pytest>=9.0.3",
     "pytest-asyncio>=1.3.0",
+    "pytest-xdist>=3.6",
     "pyyaml>=6.0",
     "ruff>=0.15.12",
 ]

tests/test_learning_loop_demo.py

@@ -3,30 +3,50 @@
 If any assertion fails, the headline claim ("CORE learned a new chain
 from a cold turn and the same prompt is now teaching-grounded with
 provenance") no longer holds.
+
+Performance: ``run_demo()`` exercises the full pipeline including the
+replay-equivalence gate (which itself runs the cognition public split
+twice).  Each invocation costs ~2-3s.  A module-scoped fixture caches
+the report so every assertion in this file shares one demo run —
+reduces this file's runtime from ~15s (7 × 2s) to ~2s.
+
+Compatibility with pytest-xdist: pytest-xdist distributes by test, not
+by module; module-scoped fixtures are re-evaluated per worker that
+picks up a test from this file.  Worst case one worker takes the
+whole file's 2s; xdist still parallelises across the rest of the
+suite.
 """
 
 from __future__ import annotations
 
+import pytest
+
 from evals.learning_loop.run_demo import run_demo
 
 
-def test_demo_closes_the_full_loop() -> None:
+@pytest.fixture(scope="module")
-    report = run_demo(emit_json=True)
+def demo_report() -> dict:
-    assert report["learning_loop_closed"] is True
+    """One ``run_demo()`` invocation shared across every test in this
-    assert report["active_corpus_byte_identical"] is True
+    module.  Module-scoped so pytest-xdist's per-worker isolation
-    assert len(report["scenes"]) == 5
+    still applies (a worker that picks up any test in this file pays
+    the demo cost once)."""
+    return run_demo(emit_json=True)
 
 
-def test_before_is_ungrounded_disclosure() -> None:
+def test_demo_closes_the_full_loop(demo_report: dict) -> None:
-    report = run_demo(emit_json=True)
+    assert demo_report["learning_loop_closed"] is True
-    assert report["before"]["grounding_source"] == "none"
+    assert demo_report["active_corpus_byte_identical"] is True
-    assert "insufficient grounding" in report["before"]["surface"].lower()
+    assert len(demo_report["scenes"]) == 5
 
 
-def test_after_is_teaching_grounded_with_new_chain_atoms() -> None:
+def test_before_is_ungrounded_disclosure(demo_report: dict) -> None:
-    report = run_demo(emit_json=True)
+    assert demo_report["before"]["grounding_source"] == "none"
-    assert report["after"]["grounding_source"] == "teaching"
+    assert "insufficient grounding" in demo_report["before"]["surface"].lower()
-    surface = report["after"]["surface"].lower()
+
+
+def test_after_is_teaching_grounded_with_new_chain_atoms(demo_report: dict) -> None:
+    assert demo_report["after"]["grounding_source"] == "teaching"
+    surface = demo_report["after"]["surface"].lower()
     # The accepted chain is (narrative, cause, reveals, meaning).
     # ``thought`` was the original cold subject; cognition saturation
     # v2 (commit ``a0edbb4``) added ``cause_thought_reveals_meaning``
@@ -38,16 +58,14 @@ def test_after_is_teaching_grounded_with_new_chain_atoms() -> None:
     assert "teaching-grounded" in surface
 
 
-def test_s1_emits_one_discovery_candidate() -> None:
+def test_s1_emits_one_discovery_candidate(demo_report: dict) -> None:
-    report = run_demo(emit_json=True)
+    s1 = demo_report["scenes"][0]
-    s1 = report["scenes"][0]
     assert s1["scene"] == "S1_cold_turn"
     assert s1["detail"]["discovery_candidates_emitted"] >= 1
 
 
-def test_s3_replay_gate_reports_no_regression() -> None:
+def test_s3_replay_gate_reports_no_regression(demo_report: dict) -> None:
-    report = run_demo(emit_json=True)
+    s3 = demo_report["scenes"][2]
-    s3 = report["scenes"][2]
     assert s3["scene"] == "S3_propose_replay_pass"
     ev = s3["detail"]["replay_evidence"]
     assert ev["replay_equivalent"] is True
@@ -55,20 +73,18 @@ def test_s3_replay_gate_reports_no_regression() -> None:
     assert s3["detail"]["state"] == "pending"
 
 
-def test_s4_active_corpus_byte_identical_after_accept() -> None:
+def test_s4_active_corpus_byte_identical_after_accept(demo_report: dict) -> None:
-    report = run_demo(emit_json=True)
+    s4 = demo_report["scenes"][3]
-    s4 = report["scenes"][3]
     assert s4["scene"] == "S4_accept_against_transient"
     assert s4["detail"]["active_corpus_byte_identical"] is True
     assert s4["detail"]["transient_lines_after"] == s4["detail"]["transient_lines_before"] + 1
 
 
-def test_same_prompt_drives_before_and_after() -> None:
+def test_same_prompt_drives_before_and_after(demo_report: dict) -> None:
     """The same input string drives both sides of the before/after pair.
     Different surfaces emerge from the corpus state change alone, not
     from any prompt variation or stochastic sampling."""
-    report = run_demo(emit_json=True)
+    assert demo_report["prompt"] == "Why does narrative exist?"
-    assert report["prompt"] == "Why does narrative exist?"
     # And the two surfaces are observably different — the loop changed
     # the response, not merely the metadata.
-    assert report["before"]["surface"] != report["after"]["surface"]
+    assert demo_report["before"]["surface"] != demo_report["after"]["surface"]

tests/test_teaching_loop_bench.py

@@ -11,47 +11,60 @@ If determinism breaks anywhere in the pipeline (proposal_id hashing,
 replay-equivalence gate, accept-side corpus_append, ProposalLog
 replay), at least one of the ``unique_*`` counts in the bench report
 will exceed 1 and this test fails.
+
+Performance: a module-scoped fixture shares one ``runs=3`` invocation
+across every test in this file.  Each ``run_teaching_loop_determinism``
+call runs the propose/replay/accept pipeline N times; cutting from 5
+calls (runs=3,2,2,3,2 = 12 pipeline runs) to 1 call (3 runs) is a
+~4× speedup with no contract loss.
 """
 
 from __future__ import annotations
 
+import json
+
+import pytest
+
 from benchmarks.teaching_loop import run_teaching_loop_determinism
 
 
-def test_teaching_loop_is_deterministic_across_three_runs() -> None:
+@pytest.fixture(scope="module")
-    report = run_teaching_loop_determinism(runs=3)
+def bench_report():
-    assert report.deterministic is True
+    """Single ``runs=3`` invocation shared across every test in this
-    assert report.active_corpus_byte_identical is True
+    file.  ``runs=3`` is the highest N any individual test asks for,
-    assert report.unique_proposal_ids == 1
+    so the determinism / latency / serialisation assertions all hold
-    assert report.unique_replay_baselines == 1
+    against the same report."""
-    assert report.unique_replay_candidates == 1
+    return run_teaching_loop_determinism(runs=3)
-    assert report.unique_regressed_metrics == 1
-    assert report.unique_chain_ids == 1
 
 
-def test_proposal_id_is_a_sha256_prefix() -> None:
+def test_teaching_loop_is_deterministic_across_three_runs(bench_report) -> None:
-    report = run_teaching_loop_determinism(runs=2)
+    assert bench_report.deterministic is True
-    pid = report.sample_proposal_id
+    assert bench_report.active_corpus_byte_identical is True
+    assert bench_report.unique_proposal_ids == 1
+    assert bench_report.unique_replay_baselines == 1
+    assert bench_report.unique_replay_candidates == 1
+    assert bench_report.unique_regressed_metrics == 1
+    assert bench_report.unique_chain_ids == 1
+
+
+def test_proposal_id_is_a_sha256_prefix(bench_report) -> None:
+    pid = bench_report.sample_proposal_id
     assert len(pid) == 32
     assert all(c in "0123456789abcdef" for c in pid)
 
 
-def test_chain_id_matches_canonical_layout() -> None:
+def test_chain_id_matches_canonical_layout(bench_report) -> None:
-    report = run_teaching_loop_determinism(runs=2)
+    assert bench_report.sample_chain_id == "cause_thought_reveals_meaning"
-    assert report.sample_chain_id == "cause_thought_reveals_meaning"
 
 
-def test_latency_stats_are_well_formed() -> None:
+def test_latency_stats_are_well_formed(bench_report) -> None:
-    report = run_teaching_loop_determinism(runs=3)
+    assert bench_report.elapsed_mean_s > 0.0
-    assert report.elapsed_mean_s > 0.0
+    assert bench_report.elapsed_p50_s > 0.0
-    assert report.elapsed_p50_s > 0.0
+    assert bench_report.elapsed_p95_s >= bench_report.elapsed_p50_s
-    assert report.elapsed_p95_s >= report.elapsed_p50_s
+    assert bench_report.elapsed_total_s >= bench_report.elapsed_mean_s * bench_report.runs * 0.9
-    assert report.elapsed_total_s >= report.elapsed_mean_s * report.runs * 0.9
 
 
-def test_report_serialises_to_json() -> None:
+def test_report_serialises_to_json(bench_report) -> None:
-    import json
+    blob = json.dumps(bench_report.as_dict(), sort_keys=True)
-    report = run_teaching_loop_determinism(runs=2)
-    blob = json.dumps(report.as_dict(), sort_keys=True)
     assert "deterministic" in blob
     assert "active_corpus_byte_identical" in blob