"""Bench-side aggregation tests for capacity metrics + deflection rate. The bench harness in ``bench/qa_sweep.py`` is the analytics surface for QA-quality work — it consumes per-row ``prompt_chars_*`` / ``answer_chars`` fields from the query result and renders a markdown summary including strict-rate-by-size buckets. These tests pin the aggregation behavior on synthetic rows so the rendering doesn't drift silently. """ from __future__ import annotations import importlib.util import sys from pathlib import Path import pytest @pytest.fixture(scope="module") def qa_sweep(): """Import bench/qa_sweep.py as a module despite living outside the arborist package. Avoids polluting the package or requiring a pip-install of the bench harness.""" bench_path = Path(__file__).parent.parent / "bench" / "qa_sweep.py" spec = importlib.util.spec_from_file_location("qa_sweep_under_test", bench_path) mod = importlib.util.module_from_spec(spec) sys.modules[spec.name] = mod spec.loader.exec_module(mod) return mod def _row(**overrides) -> dict: """Synthetic bench-row fixture with sensible defaults for every field the summarizer/renderer reads.""" base = { "question": "q", "answer_mode": "claim_lattice", "status": "cache_miss_then_written", "audit_mode": "STRICT", "n_quotes": 1, "n_verified": 1, "ratio": 1.0, "verifier_method": "claim_lattice", "lookup_path": "miss", "failure_stage": None, "lazy_anchor_ratio": None, "pointer_id_distribution": None, "cache_key": "abc123", "n_sources": 1, "elapsed_s": 1.0, "deflection_kind": "on_topic", "subject_anchor": "x", "subject_in_answer": True, "prompt_chars_total": 5000, "prompt_chars_evidence": 4000, "prompt_chars_system": 800, "prompt_chars_question": 30, "answer_chars": 100, # Ticket #000008 — bench harness extension for FORMAT_COLLAPSED # rate, violation-kind tallies, and bracket-count diagnostics. # None on `format_collapsed` means "check didn't apply" (non- # lattice modes), boolean otherwise. `violation_kinds` is a # sorted list of unique kind strings observed for the row. "format_collapsed": None, "violation_kinds": [], "answer_brackets": 0, # Phase 0.x bench fields (commit lands ahead of Phase 1 # classifier; keys stay present from Phase 0.x onwards). "answer_pointer_count": 0, "answer_chars_with_brackets": 0, "raw_meaningful_line_count": 0, "quantifier_intensity": None, "quantifier_matched_token": None, "scope_bound_hint": None, "claim_cap_applied": None, "model_profile_id": "test-model", # Ticket #000010 — preflight bench fields (defaults match a # neutral, well-formed-question run). "preflight_logical_statuses": ["well_formed"], "preflight_question_shape": "single_fact", "preflight_result": "PREFLIGHT_OK", "preflight_temporal_sensitivity": "low", "preflight_has_false_premise": False, "preflight_has_contradiction": False, "preflight_corpus_requirement": "encyclopedic", # #000009 §7.2 preflight_hash prefix (mirrors cache_key # truncation pattern). Defaults to empty for non-preflight # rows or legacy fixture rows. "preflight_hash": "", "directive_compliance": { "D2_pointer_clauses": True, "D3_cti_substrate_ready": True, "D4_evidence_map_bound": True, "D6_warrant_fired": True, "D7_honest_label": True, }, "error": None, } base.update(overrides) return base def test_summarize_counts_verdicts_by_mode(qa_sweep): rows = [ _row(answer_mode="quote", audit_mode="STRICT"), _row(answer_mode="quote", audit_mode="HYBRID"), _row(answer_mode="claim_lattice", audit_mode="STRICT"), _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED"), ] summary = qa_sweep._summarize(rows) assert summary["quote"]["STRICT"] == 1 assert summary["quote"]["HYBRID"] == 1 assert summary["claim_lattice"]["STRICT"] == 1 assert summary["claim_lattice"]["UNGROUNDED"] == 1 def test_summarize_counts_deflections_only_on_grounded_rows(qa_sweep): """A deflection_kind=='deflection' on UNGROUNDED isn't a real topic shift (no answer to deflect with). Only STRICT/HYBRID rows with deflection_kind='deflection' should count toward the deflections column.""" rows = [ _row(answer_mode="claim_lattice", audit_mode="STRICT", deflection_kind="deflection"), _row(answer_mode="claim_lattice", audit_mode="HYBRID", deflection_kind="deflection"), _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED", deflection_kind="deflection"), _row(answer_mode="claim_lattice", audit_mode="STRICT", deflection_kind="on_topic"), ] summary = qa_sweep._summarize(rows) # 2 grounded deflections (STRICT + HYBRID); UNGROUNDED + on_topic don't count. assert summary["claim_lattice"]["deflections"] == 2 def test_render_markdown_includes_summary_table_with_capacity_metrics(qa_sweep): rows = [ _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=5000), _row(answer_mode="claim_lattice", audit_mode="HYBRID", prompt_chars_total=10000), ] summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-01T00-00-00Z", ["claim_lattice"], ["q"], n_samples=2, ) # Headline table has the deflections column (added before capacity work). assert "deflections" in md assert "strict-rate" in md # Strict-rate-by-prompt-size section exists. assert "strict-rate by prompt size" in md # Buckets are present in the markdown body. assert "<8KB" in md assert "8-16KB" in md def test_render_markdown_buckets_strict_rate_by_size(qa_sweep): """Two STRICT rows in <8KB bucket + two UNGROUNDED rows in 8-16KB bucket → <8KB strict-rate=1.0, 8-16KB strict-rate=0.0.""" rows = [ _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=4000), _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=5000), _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED", prompt_chars_total=10000), _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED", prompt_chars_total=12000), ] summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-01T00-00-00Z", ["claim_lattice"], ["q"], n_samples=4, ) # Find the size-bucket section; verify both buckets show their # expected strict-rate. Markdown table cell format is `| value |`. bucket_section = md.split("strict-rate by prompt size")[1] # <8KB row: 2 runs, 2 STRICT, rate 1.00 lines_8kb = [ L for L in bucket_section.splitlines() if "<8KB" in L and "claim_lattice" in L ] assert lines_8kb, f"missing <8KB row in:\n{bucket_section[:600]}" assert " 1.00 " in lines_8kb[0] # 8-16KB row: 2 runs, 0 STRICT, rate 0.00 lines_16kb = [ L for L in bucket_section.splitlines() if "8-16KB" in L and "claim_lattice" in L ] assert lines_16kb, f"missing 8-16KB row in:\n{bucket_section[:600]}" assert " 0.00 " in lines_16kb[0] def test_render_markdown_recommended_context_budget_section(qa_sweep): """Bench surfaces a per-mode peak-strict-rate bucket as the 'recommended context budget' — operator-readable, not auto- applied. Minimum sample size of 5 runs/bucket; smaller samples fall through as 'insufficient samples'.""" # 6 STRICT in <8KB bucket + 6 UNGROUNDED in 16-32KB bucket # → <8KB has strict-rate 1.0, 16-32KB has 0.0; <8KB recommended. rows = [] for _ in range(6): rows.append(_row( answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=4000, )) for _ in range(6): rows.append(_row( answer_mode="claim_lattice", audit_mode="UNGROUNDED", prompt_chars_total=20000, )) summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-02T00-00-00Z", ["claim_lattice"], ["q"], n_samples=12, ) assert "recommended context budget (learned from this bench)" in md rec_section = md.split("recommended context budget")[1] # Peak bucket is <8KB at 100%. assert "<8KB" in rec_section assert "1.00" in rec_section # 16-32KB bucket (0% strict) does NOT show as the peak. rec_table_lines = [ L for L in rec_section.splitlines() if "claim_lattice" in L and "<8KB" in L ] assert rec_table_lines, f"missing recommended-bucket row:\n{rec_section[:500]}" def test_render_markdown_recommended_budget_skips_undersampled(qa_sweep): """Buckets with fewer than 5 runs don't qualify as 'recommended' — the recommendation needs statistical weight.""" # 3 STRICT in <8KB (under sample-size floor) + 3 STRICT in # 16-32KB (also under floor). All buckets have <5 runs → # 'insufficient samples' fallback. rows = [ _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=4000), _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=4000), _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=4000), _row(answer_mode="claim_lattice", audit_mode="UNGROUNDED", prompt_chars_total=20000), ] summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-02T00-00-00Z", ["claim_lattice"], ["q"], n_samples=4, ) rec_section = md.split("recommended context budget")[1] assert "insufficient samples" in rec_section def test_render_markdown_log_scale_buckets_cover_giant_context(qa_sweep): """Log-scale buckets reach 1M+ chars so the same bench harness covers 8B-class models (Hermes 82K context) through 1M-context models (Gemini / Claude). A 500K-char prompt lands in the 512K-1M bucket without code change.""" rows = [] for _ in range(6): rows.append(_row( answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=600_000, # 600KB → 512K-1M bucket )) summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-02T00-00-00Z", ["claim_lattice"], ["q"], n_samples=6, ) assert "512K-1M" in md rec_section = md.split("recommended context budget")[1] # 600K-char prompts land in the giant bucket; recommendation # should pick it. assert "512K-1M" in rec_section def test_render_markdown_skips_empty_buckets(qa_sweep): """Bucket with zero runs in a mode shouldn't render an empty row — the table should only include cells with data.""" rows = [ _row(answer_mode="claim_lattice", audit_mode="STRICT", prompt_chars_total=5000), ] summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-01T00-00-00Z", ["claim_lattice"], ["q"], n_samples=1, ) bucket_section = md.split("strict-rate by prompt size")[1] # Only <8KB has data; >=64KB / 32-64KB / 16-32KB / 8-16KB rows # should be absent. for missing_label in (">=64KB", "32-64KB", "16-32KB", "8-16KB"): assert missing_label not in bucket_section, ( f"empty bucket {missing_label} leaked into rendering:\n" f"{bucket_section[:600]}" ) # --------------------------------------------------------------------------- # directive coverage (seven-point program) # --------------------------------------------------------------------------- def test_summarize_aggregates_directive_pass_counts_per_mode(qa_sweep): """Per-row directive_compliance booleans aggregate into per-mode pass counts. A 2-row sample with one D2 fail should report 1/2.""" rows = [ _row( answer_mode="claim_lattice", directive_compliance={ "D2_pointer_clauses": True, "D3_cti_substrate_ready": True, "D4_evidence_map_bound": True, "D6_warrant_fired": True, "D7_honest_label": True, }, ), _row( answer_mode="claim_lattice", directive_compliance={ "D2_pointer_clauses": False, # the failing row "D3_cti_substrate_ready": True, "D4_evidence_map_bound": True, "D6_warrant_fired": True, "D7_honest_label": True, }, ), ] summary = qa_sweep._summarize(rows) dp = summary["claim_lattice"]["directive_pass"] assert dp["D2_pointer_clauses"] == 1 assert dp["D3_cti_substrate_ready"] == 2 assert dp["D4_evidence_map_bound"] == 2 assert dp["D6_warrant_fired"] == 2 assert dp["D7_honest_label"] == 2 def test_render_markdown_directive_coverage_section(qa_sweep): """The markdown summary includes a 'directive coverage' table showing per-mode pass counts per directive.""" rows = [_row(answer_mode="claim_lattice")] summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-01T00-00-00Z", ["claim_lattice"], ["q"], n_samples=1, ) assert "directive coverage (seven-point program)" in md # Header row contains the directive abbreviations. assert "D2 pointer" in md assert "D3 cti-ready" in md assert "D4 ev-map bound" in md assert "D6 warrant" in md assert "D7 honest label" in md # Body cell shows fraction (1/1) for the synthetic row. coverage_section = md.split("directive coverage")[1] # Per-mode row with 100% coverage on each directive. assert "1/1 (100%)" in coverage_section def test_directive_compliance_helper_marks_quote_mode_d2_false(qa_sweep): """Quote-mode rows fail D2 by definition (D2 demands lattice answer mode).""" result = { "audit_mode": "STRICT", "verifier_method": "quote", "run_dag_root": None, # quote mode predates the run-DAG split } dc = qa_sweep._directive_compliance( answer_mode="quote", result=result, err=None, ) assert dc["D2_pointer_clauses"] is False assert dc["D3_cti_substrate_ready"] is False # not lattice mode assert dc["D4_evidence_map_bound"] is False # no run_dag_root assert dc["D6_warrant_fired"] is False # D7: quote-mode STRICT keeps the STRICT label; the four-rung # ladder (POINTER-LINKED / ANCHOR-WARRANTED / EVIDENCE-WARRANTED) # applies to lattice modes only. Quote mode passes D7 vacuously. assert dc["D7_honest_label"] is True def test_directive_compliance_helper_marks_lattice_mode_d2_true(qa_sweep): """Lattice-mode rows pass D2.""" result = { "audit_mode": "STRICT", "verifier_method": "claim_lattice", "run_dag_root": "abc" * 21, } dc = qa_sweep._directive_compliance( answer_mode="claim_lattice", result=result, err=None, ) assert dc["D2_pointer_clauses"] is True assert dc["D3_cti_substrate_ready"] is True assert dc["D4_evidence_map_bound"] is True assert dc["D6_warrant_fired"] is True assert dc["D7_honest_label"] is True def test_directive_compliance_returns_empty_on_error_row(qa_sweep): """Error rows don't carry directive signal — no per-row check can succeed when the run threw.""" dc = qa_sweep._directive_compliance( answer_mode="claim_lattice", result={}, err="some error", ) assert dc == {} # Ticket #000008 — bench harness extension. The summarizer must: # - count FORMAT_COLLAPSED firings per mode (only explicit True; # None means the check didn't apply, not a counter increment) # - tally per-violation-kind counts so we can see which kinds # dominate which mode # - aggregate raw-output bracket counts on lattice rows so we # can chart "model is following the pointer protocol" vs # FORMAT_COLLAPSED at aggregate scale # The renderer must surface those tallies in a dedicated section. def test_summarize_counts_format_collapsed_only_on_explicit_true(qa_sweep): """format_collapsed=None means the check didn't run (non-lattice rows). Only explicit True counts. False counts as not-collapsed.""" rows = [ _row(answer_mode="claim_lattice_pointer", format_collapsed=True), _row(answer_mode="claim_lattice_pointer", format_collapsed=False), _row(answer_mode="claim_lattice_pointer", format_collapsed=None), _row(answer_mode="quote", format_collapsed=None), ] summary = qa_sweep._summarize(rows) assert summary["claim_lattice_pointer"]["format_collapses"] == 1 assert summary["quote"]["format_collapses"] == 0 def test_summarize_tallies_violation_kinds_per_mode(qa_sweep): """Each kind counts once per row even if the same kind fires on multiple claims. Different rows in the same mode accumulate.""" rows = [ _row(answer_mode="claim_lattice_pointer", violation_kinds=["FORMAT_COLLAPSED", "TITLE_MISMATCH"]), _row(answer_mode="claim_lattice_pointer", violation_kinds=["TITLE_MISMATCH"]), _row(answer_mode="claim_lattice", violation_kinds=["WARRANT_MISSING"]), ] summary = qa_sweep._summarize(rows) pointer_counts = summary["claim_lattice_pointer"]["violation_kind_counts"] assert pointer_counts["FORMAT_COLLAPSED"] == 1 assert pointer_counts["TITLE_MISMATCH"] == 2 json_counts = summary["claim_lattice"]["violation_kind_counts"] assert json_counts["WARRANT_MISSING"] == 1 def test_summarize_aggregates_brackets_only_on_lattice_modes(qa_sweep): """Quote/span/entity/paraphrase rows always record 0 brackets; averaging them in would skew the lattice-mode signal. Only claim_lattice* modes contribute to the bracket-count aggregate.""" rows = [ _row(answer_mode="claim_lattice_pointer", answer_brackets=10), _row(answer_mode="claim_lattice_pointer", answer_brackets=20), _row(answer_mode="quote", answer_brackets=0), ] summary = qa_sweep._summarize(rows) pointer = summary["claim_lattice_pointer"] assert pointer["answer_brackets_n"] == 2 assert pointer["answer_brackets_sum"] == 30 quote = summary["quote"] assert quote["answer_brackets_n"] == 0 assert quote["answer_brackets_sum"] == 0 def test_render_markdown_includes_format_collapse_section(qa_sweep): """A bench summary with at least one FORMAT_COLLAPSED row should surface a 'format-collapse' section with per-mode tallies.""" rows = [ _row(answer_mode="claim_lattice_pointer", format_collapsed=True, violation_kinds=["FORMAT_COLLAPSED"], answer_brackets=0), _row(answer_mode="claim_lattice_pointer", format_collapsed=False, violation_kinds=[], answer_brackets=14), ] summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-02T00-00-00Z", ["claim_lattice_pointer"], ["q"], n_samples=2, ) assert "format-collapse" in md.lower() # FORMAT_COLLAPSED kind appears as a column header when present. assert "FORMAT_COLLAPSED" in md # Per-mode rate cell shows 1/2. assert "1/2" in md def test_render_markdown_handles_no_violations(qa_sweep): """Renderer must not crash when no violations fired across the sweep — the violation-kind union is empty, so the table degrades to mode + format-collapse + mean-brackets columns only.""" rows = [_row(answer_mode="quote", violation_kinds=[])] summary = qa_sweep._summarize(rows) md = qa_sweep._render_markdown( rows, summary, "2026-05-02T00-00-00Z", ["quote"], ["q"], n_samples=1, ) # No exception, section header still present, no kind columns. assert "format-collapse" in md.lower() # Ticket #000008 Phase 0.x — bracket diagnostics helpers. Mirror the # verifier's FORMAT_COLLAPSED detector inputs so bench-side and # verifier-side numbers agree. def test_bracket_diagnostics_empty_returns_zeros(qa_sweep): assert qa_sweep._bracket_diagnostics("") == (0, 0, 0, 0) assert qa_sweep._bracket_diagnostics(None) == (0, 0, 0, 0) def test_bracket_diagnostics_single_pointer(qa_sweep): raw = "Tyrannosaurus rex appears in the climactic scene. [E1]\n" bracket_count, distinct, chars, lines = qa_sweep._bracket_diagnostics(raw) assert bracket_count == 1 assert distinct == 1 # `[E1]` = 4 characters assert chars == 4 assert lines == 1 def test_bracket_diagnostics_multi_pointer_in_one_bracket(qa_sweep): """`[E1, E2]` is one bracket region but two distinct pointer ids. `answer_brackets` counts `[E\\d+` openings (1); `answer_pointer _count` counts unique ids (2).""" raw = "Claim text. [E1, E2]\n" bracket_count, distinct, chars, _ = qa_sweep._bracket_diagnostics(raw) assert bracket_count == 1 assert distinct == 2 # `[E1, E2]` = 8 characters assert chars == 8 def test_bracket_diagnostics_multiple_pointers_separate_brackets(qa_sweep): # Each line must be >20 chars to count as meaningful (matches the # verifier FORMAT_COLLAPSED threshold). raw = ( "Tyrannosaurus rex appears in the first scene. [E1]\n" "Velociraptors stalk the workers in the kitchen. [E2]\n" "The T-rex breaks out of the paddock dramatically. [E1]\n" ) bracket_count, distinct, _, lines = qa_sweep._bracket_diagnostics(raw) assert bracket_count == 3 # three `[E\d+` openings assert distinct == 2 # only E1 + E2 unique assert lines == 3 def test_bracket_diagnostics_format_collapsed_shape(qa_sweep): """The 2026-05-02 winners-of-all-major-sports case: 5+ meaningful prose lines with zero brackets. Confirms the bench reports the same denominator the verifier's FORMAT_COLLAPSED gate uses.""" raw = "\n".join( f"The 19{i} FINA Men's Water Polo World Cup was won by Hungary." for i in range(70, 80) ) bracket_count, distinct, chars, lines = qa_sweep._bracket_diagnostics(raw) assert bracket_count == 0 assert distinct == 0 assert chars == 0 assert lines >= 5