From 34295e55ce39e473bb0481f6322ee50e36690ee6 Mon Sep 17 00:00:00 2001 From: Shay Date: Mon, 18 May 2026 16:12:27 -0700 Subject: [PATCH] perf(test-infra): pytest-xdist + module-scoped demo fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 `` 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 | 27 ++++++++++++ pyproject.toml | 1 + tests/test_learning_loop_demo.py | 68 +++++++++++++++++++------------ tests/test_teaching_loop_bench.py | 63 ++++++++++++++++------------ 4 files changed, 108 insertions(+), 51 deletions(-) diff --git a/core/cli.py b/core/cli.py index 938f17ba..17f38b74 100644 --- a/core/cli.py +++ b/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 `` 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) diff --git a/pyproject.toml b/pyproject.toml index b95a73be..ae6924bb 100644 --- a/pyproject.toml +++ b/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", ] diff --git a/tests/test_learning_loop_demo.py b/tests/test_learning_loop_demo.py index 8e9b7db5..d680fc0a 100644 --- a/tests/test_learning_loop_demo.py +++ b/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: - report = run_demo(emit_json=True) - assert report["learning_loop_closed"] is True - assert report["active_corpus_byte_identical"] is True - assert len(report["scenes"]) == 5 +@pytest.fixture(scope="module") +def demo_report() -> dict: + """One ``run_demo()`` invocation shared across every test in this + module. Module-scoped so pytest-xdist's per-worker isolation + 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: - report = run_demo(emit_json=True) - assert report["before"]["grounding_source"] == "none" - assert "insufficient grounding" in report["before"]["surface"].lower() +def test_demo_closes_the_full_loop(demo_report: dict) -> None: + assert demo_report["learning_loop_closed"] is True + assert demo_report["active_corpus_byte_identical"] is True + assert len(demo_report["scenes"]) == 5 -def test_after_is_teaching_grounded_with_new_chain_atoms() -> None: - report = run_demo(emit_json=True) - assert report["after"]["grounding_source"] == "teaching" - surface = report["after"]["surface"].lower() +def test_before_is_ungrounded_disclosure(demo_report: dict) -> None: + assert demo_report["before"]["grounding_source"] == "none" + assert "insufficient grounding" in demo_report["before"]["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: - report = run_demo(emit_json=True) - s1 = report["scenes"][0] +def test_s1_emits_one_discovery_candidate(demo_report: dict) -> None: + s1 = demo_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: - report = run_demo(emit_json=True) - s3 = report["scenes"][2] +def test_s3_replay_gate_reports_no_regression(demo_report: dict) -> None: + s3 = demo_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: - report = run_demo(emit_json=True) - s4 = report["scenes"][3] +def test_s4_active_corpus_byte_identical_after_accept(demo_report: dict) -> None: + s4 = demo_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 report["prompt"] == "Why does narrative exist?" + assert demo_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"] diff --git a/tests/test_teaching_loop_bench.py b/tests/test_teaching_loop_bench.py index 65b9349b..b46d51e8 100644 --- a/tests/test_teaching_loop_bench.py +++ b/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: - report = run_teaching_loop_determinism(runs=3) - assert report.deterministic is True - assert report.active_corpus_byte_identical is True - assert report.unique_proposal_ids == 1 - assert report.unique_replay_baselines == 1 - assert report.unique_replay_candidates == 1 - assert report.unique_regressed_metrics == 1 - assert report.unique_chain_ids == 1 +@pytest.fixture(scope="module") +def bench_report(): + """Single ``runs=3`` invocation shared across every test in this + file. ``runs=3`` is the highest N any individual test asks for, + so the determinism / latency / serialisation assertions all hold + against the same report.""" + return run_teaching_loop_determinism(runs=3) -def test_proposal_id_is_a_sha256_prefix() -> None: - report = run_teaching_loop_determinism(runs=2) - pid = report.sample_proposal_id +def test_teaching_loop_is_deterministic_across_three_runs(bench_report) -> None: + assert bench_report.deterministic is True + 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: - report = run_teaching_loop_determinism(runs=2) - assert report.sample_chain_id == "cause_thought_reveals_meaning" +def test_chain_id_matches_canonical_layout(bench_report) -> None: + assert bench_report.sample_chain_id == "cause_thought_reveals_meaning" -def test_latency_stats_are_well_formed() -> None: - report = run_teaching_loop_determinism(runs=3) - assert report.elapsed_mean_s > 0.0 - assert report.elapsed_p50_s > 0.0 - assert report.elapsed_p95_s >= report.elapsed_p50_s - assert report.elapsed_total_s >= report.elapsed_mean_s * report.runs * 0.9 +def test_latency_stats_are_well_formed(bench_report) -> None: + assert bench_report.elapsed_mean_s > 0.0 + assert bench_report.elapsed_p50_s > 0.0 + assert bench_report.elapsed_p95_s >= bench_report.elapsed_p50_s + assert bench_report.elapsed_total_s >= bench_report.elapsed_mean_s * bench_report.runs * 0.9 -def test_report_serialises_to_json() -> None: - import json - report = run_teaching_loop_determinism(runs=2) - blob = json.dumps(report.as_dict(), sort_keys=True) +def test_report_serialises_to_json(bench_report) -> None: + blob = json.dumps(bench_report.as_dict(), sort_keys=True) assert "deterministic" in blob assert "active_corpus_byte_identical" in blob