Private-chain (SM1) adversary in the per-node engine

Sec 6.8 recorded that "the per-node engine has no private-chain strategy", which is
why every selfish result came from the global race model with uncle recovery as
a free knob eta -- and why open item 5 (does the uncle cap need margin under
attack-inflated orphaning?) could not be sized: a knob has no queue to overflow.

adversary_strategy="selfish" adds it. The coalition mines one shared private
chain and releases under the classic SM1 rules in (a, h) form: adopt when the
public chain wins, match at equal length, override at a one-block lead, else
wait. Only VISIBILITY is modelled -- the coalition's mining needs no special
case, because a member's fork choice already builds on the private tip whenever
it leads (that tip has the greatest height among blocks the member can see) and
falls back to the public chain exactly when the public chain overtakes, which
is the adopt branch. So the private chain forms, extends and is abandoned
emergently, and the code that had to be written is the arrival matrix.

Design notes worth keeping:
- Private blocks reuse the sentinel `withhold` already had (never-arrives), so
  the existing exclusions from canonical-tip selection apply unchanged; release
  flips it back and gossips DIRECTLY from the producer, bypassing Blend, since
  an adversary has no privacy budget to respect and wants the race won.
- A private chain breaks the windowed horizon's premise (a hidden block is old
  enough to look fully-propagated while no honest node has it, and it becomes
  visible LATER, which the one-way frontier pointer cannot revisit), so selfish
  forces the exact full scan and full matrix.
- Blocks still hidden at epoch end are abandoned and hidden from the coalition
  too, or the canonical-tip search would crown a chain no honest node saw.

Validated against Eyal-Sirer at sub-slot latency: revenue share 0.0356 vs an
exact 0.0356 at alpha = 0.1, and above the closed form at higher alpha by just
the margin the alpha_eff fork-amplification correction predicts (0.498 vs 0.484
at alpha = 0.4, with fork rate 0.38).

Adds p_ref_honest: the reference rate over orphans produced OUTSIDE the
coalition. Under a private-chain attack this diverges sharply from p_ref, and
only the honest one measures the repair the report credits to uncle counting --
an attacker's own discarded blocks are its loss to bear.

test_fork unpacks fork_stats positionally, so its three call sites take the new
fifth value. 247 tests pass.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tools/simulators/tsi/tsi-sim-pernode/scripts/selfish_uncle_margin.py

@@ -0,0 +1,127 @@
+"""Does the uncle cap need margin under a private-chain attack? — REPORT §8.3 item 5.
+
+Item 5: "Under attack-inflated orphaning the honest-load cap may need extra margin (owed uncles
+beyond `U` defer and can age out of `W`); this report does not size it." It could not be sized
+before, because the per-node engine had no private-chain strategy (§6.8) — the selfish results
+came from a global race model in which uncle recovery is a free knob, not a queue with a cap.
+
+With `adversary_strategy="selfish"` in the engine, the whole loop is present: the attack orphans
+honest blocks in runs, the survivors queue for the `U` uncle slots of each canonical block, and
+whatever does not drain within `W` ages out. This sweeps the cap against the attack to find the
+smallest `U` that still recovers, and compares it to the honest rule `U = ceil(rho) + 1`.
+
+Three quantities separate the two failure modes the item conflates:
+
+  * ``p_ref_honest`` — of the honest blocks the attacker orphaned, how many got referenced at
+    all. Falls for TWO different reasons, which is why the next column matters.
+  * ``deep_ref_share`` — the share of examined references rejected by the first-fork rule. An
+    override discards a *chain*, and only its first block is countable (§2.1), so this isolates
+    "unreferenceable by construction" from "queue too small".
+  * ``D_hat/D`` — what the estimator actually lands on, the thing the cap is sized to protect.
+
+If raising `U` lifts recovery, the cap is the binding constraint and item 5 needs a bigger
+number. If it does not, the loss is structural and no cap buys it back.
+
+Run:  python scripts/selfish_uncle_margin.py   (writes runs/selfish_uncle_margin.parquet)
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+from joblib import Parallel, delayed
+
+from tsi_sim.config import SimConfig
+from tsi_sim.engine import run_trajectory
+from tsi_sim.memguard import ArrivalMatrixTooLarge
+
+HERE = Path(__file__).resolve().parent.parent
+RUNS = HERE / "runs"
+RUNS.mkdir(exist_ok=True)
+
+EPOCHS = 16
+REPS = 8
+N_JOBS = 6
+
+BASE = dict(n_nodes=1000, stake_dist="pareto", topology="blend", degree=6,
+            link_latency_mean=0.5, link_latency_dist="geo", blend_hops=3,
+            k=256, epochs=EPOCHS, genesis_d_factor=0.5, early_stop=False,
+            adversary_strategy="selfish")
+
+ALPHAS = [0.0, 0.2, 0.3, 0.4]
+DELAYS = [8.0, 16.0]          # rho ~ 0.56 (design point) and ~1.0 (the load boundary)
+CAPS = [1, 2, 3, 4]           # spec allows up to MAX_UNCLES = 4
+WINDOWS = [10, 20]            # W = 10/f (recommended) and the 20/f widening of §3.4
+
+
+def _cell(alpha: float, delay: float, u: int, w: int, rep: int) -> dict:
+    cfg = SimConfig(**BASE, blend_delay_max=delay, max_uncles=u, window_absorption=w,
+                    adversary_frac=alpha, replicate=rep)
+    row = dict(alpha=alpha, blend_delay_max=delay, max_uncles=u, window_absorption=w, rep=rep)
+    try:
+        t = pd.DataFrame(run_trajectory(cfg))
+        t = t[t.epoch >= EPOCHS // 2]
+        row |= dict(collapsed=False,
+                    mean_ratio=float(t.mean_ratio.mean()),
+                    fork_rate=float(t.fork_rate.mean()),
+                    p_ref=float(t.p_ref.mean()),
+                    p_ref_honest=float(t.p_ref_honest.mean()),
+                    deep_ref_share=float(t.deep_ref_share.mean()),
+                    max_reorg_depth=int(t.max_reorg_depth.max()),
+                    adv_share=float(t.adv_blocks.sum()
+                                    / max(t.adv_blocks.sum() + t.honest_blocks.sum(), 1)))
+    except ArrivalMatrixTooLarge:
+        row |= dict(collapsed=True)
+    return row
+
+
+def sweep() -> pd.DataFrame:
+    jobs = [(a, d, u, w, r) for a in ALPHAS for d in DELAYS for u in CAPS
+            for w in WINDOWS for r in range(REPS)]
+    df = pd.DataFrame(Parallel(n_jobs=N_JOBS, backend="loky", inner_max_num_threads=1)(
+        delayed(_cell)(a, d, u, w, r) for a, d, u, w, r in jobs))
+    df.to_parquet(RUNS / "selfish_uncle_margin.parquet", index=False)
+    return df
+
+
+BAR = 0.98        # the §3.6 recovery bar, as a fraction of the true stake
+
+
+def report(df: pd.DataFrame) -> None:
+    ok = df[~df.collapsed]
+    for w in WINDOWS:
+        print(f"\n=== W = {w} block-intervals ===")
+        print(f"{'delta':>6} {'alpha':>6} | " + "  ".join(f"U={u}" for u in CAPS)
+              + " |  smallest U >= bar   p_ref_h  deep_ref  fork")
+        for d in DELAYS:
+            for a in ALPHAS:
+                g = ok[(ok.window_absorption == w) & (ok.blend_delay_max == d) & (ok.alpha == a)]
+                if g.empty:
+                    continue
+                cells, best = [], None
+                for u in CAPS:
+                    gu = g[g.max_uncles == u]
+                    m = gu.mean_ratio.mean() if len(gu) else np.nan
+                    cells.append(f"{m:.3f}")
+                    if best is None and m >= BAR:
+                        best = u
+                ref = g[g.max_uncles == max(CAPS)]
+                print(f"{d:6.1f} {a:6.2f} | " + "  ".join(cells)
+                      + f" |  {str(best):>4}            {ref.p_ref_honest.mean():7.3f}"
+                      + f"  {ref.deep_ref_share.mean():8.3f}  {ref.fork_rate.mean():5.3f}")
+    n_col = int(df.collapsed.sum())
+    if n_col:
+        print(f"\n{n_col} of {len(df)} runs collapsed into the §6.2 branch (excluded above)")
+
+
+def main() -> None:
+    print(f"=== selfish uncle-margin sweep ({len(ALPHAS)*len(DELAYS)*len(CAPS)*len(WINDOWS)*REPS}"
+          f" runs; recovery bar {BAR}) ===")
+    report(sweep())
+    print(f"\nwrote {RUNS}/selfish_uncle_margin.parquet")
+
+
+if __name__ == "__main__":
+    main()

tools/simulators/tsi/tsi-sim-pernode/src/tsi_sim/blocktree.py

@@ -174,6 +174,76 @@ def _max_span_blocks(active_slots: np.ndarray, counts: np.ndarray, span: float)
     return best
 
 
+class _SelfishCoalition:
+    """Eyal–Sirer SM1 private-chain state, driven from the coalition's shared view.
+
+    The coalition mines one private chain and releases it under the classic SM1 rules, in terms
+    of ``a`` = unreleased private blocks since the fork and ``h`` = public blocks since the fork
+    as the coalition sees them:
+
+        h > a            adopt    — the public chain won; the private blocks are dead
+        h == a  (a > 0)  match    — release all; the two chains race at equal length
+        h == a - 1 (a>=2) override — release all; the public ``h`` blocks are orphaned
+        h < a - 1        wait     — stay hidden and keep the lead
+
+    Only *visibility* is modelled here; the coalition's **mining** needs no special case. A
+    coalition member's fork choice already builds on the private tip whenever the private chain
+    leads, because that tip has the greatest height among the blocks that member can see — and
+    it falls back to the public chain exactly when the public chain overtakes, which is the
+    "adopt" branch. So the private chain forms, extends and is abandoned emergently.
+
+    The coalition is treated as one entity that shares its view instantly: a member learns a
+    public block as soon as the *earliest* member does, and sees every private block at its
+    production slot. Both are best-case-for-the-adversary, which is the direction we want for a
+    bound on the damage.
+    """
+
+    def __init__(self, coal_idx: np.ndarray, n_blocks: int, E: int) -> None:
+        self.coal = coal_idx
+        self.priv: list[int] = []          # unreleased private blocks, oldest first
+        self.fork_height = 0               # height of the block the private chain forked from
+        self.unreleased = np.zeros(n_blocks, dtype=bool)
+        self.coal_arr = np.full(n_blocks, float(E) + 1.0)   # when the coalition learns of a block
+        self.coal_arr[0] = 0.0
+        self.n_released = 0                # blocks made public by a release
+        self.n_abandoned = 0               # private blocks the coalition gave up on
+        self.n_override = 0                # releases that orphaned >=1 honest block
+
+    def note_block(self, b: int, arrival_at_coalition: float) -> None:
+        self.coal_arr[b] = arrival_at_coalition
+
+    def add_private(self, b: int, t: int, parent_height: int) -> None:
+        if not self.priv:                  # opening a new private chain: record its fork height
+            self.fork_height = parent_height
+        self.priv.append(b)
+        self.unreleased[b] = True
+        self.coal_arr[b] = float(t)        # shared inside the coalition immediately
+
+    def public_height(self, t: int, height: np.ndarray, nb: int) -> int:
+        """Best height the coalition can see on the PUBLIC chain (private blocks excluded)."""
+        vis = (self.coal_arr[:nb] <= t) & (~self.unreleased[:nb])
+        return int(height[:nb][vis].max()) if vis.any() else 0
+
+    def decide(self, t: int, height: np.ndarray, nb: int) -> list[int]:
+        """Apply the SM1 rule; return the private blocks to release now (possibly empty)."""
+        a = len(self.priv)
+        if a == 0:
+            return []
+        h = self.public_height(t, height, nb) - self.fork_height
+        if h > a:                                  # adopt: the public chain won outright
+            self.n_abandoned += a
+            self.priv.clear()
+            return []
+        if h == a or (h == a - 1 and a >= 2):      # match / override: publish the whole chain
+            out = self.priv
+            self.priv = []
+            self.n_released += len(out)
+            if h >= 1:
+                self.n_override += 1
+            return out
+        return []                                  # wait
+
+
 def build_tree_pernode(
     active_slots: np.ndarray,
     winners_per_slot: list[np.ndarray],
@@ -237,7 +307,12 @@ def build_tree_pernode(
     key[0] = np.int64(0) * c1 - np.int64(-1) * c2 - np.int64(0)
     NEG = np.iinfo(np.int64).min
 
-    windowed = bool(config.windowed_fork_choice)
+    selfish = (adversary_mask is not None and config.adversary_frac > 0.0
+               and config.adversary_strategy == "selfish")
+    # A private chain breaks the windowed horizon's premise: an unreleased block is old enough to
+    # be "fully propagated" while no honest node has it, and it becomes visible LATER (on release),
+    # which the one-way frontier pointer can never revisit. So selfish runs the exact full scan.
+    windowed = bool(config.windowed_fork_choice) and not selfish
     if not windowed:
         horizon = float(E)                             # full scan (gb unused)
     elif config.topology == "blend":
@@ -264,6 +339,7 @@ def build_tree_pernode(
     withholding = (adversary_mask is not None and config.adversary_frac > 0.0
                    and config.adversary_strategy == "withhold")
     if config.prune_arrival and windowed and config.jitter_mean == 0.0 and not withholding:
+        # (selfish already cleared `windowed`, so it never reaches the pruned path either)
         return _build_pruned(active_slots, winners_per_slot, path_latency, config, rng,
                              slot, parent, height, leader, uncles, key, c1, c2,
                              float(horizon), n_blocks, E, n, adversary_mask)
@@ -289,6 +365,8 @@ def build_tree_pernode(
     gb_id = 0
     fp_idx = 1          # frontier pointer over fully-propagated blocks
 
+    coalition = _SelfishCoalition(np.nonzero(adversary_mask)[0], n_blocks, E) if selfish else None
+
     nb = 1
     for si in range(active_slots.shape[0]):
         t = int(active_slots[si])
@@ -338,12 +416,46 @@ def build_tree_pernode(
             if hide:
                 A[:, b] = float(E) + 1.0                     # withheld: never arrives -> orphan
                 withheld[b] = True
+            elif coalition is not None and adv:
+                # Private: visible to the whole coalition at once, invisible to everyone else
+                # until released. Kept off the honest side by the same sentinel `withhold` uses.
+                A[:, b] = float(E) + 1.0
+                A[coalition.coal, b] = max(float(t), float(A[v, p_id]))
+                withheld[b] = True                           # flipped back on release
+                coalition.add_private(b, t, int(height[p_id]))
             else:
                 np.maximum(col, A[:, p_id], out=col)
                 A[:, b] = col
                 A[v, b] = max(float(t), float(A[v, p_id]))   # producer sees own block at its slot
+                if coalition is not None:
+                    coalition.note_block(b, float(A[coalition.coal, b].min()))
             nb += 1
 
+        if coalition is not None:
+            for rb in coalition.decide(t, height, nb):
+                # Release by DIRECT gossip from the producer, bypassing the Blend cascade: the
+                # adversary has no privacy budget to respect and wants the race won, so this is
+                # its fastest legal publication. Oldest first, so each block's parent arrival is
+                # already final when the no-earlier-than-parent clamp is applied.
+                prod = int(leader[rb])
+                rel = float(t) + path_latency[prod]
+                np.maximum(rel, A[:, int(parent[rb])], out=rel)
+                np.minimum(rel, A[:, rb], out=rel)           # coalition already had it privately
+                A[:, rb] = rel
+                withheld[rb] = False
+                coalition.unreleased[rb] = False
+
+    if coalition is not None and coalition.priv:
+        # Private blocks still hidden when the epoch ends are abandoned: the race they were held
+        # for is over, so they can never be cashed in. Hide them from the coalition too, or the
+        # canonical-tip search (which takes the best tip ANY node holds) would crown a chain no
+        # honest node ever saw and credit it phantom blocks.
+        stranded = np.array(coalition.priv, dtype=np.int64)
+        A[:, stranded] = float(E) + 1.0
+        withheld[stranded] = True
+        coalition.n_abandoned += len(coalition.priv)
+        coalition.priv.clear()
+
     tree = BlockTree(slot=slot, parent=parent, height=height, leader=leader, uncles=uncles)
     return tree, A
 

tools/simulators/tsi/tsi-sim-pernode/src/tsi_sim/config.py

@@ -30,7 +30,12 @@ InitDest = Literal["common", "heterogeneous"]
 #  "withhold" — never gossips its blocks (they are orphaned, its won slots become gaps in the
 #     canonical chain), so the counted density drops ~adversary_frac and TSI deflates D_est toward
 #     the reduced ACTIVE stake. Stronger, but the withheld blocks earn nothing (griefing/grinding).
-AdversaryStrategy = Literal["suppress", "withhold"]
+#  "selfish" — mines a PRIVATE chain and releases it to orphan honest blocks (Eyal-Sirer SM1).
+#     Unlike "withhold" (which discards its blocks — abstention, a dead loss), this recovers the
+#     forfeit by displacing honest work, and is the one profitable lever (report §6.6). Its
+#     estimator damage is what the countable uncle rule can only partly repair, because an
+#     override discards a CHAIN of honest blocks and only the first is referenceable (§2.1).
+AdversaryStrategy = Literal["suppress", "withhold", "selfish"]
 # WHICH nodes make up that coalition, at the same total stake:
 #  "random" — a uniformly random set grown until its stake reaches adversary_frac (the default; the
 #     block share is then smooth in adversary_frac, which is all the density levers depend on);
@@ -270,8 +275,8 @@ class SimConfig:
         if self.adversary_selection not in ("random", "whale"):
             raise ValueError(f"adversary_selection must be random|whale, got "
                              f"{self.adversary_selection!r}")
-        if self.adversary_strategy not in ("suppress", "withhold"):
-            raise ValueError(f"adversary_strategy must be suppress|withhold, got "
+        if self.adversary_strategy not in ("suppress", "withhold", "selfish"):
+            raise ValueError(f"adversary_strategy must be suppress|withhold|selfish, got "
                              f"{self.adversary_strategy!r}")
         checks = {
             "n_nodes": self.n_nodes >= 1,

tools/simulators/tsi/tsi-sim-pernode/src/tsi_sim/epoch.py

@@ -30,6 +30,7 @@ class EpochResult:
     max_reorg_depth: int      # deepest maximal orphan branch (blocks a reorg would discard)
     mean_reorg_depth: float   # mean maximal-orphan-branch depth
     p_ref: float              # emergent reference rate: in-window orphans referenced as uncles
+    p_ref_honest: float       # ...restricted to orphans produced OUTSIDE the coalition
     deep_ref_share: float     # share of examined references rejected by the parent-on-chain
                               # (first-fork) counting rule; 0 under the old model
 
@@ -112,7 +113,8 @@ def simulate_epoch(
 
     attribution = coalition_mask if coalition_mask is not None else adversary_mask
     adv_blocks, honest_blocks = _canonical_producer_split(tree, A, attribution, T, E)
-    fork_rate, max_reorg_depth, mean_reorg_depth, p_ref = fork.fork_stats(tree, A, T, cutoff=E)
+    fork_rate, max_reorg_depth, mean_reorg_depth, p_ref, p_ref_honest = fork.fork_stats(
+        tree, A, T, cutoff=E, coalition_mask=attribution)
     ref_total = int(ms.ref_total.sum())
     deep_ref_share = (int(ms.ref_deep.sum()) / ref_total) if ref_total else 0.0
 
@@ -123,5 +125,5 @@ def simulate_epoch(
         mean_orphan_rate=float(ms.orphan_rate.mean()),
         adv_blocks=adv_blocks, honest_blocks=honest_blocks,
         fork_rate=fork_rate, max_reorg_depth=max_reorg_depth, mean_reorg_depth=mean_reorg_depth,
-        p_ref=p_ref, deep_ref_share=deep_ref_share,
+        p_ref=p_ref, p_ref_honest=p_ref_honest, deep_ref_share=deep_ref_share,
     )

tools/simulators/tsi/tsi-sim-pernode/src/tsi_sim/fork.py

@@ -18,17 +18,24 @@ import numpy as np
 from .blocktree import BlockTree
 
 
-def fork_stats(tree: BlockTree, A, T: int, cutoff: int) -> tuple[float, int, float, float]:
-    """Return ``(fork_rate, max_reorg_depth, mean_reorg_depth, p_ref)`` over in-window blocks.
+def fork_stats(tree: BlockTree, A, T: int, cutoff: int,
+               coalition_mask=None) -> tuple[float, int, float, float, float]:
+    """Return ``(fork_rate, max_reorg_depth, mean_reorg_depth, p_ref, p_ref_honest)``.
 
     ``p_ref`` is the emergent **reference rate**: the fraction of in-window orphans that some
     canonical block references as an uncle — the quantity the §6.8 soft-inclusion argument
     assumes is high. ``A`` is the arrival matrix (full ``np.ndarray`` or pruned): only used to
     exclude withheld blocks (which reach no node) from canonical-tip selection.
+
+    ``p_ref_honest`` restricts that to orphans produced by nodes OUTSIDE ``coalition_mask``.
+    Under a private-chain attack the two diverge and only the honest one measures the repair the
+    report credits to uncle counting: an attacker's own discarded blocks are its loss to bear,
+    and counting them would flatter `p_ref` with orphans nobody is owed. Equal to ``p_ref`` when
+    no mask is given.
     """
     nb = tree.n_blocks
     if nb <= 1:
-        return 0.0, 0, 0.0, 1.0
+        return 0.0, 0, 0.0, 1.0, 1.0
     ids = np.arange(nb)
     if isinstance(A, np.ndarray):
         arrived = (A <= cutoff).any(axis=0)
@@ -48,7 +55,7 @@ def fork_stats(tree: BlockTree, A, T: int, cutoff: int) -> tuple[float, int, flo
     in_win = (tree.slot >= 0) & (tree.slot < T)
     total = int(in_win.sum())
     if total == 0:
-        return 0.0, 0, 0.0, 1.0
+        return 0.0, 0, 0.0, 1.0, 1.0
 
     # depth[b] = length of the non-canonical run ending at b (0 if canonical). Parent-before-child
     # holds because a block\'s parent has a strictly smaller id (built earlier).
@@ -75,4 +82,11 @@ def fork_stats(tree: BlockTree, A, T: int, cutoff: int) -> tuple[float, int, flo
             referenced[u] = True
     ref_orphans = int((orphan_in_win & referenced).sum())
     p_ref = ref_orphans / n_orphan if n_orphan else 1.0
-    return fork_rate, max_depth, mean_depth, p_ref
+
+    if coalition_mask is None:
+        p_ref_honest = p_ref
+    else:
+        honest_orphan = orphan_in_win & ~np.asarray(coalition_mask)[tree.leader]
+        n_ho = int(honest_orphan.sum())
+        p_ref_honest = (int((honest_orphan & referenced).sum()) / n_ho) if n_ho else 1.0
+    return fork_rate, max_depth, mean_depth, p_ref, p_ref_honest

tools/simulators/tsi/tsi-sim-pernode/src/tsi_sim/metrics.py

@@ -64,6 +64,7 @@ def divergence_row(
         max_reorg_depth=er.max_reorg_depth,
         mean_reorg_depth=er.mean_reorg_depth,
         p_ref=er.p_ref,
+        p_ref_honest=er.p_ref_honest,
         deep_ref_share=er.deep_ref_share,
     )
     return row

tools/simulators/tsi/tsi-sim-pernode/tests/test_fork.py

@@ -23,14 +23,14 @@ def make_tree(slots, parents, heights):
 def test_no_forks():
     # a straight chain 1->2->3, no orphans
     tree = make_tree([-1, 0, 1, 2], [-1, 0, 1, 2], [0, 1, 2, 3])
-    fr, mx, mn, pr = fork_stats(tree, None, T=10, cutoff=100)
+    fr, mx, mn, pr, _ = fork_stats(tree, None, T=10, cutoff=100)
     assert fr == 0.0 and mx == 0 and mn == 0.0
 
 
 def test_single_orphan_depth_one():
     # canonical 1(s0),2(s1),4(s3); orphan 3(s2) hangs off block1 -> branch depth 1
     tree = make_tree([-1, 0, 1, 2, 3], [-1, 0, 1, 1, 2], [0, 1, 2, 2, 3])
-    fr, mx, mn, pr = fork_stats(tree, None, T=10, cutoff=100)
+    fr, mx, mn, pr, _ = fork_stats(tree, None, T=10, cutoff=100)
     assert mx == 1
     assert abs(fr - 1 / 4) < 1e-9        # 1 orphan of 4 in-window blocks
 
@@ -39,7 +39,7 @@ def test_deep_orphan_branch():
     # canonical spine 1..3 (heights 1,2,3); a 2-deep orphan branch 4->5 off block1
     # blocks: 0 gen; 1(s0,h1),2(s1,h2),3(s2,h3) canonical; 4(s1,h2)->1, 5(s2,h3)->4 orphan
     tree = make_tree([-1, 0, 1, 2, 1, 2], [-1, 0, 1, 2, 1, 4], [0, 1, 2, 3, 2, 3])
-    fr, mx, mn, pr = fork_stats(tree, None, T=10, cutoff=100)
+    fr, mx, mn, pr, _ = fork_stats(tree, None, T=10, cutoff=100)
     assert mx == 2                        # branch 4->5 is 2 deep
     assert abs(fr - 2 / 5) < 1e-9         # 2 orphans of 5
 

tools/simulators/tsi/tsi-sim-pernode/tests/test_selfish_engine.py

@@ -0,0 +1,174 @@
+"""The private-chain (SM1) adversary inside the per-node engine (§6.6, open item 5).
+
+§6.8 recorded that "the per-node engine has no private-chain strategy", which is why the
+selfish results came from the global race model with uncle recovery as a free knob. These
+tests pin the engine version: that it leaves every honest result untouched, that its blocks
+are conserved, and that it actually orphans honest work rather than merely hiding its own.
+"""
+
+import numpy as np
+import pytest
+
+from tsi_sim.blocktree import build_tree_pernode
+from tsi_sim.config import SimConfig
+from tsi_sim.engine import _adversary_mask, run_trajectory
+from tsi_sim.rng import rng_for
+from tsi_sim.stake import make_stake
+
+BASE = dict(n_nodes=200, stake_dist="pareto", topology="blend", degree=6,
+            link_latency_mean=0.5, link_latency_dist="geo", blend_hops=3, blend_delay_max=8.0,
+            max_uncles=2, k=32, epochs=4, genesis_d_factor=0.5, early_stop=False)
+
+
+def _traj(**over):
+    return run_trajectory(SimConfig(**{**BASE, **over}))
+
+
+def test_selfish_at_zero_stake_keeps_the_honest_fast_paths():
+    """With no coalition, `selfish` must not disturb the honest engine at all.
+
+    Note it is NOT bit-identical to `suppress` at frac = 0: adversary_strategy sits in the base
+    RNG key, so switching it reseeds the run even though the field is inert without a coalition.
+    That is pre-existing and harmless (both are valid honest runs), so the invariant worth
+    pinning is the one that protects committed results — that the windowed fork choice and the
+    arrival prune, which `selfish` disables when it IS active, stay enabled and stay exact here.
+    """
+    exact = _traj(adversary_strategy="selfish", adversary_frac=0.0,
+                  windowed_fork_choice=False, prune_arrival=False)
+    fast = _traj(adversary_strategy="selfish", adversary_frac=0.0)
+    assert [r["mean_ratio"] for r in fast] == [r["mean_ratio"] for r in exact]
+    assert max(r["range_ratio"] for r in fast) == 0.0        # honest run: nodes agree exactly
+
+
+def test_selfish_key_is_distinct_from_the_other_strategies():
+    # adversary_strategy already sits in the base key, so no historical seed moves; this just
+    # pins that the new value is not silently aliased onto an existing stream.
+    keys = {s: SimConfig(**BASE, adversary_frac=0.3, adversary_strategy=s).key()
+            for s in ("suppress", "withhold", "selfish")}
+    assert len(set(keys.values())) == 3
+
+
+def test_selfish_is_deterministic():
+    a = _traj(adversary_frac=0.3, adversary_strategy="selfish")
+    b = _traj(adversary_frac=0.3, adversary_strategy="selfish")
+    assert [r["mean_ratio"] for r in a] == [r["mean_ratio"] for r in b]
+
+
+def _tree(**over):
+    cfg = SimConfig(**{**BASE, **over})
+    stake = make_stake(cfg, rng_for(cfg))
+    mask = _adversary_mask(cfg, stake)
+    from tsi_sim import lottery, topology
+    root = __import__("tsi_sim.rng", fromlist=["seedseq_for"]).seedseq_for(cfg)
+    kids = root.spawn(cfg.epochs + 3)
+    pl = topology.build_path_latency(cfg, np.random.default_rng(kids[1]))
+    d_est = np.full(cfg.n_nodes, cfg.genesis_d_factor * stake.sum())
+    p = lottery.win_probs(stake, d_est, cfg.f)
+    ws, wn = lottery.sample_wins(p, cfg.epoch_len, np.random.default_rng(kids[3]))
+    slots, groups = lottery.group_by_slot(ws, wn)
+    tree, A = build_tree_pernode(slots, groups, pl, cfg, np.random.default_rng(kids[4]),
+                                 adversary_mask=mask)
+    return cfg, tree, A, mask
+
+
+def test_private_blocks_are_invisible_to_honest_nodes_while_hidden():
+    cfg, tree, A, mask = _tree(adversary_frac=0.3, adversary_strategy="selfish")
+    E = cfg.epoch_len
+    honest = ~mask
+    # Every block is either public (some honest node has it) or hidden from ALL honest nodes.
+    reaches_honest = (A[honest] <= E).any(axis=0)
+    hidden = ~reaches_honest
+    hidden[0] = False
+    # a hidden block is never a partial leak: no honest node holds it
+    assert not (A[honest][:, hidden] <= E).any()
+    # and every hidden block was produced by the coalition, never by an honest node
+    assert mask[tree.leader[hidden]].all()
+
+
+def test_released_blocks_never_precede_their_parent():
+    # The release path applies its own no-earlier-than-parent clamp; a violation would let a
+    # node build on a child before its parent and corrupt the tree.
+    cfg, tree, A, mask = _tree(adversary_frac=0.3, adversary_strategy="selfish")
+    for b in range(1, tree.n_blocks):
+        p = int(tree.parent[b])
+        assert (A[:, b] >= A[:, p] - 1e-9).all(), f"block {b} precedes parent {p}"
+
+
+def _honest_orphans_in_window(cfg, tree, A, mask) -> int:
+    """In-window orphans produced by NON-coalition nodes — the displaced honest work."""
+    E, T = cfg.epoch_len, cfg.period_T
+    nb = tree.n_blocks
+    ids = np.arange(nb)
+    arrived = (A <= E).any(axis=0)
+    arrived[0] = True
+    h = np.where(arrived, tree.height, np.iinfo(np.int64).min)
+    best = int(np.lexsort((-ids, -tree.slot, h))[-1])
+    canonical = np.zeros(nb, dtype=bool)
+    b = best
+    while b > 0:
+        canonical[b] = True
+        b = int(tree.parent[b])
+    in_win = (tree.slot >= 0) & (tree.slot < T)
+    return int((in_win & ~canonical & ~mask[tree.leader]).sum())
+
+
+def test_selfish_displaces_honest_work_where_withholding_only_hides_its_own():
+    # This is the distinction between the two levers, and the reason only one of them is
+    # profitable: withholding discards the coalition's OWN blocks (a dead loss, and honest
+    # blocks keep their places), while a private chain overrides HONEST blocks off the chain.
+    # Compare the honest orphan count at matched stake -- not fork_rate, which counts the
+    # withholder's own vanished blocks as orphans too and so runs high for the wrong reason.
+    kw = dict(adversary_frac=0.4, max_uncles=0)
+    cfg_s, tree_s, A_s, mask_s = _tree(adversary_strategy="selfish", **kw)
+    cfg_w, tree_w, A_w, mask_w = _tree(adversary_strategy="withhold", **kw)
+    assert (_honest_orphans_in_window(cfg_s, tree_s, A_s, mask_s)
+            > _honest_orphans_in_window(cfg_w, tree_w, A_w, mask_w))
+
+
+def test_selfish_deflates_the_estimate_below_the_honest_baseline():
+    tail = slice(2, None)
+    honest = np.mean([r["mean_ratio"] for r in _traj(max_uncles=0)[tail]])
+    selfish = np.mean([r["mean_ratio"] for r in
+                       _traj(adversary_frac=0.35, adversary_strategy="selfish",
+                             max_uncles=0)[tail]])
+    assert selfish < honest
+
+
+def test_uncle_counting_repairs_part_of_the_selfish_deflation():
+    # The §6.6 claim, now measurable in the engine rather than through the free knob eta:
+    # uncles recover some of the loss, and (per §6.6/fig36) not all of it.
+    tail = slice(2, None)
+    d0 = np.mean([r["mean_ratio"] for r in
+                  _traj(adversary_frac=0.35, adversary_strategy="selfish", max_uncles=0)[tail]])
+    d2 = np.mean([r["mean_ratio"] for r in
+                  _traj(adversary_frac=0.35, adversary_strategy="selfish", max_uncles=2)[tail]])
+    assert d2 > d0
+
+
+def test_p_ref_honest_defaults_to_p_ref_without_a_coalition():
+    from tsi_sim.fork import fork_stats
+    cfg, tree, A, _ = _tree()
+    *_, p_ref, p_ref_h = fork_stats(tree, A, cfg.period_T, cutoff=cfg.epoch_len)
+    assert p_ref == p_ref_h
+
+
+@pytest.mark.parametrize("strategy", ["selfish", "withhold"])
+def test_hidden_blocks_are_excluded_from_the_canonical_chain(strategy):
+    # A chain no honest node ever saw must not be crowned canonical, or it would collect
+    # phantom rewards and phantom density.
+    from tsi_sim.epoch import _canonical_producer_split
+    cfg, tree, A, mask = _tree(adversary_frac=0.4, adversary_strategy=strategy)
+    E, T = cfg.epoch_len, cfg.period_T
+    adv, hon = _canonical_producer_split(tree, A, mask, T, E)
+    reaches_honest = (A[~mask] <= E).any(axis=0)
+    # walk the chosen canonical tip: every block on it is public
+    ids = np.arange(tree.n_blocks)
+    arrived = (A <= E).any(axis=0)
+    arrived[0] = True
+    h = np.where(arrived, tree.height, np.iinfo(np.int64).min)
+    best = int(np.lexsort((-ids, -tree.slot, h))[-1])
+    b = best
+    while b > 0:
+        assert reaches_honest[b], f"canonical block {b} was never public"
+        b = int(tree.parent[b])
+    assert adv + hon > 0