#000037 follow-through: persist falsification proposals + harvest live stream

Two coupled changes that wire the §13 Step 11 proposal stream from
in-memory-and-discarded to persisted-and-harvestable.

(A) emit_controller_events now writes a 4th event kind
controller_falsification_proposal, one row per
decision.falsification_proposals entry. Body carries branch_id +
witness_divergence + reason; label column carries the reason for
terminal-table inspection. Deterministic ordering by (branch_id,
witness_divergence) so canonical-body hashes are stable. Idempotent
under the existing UNIQUE (event_kind, body_hash) constraint. The
audit-chain semantics remain unchanged (still a sibling table; no
event_hash preimage entry). The CLI inspector's --kind choices gain
the new event kind so operators can filter for it directly.

(B) bench/scripts/harvest_falsification_proposals.py gains a third
source bucket CONTROLLER_PROPOSAL alongside the existing HYBRID +
UNGROUNDED providence_cache buckets. Reads
controller_falsification_proposal rows from controller_events,
extracts the 16-char cache_key prefix from branch_id (qa:<prefix>
pattern from the QA-runner advisory), joins back to providence_cache
for fixture enrichment (answer_text + audit_mode + verifier_method),
and tags _harvest_meta.harvested_from = "controller_events" so the
two source paths stay distinguishable in the fixture pack. Dedup
against the providence_cache buckets by fixture id.

Today this typically yields 0 new fixtures because (i) no live QA
has fired since Phase 2 wiring landed, and (ii) the QA-runner
single-branch advisory's proposals overlap providence_cache content
the harvester already finds. Real net value comes from Phase 3
(#000045) sweep emissions, which will produce multi-branch chunk
proposals that providence_cache rows can't predict.

Tests: 3 new in tests/test_prometheus_audit.py (proposal-row
emission, no-proposal no-row, idempotency); 1 new in
tests/test_bench_batteries.py (synthetic qa.db with both
providence_cache + controller_events rows; asserts the
controller_events bucket surfaces a fixture invisible to the
divergence-thresholded providence_cache buckets). The pre-existing
harvested-pack-runs-clean test relaxes its harvested_from pin from
"providence_cache"-only to {"providence_cache", "controller_events"}.

arborist/cli.py

@@ -5026,6 +5026,7 @@ def build_parser() -> argparse.ArgumentParser:
             "controller_decision",
             "controller_difficulty",
             "controller_budget_allocation",
+            "controller_falsification_proposal",
         ],
         default=None,
         help="filter to one event_kind",

arborist/substrate/prometheus_audit.py

@@ -9,12 +9,17 @@ prune controller_events without integrity risk.
 The decision shape is Phase 1's territory; this module never
 computes decisions, only persists them.
 
-Three event kinds emitted per ControllerDecision:
+Four event kinds emitted per ControllerDecision:
-  controller_decision           — one row, label + entropy +
+  controller_decision               — one row, label + entropy +
-                                  selected_branch_id
+                                      selected_branch_id
-  controller_difficulty         — one row, difficulty_next
+  controller_difficulty             — one row, difficulty_next
-  controller_budget_allocation  — one row per branch with nonzero
+  controller_budget_allocation      — one row per branch with nonzero
-                                  allocation
+                                      allocation
+  controller_falsification_proposal — one row per §13 Step 11 proposal
+                                      (witness_divergence ≥ threshold);
+                                      lets a downstream harvester read
+                                      the proposal stream from the
+                                      audit table
 
 Idempotency is enforced by the table's
 ``UNIQUE (event_kind, body_hash)`` constraint. body_hash is the
@@ -197,4 +202,38 @@ def emit_controller_events(
         if eid is not None:
             written.append(eid)
 
+    # --- 4. per-proposal controller_falsification_proposal rows -------
+    # §13 Step 11 proposals are emitted in-memory by `controller_decide`
+    # but were previously discarded at the persist boundary. Persisting
+    # them as a 4th event kind closes the live-harvest loop with
+    # `bench/scripts/harvest_falsification_proposals.py` — the harvester
+    # can read this stream alongside its providence_cache scan.
+    # Deterministic ordering by (branch_id, witness_divergence) so the
+    # canonical body hashes are stable across tuple-iteration variance.
+    proposals = tuple(decision.falsification_proposals)
+    for prop in sorted(
+        proposals, key=lambda p: (p.branch_id, p.witness_divergence)
+    ):
+        proposal_body = {
+            "kind": "controller_falsification_proposal",
+            "organism_root": organism_root,
+            "branch_id": prop.branch_id,
+            "witness_divergence": float(prop.witness_divergence),
+            "reason": prop.reason,
+        }
+        eid = _insert_row(
+            conn,
+            organism_root=organism_root,
+            branch_id=prop.branch_id,
+            event_kind="controller_falsification_proposal",
+            label=prop.reason,
+            entropy=None,
+            difficulty=None,
+            allocation=None,
+            body=proposal_body,
+            ts=ts,
+        )
+        if eid is not None:
+            written.append(eid)
+
     return written

bench/scripts/harvest_falsification_proposals.py

@@ -22,10 +22,19 @@ Sample policy (`SAMPLE_PER_BUCKET = 20`):
   ``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.
 
-This produces 40 corpus-derived fixtures total. The pack expands
+The first two buckets produce 40 corpus-derived fixtures by default;
-when the harvest is re-run against a shard with more high-divergence
+the third grows the pack as the controller_events stream fills.
-rows; idempotency on re-harvest is the caller's job (the runner
+Idempotency on re-harvest is the caller's job (the runner
 truncates+rewrites by default).
 
 Mapping cache row → fixture:
@@ -137,12 +146,77 @@ def _build_fixture(row: sqlite3.Row, divergence: float) -> dict:
     }
 
 
+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."""
+    """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}")
 
@@ -169,14 +243,39 @@ def harvest(
         if audit_mode in buckets:
             buckets[audit_mode].append((div, row))
 
-    conn.close()
-
     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
 
 
@@ -191,16 +290,21 @@ def write_pack(fixtures: list[dict], out_path: Path) -> None:
             "task_count": len(fixtures),
             "notes": (
                 "Corpus-derived 5F falsification fixtures harvested "
-                "from #000037 Phase 1 FalsificationFixtureProposal "
+                "from two sources: (i) #000037 Phase 1 divergence-"
-                "records. Each row is a providence_cache entry with "
+                "based providence_cache rows (HYBRID + UNGROUNDED "
-                "witness_divergence (n_unverified / n_quotes) >= "
+                "audit modes, witness_divergence >= "
-                f"{DEFAULT_DIVERGENCE_THRESHOLD}. Stratified sample: "
+                f"{DEFAULT_DIVERGENCE_THRESHOLD}, top-"
-                f"top-{SAMPLE_PER_BUCKET}-by-cache_key for HYBRID + "
+                f"{SAMPLE_PER_BUCKET}-by-cache_key per bucket); (ii) "
-                "UNGROUNDED audit modes (the only modes that produce "
+                "live controller_events stream (event_kind = "
-                "high-divergence rows; STRICT and CANONICAL_PROJECTION "
+                "controller_falsification_proposal, persisted by "
-                "rows have divergence ~0 by construction). Embedded "
+                "Phase 2 audit writes), branch_id-keyed with 16-char "
-                "runner path: `observed_violations` is the verifier-"
+                "cache_key prefix join back to providence_cache for "
-                "shape signal; `answer_text` preserved verbatim 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`."
             ),
         }
@@ -254,11 +358,17 @@ def main(argv: list[str] | None = None) -> int:
         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
 
 

tests/test_bench_batteries.py

@@ -693,7 +693,13 @@ def test_5f_falsification_harvested_pack_runs_clean():
             row = json.loads(line)
             meta = row.get("_harvest_meta", {})
             assert meta.get("source_ticket") == "#000037 §13 step 11"
-            assert meta.get("harvested_from") == "providence_cache"
+            # Two harvest sources now: providence_cache directly
+            # (Phase 1 buckets) and controller_events stream (live
+            # Phase 2 advisory writes).
+            assert meta.get("harvested_from") in (
+                "providence_cache",
+                "controller_events",
+            )
             assert isinstance(meta.get("witness_divergence"), (int, float))
             assert meta["witness_divergence"] >= 0.5
             assert meta.get("audit_mode_at_harvest") in ("HYBRID", "UNGROUNDED")
@@ -722,6 +728,86 @@ def test_5f_falsification_harvested_pack_widens_motif_coverage():
     assert "UNGROUNDED" in motifs
 
 
+def test_harvest_includes_controller_events_bucket(tmp_path):
+    """The harvester pulls a third bucket from the live
+    controller_events stream (event_kind =
+    controller_falsification_proposal) in addition to the two
+    providence_cache buckets. Builds a synthetic qa.db with both
+    sources and asserts the controller_events-derived fixture appears
+    with the right provenance."""
+    import sqlite3 as _sqlite3
+    from bench.scripts.harvest_falsification_proposals import harvest
+
+    qa_db = tmp_path / "qa.db"
+    conn = _sqlite3.connect(qa_db)
+    conn.row_factory = _sqlite3.Row
+    # Minimal providence_cache schema for the harvest path.
+    conn.executescript(
+        "CREATE TABLE providence_cache ("
+        " cache_key TEXT PRIMARY KEY, audit_mode TEXT, n_quotes INTEGER,"
+        " n_verified INTEGER, unverified_quotes TEXT, verifier_method TEXT,"
+        " falsification_state TEXT, answer_text TEXT);"
+        "CREATE TABLE controller_events ("
+        " event_id INTEGER PRIMARY KEY AUTOINCREMENT,"
+        " organism_root TEXT NOT NULL, branch_id TEXT, event_kind TEXT NOT NULL,"
+        " label TEXT, entropy REAL, difficulty REAL, allocation REAL,"
+        " body_blob TEXT NOT NULL, body_hash TEXT NOT NULL,"
+        " recorded_at INTEGER NOT NULL,"
+        " UNIQUE (event_kind, body_hash));"
+    )
+    # Providence row whose cache_key matches a future
+    # controller_events branch_id prefix. Divergence < 0.5 so the row
+    # is INVISIBLE to the providence buckets — it surfaces ONLY via the
+    # controller_events stream.
+    conn.execute(
+        "INSERT INTO providence_cache VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            "ce-only-cache-key-12345678abcdef",
+            "HYBRID",
+            10, 9,
+            json.dumps(["one quote"]),  # 1/10 = 0.1 < threshold
+            "claim_lattice",
+            "live",
+            "Synthetic answer text from a controller-events row.",
+        ),
+    )
+    # Matching controller_falsification_proposal row. Divergence here
+    # is the high signal that surfaced the proposal at controller-decide
+    # time (sweep mode would commonly push such rows).
+    conn.execute(
+        "INSERT INTO controller_events VALUES "
+        "(NULL, ?, ?, ?, ?, NULL, NULL, NULL, ?, ?, ?)",
+        (
+            "qa:ce-only-cache-key-12345678abcdef",
+            "qa:ce-only-cache-key",  # 16-char prefix of the cache_key
+            "controller_falsification_proposal",
+            "WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD",
+            json.dumps({
+                "kind": "controller_falsification_proposal",
+                "organism_root": "qa:ce-only-cache-key-12345678abcdef",
+                "branch_id": "qa:ce-only-cache-key",
+                "witness_divergence": 0.8,
+                "reason": "WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD",
+            }),
+            "deadbeef" * 8,  # placeholder body_hash for the test
+            1700000000,
+        ),
+    )
+    conn.commit()
+    conn.close()
+
+    fixtures = harvest(qa_db, threshold=0.5, sample_per_bucket=20)
+    ce_fixtures = [
+        fx for fx in fixtures
+        if fx["_harvest_meta"]["harvested_from"] == "controller_events"
+    ]
+    assert len(ce_fixtures) == 1
+    fx = ce_fixtures[0]
+    assert fx["_harvest_meta"]["controller_organism_root"].startswith("qa:")
+    assert fx["_harvest_meta"]["witness_divergence"] == 0.8
+    assert fx["id"] == "5f-fal-harvested-ce-only-cache-ke"
+
+
 def test_5f_falsification_live_helper_calls_real_verifier():
     """UNGROUNDED + STRICT_SPAN signals come straight from verify_quotes."""
     from bench.batteries.b_5f import _live_falsification_violations

tests/test_prometheus_audit.py

@@ -52,6 +52,14 @@ from arborist.substrate.prometheus_audit import (
 # landing.
 
 
+@dataclass(frozen=True)
+class StubProposal:
+    organism_root: str
+    branch_id: str
+    witness_divergence: float
+    reason: str
+
+
 @dataclass(frozen=True)
 class StubDecision:
     selected_branch_id: str | None
@@ -63,6 +71,7 @@ class StubDecision:
     notes: tuple
     memory_proposals: tuple = ()
     selfmodel_proposals: tuple = ()
+    falsification_proposals: tuple = ()
     advisory_events: tuple = ()
 
 
@@ -388,6 +397,97 @@ def test_query_by_kind_returns_rows(shard):
     assert n_alloc == 2
 
 
+# ---------------------------------------------------------------------
+# §13 Step 11 proposal persistence (4th event kind)
+# ---------------------------------------------------------------------
+
+
+def test_emit_persists_falsification_proposal_rows(shard):
+    """A decision carrying §13 Step 11 falsification proposals writes
+    one ``controller_falsification_proposal`` row per proposal."""
+    conn, _ = shard
+    dec = _stub_decision(
+        falsification_proposals=(
+            StubProposal(
+                organism_root="qa:abcd1234abcd1234abcd",
+                branch_id="qa:abcd1234abcd1234",
+                witness_divergence=0.66,
+                reason="WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD",
+            ),
+            StubProposal(
+                organism_root="qa:abcd1234abcd1234abcd",
+                branch_id="qa:other-branch",
+                witness_divergence=0.75,
+                reason="WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD",
+            ),
+        ),
+    )
+    with transaction(conn):
+        emit_controller_events(conn, dec, "qa:abcd1234abcd1234abcd")
+
+    rows = conn.execute(
+        "SELECT branch_id, label FROM controller_events "
+        "WHERE event_kind='controller_falsification_proposal' "
+        "ORDER BY branch_id"
+    ).fetchall()
+    assert len(rows) == 2
+    # Sorted-by-branch_id ordering pin matches emit_controller_events'
+    # canonical-body discipline.
+    assert rows[0]["branch_id"] == "qa:abcd1234abcd1234"
+    assert rows[1]["branch_id"] == "qa:other-branch"
+    # ``label`` carries the proposal reason for terminal-table inspection.
+    assert rows[0]["label"] == "WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD"
+
+
+def test_emit_no_proposals_writes_no_proposal_rows(shard):
+    """Empty falsification_proposals tuple → no proposal rows. The
+    other three event kinds are unaffected."""
+    conn, _ = shard
+    dec = _stub_decision(falsification_proposals=())
+    with transaction(conn):
+        emit_controller_events(conn, dec, "qa:no-proposals")
+    n = conn.execute(
+        "SELECT COUNT(*) FROM controller_events "
+        "WHERE event_kind='controller_falsification_proposal'"
+    ).fetchone()[0]
+    assert n == 0
+    # Other rows still present.
+    n_other = conn.execute(
+        "SELECT COUNT(*) FROM controller_events "
+        "WHERE event_kind != 'controller_falsification_proposal'"
+    ).fetchone()[0]
+    assert n_other >= 3  # decision + difficulty + ≥1 allocation
+
+
+def test_emit_proposal_rows_idempotent(shard):
+    """Re-emitting the same decision (same proposals) is a no-op under
+    the UNIQUE (event_kind, body_hash) constraint."""
+    conn, _ = shard
+    proposals = (
+        StubProposal(
+            organism_root="qa:idem-root",
+            branch_id="qa:idem-branch",
+            witness_divergence=0.55,
+            reason="WITNESS_DIVERGENCE_EXCEEDS_THRESHOLD",
+        ),
+    )
+    dec = _stub_decision(falsification_proposals=proposals)
+    with transaction(conn):
+        emit_controller_events(conn, dec, "qa:idem-root")
+        first_n = conn.execute(
+            "SELECT COUNT(*) FROM controller_events "
+            "WHERE event_kind='controller_falsification_proposal'"
+        ).fetchone()[0]
+    with transaction(conn):
+        emit_controller_events(conn, dec, "qa:idem-root")
+        second_n = conn.execute(
+            "SELECT COUNT(*) FROM controller_events "
+            "WHERE event_kind='controller_falsification_proposal'"
+        ).fetchone()[0]
+    assert first_n == 1
+    assert second_n == first_n
+
+
 # ---------------------------------------------------------------------
 # Phase 2 integration: arborist.qa.runner._emit_qa_controller_advisory
 # wires real QA verdicts → controller_events rows