From 0798db7941882f307f740f468ae95265b4b4c635 Mon Sep 17 00:00:00 2001 From: leetcrypt Date: Sun, 28 Jun 2026 23:23:17 -0700 Subject: [PATCH] =?UTF-8?q?feat(operator):=20harness-mode=20operator=20?= =?UTF-8?q?=E2=80=94=20drive=20a=20room=20with=20any=20function-calling=20?= =?UTF-8?q?model=20(P2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- cmd_chat/operator/__main__.py | 102 ++++++++ cmd_chat/operator/harness.py | 449 +++++++++++++++++++++++++++++++++ tests/test_operator_harness.py | 219 ++++++++++++++++ 3 files changed, 770 insertions(+) create mode 100644 cmd_chat/operator/harness.py create mode 100644 tests/test_operator_harness.py diff --git a/cmd_chat/operator/__main__.py b/cmd_chat/operator/__main__.py index 0ba45be..1955369 100644 --- a/cmd_chat/operator/__main__.py +++ b/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": diff --git a/cmd_chat/operator/harness.py b/cmd_chat/operator/harness.py new file mode 100644 index 0000000..a791f49 --- /dev/null +++ b/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: ' " + "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 diff --git a/tests/test_operator_harness.py b/tests/test_operator_harness.py new file mode 100644 index 0000000..89d9adb --- /dev/null +++ b/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"