diff --git a/cmd_chat/operator/__main__.py b/cmd_chat/operator/__main__.py index b4b35e9..1ecf18e 100644 --- a/cmd_chat/operator/__main__.py +++ b/cmd_chat/operator/__main__.py @@ -37,6 +37,12 @@ def _add_conn_args(p: argparse.ArgumentParser) -> None: p.add_argument("--trigger", default=None, help="extra phrase that marks a message 'addressed' to us " "(the operator name / @name always count)") + p.add_argument("--depth", type=int, default=1, + help="recursion depth budget (levels of nested operators)") + p.add_argument("--fanout", type=int, default=2, + help="recursion fan-out budget (children per level)") + p.add_argument("--cost", type=float, default=5.0, + help="soft cost ceiling (USD) carried to children") def _build_parser() -> argparse.ArgumentParser: @@ -81,6 +87,24 @@ def _build_parser() -> argparse.ArgumentParser: gt.add_argument("--out", default=None, help="write to this local path (else stdout)") gt.add_argument("--session", default=None) + sp = sub.add_parser("spawn", help="spawn a nested Claude operator (budgeted)") + sp.add_argument("objective", help="what the nested operator should achieve") + sp.add_argument("--room-host", required=True) + sp.add_argument("--room-port", type=int, required=True) + sp.add_argument("--room-name", default="operator") + sp.add_argument("--stop", action="append", default=None, + help="a stop condition (repeatable)") + sp.add_argument("--target", choices=["host"], default="host") + sp.add_argument("--config-dir", default=None, + help="isolated CLAUDE_CONFIG_DIR for the child") + sp.add_argument("--allow-creds", action="store_true", + help="carry our credentials to the child (default: off)") + sp.add_argument("--skip-permissions", action="store_true", + help="run the child unattended (--dangerously-skip-permissions)") + sp.add_argument("--go", action="store_true", + help="actually launch (default is a dry-run plan)") + sp.add_argument("--session", default=None) + ky = sub.add_parser("keys", help="inject keystrokes into the room's shared PTY") ky.add_argument("keys", nargs="*", help="tokens: literal text, named keys (enter, ctrl-c, esc, " @@ -96,10 +120,12 @@ def _build_parser() -> argparse.ArgumentParser: # ── daemon ─────────────────────────────────────────────────────────────── def _run_serve(args) -> int: from .bridge import OperatorBridge # heavy imports only on the daemon path + from .bootstrap import Budget + budget = Budget(depth=args.depth, fanout=args.fanout, cost_usd=args.cost) bridge = OperatorBridge( args.host, args.port, name=args.user, password=args.password, insecure=args.insecure, no_tls=args.no_tls, - session=args.session, trigger=args.trigger) + session=args.session, trigger=args.trigger, budget=budget) try: bridge.run() except KeyboardInterrupt: @@ -139,6 +165,8 @@ def _run_up(args) -> int: cmd += ["--session", args.session] if args.trigger: cmd += ["--trigger", args.trigger] + cmd += ["--depth", str(args.depth), "--fanout", str(args.fanout), + "--cost", str(args.cost)] log = open(sess.log_path, "a") # noqa: SIM115 — handed to the child proc = subprocess.Popen(cmd, stdout=log, stderr=log, stdin=subprocess.DEVNULL, @@ -267,6 +295,18 @@ def _run_get(args) -> int: return 0 +def _run_spawn(args) -> int: + resp = _client_request(args, { + "op": "spawn", "objective": args.objective, + "room": {"host": args.room_host, "port": args.room_port, + "name": args.room_name}, + "stop": args.stop, "target": args.target, + "config_dir": args.config_dir, "allow_creds": args.allow_creds, + "skip_permissions": args.skip_permissions, "dry_run": not args.go}) + print(json.dumps(resp, indent=2)) + return 0 if resp.get("ok") else 1 + + def _run_keys(args) -> int: from . import sandbox as sbx if args.help_keys: @@ -309,6 +349,8 @@ def main(argv: list[str] | None = None) -> int: return _run_get(args) if verb == "keys": return _run_keys(args) + if verb == "spawn": + return _run_spawn(args) if verb in ("roster", "status", "down"): return _run_simple(args, verb) return 2 diff --git a/cmd_chat/operator/bootstrap.py b/cmd_chat/operator/bootstrap.py new file mode 100644 index 0000000..4ee0074 --- /dev/null +++ b/cmd_chat/operator/bootstrap.py @@ -0,0 +1,215 @@ +"""Bootstrap + recursive-spawn primitives for the operator bridge (Phase 5). + +Goal: from inside one room (or a sandbox/VM the operator owns), stand up a +*nested* Claude Code session that runs the `hh-operator` skill against another +room — a tree of operators. Doing that safely needs four things, all here: + +1. **Detect** the target system (OS / arch / package manager / what's already + installed) so install is a decision, not a guess. +2. **Install** Claude Code via a *known* package name (npm `@anthropic-ai/ + claude-code`) or an explicitly-configured installer — never a guessed URL. +3. **Authenticate** by carrying the parent's own credentials into the child — + but **gated off by default**: a child auths itself unless the operator + explicitly opts in (`allow_creds`). We never exfiltrate creds to a system we + don't own, and never enable this implicitly. +4. **Budget** the recursion: hard **depth** and **fan-out** caps plus a soft + **cost** ceiling, decremented on every descent, refusing at zero. This is the + guardrail that stops a runaway tree. + +This module is pure/decision-making: it builds argv, plans, and budgets. It does +not itself run network installs or launch detached Claude sessions — the bridge +does that behind explicit flags, so unit tests can exercise all the logic with +no side effects. +""" + +from __future__ import annotations + +import os +import platform +import shutil +from dataclasses import dataclass, replace +from pathlib import Path + +# Documented install package — NOT a guessed URL. Override the whole install +# command with $HH_CLAUDE_INSTALL when a site has its own channel. +CLAUDE_NPM_PACKAGE = "@anthropic-ai/claude-code" + + +class BudgetExhausted(RuntimeError): + """A spawn was refused because a recursion budget hit zero.""" + + +@dataclass(frozen=True) +class Budget: + """Hard caps on a recursion tree. + + - ``depth`` — how many more levels may be spawned below here (0 = leaf). + - ``fanout`` — how many children *this* node may still spawn. + - ``cost_usd`` — soft ceiling carried down and surfaced to the child as a + directive (Claude Code has no native $ cap, so this is advisory; depth and + fanout are the hard stops). + """ + depth: int = 1 + fanout: int = 2 + cost_usd: float = 5.0 + + def can_spawn(self) -> bool: + return self.depth > 0 and self.fanout > 0 + + def descend(self) -> "Budget": + """Consume one child slot at this level and return the budget the child + inherits (one level shallower, its own fresh fanout). Raises when this + node can't spawn any more.""" + if self.depth <= 0: + raise BudgetExhausted("depth budget exhausted — cannot nest deeper") + if self.fanout <= 0: + raise BudgetExhausted("fan-out budget exhausted at this level") + # The child gets depth-1 and the same fanout/cost to share among its own + # children; *this* node records one fewer remaining child via `spent`. + return Budget(depth=self.depth - 1, fanout=self.fanout, cost_usd=self.cost_usd) + + def spent(self) -> "Budget": + """Return this node's budget after using one of its child slots.""" + return replace(self, fanout=self.fanout - 1) + + def as_dict(self) -> dict: + return {"depth": self.depth, "fanout": self.fanout, "cost_usd": self.cost_usd} + + +# ── detection ─────────────────────────────────────────────────────────────── +def _distro_id() -> str: + try: + for line in Path("/etc/os-release").read_text().splitlines(): + if line.startswith("ID="): + return line.split("=", 1)[1].strip().strip('"') + except OSError: + pass + return "" + + +def _pkg_manager() -> str | None: + for mgr in ("apt-get", "dnf", "yum", "apk", "pacman", "brew"): + if shutil.which(mgr): + return mgr + return None + + +def detect_system() -> dict: + """Snapshot what the bootstrap needs to decide install strategy. Pure read — + no mutation, safe to call anywhere (including inside a sandbox via exec).""" + return { + "os": platform.system().lower(), # linux / darwin + "arch": platform.machine(), # x86_64 / aarch64 … + "distro": _distro_id(), # ubuntu / debian / alpine … + "pkg_manager": _pkg_manager(), + "has_node": shutil.which("node") is not None, + "has_npm": shutil.which("npm") is not None, + "claude": shutil.which("claude"), # path or None + } + + +def claude_present() -> bool: + return shutil.which("claude") is not None + + +def install_plan(sysinfo: dict | None = None) -> list[list[str]]: + """Ordered candidate commands to get the `claude` CLI onto the system. + + Honours $HH_CLAUDE_INSTALL (a full shell command) first so a site can pin + its own channel. Otherwise prefers the documented npm package; if npm is + missing, prefix the distro's node install. We never fabricate a download + URL — if none of these apply the caller surfaces "install claude manually".""" + override = os.environ.get("HH_CLAUDE_INSTALL") + if override: + return [["sh", "-c", override]] + info = sysinfo or detect_system() + plans: list[list[str]] = [] + if info.get("has_npm"): + plans.append(["npm", "install", "-g", CLAUDE_NPM_PACKAGE]) + else: + mgr = info.get("pkg_manager") + node_install = { + "apt-get": ["apt-get", "install", "-y", "nodejs", "npm"], + "dnf": ["dnf", "install", "-y", "nodejs", "npm"], + "yum": ["yum", "install", "-y", "nodejs", "npm"], + "apk": ["apk", "add", "nodejs", "npm"], + "pacman": ["pacman", "-Sy", "--noconfirm", "nodejs", "npm"], + "brew": ["brew", "install", "node"], + }.get(mgr) + if node_install: + plans.append(node_install) + plans.append(["npm", "install", "-g", CLAUDE_NPM_PACKAGE]) + return plans + + +# ── credentials (gated) ───────────────────────────────────────────────────── +def creds_source() -> Path | None: + """The parent session's own OAuth/API credentials file, if present.""" + p = Path.home() / ".claude" / ".credentials.json" + return p if p.exists() else None + + +def plan_creds(dest_home: str | os.PathLike, *, allow: bool) -> dict: + """Decide whether to carry credentials into a child's config dir. + + Default (``allow=False``) → carry nothing; the child must authenticate + itself (`claude` interactive login or its own $ANTHROPIC_API_KEY). Only when + the operator *explicitly* opts in do we stage a copy of our own creds — and + only into a path we were handed. This function plans; the bridge performs the + copy so the side effect stays behind one audited flag.""" + if not allow: + return {"staged": False, "reason": "creds gating on — child authenticates itself"} + src = creds_source() + if src is None: + return {"staged": False, "reason": "no parent credentials to carry"} + dest = Path(dest_home) / ".claude" / ".credentials.json" + return {"staged": True, "src": str(src), "dest": str(dest)} + + +# ── directive + run argv ──────────────────────────────────────────────────── +def compose_directive(objective: str, room: dict, budget: Budget, + stop: list[str] | None = None) -> str: + """The prompt the nested Claude runs under `claude -p`. Tells it to use the + hh-operator skill, where to join, what to achieve, when to stop, and the + recursion budget it inherits (so it self-limits its own spawns).""" + host, port = room.get("host", "?"), room.get("port", "?") + name = room.get("name", "operator") + stop = stop or ["objective met", "owner says stop/leave", "idle past remit"] + lines = [ + "Use the hh-operator skill. You are an autonomous operator joining a " + "hack-house room.", + f"Join: `hh-bridge up {host} {port} {name}` " + f"(password via room config; add --no-tls for plain ws).", + f"Objective: {objective}", + "Stop conditions (leave with `hh-bridge down` when any is met): " + + "; ".join(stop) + ".", + f"Recursion budget you inherit: depth={budget.depth}, " + f"fanout={budget.fanout}, cost≈${budget.cost_usd:.2f}. " + "You may spawn at most `fanout` child operators and only while depth>0; " + "pass each child a decremented budget. Stay within the cost ceiling.", + "Drive sandboxes only after the owner grants you. Never touch a system " + "you weren't asked to.", + ] + return "\n".join(lines) + + +def build_run_argv(directive: str, *, skip_permissions: bool = False, + config_dir: str | None = None, + claude_bin: str = "claude") -> list[str]: + """argv to launch a headless nested Claude session. `skip_permissions` maps + to Claude Code's unattended flag (only for trusted, sandboxed autonomy); + `config_dir` points the child at an isolated $CLAUDE_CONFIG_DIR so its creds + and state don't collide with the parent.""" + argv = [claude_bin, "-p", directive] + if skip_permissions: + argv.append("--dangerously-skip-permissions") + return argv + + +def child_env(config_dir: str | None = None) -> dict: + """Environment for the child process. Isolates its config dir when given so + a nested session's auth/state is its own.""" + env = dict(os.environ) + if config_dir: + env["CLAUDE_CONFIG_DIR"] = str(config_dir) + return env diff --git a/cmd_chat/operator/bridge.py b/cmd_chat/operator/bridge.py index 224b0ad..7b8533d 100644 --- a/cmd_chat/operator/bridge.py +++ b/cmd_chat/operator/bridge.py @@ -25,6 +25,7 @@ import time import websockets from ..client.client import Client +from . import bootstrap as boot from . import sandbox as sbx from .session import Session @@ -39,12 +40,14 @@ class OperatorBridge(Client): def __init__(self, server: str, port: int, name: str, password: str | None = None, insecure: bool = False, no_tls: bool = False, session: str | None = None, - trigger: str | None = None): + trigger: str | None = None, budget: boot.Budget | None = None): super().__init__(server, port, username=name, password=password, insecure=insecure, no_tls=no_tls) self.name = name self.trigger = (trigger or "").strip() self.session = Session(session or name) + # Recursion budget this operator may spend spawning nested operators. + self.budget = budget or boot.Budget() # Inbox: monotonically-seq'd events + an async condition to wake waiters. self.events: list[dict] = [] @@ -302,6 +305,8 @@ class OperatorBridge(Client): return await self._op_get(req) if op == "keys": return await self._op_keys(req) + if op == "spawn": + return await self._op_spawn(req) if op == "down": asyncio.get_running_loop().call_soon(self._begin_shutdown) return {"ok": True, "stopping": True} @@ -469,6 +474,87 @@ class OperatorBridge(Client): await self._emit("keys", bytes=len(data)) return {"ok": True, "bytes": len(data)} + # ── recursive spawn (Phase 5) ──────────────────────────────────────── + async def _op_spawn(self, req: dict) -> dict: + """Stand up a nested Claude Code operator against another room. + + Guardrails, in order: + - **budget** — refuse unless this node may still spawn (depth>0 & + fanout>0); the child inherits a decremented budget, we record one spent + slot. This is the hard stop against a runaway tree. + - **creds gating** — we carry our own credentials to the child *only* when + `allow_creds` is explicitly set; otherwise the child authenticates + itself. Never implicit, never to a system we don't own. + - **dry_run** — default returns the full plan (argv, budget, creds + decision) and launches nothing, so a tree can be inspected before it's + grown. + + `target` is "host" (Popen detached here) — sandbox/VM placement is the + operator's to run via `exec` using the returned plan.""" + objective = str(req.get("objective", "")).strip() + if not objective: + return {"ok": False, "error": "spawn needs an objective"} + room = req.get("room") or {} + if not room.get("host") or not room.get("port"): + return {"ok": False, "error": "spawn needs room {host, port, name}"} + + if not self.budget.can_spawn(): + return {"ok": False, "error": "recursion budget exhausted " + f"(depth={self.budget.depth}, fanout={self.budget.fanout})"} + try: + child_budget = self.budget.descend() + except boot.BudgetExhausted as e: + return {"ok": False, "error": str(e)} + + allow_creds = bool(req.get("allow_creds", False)) + dry_run = bool(req.get("dry_run", True)) + skip_perms = bool(req.get("skip_permissions", False)) + config_dir = req.get("config_dir") + + stop = req.get("stop") or None + directive = boot.compose_directive(objective, room, child_budget, stop=stop) + argv = boot.build_run_argv(directive, skip_permissions=skip_perms, + config_dir=config_dir) + creds = boot.plan_creds(config_dir or "~", allow=allow_creds) + plan = {"argv": argv, "budget": child_budget.as_dict(), "creds": creds, + "target": req.get("target", "host"), + "claude_present": boot.claude_present()} + + if dry_run: + return {"ok": True, "dry_run": True, "plan": plan} + + if not boot.claude_present(): + return {"ok": False, "error": "claude CLI not installed", + "install_plan": boot.install_plan(), "plan": plan} + + # Carry creds only behind the explicit gate. + if creds.get("staged"): + try: + dest = boot.Path(creds["dest"]) + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_bytes(boot.Path(creds["src"]).read_bytes()) + dest.chmod(0o600) + except OSError as e: + return {"ok": False, "error": f"creds staging failed: {e}"} + + # Launch detached so the child outlives this daemon; logs to the session. + import subprocess + try: + logf = open(self.session.dir / f"spawn-{self.seq + 1}.log", "a") # noqa: SIM115 + proc = subprocess.Popen( + argv, env=boot.child_env(config_dir), + stdout=logf, stderr=logf, stdin=subprocess.DEVNULL, + start_new_session=True, close_fds=True) + except OSError as e: + return {"ok": False, "error": f"spawn launch failed: {e}"} + + self.budget = self.budget.spent() # consume one child slot here + await self._emit("spawn", objective=objective, pid=proc.pid, + budget=child_budget.as_dict(), + creds_staged=creds.get("staged", False)) + return {"ok": True, "pid": proc.pid, "plan": plan, + "remaining": self.budget.as_dict()} + def _begin_shutdown(self) -> None: self._stop.set() self.running = False diff --git a/tests/test_operator_bridge.py b/tests/test_operator_bridge.py index 131ae8c..aa4cc6d 100644 --- a/tests/test_operator_bridge.py +++ b/tests/test_operator_bridge.py @@ -15,6 +15,7 @@ from cryptography.fernet import Fernet from cmd_chat.operator.bridge import OperatorBridge from cmd_chat.operator import sandbox as sbx +from cmd_chat.operator import bootstrap as boot def _bridge(name="oracle", trigger=None): @@ -272,3 +273,93 @@ def test_exec_requires_target(): r = await b._op_exec({"cmd": ""}) assert r["ok"] is False and "empty" in r["error"] asyncio.run(go()) + + +# ── Phase 5: recursion budget guardrails ────────────────────────────────── +def test_budget_descend_and_exhaust(): + b = boot.Budget(depth=2, fanout=2, cost_usd=5.0) + assert b.can_spawn() + child = b.descend() # one level down for the child + assert child.depth == 1 and child.fanout == 2 + spent = b.spent() # this node used one child slot + assert spent.fanout == 1 and spent.depth == 2 + # depth floor: a leaf cannot nest + leaf = boot.Budget(depth=0, fanout=2) + assert not leaf.can_spawn() + try: + leaf.descend() + assert False, "expected BudgetExhausted" + except boot.BudgetExhausted: + pass + # fan-out floor + spread = boot.Budget(depth=3, fanout=0) + assert not spread.can_spawn() + + +def test_install_plan_prefers_npm_and_honors_override(monkeypatch=None): + plan = boot.install_plan({"has_npm": True, "pkg_manager": "apt-get"}) + assert plan and plan[0] == ["npm", "install", "-g", boot.CLAUDE_NPM_PACKAGE] + # no npm → distro node install then npm global + plan = boot.install_plan({"has_npm": False, "pkg_manager": "apk"}) + assert ["apk", "add", "nodejs", "npm"] in plan + # explicit override wins and is the only command + import os + os.environ["HH_CLAUDE_INSTALL"] = "echo custom-install" + try: + plan = boot.install_plan({"has_npm": True}) + assert plan == [["sh", "-c", "echo custom-install"]] + finally: + del os.environ["HH_CLAUDE_INSTALL"] + + +def test_compose_directive_carries_objective_stop_budget(): + d = boot.compose_directive( + "map the room and report", {"host": "h", "port": 9, "name": "scout"}, + boot.Budget(depth=1, fanout=2, cost_usd=3.0), stop=["task done"]) + assert "hh-operator skill" in d + assert "map the room and report" in d + assert "task done" in d + assert "depth=1" in d and "fanout=2" in d + + +def test_plan_creds_gated_off_by_default(): + off = boot.plan_creds("/tmp/child", allow=False) + assert off["staged"] is False and "gating on" in off["reason"] + + +def test_build_run_argv_flags(): + argv = boot.build_run_argv("do x", skip_permissions=True) + assert argv[:3] == ["claude", "-p", "do x"] + assert "--dangerously-skip-permissions" in argv + argv = boot.build_run_argv("do x") + assert "--dangerously-skip-permissions" not in argv + + +# ── spawn op: gating + dry-run ──────────────────────────────────────────── +def test_spawn_dry_run_returns_plan_without_launch(): + async def go(): + b = _bridge("oracle") + b.budget = boot.Budget(depth=1, fanout=1, cost_usd=2.0) + r = await b._op_spawn({"objective": "scout the room", + "room": {"host": "h", "port": 7, "name": "kid"}, + "dry_run": True}) + assert r["ok"] and r["dry_run"] is True + assert r["plan"]["budget"] == {"depth": 0, "fanout": 1, "cost_usd": 2.0} + assert r["plan"]["creds"]["staged"] is False # gated off + # dry-run must not consume the budget + assert b.budget.fanout == 1 + # no spawn event emitted on a dry run + assert [e for e in b.events if e["kind"] == "spawn"] == [] + asyncio.run(go()) + + +def test_spawn_refuses_without_objective_or_room_or_budget(): + async def go(): + b = _bridge("oracle") + assert (await b._op_spawn({"objective": ""}))["ok"] is False + assert (await b._op_spawn({"objective": "x"}))["ok"] is False # no room + b.budget = boot.Budget(depth=0, fanout=0) + r = await b._op_spawn({"objective": "x", + "room": {"host": "h", "port": 7}}) + assert r["ok"] is False and "budget exhausted" in r["error"] + asyncio.run(go())