arborist/tests/test_fork_score.py

"""Tests for ``arborist.substrate.fork_score`` — pure-function
ForkScore computation over (parent, child) BatteryResult bundles
(ticket #000012 Phase 1a).

Covers:
  - bench_result_to_metrics adapter (BatteryResult → nested dict)
  - fork_score over the three rate-deltas (5S, 5T, 5F) and the
    five non-bench inputs
  - hard-reject paths: NEG_INF_REGRESSION, per-battery
    HARD_REGRESSION_FLOOR
  - verdict band: ACCEPT (≥SIGNAL_FLOOR), MARGINAL ([0, SIGNAL_FLOOR)),
    REJECT (<0 or hard-reject)
  - empty inputs (zero parent, zero child) → score 0, verdict
    MARGINAL
  - ScoredFork.to_dict serializability
  - breakdown sum identity (score ≡ Σ breakdown)
"""

from __future__ import annotations

import math

import pytest

from arborist.substrate.fork_score import (
    HARD_REGRESSION_FLOOR,
    INFINITE_BONUS_CAP,
    SIGNAL_FLOOR,
    ScoredFork,
    bench_result_to_metrics,
    fork_score,
)
from arborist.substrate.weights import DEFAULT_WEIGHTS, WeightSet


# --- bench_result_to_metrics adapter --------------------------------


def test_bench_result_to_metrics_groups_by_battery():
    """Adapter from runner JSON to nested-dict shape."""
    payload = {
        "results": [
            {"battery": "5s", "sub_battery": "syntax",
             "metrics": {"parse_pass_rate": 0.9}},
            {"battery": "5s", "sub_battery": "semantics",
             "metrics": {"equivalence_recovery_rate": 0.85}},
            {"battery": "5t", "sub_battery": "transfer-learning",
             "metrics": {"transfer_learning_success_rate": 0.7}},
            {"battery": "5f", "sub_battery": "function",
             "metrics": {"function_pass_rate": 0.95}},
        ],
    }
    out = bench_result_to_metrics(payload)
    assert set(out.keys()) == {"5s", "5t", "5f"}
    assert out["5s"]["syntax"]["parse_pass_rate"] == 0.9
    assert out["5s"]["semantics"]["equivalence_recovery_rate"] == 0.85
    assert out["5t"]["transfer-learning"]["transfer_learning_success_rate"] == 0.7
    assert out["5f"]["function"]["function_pass_rate"] == 0.95


def test_bench_result_to_metrics_skips_unknown_battery():
    """Battery keys outside {5s, 5t, 5f} get dropped."""
    payload = {
        "results": [
            {"battery": "5r", "sub_battery": "rho",
             "metrics": {"rho_rate": 0.5}},
            {"battery": "5s", "sub_battery": "syntax",
             "metrics": {"parse_pass_rate": 0.8}},
        ],
    }
    out = bench_result_to_metrics(payload)
    assert "5r" not in out
    assert out["5s"]["syntax"]["parse_pass_rate"] == 0.8


def test_bench_result_to_metrics_empty_payload():
    """No `results` key → empty nested dicts (still keyed by battery)."""
    out = bench_result_to_metrics({})
    assert out == {"5s": {}, "5t": {}, "5f": {}}


# --- fork_score happy path ------------------------------------------


def _bench_dict(rates_by_battery_sub: dict[str, dict[str, dict[str, float]]]):
    """Helper: build a nested metrics dict directly."""
    out = {"5s": {}, "5t": {}, "5f": {}}
    for battery, subs in rates_by_battery_sub.items():
        out[battery] = {sub: dict(metrics) for sub, metrics in subs.items()}
    return out


def test_fork_score_pure_improvement_accepts():
    """Child improves on every 5S sub-battery by 10pp → score
    ≥ SIGNAL_FLOOR → ACCEPT verdict."""
    parent = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.5},
            "semantics": {"equivalence_recovery_rate": 0.5},
            "syllogism": {"step_validity_rate": 0.5},
            "synthesis": {"derivation_pass_rate": 0.5},
            "semiotics": {"invariance_under_swap": 0.5},
        },
    })
    child = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.6},
            "semantics": {"equivalence_recovery_rate": 0.6},
            "syllogism": {"step_validity_rate": 0.6},
            "synthesis": {"derivation_pass_rate": 0.6},
            "semiotics": {"invariance_under_swap": 0.6},
        },
    })
    r = fork_score(parent, child)
    assert r.verdict == "ACCEPT"
    assert r.score >= SIGNAL_FLOOR


def test_fork_score_marginal_band():
    """Child improves by less than SIGNAL_FLOOR's mapped weight
    → MARGINAL band (score in [0, SIGNAL_FLOOR))."""
    # Tiny improvement — 1pp on one sub-battery — well below the
    # signal floor when averaged.
    parent = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.50},
            "semantics": {"equivalence_recovery_rate": 0.50},
            "syllogism": {"step_validity_rate": 0.50},
            "synthesis": {"derivation_pass_rate": 0.50},
            "semiotics": {"invariance_under_swap": 0.50},
        },
    })
    child = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.51},
            "semantics": {"equivalence_recovery_rate": 0.50},
            "syllogism": {"step_validity_rate": 0.50},
            "synthesis": {"derivation_pass_rate": 0.50},
            "semiotics": {"invariance_under_swap": 0.50},
        },
    })
    r = fork_score(parent, child)
    assert r.verdict == "MARGINAL"
    assert 0 <= r.score < SIGNAL_FLOOR


def test_fork_score_zero_zero_yields_zero():
    """Zero parent + zero child → score 0 (no terms fire) → MARGINAL
    (score in [0, SIGNAL_FLOOR))."""
    r = fork_score({"5s": {}, "5t": {}, "5f": {}},
                   {"5s": {}, "5t": {}, "5f": {}})
    assert r.score == pytest.approx(0.0, abs=1e-12)
    assert r.verdict == "MARGINAL"


# --- regression / hard-reject paths ---------------------------------


def test_fork_score_hard_regression_5s_rejects():
    """Drop ≥HARD_REGRESSION_FLOOR on a 5S sub → REJECT regardless
    of overall positive score."""
    parent = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.9},
            "semantics": {"equivalence_recovery_rate": 0.5},
            "syllogism": {"step_validity_rate": 0.5},
            "synthesis": {"derivation_pass_rate": 0.5},
            "semiotics": {"invariance_under_swap": 0.5},
        },
    })
    # syntax drops by 0.20 (>= 0.05 hard floor); other subs improve.
    child = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.7},
            "semantics": {"equivalence_recovery_rate": 0.95},
            "syllogism": {"step_validity_rate": 0.95},
            "synthesis": {"derivation_pass_rate": 0.95},
            "semiotics": {"invariance_under_swap": 0.95},
        },
    })
    r = fork_score(parent, child)
    assert r.verdict == "REJECT"
    assert any("REGRESSION_5S" in f and "syntax" in f for f in r.flags)


def test_fork_score_negative_score_rejects():
    """Score < 0 (more decreases than increases) → REJECT."""
    parent = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.8},
            "semantics": {"equivalence_recovery_rate": 0.8},
            "syllogism": {"step_validity_rate": 0.8},
            "synthesis": {"derivation_pass_rate": 0.8},
            "semiotics": {"invariance_under_swap": 0.8},
        },
    })
    # All decrease by 0.04 (less than HARD_REGRESSION_FLOOR=0.05),
    # so no per-sub regression flag — but mean delta is negative
    # → score < 0 → REJECT
    child = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.76},
            "semantics": {"equivalence_recovery_rate": 0.76},
            "syllogism": {"step_validity_rate": 0.76},
            "synthesis": {"derivation_pass_rate": 0.76},
            "semiotics": {"invariance_under_swap": 0.76},
        },
    })
    r = fork_score(parent, child)
    assert r.verdict == "REJECT"
    assert r.score < 0


def test_fork_score_neg_infinite_count_in_5f_hard_rejects():
    """`adaptation_efficiency_neg_infinite_count > 0` on child →
    auto-REJECT regardless of other terms."""
    parent = _bench_dict({
        "5f": {
            "function": {"function_pass_rate": 0.5},
            "finetuning": {"adaptation_improvement_rate": 0.5},
            "falsification": {"error_detection_rate": 0.5},
            "formulate": {"structural_match_rate": 0.5},
            "feedback-loop": {"integration_coverage_rate": 0.5},
        },
    })
    child = _bench_dict({
        "5f": {
            "function": {"function_pass_rate": 0.95},
            "finetuning": {
                "adaptation_improvement_rate": 0.95,
                "adaptation_efficiency_neg_infinite_count": 1,
            },
            "falsification": {"error_detection_rate": 0.95},
            "formulate": {"structural_match_rate": 0.95},
            "feedback-loop": {"integration_coverage_rate": 0.95},
        },
    })
    r = fork_score(parent, child)
    assert r.verdict == "REJECT"
    assert any("NEG_INF_REGRESSION" in f for f in r.flags)


# --- non-bench input terms ------------------------------------------


def test_fork_score_capital_cost_penalty_subtracts():
    """capital_delta > 0 reduces the score (cost penalty)."""
    parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    child = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    no_penalty = fork_score(parent, child, capital_delta=0.0)
    with_penalty = fork_score(parent, child, capital_delta=10.0)
    # capital_delta term is -theta * capital_delta — so positive
    # capital_delta should REDUCE the score
    assert with_penalty.score < no_penalty.score


def test_fork_score_audit_completeness_adds():
    """audit_completeness > 0 increases the score (positive term)."""
    parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    child = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    base = fork_score(parent, child)
    bonus = fork_score(parent, child, audit_completeness=1.0)
    assert bonus.score > base.score


def test_fork_score_security_risk_subtracts_when_iota_positive():
    """security_risk reduces the score when the iota weight is
    non-zero. Default WeightSet sets iota=0 (security risk is opt-
    in for the validator), so we test with an explicit iota>0."""
    parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    child = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    iota_on = WeightSet(
        alpha=DEFAULT_WEIGHTS.alpha, beta=DEFAULT_WEIGHTS.beta,
        gamma=DEFAULT_WEIGHTS.gamma, delta=DEFAULT_WEIGHTS.delta,
        epsilon=DEFAULT_WEIGHTS.epsilon, zeta=DEFAULT_WEIGHTS.zeta,
        eta=DEFAULT_WEIGHTS.eta, theta=DEFAULT_WEIGHTS.theta,
        iota=1.0,                                       # turn on
        kappa=DEFAULT_WEIGHTS.kappa, lambda_=DEFAULT_WEIGHTS.lambda_,
    )
    base = fork_score(parent, child, weights=iota_on)
    risky = fork_score(parent, child, weights=iota_on, security_risk=1.0)
    assert risky.score < base.score


def test_fork_score_security_risk_inert_under_default_weights():
    """Honest documentation of the default behavior: with iota=0
    (the default), passing security_risk does NOT change the score.
    This is by design — operators must opt-in to the security-risk
    penalty by setting iota>0."""
    parent = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    child = _bench_dict({"5s": {}, "5t": {}, "5f": {}})
    assert DEFAULT_WEIGHTS.iota == 0.0, (
        "test assumes iota=0 default; update if WeightSet defaults change"
    )
    base = fork_score(parent, child)
    with_risk = fork_score(parent, child, security_risk=1.0)
    assert with_risk.score == pytest.approx(base.score, abs=1e-12)


# --- weights customization -----------------------------------------


def test_fork_score_weights_recorded_in_output():
    """Custom WeightSet flows through to the output's weights dict."""
    custom = WeightSet(alpha=0.5, beta=0.5, gamma=0.5,
                      delta=0.1, epsilon=0.1, zeta=0.1,
                      eta=1.0, theta=1.0, iota=1.0,
                      kappa=1.0, lambda_=1.0)
    r = fork_score({}, {}, weights=custom)
    assert r.weights["alpha"] == pytest.approx(0.5)
    assert r.weights["lambda_"] == pytest.approx(1.0)


def test_fork_score_default_weights_used_when_unspecified():
    r = fork_score({}, {})
    assert r.weights == DEFAULT_WEIGHTS.as_dict()


# --- ScoredFork API -------------------------------------------------


def test_scored_fork_to_dict_serializable():
    """ScoredFork.to_dict() returns JSON-serializable types."""
    r = fork_score({}, {})
    d = r.to_dict()
    import json

    json.dumps(d)  # must not raise
    assert "score" in d
    assert "verdict" in d
    assert "breakdown" in d
    assert "flags" in d
    assert "weights" in d


# --- score = sum(breakdown) closure ---------------------------------


def test_score_equals_sum_of_breakdown():
    """Closure check: score ≡ Σ breakdown values. Multiple
    parent/child configurations to widen the cone."""
    configs = [
        # Plain improvement on 5S only.
        (
            {"5s": {"syntax": {"parse_pass_rate": 0.5}}},
            {"5s": {"syntax": {"parse_pass_rate": 0.7}}},
        ),
        # Mixed deltas across all three batteries.
        (
            _bench_dict({
                "5s": {"syntax": {"parse_pass_rate": 0.6}},
                "5t": {"transfer-learning": {"transfer_learning_success_rate": 0.6}},
                "5f": {"function": {"function_pass_rate": 0.6}},
            }),
            _bench_dict({
                "5s": {"syntax": {"parse_pass_rate": 0.7}},
                "5t": {"transfer-learning": {"transfer_learning_success_rate": 0.65}},
                "5f": {"function": {"function_pass_rate": 0.55}},
            }),
        ),
        # All-zero inputs → score 0.
        ({"5s": {}, "5t": {}, "5f": {}}, {"5s": {}, "5t": {}, "5f": {}}),
    ]
    for parent, child in configs:
        r = fork_score(parent, child)
        assert r.score == pytest.approx(sum(r.breakdown.values()), abs=1e-9), (
            f"score {r.score} ≠ Σbreakdown {sum(r.breakdown.values())} on "
            f"parent={parent!r}, child={child!r}"
        )


# --- constants honored from arborist.substrate.fork_score -----------


def test_signal_floor_honored():
    """Score exactly at SIGNAL_FLOOR is ACCEPT (≥, not >)."""
    # We construct a child whose 5S delta * alpha == SIGNAL_FLOOR.
    # alpha defaults to 0.30 per WeightSet (see arborist.substrate.weights).
    # Solve for delta: delta = SIGNAL_FLOOR / alpha = 0.05 / 0.30 ≈ 0.1667.
    # Average of 5 sub-deltas; set each sub to 0.1667 to hit it.
    parent = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.5},
            "semantics": {"equivalence_recovery_rate": 0.5},
            "syllogism": {"step_validity_rate": 0.5},
            "synthesis": {"derivation_pass_rate": 0.5},
            "semiotics": {"invariance_under_swap": 0.5},
        },
    })
    target_delta = SIGNAL_FLOOR / DEFAULT_WEIGHTS.alpha
    child = _bench_dict({
        "5s": {
            "syntax": {"parse_pass_rate": 0.5 + target_delta},
            "semantics": {"equivalence_recovery_rate": 0.5 + target_delta},
            "syllogism": {"step_validity_rate": 0.5 + target_delta},
            "synthesis": {"derivation_pass_rate": 0.5 + target_delta},
            "semiotics": {"invariance_under_swap": 0.5 + target_delta},
        },
    })
    r = fork_score(parent, child)
    # Score = alpha · target_delta = SIGNAL_FLOOR exactly. Verdict ACCEPT.
    assert r.score == pytest.approx(SIGNAL_FLOOR, abs=1e-9)
    assert r.verdict == "ACCEPT"


# ---------------------------------------------------------------------
# #000047 — delta_aggregator (mean / max / sum)
# ---------------------------------------------------------------------


def test_delta_aggregator_default_is_mean():
    assert DEFAULT_WEIGHTS.delta_aggregator == "mean"


def test_aggregate_helper():
    from arborist.substrate.fork_score import _aggregate

    assert _aggregate([], "mean") == 0.0
    assert _aggregate([], "max") == 0.0
    assert _aggregate([], "sum") == 0.0
    assert _aggregate([0.1, 0.2, 0.3, 0.0, 0.0], "mean") == pytest.approx(0.12)
    assert _aggregate([0.6, 0.0, 0.0, 0.0, 0.0], "max") == pytest.approx(0.6)
    assert _aggregate([-0.1, -0.2, 0.0], "max") == 0.0  # floored at 0
    assert _aggregate([0.05] * 5, "sum") == pytest.approx(0.25)
    with pytest.raises(ValueError, match="unknown delta aggregator"):
        _aggregate([0.1], "bogus")


def test_weightset_rejects_bad_aggregator():
    with pytest.raises(ValueError, match="delta_aggregator must be one of"):
        WeightSet(delta_aggregator="median")


def test_weights_from_dict_aggregator():
    from arborist.substrate.weights import from_dict

    assert from_dict({}).delta_aggregator == "mean"
    assert from_dict({"delta_aggregator": "sum"}).delta_aggregator == "sum"
    assert from_dict({"alpha": 2.0, "delta_aggregator": "max"}).delta_aggregator == "max"
    assert from_dict({"alpha": 2.0, "delta_aggregator": "max"}).alpha == 2.0


def test_fork_score_aggregator_changes_5f_term_for_single_sub_gain():
    """A child that lifts ONE 5f sub by +0.6 (the rest flat): mean
    dilutes it to 0.12, max/sum weigh it at face value 0.6."""
    parent = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.4}}})
    child = _bench_dict({"5f": {"falsification": {"error_detection_rate": 1.0}}})
    r_mean = fork_score(parent, child)  # default mean
    r_max = fork_score(parent, child, weights=WeightSet(delta_aggregator="max"))
    r_sum = fork_score(parent, child, weights=WeightSet(delta_aggregator="sum"))
    assert r_mean.breakdown["gamma_x_delta_5f"] == pytest.approx(0.6 / 5)
    assert r_max.breakdown["gamma_x_delta_5f"] == pytest.approx(0.6)
    assert r_sum.breakdown["gamma_x_delta_5f"] == pytest.approx(0.6)


def test_fork_score_sum_vs_max_diverge_for_broad_gain():
    """Two 5f subs lifted by +0.6 each: mean 0.24, max 0.6, sum 1.2 —
    three distinct values, so sum ≠ max once the improvement is broad."""
    parent = _bench_dict({"5f": {
        "falsification": {"error_detection_rate": 0.4},
        "formulate": {"structural_match_rate": 0.4},
    }})
    child = _bench_dict({"5f": {
        "falsification": {"error_detection_rate": 1.0},
        "formulate": {"structural_match_rate": 1.0},
    }})
    assert fork_score(parent, child).breakdown["gamma_x_delta_5f"] == pytest.approx(1.2 / 5)
    assert fork_score(parent, child, weights=WeightSet(delta_aggregator="max")).breakdown["gamma_x_delta_5f"] == pytest.approx(0.6)
    assert fork_score(parent, child, weights=WeightSet(delta_aggregator="sum")).breakdown["gamma_x_delta_5f"] == pytest.approx(1.2)


def test_fork_score_records_aggregator_in_weights():
    parent = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.4}}})
    child = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.5}}})
    assert fork_score(parent, child).weights["delta_aggregator"] == "mean"
    r = fork_score(parent, child, weights=WeightSet(delta_aggregator="max"))
    assert r.weights["delta_aggregator"] == "max"


def test_hard_regression_flag_independent_of_aggregator():
    """A single-sub regression below HARD_REGRESSION_FLOOR forces
    REJECT under every aggregator — the per-sub flag is computed before
    aggregation."""
    parent = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.5}}})
    child = _bench_dict({"5f": {"falsification": {"error_detection_rate": 0.4}}})  # -0.1
    for agg in ("mean", "max", "sum"):
        r = fork_score(parent, child, weights=WeightSet(delta_aggregator=agg))
        assert r.verdict == "REJECT", agg
        assert any(f.startswith("REGRESSION_5F:") for f in r.flags), agg


# ---------------------------------------------------------------------
# Phase 1c — branch-set persistence (#000012 §7 Phase 1c)
# ---------------------------------------------------------------------


def _scored():
    """Build a representative ScoredFork via the real fork_score()."""
    parent = _bench_dict({
        "5s": {"syntax": {"parse_pass_rate": 0.5}},
    })
    child = _bench_dict({
        "5s": {"syntax": {"parse_pass_rate": 0.7}},
    })
    return fork_score(parent, child)


def test_phase1c_migration_creates_table(tmp_path):
    """Opening a connection runs the Phase 1c migration; the table
    + both indexes exist."""
    from arborist.store import connect, invalidate_migration_cache

    db = tmp_path / "shard.db"
    invalidate_migration_cache(db)
    conn = connect(db)
    try:
        row = conn.execute(
            "SELECT name FROM sqlite_master "
            "WHERE type='table' AND name='fork_score_branches'"
        ).fetchone()
        assert row is not None
        idx_names = {
            r[0]
            for r in conn.execute(
                "SELECT name FROM sqlite_master WHERE type='index' "
                "AND tbl_name='fork_score_branches'"
            ).fetchall()
        }
        assert "idx_fork_score_branches_set" in idx_names
        assert "idx_fork_score_branches_parent" in idx_names
    finally:
        conn.close()
        invalidate_migration_cache(db)


def test_phase1c_persist_branch_score_writes_one_row(tmp_path):
    from arborist.store import connect, invalidate_migration_cache, transaction
    from arborist.substrate.fork_score import (
        ESTIMATOR_VERSION,
        persist_branch_score,
    )

    db = tmp_path / "shard.db"
    invalidate_migration_cache(db)
    conn = connect(db)
    try:
        with transaction(conn):
            persist_branch_score(
                conn,
                branch_set_id="cp-1",
                branch_id="b-A",
                parent_root="parent-root-aaaa",
                child_root="child-root-A",
                scored=_scored(),
                weights_id="default",
            )
        rows = conn.execute(
            "SELECT branch_set_id, branch_id, parent_root, child_root, "
            "  verdict, weights_id, estimator_version "
            "FROM fork_score_branches"
        ).fetchall()
        assert len(rows) == 1
        r = rows[0]
        assert r["branch_set_id"] == "cp-1"
        assert r["branch_id"] == "b-A"
        assert r["parent_root"] == "parent-root-aaaa"
        assert r["child_root"] == "child-root-A"
        assert r["verdict"] in ("ACCEPT", "MARGINAL", "REJECT")
        assert r["estimator_version"] == ESTIMATOR_VERSION
    finally:
        conn.close()
        invalidate_migration_cache(db)


def test_phase1c_persist_upserts_on_pk(tmp_path):
    """Re-scoring the same (branch_set_id, branch_id) is an upsert,
    not a duplicate row. Fields refresh."""
    from arborist.store import connect, invalidate_migration_cache, transaction
    from arborist.substrate.fork_score import persist_branch_score

    db = tmp_path / "shard.db"
    invalidate_migration_cache(db)
    conn = connect(db)
    try:
        s1 = _scored()
        with transaction(conn):
            persist_branch_score(
                conn,
                branch_set_id="cp-up",
                branch_id="b-up",
                parent_root="p1",
                child_root="c1",
                scored=s1,
                weights_id="w1",
                ts=1700000000,
            )
        with transaction(conn):
            persist_branch_score(
                conn,
                branch_set_id="cp-up",
                branch_id="b-up",
                parent_root="p1",
                child_root="c2-new",
                scored=s1,
                weights_id="w2-new",
                ts=1700000999,
            )
        rows = conn.execute(
            "SELECT child_root, weights_id, recorded_at "
            "FROM fork_score_branches WHERE branch_set_id = 'cp-up'"
        ).fetchall()
        assert len(rows) == 1
        assert rows[0]["child_root"] == "c2-new"
        assert rows[0]["weights_id"] == "w2-new"
        assert rows[0]["recorded_at"] == 1700000999
    finally:
        conn.close()
        invalidate_migration_cache(db)


def test_phase1c_branch_set_density_counts_branches(tmp_path):
    """branch_set_density returns 0 / N for the queried checkpoint;
    rows under other checkpoints don't leak."""
    from arborist.store import connect, invalidate_migration_cache, transaction
    from arborist.substrate.fork_score import (
        branch_set_density,
        persist_branch_score,
    )

    db = tmp_path / "shard.db"
    invalidate_migration_cache(db)
    conn = connect(db)
    try:
        assert branch_set_density(conn, "missing") == 0
        s = _scored()
        with transaction(conn):
            for i in range(4):
                persist_branch_score(
                    conn,
                    branch_set_id="cp-A",
                    branch_id=f"branch-{i}",
                    parent_root="parent-root",
                    child_root=f"child-root-{i}",
                    scored=s,
                )
            persist_branch_score(
                conn,
                branch_set_id="cp-B",
                branch_id="lone",
                parent_root="parent-root",
                child_root="child-root-Z",
                scored=s,
            )
        # #000037 §12 Trigger 1 satisfied: ≥4 branches at "cp-A".
        assert branch_set_density(conn, "cp-A") == 4
        assert branch_set_density(conn, "cp-B") == 1
        assert branch_set_density(conn, "cp-missing") == 0
    finally:
        conn.close()
        invalidate_migration_cache(db)


def test_phase1c_breakdown_blob_round_trips_as_json(tmp_path):
    """breakdown_blob stores the per-term breakdown losslessly so a
    downstream reader can replay the verdict."""
    import json as _json
    from arborist.store import connect, invalidate_migration_cache, transaction
    from arborist.substrate.fork_score import persist_branch_score

    db = tmp_path / "shard.db"
    invalidate_migration_cache(db)
    conn = connect(db)
    try:
        scored = _scored()
        with transaction(conn):
            persist_branch_score(
                conn,
                branch_set_id="cp-blob",
                branch_id="b-blob",
                parent_root="p",
                child_root="c",
                scored=scored,
            )
        row = conn.execute(
            "SELECT breakdown_blob FROM fork_score_branches WHERE "
            "branch_set_id='cp-blob'"
        ).fetchone()
        recovered = _json.loads(row["breakdown_blob"])
        assert sum(recovered.values()) == pytest.approx(
            scored.score, abs=1e-9
        )
        assert set(recovered.keys()) == set(scored.breakdown.keys())
    finally:
        conn.close()
        invalidate_migration_cache(db)