arborist/tests/test_prometheus.py

"""Phase 1 tests for arborist.substrate.prometheus (ticket #000037).

Covers the §16.2 test list — 17 named cases — plus a small set of
sanity checks (determinism, weight-profile shape, trigger-check
positive path) that don't appear in the §16.2 list but harden the
"pure function" contract.

Discipline (§16.1):

- Pure-function — no DB, no LLM, no scheduler.
- Banned imports verified via AST inspection of the module source.
- Proposals only; no mutation API calls.
"""

from __future__ import annotations

import ast
import math
import random
from pathlib import Path

import pytest

from arborist.substrate.prometheus import (
    ABS_STDDEV_ZERO_MEAN,
    H_HIGH,
    H_LOW,
    KAPPA_MEMORY,
    N_MIN_TRIGGER_2,
    WEIGHT_PROFILES,
    BatteryDeltas,
    ControllerBranch,
    ControllerDecision,
    ControllerInput,
    ControllerWeights,
    MemoryRootUpdateProposal,
    SelfModelUpdateProposal,
    conservative_weights,
    controller_decide,
    exploratory_weights,
    phase_1_trigger_check,
    safe_weights,
)


# ---------------------------------------------------------------------
# Test helpers
# ---------------------------------------------------------------------


def _b(
    branch_id: str = "b1",
    *,
    delta_5s: float = 0.05,
    delta_5t: float = 0.05,
    delta_5f: float = 0.05,
    delta_5r: float = 0.0,
    witness_divergence: float = 0.0,
    capital_cost: float = 0.0,
    regression_penalty: float = 0.0,
    security_risk: float = 0.0,
    memory_invalidation: float = 0.0,
    selfmodel_calibration_gain: float = 0.0,
    warrant_promotion_gain: float = 0.0,
    hard_vetoes: tuple[str, ...] = (),
    payoff_b: float = 1.0,
) -> ControllerBranch:
    return ControllerBranch(
        branch_id=branch_id,
        deltas=BatteryDeltas(delta_5s, delta_5t, delta_5f, delta_5r),
        witness_divergence=witness_divergence,
        capital_cost=capital_cost,
        regression_penalty=regression_penalty,
        security_risk=security_risk,
        memory_invalidation=memory_invalidation,
        selfmodel_calibration_gain=selfmodel_calibration_gain,
        warrant_promotion_gain=warrant_promotion_gain,
        hard_vetoes=hard_vetoes,
        payoff_b=payoff_b,
    )


def _inp(
    branches: tuple[ControllerBranch, ...],
    *,
    organism_root: str = "Ω0",
    budget: int = 4,
    hermes_utilization: int = 0,
    weights: ControllerWeights | None = None,
    difficulty: float = 1.0,
    divergence_rate: float = 0.0,
    hard_regression_rate: float = 0.0,
    witness_agreement_rate: float = 1.0,
    capital_pressure: float = 0.0,
) -> ControllerInput:
    return ControllerInput(
        organism_root=organism_root,
        branches=branches,
        budget=budget,
        hermes_utilization=hermes_utilization,
        weights=weights or safe_weights(),
        difficulty=difficulty,
        divergence_rate=divergence_rate,
        hard_regression_rate=hard_regression_rate,
        witness_agreement_rate=witness_agreement_rate,
        capital_pressure=capital_pressure,
    )


# ---------------------------------------------------------------------
# §16.2 #1 — test_all_vetoed_returns_reject_or_quarantine
# ---------------------------------------------------------------------


def test_all_vetoed_returns_reject_or_quarantine():
    """Every branch vetoed → label is REJECT or QUARANTINE (or
    ESCALATE for the most severe class). selected_branch_id None,
    notes mention ALL_BRANCHES_VETOED."""
    branches = (
        _b("a", hard_vetoes=("hard_regression",)),  # REJECT class
        _b("b", hard_vetoes=("verifier_failure",)),  # REJECT class
    )
    d = controller_decide(_inp(branches))
    assert d.label == "REJECT"
    assert d.selected_branch_id is None
    assert "ALL_BRANCHES_VETOED" in d.notes

    # Mixed REJECT + QUARANTINE → QUARANTINE wins (higher severity).
    branches = (
        _b("a", hard_vetoes=("hard_regression",)),
        _b("b", hard_vetoes=("cache_drift",)),
    )
    d = controller_decide(_inp(branches))
    assert d.label == "QUARANTINE"

    # Any ESCALATE-class veto → ESCALATE wins (highest severity).
    branches = (
        _b("a", hard_vetoes=("hard_regression",)),
        _b("b", hard_vetoes=("memory_invalidation_exceeded",)),
    )
    d = controller_decide(_inp(branches))
    assert d.label == "ESCALATE"


# ---------------------------------------------------------------------
# §16.2 #2 — test_zero_budget_returns_deferred
# ---------------------------------------------------------------------


def test_zero_budget_returns_deferred():
    """B = 0 (Hermes saturated) → DEFERRED for everything, regardless
    of how good the branches look."""
    branches = (_b("a", delta_5s=0.2), _b("b", delta_5s=0.3))
    d = controller_decide(_inp(branches, budget=0))
    assert d.label == "DEFERRED"
    assert d.selected_branch_id is None
    assert all(v == 0.0 for v in d.allocations.values())
    assert "ZERO_BUDGET" in d.notes


# ---------------------------------------------------------------------
# §16.2 #3 — test_stable_softmax_no_overflow
# ---------------------------------------------------------------------


def test_stable_softmax_no_overflow():
    """Utilities ±10000 must not overflow exp() — the stable softmax
    max-shift bounds every scaled-exp in [0, 1]."""
    # Both very large positive and very large negative inputs.
    branches = (
        _b("hot", delta_5s=10000.0, delta_5t=10000.0, delta_5f=10000.0),
        _b("cold", delta_5s=-10000.0, delta_5t=-10000.0, delta_5f=-10000.0),
        _b("middle", delta_5s=0.1),
    )
    d = controller_decide(_inp(branches))
    # No NaN / inf escapes.
    assert math.isfinite(d.entropy)
    for v in d.allocations.values():
        assert math.isfinite(v)
    # Hot branch gets virtually all the budget.
    assert d.allocations["hot"] > d.allocations["cold"]
    assert d.allocations["hot"] > d.allocations["middle"]


# ---------------------------------------------------------------------
# §16.2 #4 — test_negative_payoff_gets_zero_allocation
# ---------------------------------------------------------------------


def test_negative_payoff_gets_zero_allocation():
    """Branch with payoff_b <= 0 — Kelly's "free regression" guard.
    Allocation zero; if it's the only branch the label REJECTs."""
    branches = (_b("a", delta_5s=0.1, payoff_b=-0.5),)
    d = controller_decide(_inp(branches))
    # Σ_raw == 0 → DEFERRED (the spec-defined exit when no positive
    # allocation survives). Both DEFERRED and REJECT are valid
    # "negative-payoff produces no work" outcomes — assert no positive
    # allocation either way.
    assert d.allocations["a"] == 0.0
    assert d.label in {"DEFERRED", "REJECT"}

    # When another branch has positive payoff, the negative one stays
    # at zero and the positive one selects.
    branches = (
        _b("neg", delta_5s=0.1, payoff_b=-0.5),
        _b("pos", delta_5s=0.1, payoff_b=2.0),
    )
    d = controller_decide(_inp(branches))
    assert d.allocations["neg"] == 0.0
    assert d.allocations["pos"] > 0.0
    assert d.selected_branch_id == "pos"


# ---------------------------------------------------------------------
# §16.2 #5 — test_unsupported_carrier_quarantines
# ---------------------------------------------------------------------


def test_unsupported_carrier_quarantines():
    """`unsupported_carrier` veto on the only branch → QUARANTINE."""
    branches = (_b("a", hard_vetoes=("unsupported_carrier",)),)
    d = controller_decide(_inp(branches))
    assert d.label == "QUARANTINE"
    assert "ALL_BRANCHES_VETOED" in d.notes
    assert d.veto_reasons["a"] == ("unsupported_carrier",)


# ---------------------------------------------------------------------
# §16.2 #6 — test_cache_drift_quarantines
# ---------------------------------------------------------------------


def test_cache_drift_quarantines():
    """`cache_drift` veto on the only branch → QUARANTINE."""
    branches = (_b("a", hard_vetoes=("cache_drift",)),)
    d = controller_decide(_inp(branches))
    assert d.label == "QUARANTINE"


# ---------------------------------------------------------------------
# §16.2 #7 — test_memory_invalidation_above_threshold_escalates
# ---------------------------------------------------------------------


def test_memory_invalidation_above_threshold_escalates():
    """Selected branch with memory_invalidation >= KAPPA_MEMORY →
    ESCALATE."""
    branches = (
        _b(
            "a",
            delta_5s=0.5,
            delta_5t=0.5,
            delta_5f=0.5,
            memory_invalidation=KAPPA_MEMORY,
        ),
    )
    d = controller_decide(_inp(branches))
    assert d.label == "ESCALATE"
    assert d.selected_branch_id == "a"
    assert "MEMORY_INVALIDATION_ABOVE_KAPPA" in d.notes

    # Just below threshold → no escalation.
    branches = (
        _b(
            "a",
            delta_5s=0.5,
            delta_5t=0.5,
            delta_5f=0.5,
            memory_invalidation=KAPPA_MEMORY - 0.01,
        ),
    )
    d = controller_decide(_inp(branches))
    assert d.label != "ESCALATE"


# ---------------------------------------------------------------------
# §16.2 #8 — test_high_entropy_increases_difficulty
# ---------------------------------------------------------------------


def test_high_entropy_increases_difficulty():
    """Multiple equally-attractive branches → high entropy →
    difficulty_next > difficulty (§7.1 H_norm term is positive).

    Note: under default `witness_agreement_rate=1.0` the agreement
    pull exactly cancels the entropy drive (both weight 0.1).
    Setting agreement_rate=0 isolates the entropy contribution."""
    # Three branches, identical utilities → uniform distribution →
    # H_norm = 1.0.
    branches = tuple(
        _b(f"b{i}", delta_5s=0.05, delta_5t=0.05, delta_5f=0.05)
        for i in range(3)
    )
    initial_difficulty = 1.0
    d = controller_decide(
        _inp(
            branches,
            difficulty=initial_difficulty,
            witness_agreement_rate=0.0,
        )
    )
    assert d.entropy >= H_HIGH, (
        f"three-equal-branch H_norm should be near 1.0, got {d.entropy}"
    )
    assert d.difficulty_next > initial_difficulty


# ---------------------------------------------------------------------
# §16.2 #9 — test_low_entropy_decreases_or_preserves_difficulty
# ---------------------------------------------------------------------


def test_low_entropy_decreases_or_preserves_difficulty():
    """Single overwhelmingly-good branch → low entropy → with
    `witness_agreement_rate=1.0` and zero divergence/regression
    pressure, difficulty_next <= difficulty."""
    branches = (
        _b("dominant", delta_5s=10.0, delta_5t=10.0, delta_5f=10.0),
        _b("weak", delta_5s=-5.0, delta_5t=-5.0, delta_5f=-5.0),
    )
    initial_difficulty = 5.0
    d = controller_decide(
        _inp(
            branches,
            difficulty=initial_difficulty,
            witness_agreement_rate=1.0,
            divergence_rate=0.0,
            hard_regression_rate=0.0,
        )
    )
    # Entropy near zero (dominant branch has p ≈ 1).
    assert d.entropy <= H_LOW
    # With agreement pull > entropy drive in this regime, difficulty
    # should not climb.
    assert d.difficulty_next <= initial_difficulty


# ---------------------------------------------------------------------
# §16.2 #10 — test_divergence_increases_witness_sampling_recommendation
# ---------------------------------------------------------------------


def test_divergence_increases_witness_sampling_recommendation():
    """High `divergence_rate` → higher difficulty_next (the controller's
    homeostat: divergence is the thermostat; more difficulty means
    wider witness sampling in the Phase 3 sweep)."""
    branches = (_b("a", delta_5s=0.1, delta_5t=0.1, delta_5f=0.1),)
    initial = 1.0

    low_div = controller_decide(
        _inp(branches, difficulty=initial, divergence_rate=0.0)
    )
    high_div = controller_decide(
        _inp(branches, difficulty=initial, divergence_rate=0.8)
    )
    assert high_div.difficulty_next > low_div.difficulty_next


# ---------------------------------------------------------------------
# §16.2 #11 — test_no_llm_call_in_controller
# ---------------------------------------------------------------------


def _module_imports() -> set[str]:
    """Parse the prometheus.py source and return the set of imported
    module names. Catches `import x`, `import x.y`, `from x.y import …`."""
    here = Path(__file__).resolve().parent
    src = (here.parent / "arborist" / "substrate" / "prometheus.py").read_text(
        encoding="utf-8"
    )
    tree = ast.parse(src)
    names: set[str] = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                names.add(alias.name)
        elif isinstance(node, ast.ImportFrom):
            if node.module:
                names.add(node.module)
    return names


def test_no_llm_call_in_controller():
    """Pure function — must not import the LLM gateway."""
    imports = _module_imports()
    assert "arborist.qa.client" not in imports, (
        "controller must not import the LLM client (Phase 1 is pure)"
    )


# ---------------------------------------------------------------------
# §16.2 #12 — test_controller_does_not_modify_cache_key_inputs
# ---------------------------------------------------------------------


def test_controller_does_not_modify_cache_key_inputs():
    """Phase 1 controller must not touch the cache_key path. AST
    inspection confirms `arborist.qa.keys` is not imported."""
    imports = _module_imports()
    assert "arborist.qa.keys" not in imports, (
        "controller must not import the cache_key path (Phase 1 is "
        "advisory-only over already-committed state)"
    )


# ---------------------------------------------------------------------
# §16.2 #13 — test_controller_outputs_advisory_event_only
# ---------------------------------------------------------------------


def test_controller_outputs_advisory_event_only():
    """An ACCEPT-bound input produces `advisory_events` as a tuple of
    (str, dict) pairs. Phase 2 will persist these; Phase 1 only
    builds the data structure."""
    branches = (_b("a", delta_5s=0.5, delta_5t=0.5, delta_5f=0.5),)
    d = controller_decide(_inp(branches))
    assert d.label == "ACCEPT"
    assert isinstance(d.advisory_events, tuple)
    assert len(d.advisory_events) >= 1
    for entry in d.advisory_events:
        assert isinstance(entry, tuple) and len(entry) == 2
        kind, body = entry
        assert isinstance(kind, str)
        assert isinstance(body, dict)
    # No mutation API was called — verified by `memory_proposals` /
    # `selfmodel_proposals` being plain tuples of frozen dataclasses
    # (no side-effects possible).
    assert isinstance(d.memory_proposals, tuple)
    assert isinstance(d.selfmodel_proposals, tuple)


# ---------------------------------------------------------------------
# §16.2 #14 — test_zero_mean_divergence_does_not_trigger_phase_1
# ---------------------------------------------------------------------


def test_zero_mean_divergence_does_not_trigger_phase_1():
    """Mean divergence = 0 + zero stddev → no signal → no trigger.
    Absolute-stddev fallback path used when mean is zero but variance
    is present."""
    # All-zero samples → mean = 0, stddev = 0 → no_variance_signal.
    triggered, reason = phase_1_trigger_check([0.0] * 40)
    assert triggered is False
    assert reason == "no_variance_signal"

    # Mean = 0 but nonzero stddev below the absolute threshold.
    samples = [0.0, 0.001, -0.001, 0.0] * 10  # n=40, mean=0, tiny stddev
    triggered, reason = phase_1_trigger_check(samples)
    assert triggered is False
    assert reason == "absolute_stddev_path"

    # Mean = 0 with nonzero stddev above the absolute threshold → fires.
    # Balance positive and negative so the mean stays at 0.
    samples = [0.2, -0.2] * 20  # n=40, mean=0, stddev=0.2 >= 0.05.
    triggered, reason = phase_1_trigger_check(samples)
    assert triggered is True
    assert reason == "absolute_stddev_path"


# ---------------------------------------------------------------------
# §16.2 #15 — test_small_sample_does_not_trigger_phase_1
# ---------------------------------------------------------------------


def test_small_sample_does_not_trigger_phase_1():
    """sample_count < N_min (= 30) → no trigger regardless of variance.
    Tiny samples are noise."""
    # 29 samples — high variance, below the floor.
    rng = random.Random(0)
    samples = [rng.uniform(0.0, 1.0) for _ in range(N_MIN_TRIGGER_2 - 1)]
    triggered, reason = phase_1_trigger_check(samples)
    assert triggered is False
    assert reason == "small_sample"

    # Empty sample — also small.
    triggered, reason = phase_1_trigger_check([])
    assert triggered is False
    assert reason == "small_sample"


# ---------------------------------------------------------------------
# §16.2 #16 — test_propose_not_mutate_memory_root
# ---------------------------------------------------------------------


def test_propose_not_mutate_memory_root():
    """A vetoed branch with regression_penalty > 0 yields a
    MemoryRootUpdateProposal record. Type is the proposal dataclass
    (no mutation API)."""
    branches = (
        _b(
            "regressed",
            hard_vetoes=("hard_regression",),
            regression_penalty=0.4,
        ),
        _b("live", delta_5s=0.1),
    )
    d = controller_decide(_inp(branches))
    assert len(d.memory_proposals) == 1
    prop = d.memory_proposals[0]
    assert isinstance(prop, MemoryRootUpdateProposal)
    assert prop.branch_id == "regressed"
    assert "BRANCH_REGRESSED" in prop.proposed_invalidation_tags
    # Frozen dataclass — no mutation surface.
    with pytest.raises((AttributeError, TypeError)):
        prop.proposed_invalidation_tags = ()  # type: ignore[misc]


# ---------------------------------------------------------------------
# §16.2 #17 — test_propose_not_mutate_self_model
# ---------------------------------------------------------------------


def test_propose_not_mutate_self_model():
    """A selected branch with selfmodel_calibration_gain != 0 yields a
    SelfModelUpdateProposal record. Type is the proposal dataclass
    (no SelfModel mutation API call)."""
    branches = (
        _b(
            "calibrating",
            delta_5s=0.5,
            delta_5t=0.5,
            delta_5f=0.5,
            selfmodel_calibration_gain=0.15,
        ),
    )
    d = controller_decide(_inp(branches))
    assert d.selected_branch_id == "calibrating"
    assert len(d.selfmodel_proposals) == 1
    prop = d.selfmodel_proposals[0]
    assert isinstance(prop, SelfModelUpdateProposal)
    assert prop.calibration_delta == 0.15
    assert prop.reason == "controller_phase_1_calibration"
    with pytest.raises((AttributeError, TypeError)):
        prop.calibration_delta = 0.0  # type: ignore[misc]


# ---------------------------------------------------------------------
# Additional sanity checks (not in §16.2 — harden the pure-function
# contract)
# ---------------------------------------------------------------------


def test_idempotent_byte_identical_output():
    """Pure function: two invocations with identical inputs produce
    byte-identical ControllerDecision."""
    branches = (
        _b("a", delta_5s=0.1, delta_5t=0.2, delta_5f=0.3),
        _b("b", delta_5s=-0.1, delta_5t=0.0, delta_5f=0.2, capital_cost=1.0),
    )
    inp = _inp(branches)
    d1 = controller_decide(inp)
    d2 = controller_decide(inp)
    assert d1 == d2


def test_weight_profiles_registry_complete():
    """All four named profiles present and structurally distinct."""
    assert set(WEIGHT_PROFILES) == {
        "safe",
        "conservative",
        "exploratory",
        "sweep",
    }
    safe = WEIGHT_PROFILES["safe"]
    cons = WEIGHT_PROFILES["conservative"]
    expl = WEIGHT_PROFILES["exploratory"]
    swp = WEIGHT_PROFILES["sweep"]
    # §15.2 deltas
    assert cons.gamma_5f == 1.0
    assert cons.lambda_capital_cost == 1.5
    assert cons.xi_security_risk == 4.0
    assert cons.eta_softmax_temperature == 0.75
    # §15.3 deltas
    assert expl.gamma_5f == 1.5
    assert expl.nu_witness_divergence == 0.75
    assert expl.lambda_capital_cost == 0.5
    assert expl.eta_softmax_temperature == 1.5
    # Sweep profile deltas (Phase 1.c — dry-run Finding #3)
    assert swp.gamma_5f == 1.5
    assert swp.lambda_capital_cost == 0.25
    assert swp.nu_witness_divergence == 0.5
    # Safe baseline
    assert safe.gamma_5f == 1.25


def test_phase_1_trigger_check_positive_path():
    """Coefficient-of-variation path fires when the sample has clear
    nonzero mean + comparable stddev."""
    rng = random.Random(1)
    # Mean ≈ 0.3, stddev ≈ 0.2 → CoV ≈ 0.67 >= 0.5 → fires.
    samples = [rng.uniform(0.0, 0.6) for _ in range(60)]
    triggered, reason = phase_1_trigger_check(samples)
    # Mean and stddev depend on the seeded RNG, but the seed is
    # fixed → result is deterministic. Sanity-check the reason set.
    assert reason in {
        "variance_threshold",
        "variance_below_threshold",
        "absolute_stddev_path",
    }
    if triggered:
        assert reason in {"variance_threshold", "absolute_stddev_path"}


def test_empty_branches_returns_unknown():
    """No candidate branches → UNKNOWN with NO_CANDIDATE_BRANCHES note
    (§14 row 1)."""
    d = controller_decide(_inp(branches=()))
    assert d.label == "UNKNOWN"
    assert d.selected_branch_id is None
    assert "NO_CANDIDATE_BRANCHES" in d.notes


def test_decision_is_a_controllerdecision():
    """Output type sanity."""
    branches = (_b("a", delta_5s=0.1),)
    d = controller_decide(_inp(branches))
    assert isinstance(d, ControllerDecision)


# ---------------------------------------------------------------------
# §14 row 4 — Hermes-saturation guard
# ---------------------------------------------------------------------


def test_hermes_saturation_returns_deferred():
    """When hermes_utilization >= budget, no headroom remains; the
    controller defers regardless of branch utility (§14 row 4)."""
    branches = (_b("good", delta_5s=0.5, delta_5f=0.5),)
    inp = _inp(branches, budget=4, hermes_utilization=4)
    d = controller_decide(inp)
    assert d.label == "DEFERRED"
    assert "HERMES_SATURATED" in d.notes
    # Allocations stay all-zero — no compute spent under saturation.
    assert all(v == 0.0 for v in d.allocations.values())
    # Advisory event records the saturation context for Phase 2 audit.
    kinds = [k for (k, _) in d.advisory_events]
    assert "controller_decision" in kinds
    body = next(b for (k, b) in d.advisory_events if k == "controller_decision")
    assert body["reason"] == "HERMES_SATURATED"
    assert body["hermes_utilization"] == 4
    assert body["budget"] == 4


def test_hermes_saturation_above_budget_also_defers():
    """utilization > budget (overflow case) still defers — the guard
    is ``utilization >= budget``, not just equality."""
    branches = (_b("a"),)
    d = controller_decide(_inp(branches, budget=4, hermes_utilization=8))
    assert d.label == "DEFERRED"
    assert "HERMES_SATURATED" in d.notes


def test_hermes_saturation_does_not_fire_when_headroom_exists():
    """utilization < budget → controller proceeds normally."""
    branches = (_b("good", delta_5f=0.5, payoff_b=10.0),)
    inp = _inp(branches, budget=4, hermes_utilization=1)
    d = controller_decide(inp)
    assert d.label != "DEFERRED" or "HERMES_SATURATED" not in d.notes


# ---------------------------------------------------------------------
# §6 + §14 — additional veto-class tests (replay/soft-hash) +
# defensive NUMERIC_INSTABILITY path
# ---------------------------------------------------------------------


def test_replay_window_unbounded_escalates():
    """§14 row + §6: `replay_window_unbounded` is an ESCALATE-class
    veto (T3 still open per #000036)."""
    branches = (_b("a", hard_vetoes=("replay_window_unbounded",)),)
    d = controller_decide(_inp(branches))
    assert d.label == "ESCALATE"
    assert "ALL_BRANCHES_VETOED" in d.notes


def test_soft_hash_signal_quarantines():
    """§14 row 10 + §6: `soft_hash_signal` above threshold is a
    QUARANTINE-class veto."""
    branches = (_b("a", hard_vetoes=("soft_hash_signal",)),)
    d = controller_decide(_inp(branches))
    assert d.label == "QUARANTINE"


def test_escalate_dominates_quarantine_dominates_reject():
    """When a chunk has mixed veto classes the cascade returns the
    most-severe class present (§6 fail-loud ordering)."""
    branches = (
        _b("escalate", hard_vetoes=("replay_window_unbounded",)),
        _b("quarantine", hard_vetoes=("cache_drift",)),
        _b("reject", hard_vetoes=("verifier_failure",)),
    )
    d = controller_decide(_inp(branches))
    assert d.label == "ESCALATE"


def test_quarantine_dominates_reject():
    branches = (
        _b("quarantine", hard_vetoes=("schema_mismatch",)),
        _b("reject", hard_vetoes=("hard_regression",)),
    )
    d = controller_decide(_inp(branches))
    assert d.label == "QUARANTINE"


# ---------------------------------------------------------------------
# §13 Step 11 — falsification-fixture proposal emission
# ---------------------------------------------------------------------


def test_high_divergence_emits_falsification_proposal():
    """A branch with witness_divergence ≥ threshold gets a
    FalsificationFixtureProposal record — the §3 'divergence →
    candidate fixture' funnel for 5F mining (#000025)."""
    branches = (
        _b("a", delta_5f=0.1, payoff_b=5.0),
        _b("diverged", witness_divergence=0.8, delta_5f=0.05),
    )
    d = controller_decide(_inp(branches))
    proposals = d.falsification_proposals
    assert any(p.branch_id == "diverged" for p in proposals)
    p = next(p for p in proposals if p.branch_id == "diverged")
    assert p.witness_divergence == 0.8
    assert p.reason == "WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD"
    assert p.organism_root == "Ω0"


def test_low_divergence_does_not_emit_proposal():
    """A branch under threshold doesn't produce a fixture proposal."""
    branches = (_b("a", witness_divergence=0.1, delta_5f=0.1, payoff_b=5.0),)
    d = controller_decide(_inp(branches))
    assert d.falsification_proposals == ()


def test_falsification_threshold_is_weight_tunable():
    """ControllerWeights.falsification_divergence_threshold gates the
    proposal — lowering it surfaces more proposals."""
    w_sensitive = ControllerWeights(
        falsification_divergence_threshold=0.05
    )
    branches = (_b("a", witness_divergence=0.10, delta_5f=0.1, payoff_b=5.0),)
    d = controller_decide(_inp(branches, weights=w_sensitive))
    assert len(d.falsification_proposals) == 1
    assert d.falsification_proposals[0].branch_id == "a"


def test_falsification_proposal_emitted_even_on_vetoed_branch():
    """Divergence-as-falsification-signal is independent of the
    accept/reject decision — even a vetoed branch can produce a
    fixture proposal when its divergence is high."""
    branches = (
        _b("vetoed", hard_vetoes=("cache_drift",), witness_divergence=0.9),
    )
    d = controller_decide(_inp(branches))
    # All-vetoed cascade still emits the proposal — Phase 1 surfaces
    # the divergence signal at every level.
    # (Implementation choice: proposals emit from ALL branches, not
    # just selected. Vetoed-AND-diverged is a 1-fixture deliverable.)
    assert any(p.branch_id == "vetoed" for p in d.falsification_proposals)


# ---------------------------------------------------------------------
# §14 + §5 — entropy gating becomes weight-tunable (h_low / h_high)
# ---------------------------------------------------------------------


def test_h_low_h_high_are_weight_tunable():
    """Moving entropy thresholds to ControllerWeights lets deployments
    override the §5 default (0.3 / 0.7) without patching the module."""
    # Push h_high above 1.0 so even max-entropy chunks ACCEPT.
    w = ControllerWeights(h_low=0.5, h_high=1.1)
    branches = (
        _b("a", delta_5f=0.10, payoff_b=10.0),
        _b("b", delta_5f=0.09, payoff_b=10.0),
    )
    d = controller_decide(_inp(branches, weights=w))
    # With h_high=1.1 the high-entropy branch never trips MARGINAL.
    assert d.label != "MARGINAL"


def test_kappa_memory_is_weight_tunable():
    """ControllerWeights.kappa_memory governs the escalate-threshold.

    Utility math (default safe weights):
        U = γ·δ5F − ω·memory_invalidation
          = 1.25·0.5 − 2.0·0.15
          = 0.625 − 0.3 = +0.325 > 0
    So we clear the NON_POSITIVE_UTILITY guard, and reach the kappa
    branch with `memory_invalidation=0.15 >= kappa_memory=0.1` →
    ESCALATE.
    """
    w = ControllerWeights(kappa_memory=0.1)
    branches = (
        _b("a", delta_5f=0.5, memory_invalidation=0.15, payoff_b=10.0),
    )
    d = controller_decide(_inp(branches, weights=w))
    assert d.label == "ESCALATE"
    assert "MEMORY_INVALIDATION_ABOVE_KAPPA" in d.notes


# ---------------------------------------------------------------------
# Module-level constants stay back-compat exports
# ---------------------------------------------------------------------


def test_module_level_constants_match_default_weights():
    """``H_LOW``, ``H_HIGH``, ``KAPPA_MEMORY`` are kept at module level
    for back-compat with callers that referenced them pre-v2; the
    weight-field defaults must match exactly so behavior is identical
    when neither is overridden."""
    w = ControllerWeights()
    assert w.h_low == H_LOW
    assert w.h_high == H_HIGH
    assert w.kappa_memory == KAPPA_MEMORY


# ---------------------------------------------------------------------
# Phase 1.c — kernel_cost / llm_cost split on ControllerBranch
# ---------------------------------------------------------------------


def test_effective_cost_uses_split_when_provided():
    """When ``kernel_cost`` or ``llm_cost`` > 0, ``effective_cost``
    returns their sum and IGNORES the legacy ``capital_cost`` field.
    This is the Phase 1.c migration path."""
    b = _b("a", capital_cost=1.0)
    b_split = ControllerBranch(
        branch_id="b",
        deltas=BatteryDeltas(0, 0, 0, 0),
        witness_divergence=0.0,
        capital_cost=999.0,  # SHOULD BE IGNORED
        kernel_cost=0.05,
        llm_cost=0.3,
    )
    # Legacy branch uses capital_cost directly:
    assert b.effective_cost == 1.0
    # Split branch sums kernel + llm, ignores legacy:
    assert b_split.effective_cost == 0.35


def test_effective_cost_falls_back_when_split_zero():
    """When both new fields are 0, ``effective_cost`` reads
    ``capital_cost`` — preserving pre-Phase-1.c behavior."""
    b = ControllerBranch(
        branch_id="legacy",
        deltas=BatteryDeltas(0, 0, 0, 0),
        witness_divergence=0.0,
        capital_cost=0.42,
        # kernel_cost + llm_cost both default to 0.0
    )
    assert b.effective_cost == 0.42


def test_utility_uses_effective_cost():
    """The utility formula reads effective_cost so a split-cost
    branch and a legacy-cost branch with the same TOTAL cost produce
    byte-identical utility values."""
    legacy = ControllerBranch(
        branch_id="legacy",
        deltas=BatteryDeltas(0.1, 0.1, 0.1, 0.1),
        witness_divergence=0.0,
        capital_cost=0.5,
    )
    split = ControllerBranch(
        branch_id="split",
        deltas=BatteryDeltas(0.1, 0.1, 0.1, 0.1),
        witness_divergence=0.0,
        capital_cost=0.0,
        kernel_cost=0.1,
        llm_cost=0.4,
    )
    w = safe_weights()
    from arborist.substrate.prometheus import _utility
    assert _utility(legacy, w) == _utility(split, w)


def test_kernel_only_branch_cheaper_than_llm_branch():
    """Two branches with identical signal but different cost class
    (kernel-only vs LLM-witness) produce different utilities,
    favoring the kernel-only path."""
    kernel_only = ControllerBranch(
        branch_id="kernel",
        deltas=BatteryDeltas(0, 0, 0.05, 0),
        witness_divergence=0.0,
        kernel_cost=0.05,
        llm_cost=0.0,
        payoff_b=10.0,
    )
    llm_only = ControllerBranch(
        branch_id="llm",
        deltas=BatteryDeltas(0, 0, 0.05, 0),
        witness_divergence=0.0,
        kernel_cost=0.0,
        llm_cost=1.0,
        payoff_b=10.0,
    )
    from arborist.substrate.prometheus import _utility
    w = safe_weights()
    assert _utility(kernel_only, w) > _utility(llm_only, w)


def test_split_cost_branches_route_through_controller():
    """End-to-end: a split-cost branch can be fed to controller_decide
    and the decision uses effective_cost in the utility math."""
    branches = (
        ControllerBranch(
            branch_id="kernel",
            deltas=BatteryDeltas(0, 0, 0.10, 0),
            witness_divergence=0.0,
            kernel_cost=0.05,
            llm_cost=0.0,
            payoff_b=10.0,
        ),
        ControllerBranch(
            branch_id="llm",
            deltas=BatteryDeltas(0, 0, 0.05, 0),
            witness_divergence=0.0,
            kernel_cost=0.0,
            llm_cost=1.0,
            payoff_b=1.0,
        ),
    )
    d = controller_decide(_inp(branches))
    # The kernel branch is unambiguously the winner under safe
    # weights: higher delta_5f, lower cost.
    assert d.selected_branch_id == "kernel"


def test_sweep_profile_accepts_what_safe_profile_defers():
    """The Phase-1.c sweep profile drops lambda_capital_cost from
    1.0 → 0.25 so the sweep is willing to pay capital for
    falsification discovery. A marginal-cost branch that DEFERREDs
    under safe weights ACCEPTs under sweep weights."""
    from arborist.substrate.prometheus import sweep_weights

    # Build a branch whose utility under safe weights is barely
    # negative (DEFERRED via Kelly's p_i > 0.5 unmet); under sweep
    # weights its utility flips positive.
    #   safe:  U = 1.25 * 0.10 - 1.0 * 0.30 = -0.175 (< 0)
    #   sweep: U = 1.50 * 0.10 - 0.25 * 0.30 = +0.075 (> 0)
    branches = (
        ControllerBranch(
            branch_id="marginal",
            deltas=BatteryDeltas(0, 0, 0.10, 0),
            witness_divergence=0.0,
            capital_cost=0.30,
            payoff_b=10.0,
        ),
    )
    safe_d = controller_decide(_inp(branches, weights=safe_weights()))
    sweep_d = controller_decide(_inp(branches, weights=sweep_weights()))
    # Safe-profile decision rejects the marginal branch.
    assert safe_d.label == "REJECT"
    # Sweep-profile decision accepts it.
    assert sweep_d.label in ("ACCEPT", "MARGINAL")