R7: pluggable SelectorPolicy seam + local-OSS (Ollama) confirmatory agent arm

Adds the RQ3 agent-selector backend as a reproducible, offline measurement
decision — not relay/data-plane traffic. Containment is untouched: the model
query is a local call to localhost:11434.

- selector.py: SelectorPolicy ABC seam; static/random/agent(heuristic) built-ins;
  make_policy() resolves (strategy, agent_backend); run_selection writes a
  write-once selector.json provenance sidecar.
- agent_selector.py: OllamaAgentPolicy — reproducible confirmatory arm
  (temp=0 + per-run seed, decisions cached keyed by (seed, state-hash),
  deterministic heuristic fallback on any model/parse failure, model weights
  digest pinned for the manifest). ClaudeExploratoryPolicy — EXPLORATORY-only
  stub: makes no paid call, refuses to serve as a confirmatory backend.
- provenance.py: additive selector_backend field so an agent-selected run pins
  the exact model id + weights digest.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

cmd_chat/sor/agent_selector.py

@@ -0,0 +1,216 @@
+"""R7 — Agent selector backends (RQ3 treatment arm).
+
+Two backends plug into the :class:`~cmd_chat.sor.selector.SelectorPolicy` seam:
+
+* :class:`OllamaAgentPolicy` — the **reproducible confirmatory** agent arm. It
+  asks a *local, open-source* model (served by Ollama on localhost) to rank the
+  live pool and pick a rebuild circuit. Reproducibility is engineered in, not
+  hoped for:
+    - inference is pinned (``temperature=0`` + per-run ``seed``);
+    - every decision is **cached keyed by (seed, state-hash)**, so a replay is
+      byte-identical regardless of any model nondeterminism;
+    - any model/parse/validation failure falls back to the deterministic local
+      heuristic, so a run never breaks and never silently drifts;
+    - the model id + weights **digest** are recorded for the provenance manifest.
+  This is a *local* call to ``localhost:11434`` — no external target, no relay
+  traffic; it is a measurement decision, not part of the data plane.
+
+* :class:`ClaudeExploratoryPolicy` — an **EXPLORATORY-only** hook for a Claude
+  Code / frontier arm. It is deliberately inert: it makes **no paid model call**
+  and refuses to run as a confirmatory backend (the paid frontier arm is human +
+  budget gated, GOAL autonomy envelope (c)). It exists so the seam is documented,
+  not so it can spend.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import urllib.error
+import urllib.request
+from typing import Any, Dict, List, Optional
+
+from cmd_chat.sor.selector import HeuristicAgentPolicy, SelectorContext, SelectorPolicy
+
+DEFAULT_MODEL = "qwen2.5:3b"
+DEFAULT_HOST = "http://localhost:11434"
+
+
+def _state_hash(live: List[str], ctx: SelectorContext) -> str:
+    """A stable hash of the decision state — the same live pool + churn history +
+    seed + hops always hashes identically, so it is a sound cache key."""
+    blob = json.dumps(
+        {
+            "seed": ctx.seed,
+            "hops": ctx.hops,
+            "live": sorted(live),
+            "kills": {n: ctx.kill_counts.get(n, 0) for n in sorted(live)},
+        },
+        sort_keys=True,
+    ).encode("utf-8")
+    return hashlib.sha256(blob).hexdigest()
+
+
+class OllamaAgentPolicy(SelectorPolicy):
+    """Local-OSS-model rebuild policy (reproducible confirmatory agent arm)."""
+
+    name = "agent-ollama"
+
+    def __init__(
+        self,
+        model: str = DEFAULT_MODEL,
+        host: str = DEFAULT_HOST,
+        timeout: float = 30.0,
+        cache: Optional[Dict[str, List[str]]] = None,
+    ) -> None:
+        self.model = model
+        self.host = host.rstrip("/")
+        self.timeout = timeout
+        self._cache: Dict[str, List[str]] = cache if cache is not None else {}
+        self._fallback = HeuristicAgentPolicy()
+        self._model_digest: Optional[str] = None
+        self._digest_resolved = False
+        self._decisions: List[Dict[str, Any]] = []
+
+    # -- provenance ------------------------------------------------------- #
+    def _resolve_digest(self) -> Optional[str]:
+        """Best-effort: fetch the model's content digest from Ollama so the exact
+        weights are pinned into the run manifest. Never raises."""
+        if self._digest_resolved:
+            return self._model_digest
+        self._digest_resolved = True
+        try:
+            req = urllib.request.Request(f"{self.host}/api/tags")
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                tags = json.loads(resp.read().decode("utf-8"))
+            for m in tags.get("models", []):
+                if m.get("name") == self.model or m.get("model") == self.model:
+                    self._model_digest = m.get("digest")
+                    break
+        except Exception:  # noqa: BLE001
+            self._model_digest = None
+        return self._model_digest
+
+    def provenance(self) -> Dict[str, Any]:
+        n_model = sum(1 for d in self._decisions if d.get("source") == "model")
+        n_cache = sum(1 for d in self._decisions if d.get("source") == "cache")
+        n_fallback = sum(1 for d in self._decisions if d.get("source") == "fallback")
+        return {
+            "policy": self.name,
+            "model": self.model,
+            "model_digest": self._resolve_digest(),
+            "host": self.host,
+            "options": {"temperature": 0, "seed": "per-run", "format": "json"},
+            "decisions": {
+                "total": len(self._decisions),
+                "model": n_model,
+                "cache": n_cache,
+                "fallback": n_fallback,
+            },
+        }
+
+    def decisions(self) -> List[Dict[str, Any]]:
+        """The per-state decision log (state-hash, chosen circuit, source) — folded
+        into the run provenance sidecar so an agent run is fully auditable."""
+        return list(self._decisions)
+
+    # -- decision --------------------------------------------------------- #
+    def choose(self, live: List[str], ctx: SelectorContext) -> Optional[List[str]]:
+        if len(live) < ctx.hops:
+            return None
+        key = f"{ctx.seed}:{_state_hash(live, ctx)}"
+        if key in self._cache:
+            chosen = list(self._cache[key])
+            self._decisions.append({"state": key, "chosen": chosen, "source": "cache"})
+            return chosen
+
+        chosen = self._query_model(live, ctx)
+        source = "model"
+        if not self._valid(chosen, live, ctx.hops):
+            chosen = self._fallback.choose(live, ctx)  # deterministic, never breaks
+            source = "fallback"
+        assert chosen is not None  # pool>=hops guaranteed above
+        self._cache[key] = list(chosen)
+        self._decisions.append({"state": key, "chosen": list(chosen), "source": source})
+        return list(chosen)
+
+    def _valid(self, chosen: Optional[List[str]], live: List[str], hops: int) -> bool:
+        if not isinstance(chosen, list) or len(chosen) != hops:
+            return False
+        if len(set(chosen)) != hops:
+            return False
+        pool = set(live)
+        return all(isinstance(c, str) and c in pool for c in chosen)
+
+    def _query_model(self, live: List[str], ctx: SelectorContext) -> Optional[List[str]]:
+        """Ask the local model for a rebuild circuit. Returns a list of node ids or
+        ``None`` on any transport/parse failure (caller falls back)."""
+        prompt = self._build_prompt(live, ctx)
+        payload = json.dumps(
+            {
+                "model": self.model,
+                "prompt": prompt,
+                "stream": False,
+                "format": "json",
+                "options": {"temperature": 0, "seed": ctx.seed, "num_predict": 160},
+            }
+        ).encode("utf-8")
+        try:
+            req = urllib.request.Request(
+                f"{self.host}/api/generate", data=payload,
+                headers={"Content-Type": "application/json"},
+            )
+            with urllib.request.urlopen(req, timeout=self.timeout) as resp:
+                body = json.loads(resp.read().decode("utf-8"))
+            text = body.get("response", "")
+            parsed = json.loads(text)
+        except (urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError):
+            return None
+        except Exception:  # noqa: BLE001 — never let the model break a run
+            return None
+        # Accept {"circuit":[...]} or a bare [...].
+        if isinstance(parsed, dict):
+            parsed = parsed.get("circuit") or parsed.get("nodes")
+        if not isinstance(parsed, list):
+            return None
+        return [str(x) for x in parsed][: ctx.hops]
+
+    def _build_prompt(self, live: List[str], ctx: SelectorContext) -> str:
+        kills = {n: ctx.kill_counts.get(n, 0) for n in sorted(live)}
+        return (
+            "You are a path selector for a research relay measurement. Choose the "
+            f"{ctx.hops} most churn-stable nodes to rebuild a circuit. A node that "
+            "has been killed more often is less stable; prefer fewer past kills, "
+            "break ties by node id ascending. Return ONLY strict JSON of the form "
+            '{"circuit": ["<id>", ...]} with exactly '
+            f"{ctx.hops} distinct ids chosen from the candidate pool.\n"
+            f"Candidate pool (id: past_kills): {json.dumps(kills, sort_keys=True)}\n"
+        )
+
+
+class ClaudeExploratoryPolicy(SelectorPolicy):
+    """EXPLORATORY-only Claude Code / frontier hook. Inert by design: it makes NO
+    paid model call and cannot serve as a confirmatory arm. The paid frontier arm
+    is human + budget gated (GOAL autonomy envelope (c)); wiring it live is a
+    deliberate, separately-authorized step — not something this build performs."""
+
+    name = "agent-claude-exploratory"
+
+    def __init__(self, allow_exploratory: bool = False) -> None:
+        self.allow_exploratory = allow_exploratory
+
+    def choose(self, live: List[str], ctx: SelectorContext) -> Optional[List[str]]:
+        raise NotImplementedError(
+            "Claude Code agent arm is EXPLORATORY-only and is not wired for "
+            "confirmatory runs: no paid-model call is made in this build (GOAL "
+            "envelope (c), human + budget gated). Use agent_backend='ollama' for "
+            "the reproducible local-model arm."
+        )
+
+    def provenance(self) -> Dict[str, Any]:
+        return {
+            "policy": self.name,
+            "status": "exploratory-stub",
+            "paid_calls": False,
+            "wired": False,
+        }

cmd_chat/sor/provenance.py

@@ -141,6 +141,9 @@ class RunManifest:
     worktree_root: Path = field(default_factory=lambda: Path.cwd())
     start_utc: Optional[str] = None
     stop_utc: Optional[str] = None
+    # R7 agent arm: model id + weights digest of the selector backend (e.g. the
+    # local Ollama model), so an agent-selected run pins the exact model it used.
+    selector_backend: Optional[Dict[str, Any]] = None
 
     def to_dict(self) -> Dict[str, Any]:
         if self.engine_kind not in ISOLATED_ENGINES:
@@ -161,6 +164,7 @@ class RunManifest:
                 "kind": self.engine_kind,
                 "image_digest": self.engine_image_digest,
             },
+            "selector_backend": self.selector_backend,
             "git": capture_git(self.worktree_root),
             "deps": capture_deps(self.worktree_root),
             "timestamps": {

cmd_chat/sor/selector.py

@@ -1,16 +1,20 @@
 """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:
+loses a node to a kill, **rebuilds** it from the currently-live pool. The rebuild
+decision is a pluggable :class:`SelectorPolicy` so the RQ3 arms can be swapped
+without touching the replay loop. Three built-in 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.
+  * ``random``  — seeded SELECTOR-stream pick from the live pool;
+  * ``agent``   — an agent policy. The default backend is a *local* stability
+    heuristic (prefers nodes that have churned least). A reproducible
+    local-open-source-model backend (Ollama, temp=0, seed-pinned, decision-cached)
+    lives in ``agent_selector.py`` and is selected with ``agent_backend="ollama"``.
+    The paid frontier-model / Claude Code arm is **human + budget gated** (GOAL
+    autonomy envelope (c)) and is only exposed as an EXPLORATORY stub that makes
+    no external call — it is never wired as a confirmatory arm here.
 
 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``
@@ -20,14 +24,109 @@ nodes). Rebuild activity is emitted as R3 ``rebuild_start``/``rebuild_done`` +
 
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
-from typing import Dict, List, Optional
+from pathlib import Path
+from typing import Any, 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")
+AGENT_BACKENDS = ("heuristic", "ollama", "claude")
+
+
+# --------------------------------------------------------------------------- #
+# Pluggable selection policy — the RQ3 arm seam.
+# --------------------------------------------------------------------------- #
+@dataclass
+class SelectorContext:
+    """State a policy may read when choosing a circuit. ``kill_counts`` is the
+    running per-node churn history; ``draws`` advances a random stream across
+    rebuilds so a stochastic policy stays deterministic within a run."""
+
+    seed: int
+    hops: int
+    kill_counts: Dict[str, int] = field(default_factory=dict)
+    draws: int = 0
+
+
+class SelectorPolicy(ABC):
+    """Chooses a ``hops``-length circuit from a live pool. Implementations must be
+    deterministic given ``(live, ctx)`` so a run is reproducible from its seed."""
+
+    name: str = "policy"
+
+    @abstractmethod
+    def choose(self, live: List[str], ctx: SelectorContext) -> Optional[List[str]]:
+        """Return a ``ctx.hops``-length circuit of distinct nodes drawn from
+        ``live``, or ``None`` if the pool is too small."""
+
+    def provenance(self) -> Dict[str, Any]:
+        """Auditable description of this policy for the run manifest sidecar."""
+        return {"policy": self.name}
+
+
+class StaticPolicy(SelectorPolicy):
+    name = "static"
+
+    def choose(self, live: List[str], ctx: SelectorContext) -> Optional[List[str]]:
+        if len(live) < ctx.hops:
+            return None
+        return sorted(live)[: ctx.hops]
+
+
+class RandomPolicy(SelectorPolicy):
+    name = "random"
+
+    def choose(self, live: List[str], ctx: SelectorContext) -> Optional[List[str]]:
+        if len(live) < ctx.hops:
+            return None
+        work = sorted(live)
+        s = SorRng(ctx.seed).stream(Domain.SELECTOR)
+        for _ in range(ctx.draws):  # replay to current position (determinism)
+            s.next_u64()
+        picked: List[str] = []
+        for _ in range(ctx.hops):
+            j = s.next_below(len(work))
+            picked.append(work.pop(j))
+            ctx.draws += 1
+        return picked
+
+
+class HeuristicAgentPolicy(SelectorPolicy):
+    """Local, model-free agent stand-in: prefer the nodes that have churned least
+    (fewest kills), ties broken by id. Spends nothing, calls nothing."""
+
+    name = "agent-heuristic"
+
+    def choose(self, live: List[str], ctx: SelectorContext) -> Optional[List[str]]:
+        if len(live) < ctx.hops:
+            return None
+        ranked = sorted(live, key=lambda n: (ctx.kill_counts.get(n, 0), n))
+        return ranked[: ctx.hops]
+
+
+def make_policy(strategy: str, agent_backend: str = "heuristic", **kwargs: Any) -> SelectorPolicy:
+    """Resolve a ``(strategy, agent_backend)`` pair to a concrete policy. The
+    ``ollama``/``claude`` agent backends are imported lazily so the base selector
+    carries no network/model dependency."""
+    if strategy == "static":
+        return StaticPolicy()
+    if strategy == "random":
+        return RandomPolicy()
+    if strategy == "agent":
+        if agent_backend == "heuristic":
+            return HeuristicAgentPolicy()
+        if agent_backend == "ollama":
+            from cmd_chat.sor.agent_selector import OllamaAgentPolicy
+            return OllamaAgentPolicy(**kwargs)
+        if agent_backend == "claude":
+            from cmd_chat.sor.agent_selector import ClaudeExploratoryPolicy
+            return ClaudeExploratoryPolicy(**kwargs)
+        raise ValueError(f"unknown agent backend {agent_backend!r} (of {AGENT_BACKENDS})")
+    raise ValueError(f"unknown selector strategy {strategy!r} (of {STRATEGIES})")
 
 
 @dataclass
@@ -52,6 +151,7 @@ class SelectionResult:
     drops: int = 0
     deferred: int = 0  # drops that could not be rebuilt (pool < hops at that step)
     events_sha256: Optional[str] = None
+    policy_provenance: Optional[Dict[str, Any]] = None
 
     @property
     def every_drop_rebuilt(self) -> bool:
@@ -61,45 +161,41 @@ class SelectionResult:
 
 
 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."""
+    """Picks a ``hops``-length circuit from a live pool via a pluggable
+    :class:`SelectorPolicy`. Pure and deterministic; the same (seed, policy, live
+    pool, churn history) 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
+    def __init__(
+        self,
+        strategy: str,
+        seed: int,
+        hops: int,
+        *,
+        agent_backend: str = "heuristic",
+        policy: Optional[SelectorPolicy] = None,
+        **policy_kwargs: Any,
+    ) -> None:
+        if policy is not None:
+            self._policy = policy
+            self.strategy = getattr(policy, "name", "custom")
+        else:
+            self._policy = make_policy(strategy, agent_backend, **policy_kwargs)
+            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)
+        self._ctx = SelectorContext(seed=seed, hops=hops)
+
+    @property
+    def policy(self) -> SelectorPolicy:
+        return self._policy
 
     def note_kill(self, node: str) -> None:
-        self._kill_count[node] = self._kill_count.get(node, 0) + 1
+        self._ctx.kill_counts[node] = self._ctx.kill_counts.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]
+        return self._policy.choose(list(live), self._ctx)
 
 
 def run_selection(
@@ -109,18 +205,26 @@ def run_selection(
     schedule: List[ChurnEvent],
     strategy: str = "static",
     log: Optional[EventLog] = None,
+    *,
+    agent_backend: str = "heuristic",
+    policy: Optional[SelectorPolicy] = None,
+    run_dir: Optional[Path] = None,
+    **policy_kwargs: Any,
 ) -> 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)
+    acceptance predicate. If ``log`` is given, emits R3 churn/rebuild events. If
+    ``run_dir`` is given, writes the policy provenance (model id/digest, decision
+    cache) to ``selector.json`` so an agent-backed run is auditable + replayable."""
+    sel = Selector(strategy, seed, hops, agent_backend=agent_backend,
+                   policy=policy, **policy_kwargs)
     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))
+    result = SelectionResult(sel.strategy, seed, hops, list(initial))
 
     for ev in schedule:
         if ev.kind == "kill":
@@ -160,9 +264,36 @@ def run_selection(
 
     if log is not None:
         result.events_sha256 = log.close()
+
+    result.policy_provenance = sel.policy.provenance()
+    if run_dir is not None:
+        _write_selector_sidecar(Path(run_dir), sel, result)
     return result
 
 
+def _write_selector_sidecar(run_dir: Path, sel: "Selector", result: SelectionResult) -> Path:
+    """Persist the policy provenance (+ any decision cache) to ``selector.json`` —
+    a write-once, hashable audit artifact folding the agent backend into the run's
+    provenance (model id/digest, per-state decisions)."""
+    import json
+
+    run_dir.mkdir(parents=True, exist_ok=True)
+    doc: Dict[str, Any] = {
+        "strategy": result.strategy,
+        "seed": result.seed,
+        "hops": result.hops,
+        "provenance": result.policy_provenance,
+    }
+    decisions = getattr(sel.policy, "decisions", None)
+    if callable(decisions):
+        doc["decisions"] = decisions()
+    path = run_dir / "selector.json"
+    if path.exists():
+        raise FileExistsError(f"selector.json already exists (immutable): {path}")
+    path.write_text(json.dumps(doc, sort_keys=True, indent=2) + "\n", encoding="utf-8")
+    return path
+
+
 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)."""