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>
This commit is contained in:
leetcrypt
2026-07-19 18:15:28 -07:00
parent 12fd2537e3
commit 0734d18989
4 changed files with 570 additions and 42 deletions
+216
View File
@@ -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,
}
+4
View File
@@ -141,6 +141,9 @@ class RunManifest:
worktree_root: Path = field(default_factory=lambda: Path.cwd()) worktree_root: Path = field(default_factory=lambda: Path.cwd())
start_utc: Optional[str] = None start_utc: Optional[str] = None
stop_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]: def to_dict(self) -> Dict[str, Any]:
if self.engine_kind not in ISOLATED_ENGINES: if self.engine_kind not in ISOLATED_ENGINES:
@@ -161,6 +164,7 @@ class RunManifest:
"kind": self.engine_kind, "kind": self.engine_kind,
"image_digest": self.engine_image_digest, "image_digest": self.engine_image_digest,
}, },
"selector_backend": self.selector_backend,
"git": capture_git(self.worktree_root), "git": capture_git(self.worktree_root),
"deps": capture_deps(self.worktree_root), "deps": capture_deps(self.worktree_root),
"timestamps": { "timestamps": {
+173 -42
View File
@@ -1,16 +1,20 @@
"""R7 — Path selector that rebuilds circuits under churn. """R7 — Path selector that rebuilds circuits under churn.
The selector consumes a churn schedule (``churn.py``) and, whenever a live circuit 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 loses a node to a kill, **rebuilds** it from the currently-live pool. The rebuild
strategies, all deterministic and offline: 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); * ``static`` — canonical-order pick (stable, no randomness);
* ``random`` — seeded PATH-stream pick from the live pool; * ``random`` — seeded SELECTOR-stream pick from the live pool;
* ``agent`` — a *local* stability heuristic (prefers nodes that have churned * ``agent`` — an agent policy. The default backend is a *local* stability
least). The paid frontier-model agent arm (GOAL autonomy envelope (c)) is heuristic (prefers nodes that have churned least). A reproducible
**human-gated and intentionally NOT wired here** — this strategy spends local-open-source-model backend (Ollama, temp=0, seed-pinned, decision-cached)
nothing and calls no external model; it is the offline stand-in used to lives in ``agent_selector.py`` and is selected with ``agent_backend="ollama"``.
validate the rebuild loop. 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 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`` 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 __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass, field 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.churn import ChurnEvent
from cmd_chat.sor.config import Domain, SorRng from cmd_chat.sor.config import Domain, SorRng
from cmd_chat.sor.events import EventLog from cmd_chat.sor.events import EventLog
STRATEGIES = ("static", "random", "agent") 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 @dataclass
@@ -52,6 +151,7 @@ class SelectionResult:
drops: int = 0 drops: int = 0
deferred: int = 0 # drops that could not be rebuilt (pool < hops at that step) deferred: int = 0 # drops that could not be rebuilt (pool < hops at that step)
events_sha256: Optional[str] = None events_sha256: Optional[str] = None
policy_provenance: Optional[Dict[str, Any]] = None
@property @property
def every_drop_rebuilt(self) -> bool: def every_drop_rebuilt(self) -> bool:
@@ -61,45 +161,41 @@ class SelectionResult:
class Selector: class Selector:
"""Picks a ``hops``-length circuit from a live pool per its strategy. Pure and """Picks a ``hops``-length circuit from a live pool via a pluggable
deterministic; the same (seed, strategy, live pool, churn count) always yields :class:`SelectorPolicy`. Pure and deterministic; the same (seed, policy, live
the same circuit.""" pool, churn history) always yields the same circuit."""
def __init__(self, strategy: str, seed: int, hops: int) -> None: def __init__(
if strategy not in STRATEGIES: self,
raise ValueError(f"unknown selector strategy {strategy!r} (of {STRATEGIES})") strategy: str,
self.strategy = strategy 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.seed = seed
self.hops = hops self.hops = hops
self._draws = 0 # advances the random stream across rebuilds self._ctx = SelectorContext(seed=seed, hops=hops)
self._kill_count: Dict[str, int] = {} # node -> times killed (agent signal)
@property
def policy(self) -> SelectorPolicy:
return self._policy
def note_kill(self, node: str) -> None: 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]]: def build(self, live: List[str]) -> Optional[List[str]]:
"""Return a ``hops``-length circuit from ``live``, or ``None`` if the pool """Return a ``hops``-length circuit from ``live``, or ``None`` if the pool
is too small (fewer than ``hops`` live nodes).""" is too small (fewer than ``hops`` live nodes)."""
if len(live) < self.hops: return self._policy.choose(list(live), self._ctx)
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( def run_selection(
@@ -109,18 +205,26 @@ def run_selection(
schedule: List[ChurnEvent], schedule: List[ChurnEvent],
strategy: str = "static", strategy: str = "static",
log: Optional[EventLog] = None, log: Optional[EventLog] = None,
*,
agent_backend: str = "heuristic",
policy: Optional[SelectorPolicy] = None,
run_dir: Optional[Path] = None,
**policy_kwargs: Any,
) -> SelectionResult: ) -> SelectionResult:
"""Replay ``schedule`` over ``nodes`` under ``strategy`` and rebuild the circuit """Replay ``schedule`` over ``nodes`` under ``strategy`` and rebuild the circuit
whenever a kill drops one of its nodes. Deterministic from the inputs. whenever a kill drops one of its nodes. Deterministic from the inputs.
Returns a :class:`SelectionResult`; ``result.every_drop_rebuilt`` is the R7 Returns a :class:`SelectionResult`; ``result.every_drop_rebuilt`` is the R7
acceptance predicate. If ``log`` is given, emits R3 churn/rebuild events.""" acceptance predicate. If ``log`` is given, emits R3 churn/rebuild events. If
sel = Selector(strategy, seed, hops) ``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 = {n: True for n in nodes}
live_list = [n for n in nodes if live[n]] live_list = [n for n in nodes if live[n]]
initial = sel.build(live_list) or [] initial = sel.build(live_list) or []
circuit: List[str] = list(initial) circuit: List[str] = list(initial)
result = SelectionResult(strategy, seed, hops, list(initial)) result = SelectionResult(sel.strategy, seed, hops, list(initial))
for ev in schedule: for ev in schedule:
if ev.kind == "kill": if ev.kind == "kill":
@@ -160,9 +264,36 @@ def run_selection(
if log is not None: if log is not None:
result.events_sha256 = log.close() 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 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: def _fp(node: str) -> str:
"""A short, stable fingerprint of a node id for event metadata (never the id """A short, stable fingerprint of a node id for event metadata (never the id
verbatim in case ids ever carry structure).""" verbatim in case ids ever carry structure)."""
+177
View File
@@ -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()