arborist/tests/test_bench_batteries.py

"""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 — HARD live-path Formulate tier --------------


def test_5f_formulate_hard_pack_below_ceiling():
    """Below-ceiling baseline for the Formulate sub-battery: 12 inputs
    that parse_pointer_claims should segment a particular way; the
    line/bullet-based parser mis-segments 8 of 12 (merges multi-claim
    lines, splits wrapped bullets) → rate 4/12. parse_pointer_claims
    is deterministic, so this is stable; the pinned value fires if it
    shifts."""
    res = b_5f.run_formulate(F5F / "formulate-hard-v1.jsonl")
    assert res.pass_count == 4
    assert res.fail_count == 8
    assert res.metrics["structural_match_rate"] == pytest.approx(4 / 12)
    for t in res.per_task:
        assert t.detail["source"] == "live"
    # The 8 that fail do so on claim-count mismatch (the parser put
    # multiple claims on one line, or split one across lines) — that's
    # the mis-segmentation, not a noisy pointer/text near-miss.
    for t in res.per_task:
        if not t.passed:
            assert t.detail["count_ok"] is False


# --- #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