"""R7 — Path selector that rebuilds circuits under churn. The selector consumes a churn schedule (``churn.py``) and, whenever a live circuit loses a node to a kill, **rebuilds** it from the currently-live pool. Three strategies, all deterministic and offline: * ``static`` — canonical-order pick (stable, no randomness); * ``random`` — seeded PATH-stream pick from the live pool; * ``agent`` — a *local* stability heuristic (prefers nodes that have churned least). The paid frontier-model agent arm (GOAL autonomy envelope (c)) is **human-gated and intentionally NOT wired here** — this strategy spends nothing and calls no external model; it is the offline stand-in used to validate the rebuild loop. The R7 acceptance predicate this satisfies: under a fixed churn seed the selector rebuilds every dropped circuit (as long as the live pool can supply ``hops`` nodes). Rebuild activity is emitted as R3 ``rebuild_start``/``rebuild_done`` + ``churn_kill``/``churn_spawn`` events when an :class:`EventLog` is supplied. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Dict, List, Optional from cmd_chat.sor.churn import ChurnEvent from cmd_chat.sor.config import Domain, SorRng from cmd_chat.sor.events import EventLog STRATEGIES = ("static", "random", "agent") @dataclass class Rebuild: """One rebuild triggered by a drop: the step, the node that dropped, and the replacement circuit (node ids in order).""" t: int dropped: str circuit: List[str] @dataclass class SelectionResult: """Outcome of replaying a churn schedule under a selector strategy.""" strategy: str seed: int hops: int initial_circuit: List[str] rebuilds: List[Rebuild] = field(default_factory=list) drops: int = 0 deferred: int = 0 # drops that could not be rebuilt (pool < hops at that step) events_sha256: Optional[str] = None @property def every_drop_rebuilt(self) -> bool: """True iff every drop that broke the circuit was met by a rebuild — the R7 acceptance predicate (no deferred/unhealed drops).""" return self.deferred == 0 and len(self.rebuilds) == self.drops class Selector: """Picks a ``hops``-length circuit from a live pool per its strategy. Pure and deterministic; the same (seed, strategy, live pool, churn count) always yields the same circuit.""" def __init__(self, strategy: str, seed: int, hops: int) -> None: if strategy not in STRATEGIES: raise ValueError(f"unknown selector strategy {strategy!r} (of {STRATEGIES})") self.strategy = strategy self.seed = seed self.hops = hops self._draws = 0 # advances the random stream across rebuilds self._kill_count: Dict[str, int] = {} # node -> times killed (agent signal) def note_kill(self, node: str) -> None: self._kill_count[node] = self._kill_count.get(node, 0) + 1 def build(self, live: List[str]) -> Optional[List[str]]: """Return a ``hops``-length circuit from ``live``, or ``None`` if the pool is too small (fewer than ``hops`` live nodes).""" if len(live) < self.hops: return None if self.strategy == "static": return sorted(live)[: self.hops] if self.strategy == "random": pool = sorted(live) s = SorRng(self.seed).stream(Domain.SELECTOR) for _ in range(self._draws): # replay to current position (determinism) s.next_u64() picked: List[str] = [] work = list(pool) for _ in range(self.hops): j = s.next_below(len(work)) picked.append(work.pop(j)) self._draws += 1 return picked # agent: local stability heuristic — fewest kills first, ties by id. No # external model, no spend (the paid arm is human-gated, not wired here). ranked = sorted(live, key=lambda n: (self._kill_count.get(n, 0), n)) return ranked[: self.hops] def run_selection( seed: int, nodes: List[str], hops: int, schedule: List[ChurnEvent], strategy: str = "static", log: Optional[EventLog] = None, ) -> SelectionResult: """Replay ``schedule`` over ``nodes`` under ``strategy`` and rebuild the circuit whenever a kill drops one of its nodes. Deterministic from the inputs. Returns a :class:`SelectionResult`; ``result.every_drop_rebuilt`` is the R7 acceptance predicate. If ``log`` is given, emits R3 churn/rebuild events.""" sel = Selector(strategy, seed, hops) live = {n: True for n in nodes} live_list = [n for n in nodes if live[n]] initial = sel.build(live_list) or [] circuit: List[str] = list(initial) result = SelectionResult(strategy, seed, hops, list(initial)) for ev in schedule: if ev.kind == "kill": live[ev.node] = False sel.note_kill(ev.node) if log is not None: log.emit("churn_kill", node_fp=_fp(ev.node), hop_index=ev.t) if ev.node in circuit: # The live circuit lost a hop -> must rebuild. result.drops += 1 if log is not None: log.emit("rebuild_start", circuit_id=f"t{ev.t}", hop_index=ev.t) rebuilt = sel.build([n for n in nodes if live[n]]) if rebuilt is None: result.deferred += 1 # pool too small right now circuit = [] else: circuit = rebuilt result.rebuilds.append(Rebuild(ev.t, ev.node, list(rebuilt))) if log is not None: log.emit("rebuild_done", circuit_id=f"t{ev.t}", hop_index=len(rebuilt)) else: # spawn live[ev.node] = True if log is not None: log.emit("churn_spawn", node_fp=_fp(ev.node), hop_index=ev.t) if not circuit: # A deferred drop can now be healed once the pool recovers. rebuilt = sel.build([n for n in nodes if live[n]]) if rebuilt is not None: circuit = rebuilt result.deferred = max(0, result.deferred - 1) result.rebuilds.append(Rebuild(ev.t, ev.node, list(rebuilt))) if log is not None: log.emit("rebuild_done", circuit_id=f"t{ev.t}", hop_index=len(rebuilt)) if log is not None: result.events_sha256 = log.close() return result def _fp(node: str) -> str: """A short, stable fingerprint of a node id for event metadata (never the id verbatim in case ids ever carry structure).""" import hashlib return hashlib.sha256(node.encode("utf-8")).hexdigest()[:8]