"""5S/5T/5F battery harness tests. Covers: - Phase 1a (#000021) — fixture digest stability + smoke - Phase 1b (#000023, #000024) — Syllogism/Synthesis/Semiotics + Transfer-Learning/Triangulation/Truthtables/Transitivity/Time - Phase 1a (#000025) — 5F: Function/Finetuning/Falsification/ Formulate/Feedback Loop - Carrier-metadata schema: unsupported carriers fail cleanly - Determinism: fixture_digest stable across reads """ from __future__ import annotations import json from dataclasses import asdict from pathlib import Path import pytest from bench.batteries import b_5f, b_5s, b_5t from bench.batteries.base import ( PHASE_1_CARRIERS, BatteryResult, fixture_digest, fixture_meta, iter_tasks, validate_carrier, ) REPO_ROOT = Path(__file__).resolve().parent.parent F5S = REPO_ROOT / "bench" / "fixtures" / "5s" F5T = REPO_ROOT / "bench" / "fixtures" / "5t" F5F = REPO_ROOT / "bench" / "fixtures" / "5f" # --------------------------------------------------------------------- # Phase 1a digest stability (must NOT change after Phase 1b lands) # --------------------------------------------------------------------- def test_phase1a_5s_syntax_digest_stable(): a = fixture_digest(F5S / "syntax-v1.jsonl") b = fixture_digest(F5S / "syntax-v1.jsonl") assert a == b assert len(a) == 64 def test_phase1a_5s_semantics_digest_stable(): a = fixture_digest(F5S / "semantics-v1.jsonl") b = fixture_digest(F5S / "semantics-v1.jsonl") assert a == b def test_phase1a_5t_transfer_digest_stable(): """Per #000024 hard constraint: legacy transfer-v1 digest stays pinned.""" a = fixture_digest(F5T / "transfer-v1.jsonl") b = fixture_digest(F5T / "transfer-v1.jsonl") assert a == b # --------------------------------------------------------------------- # Carrier metadata schema (#000023/#000024/#000025) # --------------------------------------------------------------------- def test_phase1_carriers_whitelist_present(): """Whitelist includes the expected Phase-1 carriers.""" for c in ("text", "claim_lattice", "memory_snapshot", "selfmodel_snapshot", "providence_record", "verifier_strategies", "propositional_logic", "relation_graph"): assert c in PHASE_1_CARRIERS def test_validate_carrier_accepts_phase_1(): assert validate_carrier({"carrier": "text"}) is None assert validate_carrier({"carrier": "claim_lattice"}) is None assert validate_carrier({}) is None # missing → defaults to text def test_validate_carrier_rejects_unsupported(): reason = validate_carrier({"carrier": "image"}) assert reason is not None assert "unsupported_carrier" in reason reason = validate_carrier({"carrier": "hidden_channel"}) assert reason is not None # --------------------------------------------------------------------- # 5S Phase 1b — Syllogism / Synthesis / Semiotics # --------------------------------------------------------------------- def test_5s_syllogism_runs_30_fixtures(): res = b_5s.run_syllogism(F5S / "syllogism-v1.jsonl") assert res.battery == "5s" assert res.sub_battery == "syllogism" assert res.pass_count + res.fail_count == 30 assert res.pass_count == 30 # all designed to pass assert res.metrics["step_validity_rate"] == 1.0 def test_5s_synthesis_runs_30_fixtures(): res = b_5s.run_synthesis(F5S / "synthesis-v1.jsonl") assert res.pass_count + res.fail_count == 30 assert res.pass_count == 30 assert res.metrics["derivation_pass_rate"] == 1.0 def test_5s_semiotics_runs_30_fixtures(): res = b_5s.run_semiotics(F5S / "semiotics-v1.jsonl") assert res.pass_count + res.fail_count == 30 assert res.pass_count == 30 assert res.metrics["invariance_under_swap"] == 1.0 def test_5s_syllogism_fixture_digest_stable(): a = fixture_digest(F5S / "syllogism-v1.jsonl") b = fixture_digest(F5S / "syllogism-v1.jsonl") assert a == b def test_5s_synthesis_fixture_digest_stable(): a = fixture_digest(F5S / "synthesis-v1.jsonl") b = fixture_digest(F5S / "synthesis-v1.jsonl") assert a == b def test_5s_semiotics_fixture_digest_stable(): a = fixture_digest(F5S / "semiotics-v1.jsonl") b = fixture_digest(F5S / "semiotics-v1.jsonl") assert a == b # --------------------------------------------------------------------- # 5T Phase 1b — Transfer Learning / Triangulation / Truthtables / # Transitivity / Time # --------------------------------------------------------------------- def test_5t_transfer_learning_runs(): res = b_5t.run_transfer_learning(F5T / "transfer-learning-v2.jsonl") assert res.battery == "5t" assert res.sub_battery == "transfer-learning" assert res.pass_count == 30 assert res.metrics["transfer_learning_success_rate"] == 1.0 def test_5t_triangulation_runs(): res = b_5t.run_triangulation(F5T / "triangulation-v1.jsonl") assert res.pass_count == 30 assert res.metrics["triangulation_agreement_rate"] == 1.0 def test_5t_truthtables_runs(): res = b_5t.run_truthtables(F5T / "truthtables-v1.jsonl") assert res.pass_count == 30 assert res.metrics["truth_table_coverage_rate"] == 1.0 def test_5t_truthtables_caps_at_n_4(): """Per ticket #000024 §4.4: N > 4 must be rejected.""" import tempfile bad = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) bad.write(json.dumps({"_meta": {"battery": "5t", "sub_battery": "truthtables", "version": "v1"}}) + "\n") bad.write(json.dumps({ "id": "5t-tt-toobig", "carrier": "claim_lattice", "domain": "propositional_logic", "pi_star_ref": "pi_truth_table_v1", "variables": ["A", "B", "C", "D", "E"], "expression": "A AND B AND C AND D AND E", "rows": [{"inputs": {v: False for v in "ABCDE"}, "expected": False}], }) + "\n") bad.close() res = b_5t.run_truthtables(Path(bad.name)) assert res.pass_count == 0 assert res.fail_count == 1 assert "exceeds cap" in res.per_task[0].detail["reason"] def test_5t_transitivity_runs(): res = b_5t.run_transitivity(F5T / "transitivity-v1.jsonl") assert res.pass_count == 30 assert res.metrics["full_chain_pass_rate"] == 1.0 def test_5t_transitivity_rejects_non_whitelisted_relation(): """Non-whitelisted relations always fail by construction.""" import tempfile bad = tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) bad.write(json.dumps({"_meta": {"battery": "5t", "sub_battery": "transitivity", "version": "v1"}}) + "\n") bad.write(json.dumps({ "id": "test", "carrier": "claim_lattice", "domain": "relation_graph", "pi_star_ref": "pi_relation_graph_v1", "edges": [ {"from": "A", "to": "B", "relation": "related_to"}, {"from": "B", "to": "C", "relation": "related_to"}, ], "query": {"from": "A", "to": "C", "relation": "related_to"}, "expected": "pass", # but it can't possibly pass on a non-transitive relation }) + "\n") bad.close() res = b_5t.run_transitivity(Path(bad.name)) # observed=fail because not in whitelist; expected=pass; mismatch. assert res.fail_count == 1 def test_5t_time_runs_against_synthetic_snapshots(): res = b_5t.run_time(F5T / "time-v1.jsonl") assert res.pass_count == 30 assert res.metrics["temporal_context_preservation_rate"] == 1.0 # --------------------------------------------------------------------- # 5F Phase 1a — Function / Finetuning / Falsification / Formulate / # Feedback Loop # --------------------------------------------------------------------- def test_5f_function_runs(): res = b_5f.run_function(F5F / "function-v1.jsonl") assert res.battery == "5f" assert res.sub_battery == "function" # Phase 1c (2026-05-09) — fixture catalog expanded 10 → 30. assert res.pass_count == 50 # Phase 1d assert res.metrics["function_pass_rate"] == 1.0 def test_5f_finetuning_runs(): res = b_5f.run_finetuning(F5F / "finetuning-v1.jsonl") assert res.pass_count == 50 # Phase 1d assert res.metrics["adaptation_improvement_rate"] == 1.0 def test_5f_falsification_runs(): res = b_5f.run_falsification(F5F / "falsification-v1.jsonl") assert res.pass_count == 62 # Phase 1e: +12 fixtures covering uncovered motifs assert res.metrics["error_detection_rate"] == 1.0 def test_5f_formulate_runs(): res = b_5f.run_formulate(F5F / "formulate-v1.jsonl") assert res.pass_count == 50 # Phase 1d assert res.metrics["structural_match_rate"] == 1.0 def test_5f_feedback_loop_runs(): res = b_5f.run_feedback_loop(F5F / "feedback-loop-v1.jsonl") assert res.pass_count == 50 # Phase 1d assert res.metrics["integration_coverage_rate"] == 1.0 # --- 5F efficiency-metric zero-cost guards (2026-05-08 review) ---- def test_efficiency_zero_cost_positive_gain_returns_infinite(): from bench.batteries.b_5f import EFFICIENCY_INFINITE, _efficiency assert _efficiency(0.5, 0) == EFFICIENCY_INFINITE def test_efficiency_zero_cost_zero_gain_returns_zero(): from bench.batteries.b_5f import EFFICIENCY_UNDEFINED, _efficiency assert _efficiency(0, 0) == EFFICIENCY_UNDEFINED assert _efficiency(0, 0) == 0.0 def test_efficiency_zero_cost_negative_gain_returns_neg_infinite(): from bench.batteries.b_5f import EFFICIENCY_INFINITE, _efficiency assert _efficiency(-0.3, 0) == -EFFICIENCY_INFINITE def test_efficiency_normal_ratio(): from bench.batteries.b_5f import _efficiency assert _efficiency(1.0, 4.0) == 0.25 assert _efficiency(2.0, 1.0) == 2.0 def test_5f_finetuning_emits_efficiency_metrics(): res = b_5f.run_finetuning(F5F / "finetuning-v1.jsonl") assert "adaptation_efficiency_mean_finite" in res.metrics assert "adaptation_efficiency_infinite_count" in res.metrics assert "adaptation_efficiency_neg_infinite_count" in res.metrics # Per-task detail carries the per-task efficiency. for t in res.per_task: assert "adaptation_efficiency" in t.detail assert "capital_cost_delta" in t.detail def test_5f_feedback_loop_emits_efficiency_metrics(): res = b_5f.run_feedback_loop(F5F / "feedback-loop-v1.jsonl") assert "feedback_efficiency_mean_finite" in res.metrics assert "feedback_efficiency_infinite_count" in res.metrics for t in res.per_task: assert "feedback_efficiency" in t.detail assert "chain_length" in t.detail # --- low-level kernel edge cases --------------------------------- def test_eval_propositional_parens_nesting(): from bench.batteries.b_5t import _eval_propositional # ((A AND B) OR C) on (T,F,T) should be (T AND F) OR T = T. assert _eval_propositional("(A AND B) OR C", {"A": True, "B": False, "C": True}) is True assert _eval_propositional("A AND (B OR C)", {"A": True, "B": False, "C": True}) is True assert _eval_propositional("NOT (A AND B)", {"A": True, "B": True}) is False def test_eval_propositional_rejects_unknown_variable(): from bench.batteries.b_5t import _eval_propositional with pytest.raises(ValueError): _eval_propositional("X AND Y", {"X": True}) # Y missing def test_eval_propositional_rejects_malformed(): from bench.batteries.b_5t import _eval_propositional with pytest.raises(ValueError): _eval_propositional("A AND", {"A": True}) # incomplete with pytest.raises(ValueError): _eval_propositional("(A OR B", {"A": True, "B": False}) # unbalanced def test_eval_propositional_xor_iff_impl(): from bench.batteries.b_5t import _eval_propositional # IMPL truth table: T→F is the only false case. for a, b in [(True, True), (True, False), (False, True), (False, False)]: expected_impl = (not a) or b expected_iff = a == b expected_xor = a != b assert _eval_propositional("A IMPL B", {"A": a, "B": b}) == expected_impl assert _eval_propositional("A IFF B", {"A": a, "B": b}) == expected_iff assert _eval_propositional("A XOR B", {"A": a, "B": b}) == expected_xor def test_walk_relation_path_handles_self_loop(): from bench.batteries.b_5t import _walk_relation_path # A→A self-loop should resolve immediately. assert _walk_relation_path([], "A", "A", "implies") is True def test_walk_relation_path_handles_cycles_without_infinite_loop(): from bench.batteries.b_5t import _walk_relation_path edges = [ {"from": "A", "to": "B", "relation": "implies"}, {"from": "B", "to": "C", "relation": "implies"}, {"from": "C", "to": "A", "relation": "implies"}, # back-edge ] # Should still find C reachable from A; no infinite loop. assert _walk_relation_path(edges, "A", "C", "implies") is True # And handle unreachable target cleanly. assert _walk_relation_path(edges, "A", "Z", "implies") is False def test_walk_relation_path_rejects_non_whitelisted_relation(): from bench.batteries.b_5t import _walk_relation_path edges = [ {"from": "A", "to": "B", "relation": "related_to"}, {"from": "B", "to": "C", "relation": "related_to"}, ] # related_to is NOT in the transitive whitelist → always False. assert _walk_relation_path(edges, "A", "C", "related_to") is False def test_content_tokens_strips_punctuation(): from bench.batteries.b_5s import _content_tokens tokens = _content_tokens("Hello, world! This is a test.") assert "hello" in tokens assert "world" in tokens assert "test" in tokens # Stopwords removed. assert "is" not in tokens assert "a" not in tokens assert "this" not in tokens def test_content_tokens_handles_unicode(): from bench.batteries.b_5s import _content_tokens tokens = _content_tokens("Bonjour, café Paris!") assert "bonjour" in tokens assert "café" in tokens assert "paris" in tokens def test_capital_cost_delta_handles_missing_budget(): """Empty resource_budget → zero cost.""" from bench.batteries.b_5f import _capital_cost_delta assert _capital_cost_delta({}) == 0.0 assert _capital_cost_delta({"resource_budget": {}}) == 0.0 assert _capital_cost_delta({ "resource_budget": {"max_compute_ms_delta": 1000} }) == 1.0 assert _capital_cost_delta({ "resource_budget": {"max_storage_delta_bytes": 5_000_000} }) == 5.0 # --- 5F Phase 1b.2 — Formulate live wire-up --------------------- def test_5f_formulate_live_path_routes_through_parse_claims(): """The new formulate-live-v1.jsonl fixtures use input_text only; runner derives produced_lattice via parse_pointer_claims and matches against expected_lattice.""" res = b_5f.run_formulate(F5F / "formulate-live-v1.jsonl") assert res.pass_count == 50 # Phase 1d assert res.metrics["structural_match_rate"] == 1.0 # Every task ran through the live path. for t in res.per_task: assert t.detail["source"] == "live" def test_5f_formulate_embedded_path_still_works(): """Phase 1a fixtures (embedded produced_lattice) keep working after the Phase 1b.2 wire-up + Phase 1c expansion. Backward compat invariant.""" res = b_5f.run_formulate(F5F / "formulate-v1.jsonl") assert res.pass_count == 50 # Phase 1d — expanded 10 → 30 for t in res.per_task: assert t.detail["source"] == "embedded" def test_5f_formulate_live_helper_uses_real_arborist_parser(tmp_path): """The live path actually calls arborist.qa.parse_claims.parse_pointer_claims — not a stub.""" from bench.batteries.b_5f import _live_produced_lattice out = _live_produced_lattice("- Hello world. [E1]\n- Second one. [E2]") assert out["claims"][0]["claim_text"] == "Hello world." assert out["claims"][0]["pointer_ids"] == ["E1"] assert out["claims"][1]["claim_text"] == "Second one." assert out["claims"][1]["pointer_ids"] == ["E2"] def test_5f_formulate_rejects_fixture_with_neither_field(tmp_path): """Task without produced_lattice OR input_text → fails cleanly.""" p = tmp_path / "bad.jsonl" p.write_text( json.dumps({"_meta": {"battery": "5f", "sub_battery": "formulate", "version": "v1"}}) + "\n" + json.dumps({ "id": "test", "carrier": "text", "domain": "claim_lattice", "pi_star_ref": "claim-lattice@v1", "expected_lattice": {"claim_count": 0, "claims": []}, }) + "\n", encoding="utf-8", ) res = b_5f.run_formulate(p) assert res.fail_count == 1 assert "produced_lattice" in res.per_task[0].detail["reason"] # --- 5F Phase 1b.2 — Feedback Loop live wire-up ----------------- def test_5f_feedback_loop_live_path_writes_real_audit_events(): """live_chain ops applied to a fresh temp shard via append_audit + memory.snapshot; expected_delta predicates verified against the resulting audit_events / memory_branch_summaries.""" res = b_5f.run_feedback_loop(F5F / "feedback-loop-live-v1.jsonl") assert res.pass_count == 50 # Phase 1d # Every passing task must report source=live. for t in res.per_task: if t.passed: assert t.detail["source"] == "live" def test_5f_feedback_loop_embedded_path_still_works(): res = b_5f.run_feedback_loop(F5F / "feedback-loop-v1.jsonl") assert res.pass_count == 50 # Phase 1d for t in res.per_task: assert t.detail["source"] == "embedded" def test_5f_live_feedback_chain_helper_appends_audit_event(tmp_path): """The live helper actually writes an audit event to a real arborist shard.""" from bench.batteries.b_5f import _live_feedback_chain state = _live_feedback_chain([ {"op": "append_audit", "event_type": "test_event", "body": {"k": "v"}}, {"op": "memory_snapshot"}, ]) assert "test_event" in state["audit_event_types"] assert "memory_snapshot_landed" in state["audit_event_types"] assert "audit-mode-distribution" in state["memory_branch_ids"] def test_5f_live_feedback_rejects_unknown_op(tmp_path): from bench.batteries.b_5f import _live_feedback_chain with pytest.raises(ValueError, match="unknown live op"): _live_feedback_chain([{"op": "fake_op"}]) def test_5f_live_feedback_chain_audit_chain_intact(): """Live helper exercises the real audit chain; chain should re-verify after the helper completes.""" import hashlib import os import tempfile from arborist.store import append_audit, connect, transaction tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) tmp.close() try: conn = connect(tmp.name) with transaction(conn): append_audit(conn, event_type="x", subject_root=None, body={"a": 1}) append_audit(conn, event_type="y", subject_root=None, body={"b": 2}) rows = conn.execute( "SELECT event_hash, prev_event_hash, body FROM audit_events ORDER BY seq" ).fetchall() prev = None for row in rows: h = hashlib.sha256() if prev is not None: h.update(bytes.fromhex(prev)) h.update(row["body"].encode("utf-8", errors="surrogatepass")) assert h.hexdigest() == row["event_hash"] prev = row["event_hash"] conn.close() finally: try: os.unlink(tmp.name) except OSError: pass # --- 5F §10.13 — feedback latency / real-workload efficiency -------- def test_5f_persisted_cost_helper(): """_persisted_cost = audit-row count + body bytes / 1e6 (the real footprint a live chain wrote, not the count of requested ops).""" from bench.batteries.b_5f import _persisted_cost assert _persisted_cost({}) == 0.0 state = { "audit_event_types": ["a", "b", "c"], "audit_event_bodies": ['{"x":1}', '{"y":2}', ""], # 7 + 7 + 0 bytes } assert _persisted_cost(state) == pytest.approx(3.0 + 14 / 1e6) def test_5f_feedback_loop_embedded_reports_no_latency(): res = b_5f.run_feedback_loop(F5F / "feedback-loop-v1.jsonl") assert res.metrics["feedback_latency_mean_seconds"] == 0.0 assert res.metrics["feedback_live_task_count"] == 0.0 for t in res.per_task: assert t.detail.get("feedback_latency_seconds") is None # Embedded cost-proxy stays len(chain) → integer-ratio efficiency. assert "persisted_cost" not in t.detail def test_5f_feedback_loop_live_reports_latency_and_persisted_cost(): res = b_5f.run_feedback_loop(F5F / "feedback-loop-live-v1.jsonl") assert res.metrics["feedback_live_task_count"] == 50.0 # Phase 1d assert res.metrics["feedback_latency_mean_seconds"] > 0.0 for t in res.per_task: if not t.passed: continue lat = t.detail["feedback_latency_seconds"] assert isinstance(lat, float) and lat >= 0.0 # Real footprint: every live chain has >= 1 append_audit op. assert t.detail["persisted_audit_rows"] >= 1 assert t.detail["persisted_cost"] >= 1.0 # feedback_efficiency now = downstream_effect / persisted_cost, # not / len(chain). downstream_effect is 1.0 when the chain # integrated, else 0.0 (an expected-"fail" task still passes). if t.detail["integrated"]: assert t.detail["feedback_efficiency"] == pytest.approx( 1.0 / t.detail["persisted_cost"] ) else: assert t.detail["feedback_efficiency"] == 0.0 # --- 5F Phase 1b.2 — Function / Finetuning / Falsification live - def test_5f_function_live_path_routes_through_parse_claims(): res = b_5f.run_function(F5F / "function-live-v1.jsonl") assert res.pass_count == 50 # Phase 1d for t in res.per_task: if t.passed: assert t.detail["source"] == "live" def test_5f_function_embedded_path_still_works(): res = b_5f.run_function(F5F / "function-v1.jsonl") assert res.pass_count == 50 # Phase 1d for t in res.per_task: assert t.detail["source"] == "embedded" def test_5f_function_live_helper_uses_real_parser(): from bench.batteries.b_5f import _live_function_produced out = _live_function_produced({"input_text": "- A claim. [E1]\n- B claim. [E2]"}) assert out["claims"][0]["claim_text"] == "A claim." assert out["claims"][0]["pointer_ids"] == ["E1"] assert out["claims"][1]["pointer_ids"] == ["E2"] def test_5f_finetuning_live_path_round_trips_selfmodel(): res = b_5f.run_finetuning(F5F / "finetuning-live-v1.jsonl") assert res.pass_count == 50 # Phase 1d for t in res.per_task: if t.passed: assert t.detail["source"] == "live" def test_5f_finetuning_embedded_path_still_works(): res = b_5f.run_finetuning(F5F / "finetuning-v1.jsonl") assert res.pass_count == 50 # Phase 1d for t in res.per_task: assert t.detail["source"] == "embedded" def test_5f_finetuning_live_helper_persists_real_selfmodel(): """The live helper writes to the SelfModel store and reads measured values back from claims_for.""" from bench.batteries.b_5f import _live_finetuning_measure out = _live_finetuning_measure({ "target_capability": "test_metric", "parent_measured_value": 0.40, "child_measured_value": 0.62, "expected_improvement_min": 0.05, "resource_budget": {"max_compute_ms_delta": 100, "max_storage_delta_bytes": 1000}, }) assert out["parent_measured"] == pytest.approx(0.40) assert out["child_measured"] == pytest.approx(0.62) assert len(out["parent_root"]) == 64 assert len(out["child_root"]) == 64 assert out["parent_root"] != out["child_root"] def test_5f_falsification_live_path_routes_through_verify_quotes(): res = b_5f.run_falsification(F5F / "falsification-live-v1.jsonl") assert res.pass_count == 50 # Phase 1d for t in res.per_task: if t.passed: assert t.detail["source"] == "live" def test_5f_falsification_embedded_path_still_works(): res = b_5f.run_falsification(F5F / "falsification-v1.jsonl") assert res.pass_count == 62 # Phase 1e: +12 fixtures covering uncovered motifs for t in res.per_task: assert t.detail["source"] == "embedded" # --- #000046 Phase 2 / #000048 step 2.4 — Formulate hard tier ----- def test_5f_formulate_hard_pack(): """The Formulate hard pack — 12 inputs `parse_pointer_claims` should segment a particular way. Was 4/12 at HEAD (the line/ bullet-only parser merged multi-claim lines / split wrapped bullets); #000048 step 2.4's clause segmenter (split on `;`, sentence boundaries, spaced dashes, conjunctions/subordinators, inline `(N)` enumerations, commas — kept only when every segment is well-pointered — plus a wrapped-bullet join) closes all 8 → 12/12. parse_pointer_claims is deterministic, so the pin is stable; if it shifts, the value fires. (The pack is now at ceiling — a harder Formulate tier would re-open below-ceiling headroom; a #000046 follow-up, out of #000048's scope.)""" res = b_5f.run_formulate(F5F / "formulate-hard-v1.jsonl") assert res.pass_count == 12 assert res.fail_count == 0 assert res.metrics["structural_match_rate"] == pytest.approx(1.0) for t in res.per_task: assert t.detail["source"] == "live" # --- #000046 Phase 1 — HARD live-path Falsification tier ---------- def test_5f_falsification_hard_pack_below_ceiling(): """The hard pack is a deliberate below-ceiling baseline: 12 near-misses whose correct verdict is UNGROUNDED. As of #000048 step 2.1 (2026-05-11): #000046's paraphrase numeric-agreement gate catches the 2 magnitude/year over-grounds (50-vs-100, 300-vs-300,000) + #000048's entity salient-token gate catches the 4 HYBRID_ENTITY over-grounds (Insulin/Berlin/1889/Pacific) → 10/12 pass; the remaining 2 are recombination STRICT_PARAPHRASE (hard-003 Mercury, hard-005 Einstein — the false claim recombines source tokens; lexical token-coverage can't tell recombination from grounding), headroom for #000048 step 2.2. verify_quotes is deterministic, so this is stable; if it shifts, the pinned value here fires (loud signal — either the verifier got better/worse, or the pack drifted).""" res = b_5f.run_falsification(F5F / "falsification-hard-v1.jsonl") assert res.pass_count == 10 assert res.fail_count == 2 assert res.metrics["error_detection_rate"] == pytest.approx(10 / 12) # All tasks route the live verifier; every fixture asserts UNGROUNDED. for t in res.per_task: assert t.detail["source"] == "live" assert t.detail["expected_reason"] == "UNGROUNDED" # The 2 that fail do so by over-grounding (STRICT_/HYBRID_), never # by the verifier saying UNGROUNDED — real over-ground misses, not # "the verifier abstained". for t in res.per_task: if not t.passed: obs = set(t.detail["observed_violations"]) assert "UNGROUNDED" not in obs assert any(o.startswith(("STRICT_", "HYBRID_")) for o in obs) def test_fork_score_positive_gamma_5f_on_hard_falsification_improvement(): """Worked example for #000046/#000048 — the loop: a real verifier improvement that lifts the falsification-hard rate produces a *positive* γ·Δ5f term in fork_score. Whether it clears SIGNAL_FLOOR (→ ACCEPT) or not (→ MARGINAL) depends on the magnitude and the ÷5 averaging dilution (#000047) — a small single-sub lift is positive-but-MARGINAL by design; a big-enough one ACCEPTs. Both shown here. The point: the bench Δ-rate carries signal it can't carry while every pack is at ceiling.""" from arborist.substrate.fork_score import fork_score hard_rate = b_5f.run_falsification( F5F / "falsification-hard-v1.jsonl" ).metrics["error_detection_rate"] assert hard_rate < 1.0 # (a) A real lift from the current hard-pack rate to 1.0 — positive, # possibly MARGINAL (the ÷5 dilution of a single-sub gain). parent_a = {"5s": {}, "5t": {}, "5f": {"falsification": {"error_detection_rate": hard_rate}}} child_a = {"5s": {}, "5t": {}, "5f": {"falsification": {"error_detection_rate": 1.0}}} sf_a = fork_score(parent_a, child_a) assert sf_a.breakdown["gamma_x_delta_5f"] > 0.0 assert sf_a.breakdown["gamma_x_delta_5f"] == pytest.approx((1.0 - hard_rate) / 5) assert sf_a.verdict in ("ACCEPT", "MARGINAL") assert not any(f.startswith(("REGRESSION_", "NEG_INF_")) for f in sf_a.flags) # (b) A bigger lift (degraded parent at 0.5 → child 1.0) clears the # floor: Δ5f = 0.5/5 = 0.1 ≥ SIGNAL_FLOOR → ACCEPT. parent_b = {"5s": {}, "5t": {}, "5f": {"falsification": {"error_detection_rate": 0.5}}} child_b = {"5s": {}, "5t": {}, "5f": {"falsification": {"error_detection_rate": 1.0}}} sf_b = fork_score(parent_b, child_b) assert sf_b.breakdown["gamma_x_delta_5f"] == pytest.approx(0.5 / 5) assert sf_b.verdict == "ACCEPT" assert not any(f.startswith(("REGRESSION_", "NEG_INF_")) for f in sf_b.flags) def test_5f_falsification_covers_every_documented_motif(): """#000025 §10.12: Falsification fixtures must cover every documented failure-motif tag. Pins motif-coverage of the falsification-v1.jsonl fixture pack against the verifier's `violations.append` registry + the soft-demote registry in arborist.cli.""" import json fixture_path = F5F / "falsification-v1.jsonl" covered: set[str] = set() with fixture_path.open() as f: for line in f: d = json.loads(line) if "expected_reason" in d: covered.add(d["expected_reason"]) documented = { # Hard violations emitted by arborist/qa/verify.py "CITATION_MISMATCH", "DEFLECTION_DETECTED", "FORMAT_COLLAPSED", "MANUAL_QUOTE_VIOLATION", "NO_EVIDENCE_POINTER", "POINTER_OVERFLOW_TRIMMED", "SCHEMA_INVALID", "SOURCE_ROLE_BLOCKED", "SUBJECT_TOKENS_ABSENT", "TITLE_MISMATCH", "TOO_MANY_CLAIMS", "TOO_MANY_EVIDENCE_IDS", "UNKNOWN_EVIDENCE_ID", "WARRANT_MISSING", # Soft-demote violations registered in arborist/cli.py "LAZY_ANCHOR_DEMOTED", "BARE_NAME_CLAIM", "BROAD_QUANTIFIER_RUNAWAY", "BROAD_QUANTIFIER_CAP_APPLIED", "BROAD_QUANTIFIER_SCOPE_UNBOUND", "BROAD_QUANTIFIER_REJECTED", } missing = documented - covered assert not missing, ( f"falsification-v1.jsonl missing {len(missing)} motif tag(s): " f"{sorted(missing)}. Add a fixture with `expected_reason` set to " "each, or remove from the documented registry if intentionally " "retired." ) def test_5f_falsification_harvested_pack_runs_clean(): """#000037 Phase 1 → 5F loop: the harvested pack at falsification-harvested-v1.jsonl was generated by bench/scripts/harvest_falsification_proposals.py from real qa.db high-divergence rows. Pin: pack runs at 100% pass-rate, all fixtures route through the embedded runner path, every fixture carries the harvest-attribution metadata.""" import json fixture_path = F5F / "falsification-harvested-v1.jsonl" res = b_5f.run_falsification(fixture_path) assert res.fail_count == 0 assert res.metrics["error_detection_rate"] == 1.0 # Every fixture must carry the harvest metadata so future # regenerations are traceable to the source providence_cache # row. with fixture_path.open() as f: next(f) # _meta header for line in f: row = json.loads(line) meta = row.get("_harvest_meta", {}) assert meta.get("source_ticket") == "#000037 §13 step 11" # Two harvest sources now: providence_cache directly # (Phase 1 buckets) and controller_events stream (live # Phase 2 advisory writes). assert meta.get("harvested_from") in ( "providence_cache", "controller_events", ) assert isinstance(meta.get("witness_divergence"), (int, float)) assert meta["witness_divergence"] >= 0.5 assert meta.get("audit_mode_at_harvest") in ("HYBRID", "UNGROUNDED") def test_5f_falsification_harvested_pack_widens_motif_coverage(): """The harvested pack introduces real-corpus motif tags that the hand-curated falsification-v1.jsonl doesn't have. Pin: the HYBRID_QUOTE + HYBRID_CLAIM_LATTICE pair appear in the harvested pack — concrete evidence that the harvest is doing widening work, not just duplicating the curated set.""" import json fixture_path = F5F / "falsification-harvested-v1.jsonl" motifs: set[str] = set() with fixture_path.open() as f: for line in f: d = json.loads(line) if "expected_reason" in d: motifs.add(d["expected_reason"]) # Harvest contains HYBRID variants surfaced from corpus that the # hand-curated set lacks. If the harvested DB drifts and these # motifs disappear, the test loud-fails — by then the harvest # rotation has lost signal and the pack should be regenerated. assert "HYBRID_QUOTE" in motifs or "HYBRID_CLAIM_LATTICE" in motifs assert "UNGROUNDED" in motifs def test_harvest_includes_controller_events_bucket(tmp_path): """The harvester pulls a third bucket from the live controller_events stream (event_kind = controller_falsification_proposal) in addition to the two providence_cache buckets. Builds a synthetic qa.db with both sources and asserts the controller_events-derived fixture appears with the right provenance.""" import sqlite3 as _sqlite3 from bench.scripts.harvest_falsification_proposals import harvest qa_db = tmp_path / "qa.db" conn = _sqlite3.connect(qa_db) conn.row_factory = _sqlite3.Row # Minimal providence_cache schema for the harvest path. conn.executescript( "CREATE TABLE providence_cache (" " cache_key TEXT PRIMARY KEY, audit_mode TEXT, n_quotes INTEGER," " n_verified INTEGER, unverified_quotes TEXT, verifier_method TEXT," " falsification_state TEXT, answer_text TEXT);" "CREATE TABLE controller_events (" " event_id INTEGER PRIMARY KEY AUTOINCREMENT," " organism_root TEXT NOT NULL, branch_id TEXT, event_kind TEXT NOT NULL," " label TEXT, entropy REAL, difficulty REAL, allocation REAL," " body_blob TEXT NOT NULL, body_hash TEXT NOT NULL," " recorded_at INTEGER NOT NULL," " UNIQUE (event_kind, body_hash));" ) # Providence row whose cache_key matches a future # controller_events branch_id prefix. Divergence < 0.5 so the row # is INVISIBLE to the providence buckets — it surfaces ONLY via the # controller_events stream. conn.execute( "INSERT INTO providence_cache VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ( "ce-only-cache-key-12345678abcdef", "HYBRID", 10, 9, json.dumps(["one quote"]), # 1/10 = 0.1 < threshold "claim_lattice", "live", "Synthetic answer text from a controller-events row.", ), ) # Matching controller_falsification_proposal row. Divergence here # is the high signal that surfaced the proposal at controller-decide # time (sweep mode would commonly push such rows). conn.execute( "INSERT INTO controller_events VALUES " "(NULL, ?, ?, ?, ?, NULL, NULL, NULL, ?, ?, ?)", ( "qa:ce-only-cache-key-12345678abcdef", "qa:ce-only-cache-key", # 16-char prefix of the cache_key "controller_falsification_proposal", "WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD", json.dumps({ "kind": "controller_falsification_proposal", "organism_root": "qa:ce-only-cache-key-12345678abcdef", "branch_id": "qa:ce-only-cache-key", "witness_divergence": 0.8, "reason": "WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD", }), "deadbeef" * 8, # placeholder body_hash for the test 1700000000, ), ) conn.commit() conn.close() fixtures = harvest(qa_db, threshold=0.5, sample_per_bucket=20) ce_fixtures = [ fx for fx in fixtures if fx["_harvest_meta"]["harvested_from"] == "controller_events" ] assert len(ce_fixtures) == 1 fx = ce_fixtures[0] assert fx["_harvest_meta"]["controller_organism_root"].startswith("qa:") assert fx["_harvest_meta"]["witness_divergence"] == 0.8 assert fx["id"] == "5f-fal-harvested-ce-only-cache-ke" def test_5f_falsification_live_helper_calls_real_verifier(): """UNGROUNDED + STRICT_SPAN signals come straight from verify_quotes.""" from bench.batteries.b_5f import _live_falsification_violations # Verbatim match → STRICT_SPAN. out_strict = _live_falsification_violations({ "answer_text": "Hydrogen has atomic number 1.", "context": "Hydrogen has atomic number 1 and is the lightest element.", }) assert "STRICT_SPAN" in out_strict # Unrelated → UNGROUNDED. out_ungrounded = _live_falsification_violations({ "answer_text": "K2 is the tallest mountain.", "context": "Mount Everest is the tallest mountain.", }) assert "UNGROUNDED" in out_ungrounded # --- 5R Phase 2 (#000021) ---------------------------------------- def test_5r_react_runs(): from bench.batteries import b_5r res = b_5r.run_react(REPO_ROOT / "bench" / "fixtures" / "5r" / "react-v1.jsonl") assert res.battery == "5r" assert res.sub_battery == "react" assert res.pass_count == 30 assert res.metrics["react_integration_rate"] == 1.0 def test_5r_rearrange_runs(): from bench.batteries import b_5r res = b_5r.run_rearrange(REPO_ROOT / "bench" / "fixtures" / "5r" / "rearrange-v1.jsonl") assert res.pass_count == 30 assert res.metrics["rearrange_invariance_rate"] == 1.0 def test_5r_restore_runs(): from bench.batteries import b_5r res = b_5r.run_restore(REPO_ROOT / "bench" / "fixtures" / "5r" / "restore-v1.jsonl") assert res.pass_count == 30 assert res.metrics["restore_retrievability_rate"] == 1.0 def test_5r_replicate_runs(): from bench.batteries import b_5r res = b_5r.run_replicate(REPO_ROOT / "bench" / "fixtures" / "5r" / "replicate-v1.jsonl") assert res.pass_count == 30 assert res.metrics["replicate_determinism_rate"] == 1.0 def test_5r_resonate_runs(): """Deterministic π* must yield distinct=1 across N runs.""" from bench.batteries import b_5r res = b_5r.run_resonate(REPO_ROOT / "bench" / "fixtures" / "5r" / "resonate-v1.jsonl") assert res.pass_count == 30 assert res.metrics["resonate_stability_rate"] == 1.0 assert res.metrics["mean_distinct_outputs"] == 1.0 def test_5r_react_rejects_unsupported_carrier(tmp_path): from bench.batteries import b_5r p = tmp_path / "bad.jsonl" p.write_text( json.dumps({"_meta": {"battery": "5r", "sub_battery": "react", "version": "v1"}}) + "\n" + json.dumps({ "id": "test", "carrier": "image", "snapshot_t0": {"facts": []}, "snapshot_t1": {"facts": []}, "expected_delta": {"added_facts": [], "removed_facts": []}, }) + "\n", encoding="utf-8", ) res = b_5r.run_react(p) assert res.fail_count == 1 assert "unsupported_carrier" in res.per_task[0].detail["reason"] # --- 5R Phase 1b.2 — React + Restore live --------------------------- def test_5r_react_live_path_writes_real_audit_events(): from bench.batteries import b_5r res = b_5r.run_react(REPO_ROOT / "bench" / "fixtures" / "5r" / "react-live-v1.jsonl") assert res.pass_count == 12 for t in res.per_task: if t.passed: assert t.detail["source"] == "live" def test_5r_react_embedded_path_still_works(): from bench.batteries import b_5r res = b_5r.run_react(REPO_ROOT / "bench" / "fixtures" / "5r" / "react-v1.jsonl") assert res.pass_count == 30 for t in res.per_task: assert t.detail["source"] == "embedded" def test_5r_restore_live_path_routes_through_audit_chain(): from bench.batteries import b_5r res = b_5r.run_restore(REPO_ROOT / "bench" / "fixtures" / "5r" / "restore-live-v1.jsonl") assert res.pass_count == 12 for t in res.per_task: if t.passed: assert t.detail["source"] == "live" def test_5r_restore_embedded_path_still_works(): from bench.batteries import b_5r res = b_5r.run_restore(REPO_ROOT / "bench" / "fixtures" / "5r" / "restore-v1.jsonl") assert res.pass_count == 30 for t in res.per_task: assert t.detail["source"] == "embedded" def test_5r_live_workspace_helper_writes_real_events(): """The shared _live_workspace_apply helper actually writes audit events to a real arborist shard.""" from bench.batteries.b_5r import _live_workspace_apply bodies = _live_workspace_apply(["alpha", "beta", "gamma"]) assert len(bodies) == 3 text = "\n".join(bodies) assert "alpha" in text assert "beta" in text assert "gamma" in text def test_finetuning_zero_cost_fixture_emits_inf(tmp_path): """Synthesize a fixture with zero resource_budget; assert inf emitted.""" p = tmp_path / "ft-zero.jsonl" p.write_text( json.dumps({"_meta": {"battery": "5f", "sub_battery": "finetuning", "version": "v1"}}) + "\n" + json.dumps({ "id": "5f-ft-zero", "carrier": "selfmodel_snapshot", "domain": "capability_transition", "pi_star_ref": "pi_selfmodel_v1", "parent_selfmodel": "P", "child_selfmodel": "C", "target_capability": "TEST", "parent_measured_value": 0.40, "child_measured_value": 0.60, "expected_improvement_min": 0.05, "resource_budget": { "max_compute_ms_delta": 0, "max_storage_delta_bytes": 0, }, "expected": "pass", }) + "\n", encoding="utf-8", ) res = b_5f.run_finetuning(p) assert res.pass_count == 1 assert res.metrics["adaptation_efficiency_infinite_count"] == 1.0 # The single task's detail should record the +inf efficiency. from bench.batteries.b_5f import EFFICIENCY_INFINITE assert res.per_task[0].detail["adaptation_efficiency"] == EFFICIENCY_INFINITE # --------------------------------------------------------------------- # Carrier rejection tests — runners fail unsupported carriers cleanly # --------------------------------------------------------------------- def _write_unsupported_fixture(path: Path, sub: str, battery: str = "5s") -> None: """Write a fixture with an unsupported carrier.""" path.write_text( json.dumps({"_meta": {"battery": battery, "sub_battery": sub, "version": "v1"}}) + "\n" + json.dumps({ "id": "test", "carrier": "image", # not in Phase-1 whitelist "domain": "scene_graph", "pi_star_ref": "image-base@v1", # the rest doesn't matter — runner rejects on carrier check "input": "x", "premises": [], "candidate_step": {"claim": "x", "uses": []}, "rule": "categorical_transitivity", "expected": "pass", }) + "\n", encoding="utf-8", ) def test_5s_syllogism_rejects_unsupported_carrier(tmp_path): p = tmp_path / "bad.jsonl" _write_unsupported_fixture(p, "syllogism") res = b_5s.run_syllogism(p) assert res.fail_count == 1 assert "unsupported_carrier" in res.per_task[0].detail["reason"] def test_5t_transfer_learning_rejects_unsupported_carrier(tmp_path): p = tmp_path / "bad.jsonl" _write_unsupported_fixture(p, "transfer-learning", battery="5t") res = b_5t.run_transfer_learning(p) assert res.fail_count == 1 assert "unsupported_carrier" in res.per_task[0].detail["reason"] def test_5f_function_rejects_unsupported_carrier(tmp_path): p = tmp_path / "bad.jsonl" _write_unsupported_fixture(p, "function", battery="5f") res = b_5f.run_function(p) assert res.fail_count == 1 assert "unsupported_carrier" in res.per_task[0].detail["reason"] # --------------------------------------------------------------------- # Runner CLI (existing test from Phase 1a) # --------------------------------------------------------------------- def test_runner_main_smoke(capsys): from bench.batteries.runner import main rc = main([ "--battery", "5s", "--sub", "syntax", "--fixtures", str(F5S / "syntax-v1.jsonl"), ]) out = capsys.readouterr().out payload = json.loads(out) assert payload["schema_version"] == "bench-result-v1" assert payload["results"][0]["battery"] == "5s" assert rc == 0 def test_runner_all_runs_full_suite(): """--all runs every battery in _DEFAULT_FIXTURES; rc=0 since all pass.""" from bench.batteries.runner import main rc = main(["--all"]) assert rc == 0