trigger_1_branch_density in bench/prometheus_sigma_trigger_probe.py read the fork_score_branches table via branch_set_density() instead of the "density check not yet implemented" stub. It groups rows by branch_set_id, fires when the most-recently-recorded checkpoint carries >= 4 branches (BRANCH_DENSITY_FLOOR), and surfaces n_checkpoints / latest_density / max_density / n_checkpoints_clearing_floor in the markdown report so section 12's "regularly" qualifier stays visible. Density sums across shards per branch_set_id. With no branch sets persisted yet the probe reports "table present but empty across shards" (data_available True, fires False) rather than a false negative. Re-ran the probe against the live shards: bench/results/prometheus-sigma-triggers-2026-05-11.md. 6 new tests in tests/test_prometheus_trigger_probe.py (probe loaded via importlib): no-table to no-data, empty-table to data-available-no-fire, latest-checkpoint->=4 to fires, earlier-dense-but-latest-sparse to no-fire, density-sums-across-shards, report-renders-density-lines. Updated #000012 Phase 1c landing receipt, #000037 section 12 Trigger 1 note, and the TICKETS.md index rows for both. Pure measurement: no mutation, no LLM call, no schema change.
This commit is contained in:
parent
da62f8047c
commit
dbe824944a
6 changed files with 388 additions and 42 deletions
181
tests/test_prometheus_trigger_probe.py
Normal file
181
tests/test_prometheus_trigger_probe.py
Normal file
|
|
@ -0,0 +1,181 @@
|
|||
"""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
|
||||
Loading…
Add table
Add a link
Reference in a new issue