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"
    assert res.pass_count == 10
    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 == 10
    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 == 10
    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 == 10
    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 == 10
    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


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