#000012 Phase 1c: branch-set persistence — fork_score_branches table + CLI

Phase 1a scores one (parent, child) fork at a time; Phase 1b is the
consensus paper. Neither persists multiple candidate branches at the
same checkpoint — and #000037 §12 Trigger 1 ("ForkScore regularly
receives ≥4 candidate branches per checkpoint") gates the multi-
branch path of the Prometheus-Σ controller on this data existing.
Phase 1c lands the missing seam.

Schema (arborist/store.py): _migrate_fork_score_branches creates the
sibling table with PK (branch_set_id, branch_id) + indexes on
branch_set_id and parent_root. Sibling — never enters
audit_events.event_hash preimage, so re-scoring or back-filling
cannot break the audit chain.

Helpers (arborist/substrate/fork_score.py): persist_branch_score
upserts one row via ON CONFLICT (branch_set_id, branch_id) DO UPDATE
so re-scoring the same fork under the same checkpoint is a clean
overwrite, not a duplicate. branch_set_density(conn, branch_set_id)
returns the count of distinct branches recorded under a checkpoint
— the function the #000037 §12 Trigger 1 probe reads.
ESTIMATOR_VERSION = "fork-score-v1" pins the producer generation on
every persisted row.

CLI (arborist/cli.py): arborist substrate score gains six new flags
(--branch-set, --branch-id, --parent-root, --child-root,
--persist-shard, --weights-id). Default off — --branch-set absent
preserves Phase 1a pure-function semantics for every existing
caller. When present, requires --parent-root and either --branch-id
or --child-root; missing inputs return exit code 2.

Tests (tests/test_fork_score.py, count 18 → 23): migration creates
the table + both indexes; persist writes one row carrying
parent/child roots + verdict + weights_id + estimator_version;
upsert on the PK refreshes child_root + weights_id + recorded_at
without duplicating; branch_set_density counts per-checkpoint and
ignores cross-set rows; breakdown_blob round-trips as canonical
JSON whose values sum to the persisted score.

Status sync: #000012 §7 Phase 1c flipped from "proposed, not yet
open" to "landed 2026-05-10" with the original proposal preserved
below as design log. TICKETS row 117 mirror-updated. AUTOCOUNT
counters in #000012 + cookbook bumped 18 → 23 plus the cookbook's
fork_score.py LOC row refreshed (298 → 386 module, 403 → 609
tests, density 1.35 → 1.58).

End-to-end smoke verified: arborist substrate score writes a
fork_score_branches row with the expected schema (verdict / weights_id
/ estimator_version) and the row survives a clean SQLite read.

arborist/cli.py

@@ -3227,6 +3227,43 @@ def _cmd_substrate_score(args: argparse.Namespace) -> int:
         p = Path(out_path)
         p.parent.mkdir(parents=True, exist_ok=True)
         p.write_text(artifact_json + "\n", encoding="utf-8")
+
+    # Phase 1c — branch-set persistence (#000012). Default off:
+    # writes only when --branch-set is present. --branch-id falls
+    # back to --child-root so a single-flag CLI works for the common
+    # case (one branch per child-root identity).
+    branch_set_id = getattr(args, "branch_set", None)
+    if branch_set_id:
+        from arborist.substrate.fork_score import persist_branch_score
+
+        parent_root = getattr(args, "parent_root", None)
+        child_root = getattr(args, "child_root", None)
+        branch_id = getattr(args, "branch_id", None) or child_root
+        if not parent_root or not branch_id:
+            sys.stderr.write(
+                "--branch-set requires --parent-root + (--branch-id "
+                "or --child-root)\n"
+            )
+            return 2
+        persist_shard = getattr(args, "persist_shard", None) or args.db
+        weights_id = getattr(args, "weights_id", None) or (
+            "default" if not args.weights else Path(args.weights).stem
+        )
+        p_conn = connect(persist_shard)
+        try:
+            with transaction(p_conn):
+                persist_branch_score(
+                    p_conn,
+                    branch_set_id=branch_set_id,
+                    branch_id=branch_id,
+                    parent_root=parent_root,
+                    child_root=child_root,
+                    scored=scored,
+                    weights_id=weights_id,
+                )
+        finally:
+            p_conn.close()
+
     # Non-zero exit on REJECT so CI can gate on it.
     return 0 if scored.verdict in ("ACCEPT", "MARGINAL") else 1
 
@@ -5396,6 +5433,43 @@ def build_parser() -> argparse.ArgumentParser:
             "/ ForkScore-aware mesh peers can ingest the artifact."
         ),
     )
+    # ----- Phase 1c (#000012 §7) — branch-set persistence ---------------
+    # Default off. When --branch-set is present a row is written to
+    # ``fork_score_branches``; absent ⇒ pure-function semantics
+    # (Phase 1a behavior preserved).
+    substrate_score.add_argument(
+        "--branch-set", dest="branch_set", default=None,
+        help=(
+            "checkpoint identity (e.g. parent_root + ts); when present, "
+            "writes one row to fork_score_branches sibling table"
+        ),
+    )
+    substrate_score.add_argument(
+        "--branch-id", dest="branch_id", default=None,
+        help="fork identifier; defaults to --child-root when omitted",
+    )
+    substrate_score.add_argument(
+        "--parent-root", dest="parent_root", default=None,
+        help="shared parent root (required when --branch-set is given)",
+    )
+    substrate_score.add_argument(
+        "--child-root", dest="child_root", default=None,
+        help="child root (nullable for in-flight branches)",
+    )
+    substrate_score.add_argument(
+        "--persist-shard", dest="persist_shard", default=None,
+        help=(
+            "SQLite path to write fork_score_branches row to; defaults "
+            "to --db (the current arborist target shard)"
+        ),
+    )
+    substrate_score.add_argument(
+        "--weights-id", dest="weights_id", default=None,
+        help=(
+            "opaque label for the WeightSet used; defaults to "
+            "'default' or the basename of --weights"
+        ),
+    )
     substrate_score.set_defaults(func=_cmd_substrate_score)
 
     # ----- memory subcommands (ticket #000017) --------------------------------

arborist/store.py

@@ -574,6 +574,7 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection:
         _migrate_memory_root(conn)
         _migrate_adapter_loss_reports(conn)
         _migrate_controller_events(conn)
+        _migrate_fork_score_branches(conn)
         _MIGRATED_SHARDS.add(cache_key)
 
     # Per-connection state — must run on EVERY open. SQLite scopes
@@ -872,6 +873,51 @@ def _migrate_controller_events(conn: sqlite3.Connection) -> None:
         )
 
 
+def _migrate_fork_score_branches(conn: sqlite3.Connection) -> None:
+    """Forward-migrate to add fork_score_branches (ticket #000012 Phase 1c).
+
+    Adds the sibling table that lets ``arborist substrate score``
+    persist multi-branch checkpoints. Sibling — does NOT enter
+    ``audit_events.event_hash`` preimage, so re-scoring or back-
+    filling cannot break the audit chain. PK ``(branch_set_id,
+    branch_id)`` so re-scoring the same fork under the same set is
+    a clean upsert, not a duplicate row. Default-off at the CLI
+    surface (``--branch-set`` flag absent ⇒ no row written).
+
+    Feeds #000037 §12 Trigger 1 — once a checkpoint accumulates ≥4
+    branches in this table, the multi-branch path of the
+    Prometheus-Σ controller has empirical data to fire on.
+    """
+    row = conn.execute(
+        "SELECT name FROM sqlite_master "
+        "WHERE type='table' AND name='fork_score_branches'"
+    ).fetchone()
+    if row is None:
+        conn.execute(
+            "CREATE TABLE fork_score_branches ("
+            " branch_set_id     TEXT NOT NULL,"
+            " branch_id         TEXT NOT NULL,"
+            " parent_root       TEXT NOT NULL,"
+            " child_root        TEXT,"
+            " score             REAL NOT NULL,"
+            " verdict           TEXT NOT NULL,"
+            " breakdown_blob    TEXT NOT NULL,"
+            " weights_id        TEXT NOT NULL,"
+            " estimator_version TEXT NOT NULL,"
+            " recorded_at       INTEGER NOT NULL,"
+            " PRIMARY KEY (branch_set_id, branch_id)"
+            ")"
+        )
+        conn.execute(
+            "CREATE INDEX idx_fork_score_branches_set "
+            "ON fork_score_branches(branch_set_id)"
+        )
+        conn.execute(
+            "CREATE INDEX idx_fork_score_branches_parent "
+            "ON fork_score_branches(parent_root)"
+        )
+
+
 def _migrate_memory_root(conn: sqlite3.Connection) -> None:
     """Forward-migrate to add memory-root tables (ticket #000017).
 

arborist/substrate/fork_score.py

@@ -23,13 +23,24 @@ Verdict thresholds:
 
 from __future__ import annotations
 
+import json
 import math
+import sqlite3
+import time
 from dataclasses import asdict, dataclass, field
 from typing import Iterable, Optional
 
 from arborist.substrate.weights import DEFAULT_WEIGHTS, WeightSet
 
 
+#: Phase 1c — version pin for rows persisted to ``fork_score_branches``.
+#: Bump when a code change would produce a different :class:`ScoredFork`
+#: from the same inputs (algorithm change, weight semantics shift,
+#: hard-regression policy change). Pinning lets a reader filter out
+#: rows produced under a prior estimator without re-scoring.
+ESTIMATOR_VERSION = "fork-score-v1"
+
+
 SIGNAL_FLOOR = 0.05
 """5-pp signal floor (matches docs/bench-maxing.md). Below this,
 score is MARGINAL — not strong enough to commit a fork."""
@@ -296,3 +307,80 @@ def fork_score(
         flags=flags,
         weights=weights.as_dict(),
     )
+
+
+# ---------------------------------------------------------------------
+# Phase 1c — branch-set persistence (#000012 §7 Phase 1c)
+# ---------------------------------------------------------------------
+
+
+def persist_branch_score(
+    conn: sqlite3.Connection,
+    *,
+    branch_set_id: str,
+    branch_id: str,
+    parent_root: str,
+    child_root: Optional[str],
+    scored: ScoredFork,
+    weights_id: str = "default",
+    estimator_version: str = ESTIMATOR_VERSION,
+    ts: Optional[int] = None,
+) -> None:
+    """Write one :class:`ScoredFork` to ``fork_score_branches``.
+
+    Upserts on ``(branch_set_id, branch_id)`` so re-scoring the same
+    fork under the same checkpoint is a clean overwrite, not a
+    duplicate row. Sibling table — never enters
+    ``audit_events.event_hash`` preimage. Caller wraps in
+    :func:`arborist.store.transaction` when batching.
+    """
+    if ts is None:
+        ts = int(time.time())
+    breakdown_blob = json.dumps(
+        scored.breakdown, sort_keys=True, separators=(",", ":")
+    )
+    conn.execute(
+        "INSERT INTO fork_score_branches ("
+        " branch_set_id, branch_id, parent_root, child_root,"
+        " score, verdict, breakdown_blob, weights_id,"
+        " estimator_version, recorded_at"
+        ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
+        " ON CONFLICT(branch_set_id, branch_id) DO UPDATE SET"
+        "   parent_root = excluded.parent_root,"
+        "   child_root = excluded.child_root,"
+        "   score = excluded.score,"
+        "   verdict = excluded.verdict,"
+        "   breakdown_blob = excluded.breakdown_blob,"
+        "   weights_id = excluded.weights_id,"
+        "   estimator_version = excluded.estimator_version,"
+        "   recorded_at = excluded.recorded_at",
+        (
+            branch_set_id,
+            branch_id,
+            parent_root,
+            child_root,
+            float(scored.score),
+            scored.verdict,
+            breakdown_blob,
+            weights_id,
+            estimator_version,
+            ts,
+        ),
+    )
+
+
+def branch_set_density(
+    conn: sqlite3.Connection, branch_set_id: str
+) -> int:
+    """Count distinct branches recorded under a checkpoint id.
+
+    Used by the #000037 §12 Trigger 1 probe to satisfy the
+    "ForkScore regularly receives ≥ 4 candidate branches per
+    checkpoint" gate.
+    """
+    row = conn.execute(
+        "SELECT COUNT(*) FROM fork_score_branches"
+        " WHERE branch_set_id = ?",
+        (branch_set_id,),
+    ).fetchone()
+    return int(row[0]) if row else 0

docs/TICKETS.md

@@ -125,7 +125,7 @@ Newest first. Update on every open/close.
 | #000015  | π* domain library + cross-domain composition   | closed · landed 2026-05-07 | 2026-05-07 | —         |
 | #000014  | SelfModel: schema, falsification, integration  | closed · landed 2026-05-07 | 2026-05-07 | —         |
 | #000013  | Spatial-temporal substrate (Merkle-AGI v7-W)   | closed · landed 2026-05-09 (substrate paper + frontier catalog + namespace stub) | 2026-05-07 | —         |
-| #000012  | Selection & consensus protocol (Merkle-AGI v8) | in progress · Phase 1a (ForkScore) landed 2026-05-08; Phase 1b (consensus paper, `docs/_source/merkle-agi-v8-consensus.rst` 834 lines) landed 2026-05-10; Phase 1c (branch-set persistence) remains proposed-not-opened | 2026-05-07 | —         |
+| #000012  | Selection & consensus protocol (Merkle-AGI v8) | in progress · Phase 1a (ForkScore) landed 2026-05-08; Phase 1b (consensus paper, `docs/_source/merkle-agi-v8-consensus.rst` 834 lines) landed 2026-05-10; Phase 1c (branch-set persistence — `fork_score_branches` sibling table, `persist_branch_score` + `branch_set_density`, 6 new CLI flags on `arborist substrate score`, default-off) landed 2026-05-10 — feeds #000037 §12 Trigger 1 | 2026-05-07 | —         |
 | #000011  | SOFT_PREFLIGHT_HINT model-assisted sidecar     | closed · landed 2026-05-04 (zero-shot full impl) | 2026-05-04 | D1 (preserves) |
 | #000010  | Meta-Cognition Preflight Guard (M0 / MCTL)     | closed · landed 2026-05-03 (Phases 1–4); DAG binding shipped via #000009 | 2026-05-03 | D1, D3    |
 | #000009  | Preflight run-DAG node binding (#000008+#000010) | closed · re-landed 2026-05-04 (§8 corrections: reject-path DAG, nested CTI clauses) | 2026-05-03 | D3, D4    |

docs/tickets/ticket-000012-selection-consensus-protocol.md

@@ -283,7 +283,7 @@ function ahead of the consensus paper:
   ([0, SIGNAL_FLOOR)), REJECT (negative score OR hard-regression
   flag OR `NEG_INF_REGRESSION` flag).
 - Reference doc: `docs/v8-fork-score.md`.
-- Tests: <!--AUTOCOUNT:tests:tests/test_fork_score.py-->18<!--/AUTOCOUNT-->
+- Tests: <!--AUTOCOUNT:tests:tests/test_fork_score.py-->23<!--/AUTOCOUNT-->
   cases in `tests/test_fork_score.py` pin the pure ScoredFork
   dataclass + scoring contract (SIGNAL_FLOOR=0.05,
   HARD_REGRESSION_FLOOR=0.05, score = sum-of-breakdown closure,
@@ -347,7 +347,48 @@ What's deliberately NOT in Phase 1a (now specified in the v8 paper):
 - Stake mechanics + economic incentives. → Part 6
 - Cross-validator ZK proof exchange. → Closure §; deferred to #000016
 
-### Phase 1c — Branch-set persistence (proposed, not yet open)
+### Phase 1c — Branch-set persistence (landed 2026-05-10)
+
+**Status:** landed. Pressure-1 satisfied (Phase 1b paper landed
+2026-05-10 — multi-branch deployment paper-spec is closed); fox
+operator-go to wire the data path so #000037 §12 Trigger 1 gains
+empirical surface. Implementation pinned below.
+
+- **Schema:** `fork_score_branches` sibling table created via
+  `arborist.store._migrate_fork_score_branches`; indexes on
+  `branch_set_id` and `parent_root`. PK `(branch_set_id, branch_id)`.
+- **Helpers:** `arborist.substrate.fork_score.persist_branch_score`
+  (upsert one row; ON CONFLICT replaces score / verdict /
+  breakdown_blob / weights_id / estimator_version / recorded_at) +
+  `branch_set_density(conn, branch_set_id)` (count distinct branches
+  for a checkpoint — the function the #000037 §12 Trigger 1 probe
+  reads).
+- **CLI:** `arborist substrate score` gains six flags
+  (`--branch-set`, `--branch-id`, `--parent-root`, `--child-root`,
+  `--persist-shard`, `--weights-id`). Default off — `--branch-set`
+  absent ⇒ pure-function semantics preserved (Phase 1a behavior
+  unchanged for every existing caller).
+- **Estimator version pin:** module-level
+  `ESTIMATOR_VERSION = "fork-score-v1"` constant; bump when a code
+  change would produce a different `ScoredFork` from the same
+  inputs (algorithm change, weight semantics, hard-regression
+  policy). Persisted on every row so a reader can filter by
+  estimator generation.
+- **Tests:** 5 new in `tests/test_fork_score.py` —
+  migration-creates-table, persist-writes-one-row, upsert-on-pk,
+  branch_set_density-counts-by-set, breakdown_blob-round-trips-as-
+  json. Test count 18 → 23.
+- **Hard constraints honored:** sibling table never enters
+  `audit_events.event_hash` preimage; no behavioral change to
+  single-validator scoring; default-off CLI; no mesh wire format
+  change; `weights_id` opaque (folding weights into a hash stays a
+  Phase 1b/wire concern).
+
+The original Phase-1c proposal text is preserved below for design-
+log continuity. Re-read it as the authoritative spec; the bullets
+above are the landing receipt.
+
+#### Original proposal (preserved)
 
 **Problem.** Phase 1a scores one ``(parent, child)`` fork at a time;
 Phase 1b is the consensus paper. Neither persists *multiple

docs/warrant-substrate-cookbook.md

@@ -607,7 +607,7 @@ than waiting for bench-time STRICT-rate drift to surface it.
   input, recommendation-text mode transitions, and the §11
   worked-example bit-for-bit (with doc-calibration update
   surfaced through the test).
-- `tests/test_fork_score.py` — <!--AUTOCOUNT:tests:tests/test_fork_score.py-->18<!--/AUTOCOUNT--> tests for v8 ForkScore
+- `tests/test_fork_score.py` — <!--AUTOCOUNT:tests:tests/test_fork_score.py-->23<!--/AUTOCOUNT--> tests for v8 ForkScore
   (#000012 Phase 1a); pins SIGNAL_FLOOR (5pp) + HARD_REGRESSION_FLOOR
   (5pp), score = sum-of-breakdown closure, security_risk inert
   under default iota=0 (opt-in), NEG_INF_REGRESSION hard-reject.
@@ -675,7 +675,7 @@ than waiting for bench-time STRICT-rate drift to surface it.
 | warrant_resolver.py | ~800 | ~430 (combined) | 0.54 |
 | warrant_chain.py | 89 | 320 (<!--AUTOCOUNT:tests:tests/test_warrant_chain.py-->9<!--/AUTOCOUNT--> tests) | 3.6 |
 | t3_bound_calculator.py | 249 | 446 (<!--AUTOCOUNT:tests:tests/test_t3_bound_calculator.py-->53<!--/AUTOCOUNT--> tests) | 1.79 |
-| fork_score.py | 298 | 403 (<!--AUTOCOUNT:tests:tests/test_fork_score.py-->18<!--/AUTOCOUNT--> tests) | 1.35 |
+| fork_score.py | 386 | 609 (<!--AUTOCOUNT:tests:tests/test_fork_score.py-->23<!--/AUTOCOUNT--> tests) | 1.58 |
 | weights.py | 73 | 180 (<!--AUTOCOUNT:tests:tests/test_weights.py-->16<!--/AUTOCOUNT--> tests) | 2.5 |
 | pi_star/protocol+registry | 124 | 280 (<!--AUTOCOUNT:tests:tests/test_pi_star_protocol_and_registry.py-->21<!--/AUTOCOUNT--> tests) | 2.3 |
 | qa/progress.py | 85 | 226 (<!--AUTOCOUNT:tests:tests/test_qa_progress.py-->31<!--/AUTOCOUNT--> tests) | 2.7 |

tests/test_fork_score.py

@@ -401,3 +401,209 @@ def test_signal_floor_honored():
     # Score = alpha · target_delta = SIGNAL_FLOOR exactly. Verdict ACCEPT.
     assert r.score == pytest.approx(SIGNAL_FLOOR, abs=1e-9)
     assert r.verdict == "ACCEPT"
+
+
+# ---------------------------------------------------------------------
+# Phase 1c — branch-set persistence (#000012 §7 Phase 1c)
+# ---------------------------------------------------------------------
+
+
+def _scored():
+    """Build a representative ScoredFork via the real fork_score()."""
+    parent = _bench_dict({
+        "5s": {"syntax": {"parse_pass_rate": 0.5}},
+    })
+    child = _bench_dict({
+        "5s": {"syntax": {"parse_pass_rate": 0.7}},
+    })
+    return fork_score(parent, child)
+
+
+def test_phase1c_migration_creates_table(tmp_path):
+    """Opening a connection runs the Phase 1c migration; the table
+    + both indexes exist."""
+    from arborist.store import connect, invalidate_migration_cache
+
+    db = tmp_path / "shard.db"
+    invalidate_migration_cache(db)
+    conn = connect(db)
+    try:
+        row = conn.execute(
+            "SELECT name FROM sqlite_master "
+            "WHERE type='table' AND name='fork_score_branches'"
+        ).fetchone()
+        assert row is not None
+        idx_names = {
+            r[0]
+            for r in conn.execute(
+                "SELECT name FROM sqlite_master WHERE type='index' "
+                "AND tbl_name='fork_score_branches'"
+            ).fetchall()
+        }
+        assert "idx_fork_score_branches_set" in idx_names
+        assert "idx_fork_score_branches_parent" in idx_names
+    finally:
+        conn.close()
+        invalidate_migration_cache(db)
+
+
+def test_phase1c_persist_branch_score_writes_one_row(tmp_path):
+    from arborist.store import connect, invalidate_migration_cache, transaction
+    from arborist.substrate.fork_score import (
+        ESTIMATOR_VERSION,
+        persist_branch_score,
+    )
+
+    db = tmp_path / "shard.db"
+    invalidate_migration_cache(db)
+    conn = connect(db)
+    try:
+        with transaction(conn):
+            persist_branch_score(
+                conn,
+                branch_set_id="cp-1",
+                branch_id="b-A",
+                parent_root="parent-root-aaaa",
+                child_root="child-root-A",
+                scored=_scored(),
+                weights_id="default",
+            )
+        rows = conn.execute(
+            "SELECT branch_set_id, branch_id, parent_root, child_root, "
+            "  verdict, weights_id, estimator_version "
+            "FROM fork_score_branches"
+        ).fetchall()
+        assert len(rows) == 1
+        r = rows[0]
+        assert r["branch_set_id"] == "cp-1"
+        assert r["branch_id"] == "b-A"
+        assert r["parent_root"] == "parent-root-aaaa"
+        assert r["child_root"] == "child-root-A"
+        assert r["verdict"] in ("ACCEPT", "MARGINAL", "REJECT")
+        assert r["estimator_version"] == ESTIMATOR_VERSION
+    finally:
+        conn.close()
+        invalidate_migration_cache(db)
+
+
+def test_phase1c_persist_upserts_on_pk(tmp_path):
+    """Re-scoring the same (branch_set_id, branch_id) is an upsert,
+    not a duplicate row. Fields refresh."""
+    from arborist.store import connect, invalidate_migration_cache, transaction
+    from arborist.substrate.fork_score import persist_branch_score
+
+    db = tmp_path / "shard.db"
+    invalidate_migration_cache(db)
+    conn = connect(db)
+    try:
+        s1 = _scored()
+        with transaction(conn):
+            persist_branch_score(
+                conn,
+                branch_set_id="cp-up",
+                branch_id="b-up",
+                parent_root="p1",
+                child_root="c1",
+                scored=s1,
+                weights_id="w1",
+                ts=1700000000,
+            )
+        with transaction(conn):
+            persist_branch_score(
+                conn,
+                branch_set_id="cp-up",
+                branch_id="b-up",
+                parent_root="p1",
+                child_root="c2-new",
+                scored=s1,
+                weights_id="w2-new",
+                ts=1700000999,
+            )
+        rows = conn.execute(
+            "SELECT child_root, weights_id, recorded_at "
+            "FROM fork_score_branches WHERE branch_set_id = 'cp-up'"
+        ).fetchall()
+        assert len(rows) == 1
+        assert rows[0]["child_root"] == "c2-new"
+        assert rows[0]["weights_id"] == "w2-new"
+        assert rows[0]["recorded_at"] == 1700000999
+    finally:
+        conn.close()
+        invalidate_migration_cache(db)
+
+
+def test_phase1c_branch_set_density_counts_branches(tmp_path):
+    """branch_set_density returns 0 / N for the queried checkpoint;
+    rows under other checkpoints don't leak."""
+    from arborist.store import connect, invalidate_migration_cache, transaction
+    from arborist.substrate.fork_score import (
+        branch_set_density,
+        persist_branch_score,
+    )
+
+    db = tmp_path / "shard.db"
+    invalidate_migration_cache(db)
+    conn = connect(db)
+    try:
+        assert branch_set_density(conn, "missing") == 0
+        s = _scored()
+        with transaction(conn):
+            for i in range(4):
+                persist_branch_score(
+                    conn,
+                    branch_set_id="cp-A",
+                    branch_id=f"branch-{i}",
+                    parent_root="parent-root",
+                    child_root=f"child-root-{i}",
+                    scored=s,
+                )
+            persist_branch_score(
+                conn,
+                branch_set_id="cp-B",
+                branch_id="lone",
+                parent_root="parent-root",
+                child_root="child-root-Z",
+                scored=s,
+            )
+        # #000037 §12 Trigger 1 satisfied: ≥4 branches at "cp-A".
+        assert branch_set_density(conn, "cp-A") == 4
+        assert branch_set_density(conn, "cp-B") == 1
+        assert branch_set_density(conn, "cp-missing") == 0
+    finally:
+        conn.close()
+        invalidate_migration_cache(db)
+
+
+def test_phase1c_breakdown_blob_round_trips_as_json(tmp_path):
+    """breakdown_blob stores the per-term breakdown losslessly so a
+    downstream reader can replay the verdict."""
+    import json as _json
+    from arborist.store import connect, invalidate_migration_cache, transaction
+    from arborist.substrate.fork_score import persist_branch_score
+
+    db = tmp_path / "shard.db"
+    invalidate_migration_cache(db)
+    conn = connect(db)
+    try:
+        scored = _scored()
+        with transaction(conn):
+            persist_branch_score(
+                conn,
+                branch_set_id="cp-blob",
+                branch_id="b-blob",
+                parent_root="p",
+                child_root="c",
+                scored=scored,
+            )
+        row = conn.execute(
+            "SELECT breakdown_blob FROM fork_score_branches WHERE "
+            "branch_set_id='cp-blob'"
+        ).fetchone()
+        recovered = _json.loads(row["breakdown_blob"])
+        assert sum(recovered.values()) == pytest.approx(
+            scored.score, abs=1e-9
+        )
+        assert set(recovered.keys()) == set(scored.breakdown.keys())
+    finally:
+        conn.close()
+        invalidate_migration_cache(db)