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:
@@ -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,
|
||||||
|
}
|
||||||
@@ -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": {
|
||||||
|
|||||||
+172
-41
@@ -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,
|
||||||
|
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.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)."""
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in New Issue
Block a user