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