arborist/bench/scripts/harvest_falsification_proposals.py

"""Harvest the §13-Step-11 falsification-fixture proposals from a live
shard into a deterministic 5F fixture pack.

Closes the Phase 1 → 5F battery loop (#000037 Phase 1 produces
:class:`FalsificationFixtureProposal` records; #000025 Phase 1e named
the documented motif set). The dry-run simulator
(``bench/scripts/prometheus_sigma_sweep_dryrun.py``) showed 576
high-divergence rows in ``~/.arborist/shards/qa.db`` at τ_qa=1d.
This script picks a representative deterministic sub-sample of those
rows and writes embedded fixture entries to a NEW pack at
``bench/fixtures/5f/falsification-harvested-v1.jsonl``.

The harvest is intentionally separate from
``falsification-v1.jsonl`` so hand-curated motif fixtures and
corpus-derived regression fixtures live in distinct packs with
distinct attribution. The harness test asserts the harvested pack
runs clean independently.

Sample policy (`SAMPLE_PER_BUCKET = 20`):

- HYBRID rows with ``witness_divergence ≥ 0.5``: top 20 by
  ``cache_key`` (lex-sorted for determinism).
- UNGROUNDED rows with ``witness_divergence ≥ 0.5``: top 20 by
  ``cache_key``.
- CONTROLLER_PROPOSAL — rows pulled from the live
  ``controller_events`` stream (event_kind =
  ``controller_falsification_proposal``, persisted by Phase 2 audit
  writes). Today's QA-runner advisory writes a single-branch
  decision per QA cycle whose proposals overlap providence_cache;
  Phase 3 sweep emissions will grow this bucket independently.
  Top 20 by branch_id, joined back to providence_cache by 16-char
  cache_key prefix for fixture enrichment. Empty bucket (zero
  rows) is the default state until live QA accumulates proposals.

The first two buckets produce 40 corpus-derived fixtures by default;
the third grows the pack as the controller_events stream fills.
Idempotency on re-harvest is the caller's job (the runner
truncates+rewrites by default).

Mapping cache row → fixture:

- ``id``: ``5f-fal-harvested-<first-16-of-cache_key>``
- ``answer_text``: preserved verbatim from the row (helps human
  readability when debugging regressions).
- ``observed_violations``: derived from audit_mode + unverified-
  quote count.
- ``expected_reason``: the audit_mode itself (UNGROUNDED /
  HYBRID_<method>).
- ``verifier_method_root``: from the row's ``verifier_method``
  column (falls back to ``"warrant-v1"`` if absent).
- ``expected``: ``"pass"``.

The fixtures route through the existing embedded-mode runner
(``arborist.qa.verify.verify_quotes`` is NOT called — see the
``"answer_text" in task and "context" in task`` branch in
``bench/batteries/b_5f.py:run_falsification``). The runner sees
``observed_violations`` directly.
"""

from __future__ import annotations

import argparse
import json
import sqlite3
import sys
from pathlib import Path

DEFAULT_QA_DB = Path.home() / ".arborist" / "shards" / "qa.db"
DEFAULT_OUT = (
    Path(__file__).resolve().parents[2]
    / "bench"
    / "fixtures"
    / "5f"
    / "falsification-harvested-v1.jsonl"
)
DEFAULT_DIVERGENCE_THRESHOLD = 0.5
SAMPLE_PER_BUCKET = 20
ANSWER_TEXT_TRUNCATE = 512


def _row_divergence(row: sqlite3.Row) -> float:
    n_q = int(row["n_quotes"] or 0)
    uv_raw = row["unverified_quotes"]
    n_uv = 0
    if uv_raw:
        try:
            n_uv = len(json.loads(uv_raw))
        except (TypeError, ValueError, json.JSONDecodeError):
            n_uv = 0
    return (n_uv / n_q) if n_q > 0 else 0.0


def _build_fixture(row: sqlite3.Row, divergence: float) -> dict:
    """Map a providence_cache row to an embedded 5F falsification fixture."""
    cache_key = row["cache_key"]
    audit_mode = (row["audit_mode"] or "UNKNOWN").upper()
    verifier_method = (row["verifier_method"] or "warrant-v1").lower()

    # Observed violations honest to the actual row data:
    #   - UNGROUNDED audit_mode → "UNGROUNDED"
    #   - HYBRID audit_mode → "HYBRID_<METHOD>" (matches the live
    #     runner's emission shape — _live_falsification_violations
    #     emits HYBRID_PARAPHRASE / HYBRID_QUOTE etc.).
    #   - Any unverified quotes → also "UNVERIFIED_QUOTE".
    observed: list[str] = []
    if audit_mode == "UNGROUNDED":
        expected_reason = "UNGROUNDED"
        observed.append("UNGROUNDED")
    elif audit_mode == "HYBRID":
        expected_reason = f"HYBRID_{verifier_method.upper()}"
        observed.append(expected_reason)
    else:
        # Should not happen given the stratification filter (we only
        # harvest HYBRID + UNGROUNDED), but stay honest if it does.
        expected_reason = audit_mode
        observed.append(audit_mode)

    if (row["unverified_quotes"] or "") not in ("", "[]"):
        observed.append("UNVERIFIED_QUOTE")

    answer_text = row["answer_text"] or ""
    if len(answer_text) > ANSWER_TEXT_TRUNCATE:
        answer_text = answer_text[: ANSWER_TEXT_TRUNCATE - 3] + "..."

    return {
        "id": f"5f-fal-harvested-{cache_key[:16]}",
        "battery": "5f",
        "sub_battery": "falsification",
        "version": "v1",
        "carrier": "providence_record",
        "domain": "claim_lattice",
        "pi_star_ref": "claim-lattice@v1",
        "answer_text": answer_text,
        "observed_violations": observed,
        "expected_reason": expected_reason,
        "verifier_method_root": verifier_method,
        "expected": "pass",
        "_harvest_meta": {
            "cache_key": cache_key,
            "witness_divergence": round(divergence, 4),
            "audit_mode_at_harvest": audit_mode,
            "harvested_from": "providence_cache",
            "harvest_threshold": DEFAULT_DIVERGENCE_THRESHOLD,
            "source_ticket": "#000037 §13 step 11",
        },
    }


def _iter_controller_proposal_keys(
    conn: sqlite3.Connection,
) -> list[tuple[str, float, str]]:
    """Pull (cache_key_prefix, witness_divergence, organism_root) tuples
    from the ``controller_falsification_proposal`` stream.

    Returns rows where ``branch_id`` matches the QA-runner-advisory
    pattern ``qa:<cache_key[:16]>``. Sorted by branch_id for
    deterministic harvest output. Empty list if the table is missing
    or no proposal rows exist.
    """
    has_table = conn.execute(
        "SELECT name FROM sqlite_master "
        "WHERE type='table' AND name='controller_events'"
    ).fetchone()
    if not has_table:
        return []
    out: list[tuple[str, float, str]] = []
    for row in conn.execute(
        "SELECT branch_id, body_blob, organism_root "
        "FROM controller_events "
        "WHERE event_kind = 'controller_falsification_proposal' "
        "ORDER BY branch_id"
    ):
        bid = row["branch_id"] or ""
        if not bid.startswith("qa:"):
            continue
        prefix = bid[3:]
        try:
            body = json.loads(row["body_blob"])
        except (TypeError, ValueError, json.JSONDecodeError):
            continue
        div = float(body.get("witness_divergence", 0.0))
        out.append((prefix, div, row["organism_root"] or ""))
    return out


def _lookup_providence_by_prefix(
    conn: sqlite3.Connection, cache_key_prefix: str
) -> sqlite3.Row | None:
    """Return the providence_cache row whose cache_key starts with
    ``cache_key_prefix`` (16-char branch_id suffix). None if no match."""
    return conn.execute(
        "SELECT cache_key, audit_mode, n_quotes, n_verified, "
        "  unverified_quotes, verifier_method, falsification_state, "
        "  answer_text "
        "FROM providence_cache "
        "WHERE cache_key LIKE ? "
        "ORDER BY cache_key LIMIT 1",
        (cache_key_prefix + "%",),
    ).fetchone()


def harvest(
    qa_db: Path,
    threshold: float = DEFAULT_DIVERGENCE_THRESHOLD,
    sample_per_bucket: int = SAMPLE_PER_BUCKET,
) -> list[dict]:
    """Return a list of fixture dicts harvested from qa_db.

    Three buckets:
      - ``HYBRID`` — providence_cache rows w/ divergence ≥ threshold
      - ``UNGROUNDED`` — providence_cache rows w/ divergence ≥ threshold
      - ``CONTROLLER_PROPOSAL`` — live controller_events stream
        (event_kind = ``controller_falsification_proposal``), joined
        back to providence_cache for fixture enrichment.

    Each bucket is independently capped at ``sample_per_bucket`` rows
    sorted deterministically (cache_key for the first two, branch_id
    for the third).
    """
    if not qa_db.exists():
        raise FileNotFoundError(f"qa.db not found: {qa_db}")

    conn = sqlite3.connect(f"file:{qa_db}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row

    buckets: dict[str, list[tuple[float, sqlite3.Row]]] = {
        "HYBRID": [],
        "UNGROUNDED": [],
    }

    for row in conn.execute(
        "SELECT cache_key, audit_mode, n_quotes, n_verified, "
        "  unverified_quotes, verifier_method, falsification_state, "
        "  answer_text "
        "FROM providence_cache "
        "WHERE falsification_state = 'live' "
        "ORDER BY cache_key"
    ):
        div = _row_divergence(row)
        if div < threshold:
            continue
        audit_mode = (row["audit_mode"] or "UNKNOWN").upper()
        if audit_mode in buckets:
            buckets[audit_mode].append((div, row))

    fixtures: list[dict] = []
    for mode in sorted(buckets):
        rows = buckets[mode][:sample_per_bucket]
        for div, row in rows:
            fixtures.append(_build_fixture(row, div))

    # Third bucket — live controller_events proposal stream. Today
    # this typically yields 0 rows (Phase 2 wiring landed but live QA
    # may not have accumulated proposals yet); Phase 3 sweep emissions
    # grow it. Dedup against the providence_cache buckets above by
    # fixture id (cache_key prefix collision → same id → set membership).
    seen_ids = {fx["id"] for fx in fixtures}
    proposals = _iter_controller_proposal_keys(conn)
    cp_taken = 0
    for prefix, div, organism_root in proposals:
        if cp_taken >= sample_per_bucket:
            break
        prov_row = _lookup_providence_by_prefix(conn, prefix)
        if prov_row is None:
            continue
        fx = _build_fixture(prov_row, div)
        # Tag the fixture's _harvest_meta to surface the controller-
        # stream provenance — distinguishes it from a pure providence-
        # cache harvest of the same row.
        fx["_harvest_meta"]["harvested_from"] = "controller_events"
        fx["_harvest_meta"]["controller_organism_root"] = organism_root
        if fx["id"] in seen_ids:
            continue
        seen_ids.add(fx["id"])
        fixtures.append(fx)
        cp_taken += 1

    conn.close()
    return fixtures


def write_pack(fixtures: list[dict], out_path: Path) -> None:
    """Write the fixture pack with a _meta header line."""
    out_path.parent.mkdir(parents=True, exist_ok=True)
    meta = {
        "_meta": {
            "battery": "5f",
            "sub_battery": "falsification",
            "version": "v1",
            "task_count": len(fixtures),
            "notes": (
                "Corpus-derived 5F falsification fixtures harvested "
                "from two sources: (i) #000037 Phase 1 divergence-"
                "based providence_cache rows (HYBRID + UNGROUNDED "
                "audit modes, witness_divergence >= "
                f"{DEFAULT_DIVERGENCE_THRESHOLD}, top-"
                f"{SAMPLE_PER_BUCKET}-by-cache_key per bucket); (ii) "
                "live controller_events stream (event_kind = "
                "controller_falsification_proposal, persisted by "
                "Phase 2 audit writes), branch_id-keyed with 16-char "
                "cache_key prefix join back to providence_cache for "
                "fixture enrichment. The third source typically yields "
                "0 rows until Phase 3 sweep emissions populate it. "
                "STRICT and CANONICAL_PROJECTION rows have divergence "
                "~0 by construction and are excluded. Embedded runner "
                "path: `observed_violations` is the verifier-shape "
                "signal; `answer_text` preserved verbatim for "
                "debugging. Regenerate via `make bench-5f-harvest`."
            ),
        }
    }

    with out_path.open("w", encoding="utf-8") as f:
        f.write(json.dumps(meta) + "\n")
        for fx in fixtures:
            f.write(json.dumps(fx) + "\n")


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(
        description="Harvest #000037 Phase 1 falsification proposals "
        "into a 5F fixture pack."
    )
    p.add_argument(
        "--qa-db",
        type=Path,
        default=DEFAULT_QA_DB,
        help=f"path to qa.db shard (default: {DEFAULT_QA_DB})",
    )
    p.add_argument(
        "--threshold",
        type=float,
        default=DEFAULT_DIVERGENCE_THRESHOLD,
        help="witness_divergence threshold for inclusion (default: 0.5)",
    )
    p.add_argument(
        "--sample-per-bucket",
        type=int,
        default=SAMPLE_PER_BUCKET,
        help="how many rows per audit_mode bucket (default: 20)",
    )
    p.add_argument(
        "--out",
        type=Path,
        default=DEFAULT_OUT,
        help=f"output fixture pack (default: {DEFAULT_OUT})",
    )
    args = p.parse_args(argv)

    fixtures = harvest(
        args.qa_db,
        threshold=args.threshold,
        sample_per_bucket=args.sample_per_bucket,
    )
    write_pack(fixtures, args.out)

    sys.stderr.write(
        f"Wrote {len(fixtures)} fixtures to {args.out}\n"
    )
    by_mode: dict[str, int] = {}
    by_source: dict[str, int] = {}
    for fx in fixtures:
        m = fx["_harvest_meta"]["audit_mode_at_harvest"]
        by_mode[m] = by_mode.get(m, 0) + 1
        src = fx["_harvest_meta"].get("harvested_from", "providence_cache")
        by_source[src] = by_source.get(src, 0) + 1
    for mode, n in sorted(by_mode.items()):
        sys.stderr.write(f"  {mode}: {n}\n")
    sys.stderr.write("by source:\n")
    for src, n in sorted(by_source.items()):
        sys.stderr.write(f"  {src}: {n}\n")
    return 0


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