feat(operator): harness-mode operator — drive a room with any function-calling model (P2)

The operator brain was Claude-only: spawn launched the claude CLI and let its
own loop drive the room. This adds the other first-class brain — a native
tool-calling loop (cmd_chat/operator/harness.py) that runs ANY function-calling
Provider (cmd_chat.ai) against an OPERATOR_TOOLS schema wired to the existing
bridge control verbs (say/exec/write/get/keys/screen/watch/manifest). No CLI
install, no creds carry.

The harness is a driver, not a new side-effect surface: every tool handler sends
the same control-socket request the hh-bridge CLI already sends, so the bridge
keeps enforcing grant-before-drive, the sandbox blast radius, and the recursion
budget. Layer-1 capabilities come from CAPABILITIES.md (the same portable
contract), and a per-agent token ceiling is enforced as a hard stop. Providers
that can't function-call are refused with a clear message rather than degrading
to prose.

New `operate` verb joins via the existing daemon socket and runs the loop with a
--profile or --provider/--model brain. Chat's native loop is left untouched
(self-contained harness over the shared Provider core, not a risky extraction).

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

cmd_chat/operator/__main__.py

@@ -109,6 +109,33 @@ def _build_parser() -> argparse.ArgumentParser:
                     help="actually launch (default is a dry-run plan)")
     sp.add_argument("--session", default=None)
 
+    # ── harness-mode operator (drive the room with any function-calling model) ──
+    op = sub.add_parser("operate",
+                        help="run a function-calling model as the operator brain "
+                             "(native loop over the bridge verbs)")
+    op.add_argument("--objective", required=True,
+                    help="what this operator should achieve in the room")
+    op.add_argument("--profile", default=None,
+                    help="named models.toml profile (e.g. local, groq-llama)")
+    op.add_argument("--provider", default=None,
+                    help="provider spec when not using --profile "
+                         "(ollama|anthropic|openai|module:Class)")
+    op.add_argument("--model", default=None, help="model name/id")
+    op.add_argument("--base-url", default=None, help="OpenAI-compatible base URL")
+    op.add_argument("--host", default=None, help="Ollama host URL")
+    op.add_argument("--stop", action="append", default=None,
+                    help="a stop condition (repeatable)")
+    op.add_argument("--max-turns", type=int, default=None,
+                    help="ceiling on model turns (default 12)")
+    op.add_argument("--token-ceiling", type=int, default=None,
+                    help="cumulative prompt+completion token hard stop (default 20000)")
+    op.add_argument("--leave-on-done", action="store_true",
+                    help="`down` the daemon when the objective is met")
+    op.add_argument("--depth", type=int, default=1)
+    op.add_argument("--fanout", type=int, default=2)
+    op.add_argument("--cost", type=float, default=5.0)
+    op.add_argument("--session", default=None)
+
     sc = sub.add_parser("screen", help="print the relayed sandbox terminal buffer")
     sc.add_argument("--tail", type=int, default=None, help="last N bytes only")
     sc.add_argument("--ansi", action="store_true", help="keep ANSI codes (raw)")
@@ -378,6 +405,79 @@ def _run_spawn(args) -> int:
     return 0 if resp.get("ok") else 1
 
 
+def _build_operate_provider(args):
+    """Resolve a Provider from --profile OR --provider/--model (+ endpoint flags).
+    Returns the provider or exits with a clear message."""
+    from cmd_chat.ai import load_profiles, make_provider, provider_from_profile, preflight
+    if args.profile:
+        profs = load_profiles()
+        prof = profs.get(args.profile)
+        if prof is None:
+            avail = ", ".join(sorted(profs)) or "(none found)"
+            print(f"no profile '{args.profile}' — available: {avail}", file=sys.stderr)
+            sys.exit(2)
+        provider = provider_from_profile(prof, name=args.profile, model=args.model,
+                                         base_url=args.base_url)
+    elif args.provider:
+        opts = {}
+        if args.base_url:
+            opts["base_url"] = args.base_url
+        if args.host:
+            opts["host"] = args.host
+        provider = make_provider(args.provider, model=args.model, **opts)
+    else:
+        print("operate needs --profile NAME or --provider SPEC --model M",
+              file=sys.stderr)
+        sys.exit(2)
+    ok, msg = preflight(provider)
+    if not ok:
+        print(f"preflight failed: {msg}", file=sys.stderr)
+        sys.exit(2)
+    return provider
+
+
+def _run_operate(args) -> int:
+    from .bootstrap import Budget
+    from .harness import OperatorHarness
+
+    sess = resolve(getattr(args, "session", None))
+    if not sess.sock_path.exists():
+        print(f"no live bridge for session '{sess.name}' (run `up` first)",
+              file=sys.stderr)
+        return 2
+    # Confirm we're connected and learn our own room name (used to filter echoes).
+    try:
+        st = request(sess.sock_path, {"op": "status"}, read_timeout=5)
+    except BridgeUnreachable as e:
+        print(str(e), file=sys.stderr)
+        return 2
+    if not st.get("connected"):
+        print("bridge is up but not connected to the room yet", file=sys.stderr)
+        return 2
+
+    provider = _build_operate_provider(args)
+    budget = Budget(depth=args.depth, fanout=args.fanout, cost_usd=args.cost)
+    kw = {}
+    if args.max_turns is not None:
+        kw["max_turns"] = args.max_turns
+    if args.token_ceiling is not None:
+        kw["token_ceiling"] = args.token_ceiling
+    harness = OperatorHarness(
+        provider=provider, sock_path=str(sess.sock_path),
+        objective=args.objective, stop=args.stop, budget=budget,
+        me=st.get("name", "operator"), **kw)
+    result = harness.run()
+    print(json.dumps({"final": result.final, "reason": result.reason,
+                      "turns": result.turns, "tokens": result.tokens,
+                      "tool_calls": result.tool_calls}, indent=2))
+    if args.leave_on_done and result.reason == "done":
+        try:
+            request(sess.sock_path, {"op": "down"}, read_timeout=5)
+        except BridgeUnreachable:
+            pass
+    return 0 if result.reason in ("done", "token-ceiling", "turn-cap") else 1
+
+
 def _run_keys(args) -> int:
     from . import sandbox as sbx
     if args.help_keys:
@@ -489,6 +589,8 @@ def main(argv: list[str] | None = None) -> int:
         return _run_keys(args)
     if verb == "spawn":
         return _run_spawn(args)
+    if verb == "operate":
+        return _run_operate(args)
     if verb == "manifest":
         return _run_manifest(args)
     if verb == "registry":

cmd_chat/operator/harness.py

@@ -0,0 +1,449 @@
+"""Harness-mode operator (Phase 2) — run *any* function-calling model as a
+hack-house operator.
+
+The CLI-subprocess operator (``bootstrap`` runner registry) spawns a vendor
+agent CLI and lets *its* own loop drive the room. This module is the other
+first-class brain: it runs the native tool-calling loop ourselves, against the
+shared Provider core (``cmd_chat.ai``), with the **operator toolset** wired to
+the bridge daemon's control verbs. So a raw Ollama / OpenAI-compatible model —
+no CLI install, no creds carry — becomes a first-class operator.
+
+It is deliberately a *driver*, not a new side-effect surface: every tool handler
+just sends the same control-socket request the `hh-bridge` CLI already sends
+(``say``/``exec``/``write``/``get``/``keys``/``screen``/``watch``/``manifest``/
+``spawn``). The bridge keeps enforcing grant-before-drive, the sandbox blast
+radius, and the recursion budget — the harness only chooses *which* verb to call.
+
+The read→think→act loop owns the **read**: it polls room events between turns and
+feeds them to the model as observations, so the model spends its turns deciding
+and acting rather than remembering to poll. Capabilities (Layer-1) come straight
+from ``CAPABILITIES.md`` via :func:`bootstrap.load_capabilities` — the same
+portable contract the Claude skill loads — so harness and CLI operators share one
+source of truth.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass, field
+from typing import Callable
+
+from cmd_chat.ai.providers import ToolsUnsupported
+
+from . import bootstrap as boot
+from .cli_client import request as _socket_request
+
+# The operator toolset — each maps 1:1 to a bridge control verb the daemon
+# already implements. Schema is the Ollama/OpenAI `tools` shape (same as the chat
+# agent's NATIVE_TOOLS), so any provider that does function calling can drive it.
+OPERATOR_TOOLS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "say",
+            "description": "Speak one line into the room. Report what you are doing "
+                           "and what you found — the room is the coordination bus.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "text": {"type": "string", "description": "The chat line to send."},
+                },
+                "required": ["text"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "exec",
+            "description": "Run a shell command in the shared sandbox out-of-band "
+                           "(does not disturb the live terminal). Returns combined "
+                           "stdout+stderr and the exit code.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "command": {"type": "string", "description": "The shell command to run."},
+                },
+                "required": ["command"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "write_file",
+            "description": "Create or overwrite a file in the sandbox with exact content.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "Destination file path."},
+                    "content": {"type": "string", "description": "Full file content."},
+                },
+                "required": ["path", "content"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "get_file",
+            "description": "Read and return the contents of a file in the sandbox.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "path": {"type": "string", "description": "File path to read."},
+                },
+                "required": ["path"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "keys",
+            "description": "Inject keystrokes into the room's shared live terminal "
+                           "(drive the PTY). Use for interactive programs; prefer "
+                           "exec for one-shot commands.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "text": {"type": "string", "description": "Literal text to type."},
+                    "enter": {"type": "boolean",
+                              "description": "Press Enter after the text (default true)."},
+                },
+                "required": ["text"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "screen",
+            "description": "Print the relayed shared-sandbox terminal buffer "
+                           "(what everyone watching the PTY sees).",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "tail": {"type": "integer",
+                             "description": "Last N bytes only (optional)."},
+                },
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "watch",
+            "description": "Block until a regex fires in room events or the terminal "
+                           "screen — how you wait on a teammate's report or a build "
+                           "result instead of sleeping.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "pattern": {"type": "string", "description": "Regex to match."},
+                    "source": {"type": "string", "enum": ["events", "screen"],
+                               "description": "Where to match (default: events)."},
+                    "timeout": {"type": "number",
+                                "description": "Give up after this many seconds."},
+                },
+                "required": ["pattern"],
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "manifest",
+            "description": "Stamp the sandbox's .hh-agent handoff record so work and "
+                           "memory hand off to the next agent or human.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "action": {"type": "string", "enum": ["push", "update"],
+                               "description": "push a fresh record or update the state."},
+                    "status": {"type": "string", "description": "Current status line."},
+                    "progress": {"type": "string", "description": "Progress note."},
+                    "done": {"type": "string", "description": "A completed item."},
+                    "todo": {"type": "string", "description": "A next item."},
+                },
+                "required": ["action"],
+            },
+        },
+    },
+]
+
+# Tool names that drive the sandbox/room mutably — used only for messaging; the
+# bridge is the real authority on grant-before-drive.
+_TOOL_NAMES = {(t["function"]["name"]) for t in OPERATOR_TOOLS}
+
+# DONE marker (same convention as the chat native loop): a text-only turn whose
+# first line starts with this means the model considers the objective met.
+_DONE_RE = re.compile(r"^\s*DONE:", re.I)
+
+DEFAULT_MAX_TURNS = 12
+DEFAULT_TOKEN_CEILING = 20000   # cumulative prompt+completion tokens (hard stop)
+DEFAULT_MAX_NUDGES = 2
+_OBS_CAP = 4096                 # bytes of any single observation fed back to the model
+
+HARNESS_SYSTEM_TAIL = (
+    "\n\nYou act ONLY by calling the provided tools (say/exec/write_file/get_file/"
+    "keys/screen/watch/manifest) — never by describing an action in prose. Work in "
+    "small steps and read each tool result before the next call. You may read and "
+    "`say` immediately, but exec/write_file/get_file/keys only work once the room "
+    "owner has granted you drive (the bridge will return an error until then — wait "
+    "and watch for the grant). When the objective is fully met, reply with a single "
+    "line starting 'DONE:' and a one-sentence summary, and do NOT call another tool."
+)
+
+
+def compose_system(objective: str, *, stop: list[str] | None = None,
+                   budget: "boot.Budget | None" = None,
+                   role_overlay: str | None = None) -> str:
+    """The harness operator's system prompt: portable Layer-1 capabilities +
+    objective + stop conditions + (optional) role overlay + recursion budget +
+    the tool-only operating rule. One screen of capabilities, behaviour overlaid —
+    no Claude-isms (the 'use the skill' line lives only in the Claude path)."""
+    stop = stop or ["objective met", "owner says stop/leave", "idle past remit"]
+    parts = [boot.load_capabilities()]
+    if role_overlay:
+        parts.append(role_overlay.strip())
+    parts.append(f"OBJECTIVE: {objective}")
+    parts.append("Stop conditions (say a short sign-off, then stop when any holds): "
+                 + "; ".join(stop) + ".")
+    if budget is not None:
+        parts.append(
+            f"Recursion budget you inherit: depth={budget.depth}, "
+            f"fanout={budget.fanout}, cost≈${budget.cost_usd:.2f}. Spawn at most "
+            "`fanout` children and only while depth>0; pass each a decremented budget.")
+    return "\n\n".join(parts) + HARNESS_SYSTEM_TAIL
+
+
+def _clip(s: str, cap: int = _OBS_CAP) -> str:
+    if len(s) <= cap:
+        return s
+    return s[:cap] + f"\n…[clipped {len(s) - cap} bytes]"
+
+
+def _format_events(events: list[dict], me: str) -> str:
+    """Turn raw bridge inbox events into a compact observation block the model
+    reads as room state. Drops our own echoes and high-volume noise."""
+    lines: list[str] = []
+    for ev in events:
+        kind = ev.get("kind")
+        if kind == "message":
+            who = ev.get("from", "?")
+            if who == me:
+                continue
+            mark = " (@you)" if ev.get("addressed") else ""
+            lines.append(f"[{who}{mark}] {ev.get('text', '')}")
+        elif kind == "acl":
+            state = "GRANTED drive" if ev.get("granted") else "drive NOT granted"
+            lines.append(f"[room] you are now {state}.")
+        elif kind == "sandbox":
+            lines.append(f"[room] sandbox {ev.get('state')} "
+                         f"({ev.get('engine') or '—'}/{ev.get('name') or '—'}).")
+        elif kind == "roster":
+            lines.append(f"[room] members: {', '.join(ev.get('users', []))}.")
+    return "\n".join(lines)
+
+
+@dataclass
+class HarnessResult:
+    final: str = ""
+    turns: int = 0
+    tokens: int = 0
+    reason: str = ""          # why the loop ended
+    tool_calls: int = 0
+
+
+@dataclass
+class OperatorHarness:
+    """Drives a room with a function-calling Provider over the bridge control
+    socket. Side-effect-free to construct; :meth:`run` does the work. ``request_fn``
+    is injectable so unit tests can drive the whole loop with no daemon/network."""
+
+    provider: object
+    sock_path: str
+    objective: str
+    stop: list[str] | None = None
+    budget: "boot.Budget | None" = None
+    role_overlay: str | None = None
+    max_turns: int = DEFAULT_MAX_TURNS
+    token_ceiling: int = DEFAULT_TOKEN_CEILING
+    max_nudges: int = DEFAULT_MAX_NUDGES
+    me: str = "operator"
+    request_fn: Callable[[str, dict, float], dict] | None = None
+    poll_timeout: float = 20.0
+
+    _cursor: int = field(default=0, init=False)
+
+    # ── control-socket plumbing ──────────────────────────────────────────
+    def _call(self, obj: dict, read_timeout: float = 35.0) -> dict:
+        fn = self.request_fn or _socket_request
+        return fn(self.sock_path, obj, read_timeout)
+
+    def _drain_events(self, *, wait: bool, timeout: float) -> list[dict]:
+        resp = self._call({"op": "read", "since": self._cursor,
+                           "wait": wait, "timeout": timeout},
+                          read_timeout=timeout + 5.0)
+        events = resp.get("events", []) or []
+        if events:
+            self._cursor = events[-1].get("seq", self._cursor)
+        elif resp.get("seq") is not None:
+            self._cursor = max(self._cursor, int(resp["seq"]))
+        return events
+
+    # ── tool dispatch (each maps to a bridge verb) ───────────────────────
+    def _run_tool(self, name: str, args: dict) -> str:
+        if name == "say":
+            r = self._call({"op": "say", "text": str(args.get("text", ""))})
+            return "(said)" if r.get("ok") else f"error: {r.get('error')}"
+        if name == "exec":
+            r = self._call({"op": "exec", "cmd": str(args.get("command", ""))})
+            out = _clip(r.get("output", ""))
+            return f"exit={r.get('rc')}\n{out}" if not r.get("error") else f"error: {r['error']}"
+        if name == "write_file":
+            r = self._call({"op": "write", "path": str(args.get("path", "")),
+                            "content": args.get("content", "")})
+            return (f"wrote {r.get('path')} ({r.get('bytes')} bytes)"
+                    if r.get("ok") else f"error: {r.get('error')}")
+        if name == "get_file":
+            import base64
+            r = self._call({"op": "get", "path": str(args.get("path", ""))})
+            if not r.get("ok"):
+                return f"error: {r.get('error')}"
+            try:
+                raw = base64.b64decode(r.get("b64", "")).decode(errors="replace")
+            except Exception:  # noqa: BLE001
+                raw = "[binary]"
+            return _clip(raw)
+        if name == "keys":
+            toks = [f"text:{args.get('text', '')}"]
+            if args.get("enter", True):
+                toks.append("enter")
+            r = self._call({"op": "keys", "keys": toks})
+            return "(keys sent)" if r.get("ok") else f"error: {r.get('error')}"
+        if name == "screen":
+            r = self._call({"op": "screen", "tail": args.get("tail"), "ansi": False})
+            return _clip(r.get("text", "")) if r.get("ok") else f"error: {r.get('error')}"
+        if name == "watch":
+            timeout = float(args.get("timeout", 30.0))
+            r = self._call({"op": "watch", "for": args.get("pattern"),
+                            "in": args.get("source", "events"), "idle": None,
+                            "timeout": timeout}, read_timeout=timeout + 5.0)
+            return _clip(json.dumps(r))
+        if name == "manifest":
+            req = {"op": "manifest", "action": args.get("action", "update")}
+            for k in ("status", "progress"):
+                if args.get(k) is not None:
+                    req[k] = args[k]
+            for k in ("done", "todo"):
+                if args.get(k):
+                    req[k] = [args[k]]
+            r = self._call(req)
+            return _clip(json.dumps(r))
+        return f"error: unknown tool '{name}'"
+
+    @staticmethod
+    def _calls_to_wire(calls: list[dict]) -> list[dict]:
+        """Round-trip the model's tool calls back into the wire `tool_calls` shape
+        so the next turn sees its own request alongside the result."""
+        return [{"type": "function",
+                 "function": {"name": c.get("name", ""),
+                              "arguments": c.get("arguments", {})}}
+                for c in calls]
+
+    # ── the loop ──────────────────────────────────────────────────────────
+    def run(self) -> HarnessResult:
+        cwt = getattr(self.provider, "complete_with_tools", None)
+        if cwt is None:
+            return HarnessResult(
+                reason="provider-no-tools",
+                final="[refused — this provider cannot do function calling; the "
+                      "harness operator needs a tool-capable model]")
+        system = compose_system(self.objective, stop=self.stop, budget=self.budget,
+                                role_overlay=self.role_overlay)
+
+        # Seed: snapshot current room state (full ring) as the first observation.
+        try:
+            seed = self._drain_events(wait=False, timeout=0.0)
+        except Exception as e:  # noqa: BLE001
+            return HarnessResult(reason="bridge-unreachable",
+                                 final=f"[bridge unreachable: {e}]")
+        obs = _format_events(seed, self.me) or "(room is quiet)"
+        messages: list[dict] = [
+            {"role": "user",
+             "content": f"You have joined the room. Current state:\n{obs}\n\n"
+                        f"Begin working toward the objective."}
+        ]
+
+        res = HarnessResult()
+        nudges = 0
+        while res.turns < self.max_turns + self.max_nudges:
+            res.turns += 1
+            try:
+                text, calls, usage = cwt(system, messages, OPERATOR_TOOLS)
+            except ToolsUnsupported as e:
+                return HarnessResult(
+                    turns=res.turns, tokens=res.tokens, reason="tools-unsupported",
+                    final=f"[refused — model rejected tool calling: {e}]")
+            except Exception as e:  # noqa: BLE001
+                res.reason, res.final = "provider-error", f"[ai error: {e}]"
+                return res
+
+            res.tokens += int((usage or {}).get("prompt_eval_count", 0) or 0)
+            res.tokens += int((usage or {}).get("eval_count", 0) or 0)
+
+            if not calls:
+                if _DONE_RE.match(text or ""):
+                    res.final = re.sub(_DONE_RE, "", text, count=1).strip()
+                    res.reason = "done"
+                    return res
+                if nudges >= self.max_nudges:
+                    res.final = (text or "").strip() or "[stopped — model stalled]"
+                    res.reason = "stalled"
+                    return res
+                nudges += 1
+                messages.append({"role": "assistant", "content": text or ""})
+                # Poll for any room activity to give the nudge fresh context.
+                fresh = []
+                try:
+                    fresh = self._drain_events(wait=True, timeout=self.poll_timeout)
+                except Exception:  # noqa: BLE001
+                    pass
+                nudge_obs = _format_events(fresh, self.me)
+                messages.append({
+                    "role": "user",
+                    "content": (("Room update:\n" + nudge_obs + "\n\n") if nudge_obs else "")
+                    + "If the objective is met, reply with a single 'DONE: <summary>' "
+                      "line. Otherwise call the next tool to make progress."})
+                continue
+
+            if res.tokens >= self.token_ceiling:
+                res.final = ("[stopped — per-agent token ceiling "
+                             f"{self.token_ceiling} reached]")
+                res.reason = "token-ceiling"
+                return res
+
+            # Echo the assistant's tool-call turn, then run each call and feed the
+            # results back as tool messages so the next turn sees the outcome.
+            messages.append({"role": "assistant", "content": text or "",
+                             "tool_calls": self._calls_to_wire(calls)})
+            for c in calls:
+                name, cargs = c.get("name", ""), c.get("arguments") or {}
+                result = self._run_tool(name, cargs)
+                res.tool_calls += 1
+                messages.append({"role": "tool", "content": _clip(result)})
+            # After acting, fold any new room events into the next observation.
+            try:
+                fresh = self._drain_events(wait=False, timeout=0.0)
+            except Exception:  # noqa: BLE001
+                fresh = []
+            obs = _format_events(fresh, self.me)
+            if obs:
+                messages.append({"role": "user", "content": "Room update:\n" + obs})
+
+        res.final = res.final or "[stopped — reached turn cap]"
+        res.reason = res.reason or "turn-cap"
+        return res

tests/test_operator_harness.py

@@ -0,0 +1,219 @@
+"""Offline tests for the harness-mode operator (Phase 2).
+
+The loop is exercised end-to-end with a scripted fake Provider (no network/model)
+and a fake control socket (no daemon), so every tool dispatch and stop condition
+is asserted deterministically.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import pytest
+
+from cmd_chat.ai.providers import ToolsUnsupported
+from cmd_chat.operator import bootstrap as boot
+from cmd_chat.operator.harness import OPERATOR_TOOLS, OperatorHarness, compose_system
+
+
+class FakeProvider:
+    """Returns a scripted (text, calls, usage) per turn. A script entry is either a
+    list of {"name","arguments"} tool calls, or a string (a text-only turn)."""
+
+    def __init__(self, script, usage=None, raise_tools=False):
+        self.script = list(script)
+        self.usage = usage or {"prompt_eval_count": 10, "eval_count": 5}
+        self.raise_tools = raise_tools
+        self.calls_seen = []
+
+    def complete_with_tools(self, system, messages, tools):
+        self.calls_seen.append((system, list(messages), tools))
+        if self.raise_tools:
+            raise ToolsUnsupported("nope")
+        step = self.script.pop(0) if self.script else "DONE: ran out of script"
+        if isinstance(step, str):
+            return step, [], self.usage
+        return "", step, self.usage
+
+
+class NoToolsProvider:
+    pass
+
+
+class FakeBridge:
+    """Records every request and answers by op with canned responses."""
+
+    def __init__(self, seed_events=None):
+        self.requests = []
+        self._seq = 0
+        self._seed = seed_events or []
+        self._seeded = False
+
+    def __call__(self, sock_path, obj, read_timeout=35.0):
+        self.requests.append(obj)
+        op = obj.get("op")
+        if op == "read":
+            if not self._seeded:
+                self._seeded = True
+                evs = []
+                for e in self._seed:
+                    self._seq += 1
+                    evs.append({"seq": self._seq, **e})
+                return {"ok": True, "events": evs, "seq": self._seq}
+            return {"ok": True, "events": [], "seq": self._seq}
+        if op == "say":
+            return {"ok": True}
+        if op == "exec":
+            return {"ok": True, "rc": 0, "output": "hello\n"}
+        if op == "write":
+            return {"ok": True, "path": obj.get("path"), "bytes": len(obj.get("content", ""))}
+        if op == "get":
+            import base64
+            return {"ok": True, "b64": base64.b64encode(b"file body").decode()}
+        if op == "keys":
+            return {"ok": True}
+        if op == "screen":
+            return {"ok": True, "text": "screen buffer"}
+        if op == "watch":
+            return {"ok": True, "matched": True}
+        if op == "manifest":
+            return {"ok": True}
+        if op == "down":
+            return {"ok": True, "stopping": True}
+        return {"ok": False, "error": f"unknown op {op}"}
+
+    def ops(self):
+        return [r.get("op") for r in self.requests]
+
+
+def _harness(provider, bridge, **kw):
+    return OperatorHarness(
+        provider=provider, sock_path="/tmp/fake.sock",
+        objective="build the thing", request_fn=bridge, me="builder", **kw)
+
+
+def test_compose_system_is_portable_no_claude_isms():
+    sysmsg = compose_system("do x", stop=["done"],
+                            budget=boot.Budget(depth=1, fanout=2, cost_usd=3.0))
+    assert "hh-operator skill" not in sysmsg          # no Claude-only delivery line
+    assert "do x" in sysmsg
+    assert "depth=1" in sysmsg and "fanout=2" in sysmsg
+    # Layer-1 capabilities are inlined.
+    assert boot.load_capabilities().split("\n", 1)[0] in sysmsg
+    # The tool-only operating rule is present.
+    assert "DONE:" in sysmsg
+
+
+def test_full_flow_drives_bridge_then_done():
+    provider = FakeProvider(script=[
+        [{"name": "say", "arguments": {"text": "starting"}}],
+        [{"name": "exec", "arguments": {"command": "echo hi"}}],
+        [{"name": "write_file", "arguments": {"path": "a.txt", "content": "x"}}],
+        "DONE: built and verified the thing",
+    ])
+    bridge = FakeBridge(seed_events=[
+        {"kind": "message", "from": "andre", "text": "go build it", "addressed": True},
+    ])
+    res = _harness(provider, bridge).run()
+    assert res.reason == "done"
+    assert res.final == "built and verified the thing"
+    assert res.tool_calls == 3
+    # Each tool was dispatched to the matching bridge verb.
+    ops = bridge.ops()
+    assert "say" in ops and "exec" in ops and "write" in ops
+    # The loop read the room (seed + post-action polls).
+    assert ops.count("read") >= 1
+
+
+def test_seed_observation_includes_room_state():
+    provider = FakeProvider(script=["DONE: nothing to do"])
+    bridge = FakeBridge(seed_events=[
+        {"kind": "message", "from": "andre", "text": "hello operator", "addressed": True},
+    ])
+    _harness(provider, bridge).run()
+    first_system, first_messages, first_tools = provider.calls_seen[0]
+    seed_obs = first_messages[0]["content"]
+    assert "hello operator" in seed_obs
+    assert first_tools is OPERATOR_TOOLS
+
+
+def test_own_messages_filtered_from_observation():
+    provider = FakeProvider(script=["DONE: ok"])
+    bridge = FakeBridge(seed_events=[
+        {"kind": "message", "from": "builder", "text": "my own echo", "addressed": False},
+        {"kind": "message", "from": "andre", "text": "real input", "addressed": False},
+    ])
+    _harness(provider, bridge).run()
+    seed_obs = provider.calls_seen[0][1][0]["content"]
+    assert "real input" in seed_obs
+    assert "my own echo" not in seed_obs
+
+
+def test_get_file_tool_decodes_content():
+    provider = FakeProvider(script=[
+        [{"name": "get_file", "arguments": {"path": "a.txt"}}],
+        "DONE: read it",
+    ])
+    bridge = FakeBridge()
+    res = _harness(provider, bridge).run()
+    assert res.reason == "done"
+    # The tool result (decoded file body) is fed back as a tool message.
+    _, msgs, _ = provider.calls_seen[-1]
+    tool_msgs = [m for m in msgs if m.get("role") == "tool"]
+    assert any("file body" in m["content"] for m in tool_msgs)
+
+
+def test_keys_tool_maps_to_text_token_and_enter():
+    provider = FakeProvider(script=[
+        [{"name": "keys", "arguments": {"text": "make build", "enter": True}}],
+        "DONE: drove the pty",
+    ])
+    bridge = FakeBridge()
+    _harness(provider, bridge).run()
+    keys_req = next(r for r in bridge.requests if r.get("op") == "keys")
+    assert keys_req["keys"] == ["text:make build", "enter"]
+
+
+def test_token_ceiling_stops_the_loop():
+    provider = FakeProvider(
+        script=[[{"name": "exec", "arguments": {"command": "sleep 1"}}]] * 50,
+        usage={"prompt_eval_count": 9000, "eval_count": 2000})
+    bridge = FakeBridge()
+    res = _harness(provider, bridge, token_ceiling=10000).run()
+    assert res.reason == "token-ceiling"
+    assert res.tokens >= 10000
+
+
+def test_tools_unsupported_refuses_clearly():
+    provider = FakeProvider(script=[], raise_tools=True)
+    bridge = FakeBridge()
+    res = _harness(provider, bridge).run()
+    assert res.reason == "tools-unsupported"
+    assert "refused" in res.final.lower()
+
+
+def test_provider_without_tool_support_refuses():
+    res = _harness(NoToolsProvider(), FakeBridge()).run()
+    assert res.reason == "provider-no-tools"
+    assert "function calling" in res.final.lower()
+
+
+def test_stall_then_nudge_then_done():
+    # A text-only non-DONE turn nudges; the next turn finishes.
+    provider = FakeProvider(script=[
+        "let me think about this",      # stall → nudge
+        "DONE: finished after a nudge",
+    ])
+    bridge = FakeBridge()
+    res = _harness(provider, bridge).run()
+    assert res.reason == "done"
+    assert res.final == "finished after a nudge"
+    assert res.turns == 2
+
+
+def test_stall_exhausts_nudges_and_stops():
+    provider = FakeProvider(script=["thinking"] * 10)
+    bridge = FakeBridge()
+    res = _harness(provider, bridge, max_nudges=2).run()
+    assert res.reason == "stalled"