arborist/bench/prometheus_sigma_trigger_probe.py

"""Prometheus-Σ §12 trigger-probe — ticket #000037 Phase 0 evidence-gather.

Phase 1 of #000037 (the recursive falsification controller) gates on
one of four §12 triggers firing. Triggers 1-3 are measurable from
already-committed state; trigger 4 is operator decision. This script
walks every shard's ``audit_events`` and ``capital_ledger`` tables,
computes the §12 quantities, and emits a markdown report.

Pure measurement. No mutation. No LLM call. No schema change. The
report is the deliverable; fox uses it to decide go/no-go on Phase 1.

The probe is defensive about data-availability: it explicitly reports
"no data" when a trigger structurally cannot fire yet (e.g. trigger 1
on a single-validator deployment with no fork-score cache table)
rather than synthesising a false signal.
"""

from __future__ import annotations

import argparse
import json
import math
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path

# §12 thresholds, transcribed verbatim from the ticket.
DIVERGENCE_N_MIN = 30
DIVERGENCE_VARIANCE_RATIO_FLOOR = 0.5
DIVERGENCE_ABS_STDDEV_FLOOR = 0.10
WITNESS_COST_SHARE_FLOOR = 0.30
DIVERGENCE_EPSILON = 1e-6  # max(mean, ε) guard against div-by-zero


def _read_witness_events(shards: list[Path]) -> list[dict]:
    """Pull every ``providence_canonical_witness`` body across shards."""
    bodies: list[dict] = []
    for sp in shards:
        try:
            conn = sqlite3.connect(sp)
        except sqlite3.OperationalError:
            continue
        try:
            cur = conn.execute(
                "SELECT body, ts FROM audit_events "
                "WHERE event_type = 'providence_canonical_witness' "
                "ORDER BY seq"
            )
            for row in cur:
                try:
                    body = json.loads(row[0])
                except (TypeError, json.JSONDecodeError):
                    continue
                body["__ts"] = row[1]
                body["__shard"] = sp.name
                bodies.append(body)
        except sqlite3.OperationalError:
            # Pre-#000028 shard with no audit_events table or schema gap;
            # not a probe-failure mode.
            pass
        finally:
            conn.close()
    return bodies


def _read_capital_ledger(shards: list[Path]) -> list[dict]:
    """Aggregate every ``capital_ledger`` row across shards."""
    rows: list[dict] = []
    for sp in shards:
        try:
            conn = sqlite3.connect(sp)
            conn.row_factory = sqlite3.Row
        except sqlite3.OperationalError:
            continue
        try:
            cur = conn.execute(
                "SELECT op_type, living, material, financial, "
                "       intellectual, experiential, social, cultural, "
                "       spiritual, recorded_at "
                "FROM capital_ledger"
            )
            for r in cur:
                d = dict(r)
                d["__shard"] = sp.name
                rows.append(d)
        except sqlite3.OperationalError:
            pass
        finally:
            conn.close()
    return rows


def _has_fork_score_table(shards: list[Path]) -> bool:
    for sp in shards:
        try:
            conn = sqlite3.connect(sp)
            row = conn.execute(
                "SELECT name FROM sqlite_master "
                "WHERE type='table' AND name LIKE '%fork_score%' LIMIT 1"
            ).fetchone()
            conn.close()
            if row:
                return True
        except sqlite3.OperationalError:
            continue
    return False


# -------------------------------------------------------------- triggers


def trigger_1_branch_density(shards: list[Path]) -> dict:
    """Trigger 1 — branch density: ≥4 candidate branches per checkpoint.

    Requires multi-branch ForkScore (#000012) to be persisting candidate
    sets. Phase 1a (closed 2026-05-08) is single-validator, so no
    branch-set table exists; the trigger structurally cannot fire yet.
    """
    has_table = _has_fork_score_table(shards)
    return {
        "trigger": 1,
        "name": "branch density (≥4 branches/checkpoint)",
        "fires": False,
        "reason": (
            "no fork_score branch-set table found across shards; "
            "ForkScore Phase 1a is single-validator. Multi-branch "
            "fork-score persistence is the prerequisite — see #000012."
        )
        if not has_table
        else "branch-set table found but density check not yet implemented",
        "data_available": has_table,
    }


def _divergence_score(agreement_label: str | None) -> float | None:
    """Map a #000028 §1.2 agreement label to a binary LLM-divergence
    score for trigger 2.

    Returns:
        - 0.0 when the LLM converged with the kernel
        - 1.0 when the LLM diverged from the kernel
        - None when the LLM modality was absent / unparseable, so the
          event tells us nothing about LLM divergence.

    The N_min=30 sample population is the set of events that produced
    a 0/1 score. KERNEL-CACHE-AGREE excludes LLM (unparseable) so it
    contributes no signal to trigger 2; CACHE-DRIFT means K=L (LLM
    agreed with kernel) so it counts as 0.0.
    """
    if not agreement_label:
        return None
    label = agreement_label.upper()
    LLM_AGREE = {
        "STRICT-WITNESSED",   # K=C=L
        "KERNEL-LLM-AGREE",   # no cache; K=L
        "CACHE-DRIFT",        # K=L; cache stale (counts as agree for LLM)
    }
    LLM_DIVERGED = {
        "LLM-DIVERGED",            # K=C; L≠K
        "KERNEL-LLM-DIVERGED",     # no cache; L≠K
        "LLM-AND-CACHE-DIVERGED",  # only K matches itself
    }
    LLM_ABSENT = {
        "KERNEL-CACHE-AGREE",      # LLM unparseable; no signal
    }
    if label in LLM_AGREE:
        return 0.0
    if label in LLM_DIVERGED:
        return 1.0
    if label in LLM_ABSENT:
        return None
    # Unknown label — substring fallback so renames don't silently
    # exclude data. Errs on the side of inclusion.
    if "DIVERGED" in label or "DRIFT_LLM" in label:
        return 1.0
    if "AGREE" in label or "WITNESSED" in label:
        return 0.0
    return None


def _stddev(samples: list[float]) -> float:
    n = len(samples)
    if n < 2:
        return 0.0
    mean = sum(samples) / n
    sq = sum((s - mean) ** 2 for s in samples)
    return math.sqrt(sq / (n - 1))


def trigger_2_divergence_variance(events: list[dict]) -> dict:
    """Trigger 2 — divergence variance: N≥30, ratio>0.5 or abs>0.10.

    Concrete signal from §12:
        sample_count >= N_min
        AND (stddev/max(mean, ε) > 0.5  OR  absolute stddev > 0.10)
    """
    scores = [
        s for s in (_divergence_score(e.get("agreement_label")) for e in events)
        if s is not None
    ]
    n = len(scores)
    if n == 0:
        return {
            "trigger": 2,
            "name": "divergence variance",
            "fires": False,
            "reason": (
                "no providence_canonical_witness events with kernel-vs-LLM "
                "outcomes found; #000028 not exercised on this corpus yet"
            ),
            "data_available": False,
            "n": 0,
        }
    mean = sum(scores) / n
    sd = _stddev(scores)
    ratio = sd / max(mean, DIVERGENCE_EPSILON)
    abs_fired = sd > DIVERGENCE_ABS_STDDEV_FLOOR
    ratio_fired = ratio > DIVERGENCE_VARIANCE_RATIO_FLOOR
    n_ok = n >= DIVERGENCE_N_MIN
    fires = n_ok and (abs_fired or ratio_fired)
    if not n_ok:
        reason = f"only {n} samples; N_min = {DIVERGENCE_N_MIN}"
    elif fires:
        which = []
        if ratio_fired:
            which.append(f"ratio {ratio:.3f} > {DIVERGENCE_VARIANCE_RATIO_FLOOR}")
        if abs_fired:
            which.append(f"abs σ {sd:.3f} > {DIVERGENCE_ABS_STDDEV_FLOOR}")
        reason = "; ".join(which)
    else:
        reason = (
            f"N={n} ≥ {DIVERGENCE_N_MIN}, but "
            f"ratio {ratio:.3f} ≤ {DIVERGENCE_VARIANCE_RATIO_FLOOR} "
            f"and abs σ {sd:.3f} ≤ {DIVERGENCE_ABS_STDDEV_FLOOR}"
        )
    return {
        "trigger": 2,
        "name": "divergence variance",
        "fires": fires,
        "reason": reason,
        "data_available": True,
        "n": n,
        "mean": round(mean, 4),
        "stddev": round(sd, 4),
        "ratio": round(ratio, 4),
    }


def trigger_3_witness_cost_share(rows: list[dict]) -> dict:
    """Trigger 3 — witness cost share: witness_compute / total_compute > 0.30.

    "Compute" is read as the ``material`` column (kWh proxy), which is
    the closest analogue to "compute" in the 8-form ledger. The ticket
    leaves the form vague; surfacing both ``material`` and ``financial``
    in the report so fox can choose if the threshold should be against
    a different form.

    Per §12 the trigger requires #000026 baseline to exist (it does;
    bench/results/real-shard-baseline.md, 2026-05-08). The ledger
    here is whatever has been recorded to date — not a controlled
    workload — so the ratio is informational pending a controlled
    sweep.
    """
    if not rows:
        return {
            "trigger": 3,
            "name": "witness cost share (material > 0.30)",
            "fires": False,
            "reason": "capital_ledger empty across all shards",
            "data_available": False,
            "witness_material": 0.0,
            "total_material": 0.0,
        }
    witness_material = sum(
        r["material"] or 0.0 for r in rows if r["op_type"] == "canonical_witness"
    )
    total_material = sum(r["material"] or 0.0 for r in rows)
    witness_financial = sum(
        r["financial"] or 0.0 for r in rows if r["op_type"] == "canonical_witness"
    )
    total_financial = sum(r["financial"] or 0.0 for r in rows)
    if total_material == 0.0:
        return {
            "trigger": 3,
            "name": "witness cost share (material > 0.30)",
            "fires": False,
            "reason": "total compute denominator is zero (per §12: do not divide)",
            "data_available": False,
            "witness_material": witness_material,
            "total_material": 0.0,
        }
    ratio = witness_material / total_material
    fires = ratio > WITNESS_COST_SHARE_FLOOR
    return {
        "trigger": 3,
        "name": "witness cost share (material > 0.30)",
        "fires": fires,
        "reason": (
            f"witness material / total material = "
            f"{witness_material:.6f} / {total_material:.6f} = {ratio:.3f}"
        ),
        "data_available": True,
        "witness_material": round(witness_material, 6),
        "total_material": round(total_material, 6),
        "witness_financial": round(witness_financial, 6),
        "total_financial": round(total_financial, 6),
        "ratio": round(ratio, 4),
        "n_witness_rows": sum(
            1 for r in rows if r["op_type"] == "canonical_witness"
        ),
        "n_total_rows": len(rows),
        "caveat": (
            "ledger reflects ad-hoc activity, not a controlled #000026 sweep — "
            "ratio is informational. Re-run after a #000026 baseline sweep "
            "for a workload-anchored answer."
        ),
    }


def trigger_4_operator_need() -> dict:
    return {
        "trigger": 4,
        "name": "operator mission need",
        "fires": None,
        "reason": "operator decision; not measurable from committed state",
        "data_available": False,
    }


# -------------------------------------------------------------- report


def render_markdown(
    shards_dir: Path,
    shards: list[Path],
    triggers: list[dict],
    events: list[dict],
    rows: list[dict],
) -> str:
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    out = []
    out.append("# Prometheus-Σ §12 trigger probe")
    out.append("")
    out.append(f"**Date:** {now}")
    out.append(f"**Shards directory:** `{shards_dir}`")
    out.append(f"**Shards walked:** {len(shards)} "
               f"({', '.join(s.name for s in shards)})")
    out.append("")
    out.append(
        "Phase 1 of ticket #000037 gates on one §12 trigger firing. "
        "This probe walks `audit_events` + `capital_ledger` and reports "
        "the empirical state of triggers 1–3. Trigger 4 is operator-stated."
    )
    out.append("")
    out.append("## Verdict")
    out.append("")
    out.append("| # | Trigger | Fires? | Reason |")
    out.append("|---|---------|--------|--------|")
    for t in triggers:
        fires = t["fires"]
        if fires is True:
            mark = "**YES**"
        elif fires is False:
            mark = "no"
        else:
            mark = "n/a"
        out.append(f"| {t['trigger']} | {t['name']} | {mark} | {t['reason']} |")
    out.append("")
    any_fires = any(t["fires"] is True for t in triggers)
    if any_fires:
        out.append("**Phase 1 trigger has fired.** Prometheus-Σ Phase 1 may proceed.")
    else:
        out.append(
            "**No measurable §12 trigger has fired yet.** Phase 1 of #000037 "
            "remains paper-only; trigger 4 (operator-stated need) bypasses the "
            "measured-pressure gates if fox has explicit mission need."
        )
    out.append("")
    out.append("## Trigger 1 — branch density")
    out.append("")
    t1 = triggers[0]
    out.append(f"- Data available: {t1['data_available']}")
    out.append(f"- {t1['reason']}")
    out.append("")
    out.append("## Trigger 2 — divergence variance")
    out.append("")
    t2 = triggers[1]
    if t2.get("data_available"):
        out.append(f"- Sample count (N): {t2['n']} "
                   f"(N_min = {DIVERGENCE_N_MIN})")
        out.append(f"- Mean divergence rate: {t2['mean']}")
        out.append(f"- Stddev: {t2['stddev']}")
        out.append(f"- Ratio (σ/mean): {t2['ratio']}")
        out.append(f"- Threshold ratio: > {DIVERGENCE_VARIANCE_RATIO_FLOOR} "
                   f"OR absolute σ > {DIVERGENCE_ABS_STDDEV_FLOOR}")
    out.append(f"- {t2['reason']}")
    if events:
        from collections import Counter
        agree_counts = Counter(e.get("agreement_label", "?") for e in events)
        out.append("")
        out.append("Agreement-label distribution across all shards:")
        out.append("")
        out.append("| label | count |")
        out.append("|-------|-------|")
        for k, v in sorted(agree_counts.items(), key=lambda kv: (-kv[1], kv[0])):
            out.append(f"| `{k}` | {v} |")
    out.append("")
    out.append("## Trigger 3 — witness cost share")
    out.append("")
    t3 = triggers[2]
    if t3.get("data_available"):
        out.append(f"- Witness material (kWh proxy): "
                   f"{t3['witness_material']}")
        out.append(f"- Total material: {t3['total_material']}")
        out.append(f"- Ratio: {t3['ratio']} "
                   f"(threshold > {WITNESS_COST_SHARE_FLOOR})")
        out.append(f"- Witness financial: {t3['witness_financial']}")
        out.append(f"- Total financial: {t3['total_financial']}")
        out.append(f"- Witness rows: {t3['n_witness_rows']}")
        out.append(f"- Total ledger rows: {t3['n_total_rows']}")
    out.append(f"- {t3['reason']}")
    if t3.get("caveat"):
        out.append("")
        out.append(f"**Caveat:** {t3['caveat']}")
    out.append("")
    out.append("## Trigger 4 — operator mission need")
    out.append("")
    out.append("- Operator-stated; not measurable from committed state.")
    out.append("- Bypasses §12 measured-pressure gates per ticket §12.")
    out.append("")
    return "\n".join(out)


# -------------------------------------------------------------- main


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument(
        "--shards-dir",
        type=Path,
        default=Path.home() / ".arborist" / "shards",
        help="Directory of *.db shard files (default: ~/.arborist/shards)",
    )
    p.add_argument(
        "--out",
        type=Path,
        default=None,
        help="Write markdown report to this path. Default: stdout.",
    )
    args = p.parse_args(argv)

    if not args.shards_dir.exists():
        print(f"shards-dir not found: {args.shards_dir}", file=sys.stderr)
        return 2

    shards = sorted(args.shards_dir.glob("*.db"))
    if not shards:
        print(f"no .db files in {args.shards_dir}", file=sys.stderr)
        return 2

    events = _read_witness_events(shards)
    rows = _read_capital_ledger(shards)

    triggers = [
        trigger_1_branch_density(shards),
        trigger_2_divergence_variance(events),
        trigger_3_witness_cost_share(rows),
        trigger_4_operator_need(),
    ]

    md = render_markdown(args.shards_dir, shards, triggers, events, rows)
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(md, encoding="utf-8")
        print(f"wrote {args.out}")
    else:
        print(md)
    return 0


if __name__ == "__main__":
    sys.exit(main())