diff --git a/cmd_chat/sor/agent_selector.py b/cmd_chat/sor/agent_selector.py new file mode 100644 index 0000000..5dacc18 --- /dev/null +++ b/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": ["", ...]} 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, + } diff --git a/cmd_chat/sor/provenance.py b/cmd_chat/sor/provenance.py index 431dcbd..b8aed21 100644 --- a/cmd_chat/sor/provenance.py +++ b/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": { diff --git a/cmd_chat/sor/selector.py b/cmd_chat/sor/selector.py index 98fff0f..8b4f470 100644 --- a/cmd_chat/sor/selector.py +++ b/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).""" diff --git a/tests/test_sor_agent_selector.py b/tests/test_sor_agent_selector.py new file mode 100644 index 0000000..89d9cbf --- /dev/null +++ b/tests/test_sor_agent_selector.py @@ -0,0 +1,177 @@ +"""R7 — pluggable SelectorPolicy seam + local-OSS (Ollama) agent arm. + +Covers the RQ3 agent-backend build detail: + * the policy seam resolves static/random/agent(heuristic) identically to the + prior inline strategies (no behavioural regression); + * the Ollama agent backend picks a valid rebuild circuit, is decision-cached + (replay-deterministic), falls back to the local heuristic on any model/parse + failure (never breaks a run), and records model+digest provenance; + * the Claude Code arm is EXPLORATORY-only — it refuses to run and makes no paid + call. + +The live Ollama cell is skipped where no local server/model is present, so the +suite stays green offline. This is a *measurement* decision path (a local call to +localhost), not relay traffic — containment is untouched. +""" + +import json +import urllib.request + +import pytest + +from cmd_chat.sor.agent_selector import ( + DEFAULT_MODEL, + ClaudeExploratoryPolicy, + OllamaAgentPolicy, +) +from cmd_chat.sor.churn import churn_schedule +from cmd_chat.sor.selector import ( + HeuristicAgentPolicy, + RandomPolicy, + SelectorContext, + StaticPolicy, + make_policy, + run_selection, +) + +NODES = [f"n{i}" for i in range(8)] + + +def _ollama_ready() -> bool: + try: + with urllib.request.urlopen("http://localhost:11434/api/tags", timeout=5) as r: + tags = json.loads(r.read().decode()) + except Exception: + return False + names = {m.get("name") for m in tags.get("models", [])} + return DEFAULT_MODEL in names + + +live_ollama = pytest.mark.skipif( + not _ollama_ready(), + reason=f"requires a local Ollama server with {DEFAULT_MODEL}", +) + + +# --------------------------------------------------------------------------- # +# Policy seam. +# --------------------------------------------------------------------------- # +def test_make_policy_resolves_builtins(): + assert isinstance(make_policy("static"), StaticPolicy) + assert isinstance(make_policy("random"), RandomPolicy) + assert isinstance(make_policy("agent", "heuristic"), HeuristicAgentPolicy) + assert isinstance(make_policy("agent", "ollama"), OllamaAgentPolicy) + + +def test_make_policy_rejects_unknown(): + with pytest.raises(ValueError): + make_policy("nope") + with pytest.raises(ValueError): + make_policy("agent", "frontier-gpt") + + +def test_heuristic_prefers_least_churned(): + ctx = SelectorContext(seed=1, hops=3, kill_counts={"n0": 5, "n1": 5, "n2": 0}) + chosen = HeuristicAgentPolicy().choose(NODES, ctx) + # Nodes with 0 kills (n2..n7) outrank the twice-killed n0/n1. + assert "n0" not in chosen and "n1" not in chosen + assert len(chosen) == 3 and len(set(chosen)) == 3 + + +# --------------------------------------------------------------------------- # +# Ollama agent backend — fallback path is testable without any server. +# --------------------------------------------------------------------------- # +def test_agent_falls_back_deterministically_when_model_unreachable(): + # Point at a dead port -> every model query fails -> deterministic heuristic. + pol = OllamaAgentPolicy(host="http://127.0.0.1:1", timeout=0.5) + ctx = SelectorContext(seed=7, hops=3, kill_counts={"n0": 9}) + chosen = pol.choose(NODES, ctx) + assert chosen == HeuristicAgentPolicy().choose(NODES, ctx) + assert pol.decisions()[-1]["source"] == "fallback" + # And it is cached: a second identical call is served from cache. + again = pol.choose(NODES, ctx) + assert again == chosen + assert pol.decisions()[-1]["source"] == "cache" + + +def test_agent_validity_guard_rejects_out_of_pool(): + pol = OllamaAgentPolicy() + assert pol._valid(["n0", "n1", "n2"], NODES, 3) is True + assert pol._valid(["n0", "n0", "n2"], NODES, 3) is False # dup + assert pol._valid(["n0", "n1"], NODES, 3) is False # wrong length + assert pol._valid(["n0", "n1", "zzz"], NODES, 3) is False # not in pool + assert pol._valid("nope", NODES, 3) is False + + +def test_agent_provenance_shape(): + pol = OllamaAgentPolicy(host="http://127.0.0.1:1", timeout=0.5) + ctx = SelectorContext(seed=1, hops=3) + pol.choose(NODES, ctx) + prov = pol.provenance() + assert prov["policy"] == "agent-ollama" + assert prov["model"] == DEFAULT_MODEL + assert prov["options"]["temperature"] == 0 + assert prov["decisions"]["total"] == 1 + + +# --------------------------------------------------------------------------- # +# Claude arm — EXPLORATORY only, never spends. +# --------------------------------------------------------------------------- # +def test_claude_arm_is_exploratory_and_refuses(): + pol = ClaudeExploratoryPolicy() + with pytest.raises(NotImplementedError): + pol.choose(NODES, SelectorContext(seed=1, hops=3)) + prov = pol.provenance() + assert prov["paid_calls"] is False and prov["wired"] is False + + +def test_run_selection_agent_backend_via_string(): + # agent+heuristic through run_selection rebuilds every dropped circuit and + # records policy provenance. + seed = 0xBADC0DE + sched = churn_schedule(seed, NODES, steps=60, kill_prob_pct=35) + res = run_selection(seed, NODES, hops=3, schedule=sched, + strategy="agent", agent_backend="heuristic") + assert res.strategy == "agent" + assert res.every_drop_rebuilt + assert res.policy_provenance["policy"] == "agent-heuristic" + + +def test_run_selection_writes_selector_sidecar(tmp_path): + seed = 42 + sched = churn_schedule(seed, NODES, steps=20) + res = run_selection(seed, NODES, hops=3, schedule=sched, + strategy="agent", agent_backend="ollama", + host="http://127.0.0.1:1", timeout=0.5, + run_dir=tmp_path) + sidecar = json.loads((tmp_path / "selector.json").read_text()) + assert sidecar["provenance"]["policy"] == "agent-ollama" + assert "decisions" in sidecar # per-state audit log present + assert res.policy_provenance["model"] == DEFAULT_MODEL + + +# --------------------------------------------------------------------------- # +# Live cell — real local model (skipped if Ollama absent). +# --------------------------------------------------------------------------- # +@live_ollama +def test_live_ollama_picks_valid_circuit_and_pins_digest(): + pol = OllamaAgentPolicy() + ctx = SelectorContext(seed=123, hops=3, kill_counts={"n0": 3, "n1": 1}) + chosen = pol.choose(NODES, ctx) + assert chosen is not None and len(chosen) == 3 and len(set(chosen)) == 3 + assert all(c in set(NODES) for c in chosen) + # Cached replay is byte-identical regardless of model nondeterminism. + assert pol.choose(NODES, ctx) == chosen + # The exact model weights are pinned for provenance. + assert pol.provenance()["model_digest"] + + +@live_ollama +def test_live_ollama_run_selection_rebuilds_every_drop(tmp_path): + seed = 2024 + sched = churn_schedule(seed, NODES, steps=40, kill_prob_pct=35) + res = run_selection(seed, NODES, hops=3, schedule=sched, + strategy="agent", agent_backend="ollama", + run_dir=tmp_path) + assert res.every_drop_rebuilt + assert (tmp_path / "selector.json").exists()