arborist/tests/test_prometheus_trigger_probe.py

"""Probe-side tests for ``bench/prometheus_sigma_trigger_probe.py``
trigger 1 (branch density), wired to the #000012 Phase 1c
``fork_score_branches`` table.

Trigger 1 was a stub until #000012 Phase 1c landed the
``fork_score_branches`` sibling table and the ``branch_set_density``
helper. These tests pin the wired behaviour: no-table → no data;
table-present-but-empty → data available, does not fire; latest
checkpoint with ≥4 branches → fires.

The probe lives outside the ``arborist`` package, so it is loaded via
``importlib`` (same pattern as ``tests/test_bench_qa_sweep.py``).
"""
from __future__ import annotations

import importlib.util
import sys
from pathlib import Path

import pytest

from arborist.store import connect, invalidate_migration_cache, transaction
from arborist.substrate.fork_score import ScoredFork, persist_branch_score


@pytest.fixture(scope="module")
def probe():
    path = (
        Path(__file__).parent.parent
        / "bench"
        / "prometheus_sigma_trigger_probe.py"
    )
    spec = importlib.util.spec_from_file_location(
        "prometheus_trigger_probe_under_test", path
    )
    mod = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = mod
    spec.loader.exec_module(mod)
    return mod


def _scored(score: float = 0.5, verdict: str = "ACCEPT") -> ScoredFork:
    return ScoredFork(
        score=score, verdict=verdict, breakdown={"t": score}, flags=[], weights={}
    )


def _make_shard(tmp_path: Path, name: str) -> Path:
    """Create a shard DB with the Phase 1c migration applied."""
    db = tmp_path / name
    invalidate_migration_cache(db)
    conn = connect(db)  # runs _migrate_fork_score_branches
    conn.close()
    invalidate_migration_cache(db)
    return db


def _persist(db: Path, *, branch_set_id: str, branch_id: str, ts: int) -> None:
    invalidate_migration_cache(db)
    conn = connect(db)
    try:
        with transaction(conn):
            persist_branch_score(
                conn,
                branch_set_id=branch_set_id,
                branch_id=branch_id,
                parent_root="parent-root",
                child_root=f"child-{branch_id}",
                scored=_scored(),
                ts=ts,
            )
    finally:
        conn.close()
        invalidate_migration_cache(db)


def test_trigger1_no_table_reports_no_data(probe, tmp_path):
    """A shard with no fork_score_branches table → data unavailable,
    does not fire, points at #000012 Phase 1c."""
    import sqlite3

    db = tmp_path / "bare.db"
    sqlite3.connect(db).close()  # empty DB, no migration
    out = probe.trigger_1_branch_density([db])
    assert out["trigger"] == 1
    assert out["fires"] is False
    assert out["data_available"] is False
    assert out["n_checkpoints"] == 0
    assert "#000012" in out["reason"]


def test_trigger1_table_empty_reports_data_available_no_fire(probe, tmp_path):
    """Table present (migration ran) but no branch sets persisted →
    data available, does not fire."""
    db = _make_shard(tmp_path, "empty.db")
    out = probe.trigger_1_branch_density([db])
    assert out["fires"] is False
    assert out["data_available"] is True
    assert out["n_checkpoints"] == 0
    assert "empty" in out["reason"]


def test_trigger1_fires_when_latest_checkpoint_has_floor_branches(probe, tmp_path):
    """Latest checkpoint (by recorded_at) carries ≥4 branches → fires;
    reason names the checkpoint and the floor count."""
    db = _make_shard(tmp_path, "dense.db")
    # Older checkpoint with only 1 branch — should NOT be the one the
    # trigger reads.
    _persist(db, branch_set_id="cp-old", branch_id="solo", ts=1_700_000_000)
    # Newer checkpoint with 4 branches — this is the latest.
    for i in range(4):
        _persist(
            db, branch_set_id="cp-new", branch_id=f"b{i}", ts=1_700_001_000 + i
        )
    out = probe.trigger_1_branch_density([db])
    assert out["fires"] is True
    assert out["data_available"] is True
    assert out["n_checkpoints"] == 2
    assert out["latest_branch_set_id"] == "cp-new"
    assert out["latest_density"] == 4
    assert out["max_density"] == 4
    assert out["n_checkpoints_clearing_floor"] == 1
    assert "cp-new" in out["reason"]


def test_trigger1_no_fire_when_latest_checkpoint_below_floor(probe, tmp_path):
    """An earlier checkpoint clearing the floor does NOT make the
    trigger fire if the *latest* checkpoint is below it — §12 reads
    the most recent checkpoint."""
    db = _make_shard(tmp_path, "mixed.db")
    for i in range(5):  # old, dense checkpoint
        _persist(
            db, branch_set_id="cp-dense", branch_id=f"d{i}", ts=1_700_000_000 + i
        )
    for i in range(2):  # newest checkpoint, sparse
        _persist(
            db, branch_set_id="cp-sparse", branch_id=f"s{i}", ts=1_700_009_000 + i
        )
    out = probe.trigger_1_branch_density([db])
    assert out["fires"] is False
    assert out["data_available"] is True
    assert out["latest_branch_set_id"] == "cp-sparse"
    assert out["latest_density"] == 2
    assert out["max_density"] == 5
    assert out["n_checkpoints_clearing_floor"] == 1
    assert "max density seen: 5" in out["reason"]


def test_trigger1_density_sums_across_shards(probe, tmp_path):
    """A checkpoint split across two shards sums to its full density."""
    db_a = _make_shard(tmp_path, "a.db")
    db_b = _make_shard(tmp_path, "b.db")
    _persist(db_a, branch_set_id="cp-split", branch_id="b0", ts=1_700_000_000)
    _persist(db_a, branch_set_id="cp-split", branch_id="b1", ts=1_700_000_001)
    _persist(db_b, branch_set_id="cp-split", branch_id="b2", ts=1_700_000_002)
    _persist(db_b, branch_set_id="cp-split", branch_id="b3", ts=1_700_000_003)
    out = probe.trigger_1_branch_density([db_a, db_b])
    assert out["n_checkpoints"] == 1
    assert out["latest_density"] == 4
    assert out["fires"] is True


def test_trigger1_renders_density_lines_in_report(probe, tmp_path):
    """render_markdown surfaces the density numbers when a checkpoint
    exists."""
    db = _make_shard(tmp_path, "render.db")
    for i in range(4):
        _persist(
            db, branch_set_id="cp-r", branch_id=f"b{i}", ts=1_700_000_000 + i
        )
    triggers = [
        probe.trigger_1_branch_density([db]),
        probe.trigger_2_divergence_variance([]),
        probe.trigger_3_witness_cost_share([]),
        probe.trigger_4_operator_need(),
    ]
    md = probe.render_markdown(tmp_path, [db], triggers, [], [])
    assert "## Trigger 1 — branch density" in md
    assert "Latest checkpoint: `cp-r`" in md
    assert "Recorded checkpoints: 1" in md
    assert "Checkpoints clearing the floor: 1 / 1" in md