"""Bounded session launcher: runs `claude --bare -p` on an executor node over SSH, streams stream-json to an immutable raw file, normalises events into the store, enforces a wall-clock limit by killing the remote process (matched by its unique session id), and supports one bounded resume (arm A4). The API key is read on the node from ~/wp42/.env; it is never logged. Recursion guard: workers receive no Agent/Task tool; the launcher refuses to start if WP42_DEPTH is set in its own environment.""" from __future__ import annotations import os, shlex, subprocess, threading, time from .contracts import utcnow from .events import normalise, parse_result_line from .config import NODE_ENV_FILE EXECUTOR = "claude" class LaunchError(RuntimeError): pass def build_argv(prompt_file: str, system_append_file: str, model: str, session_id: str, max_budget_usd: float, max_turns: int, tools: list[str], disallowed: list[str], resume: bool = False, effort: str | None = None) -> list[str]: argv = [EXECUTOR, "--bare", "-p", f"$(cat {shlex.quote(prompt_file)})", "--output-format", "stream-json", "--verbose", "--model", model, "--tools", ",".join(tools), "--allowedTools", *tools, "--permission-prompts", "none", "--max-budget-usd", f"{max_budget_usd:.2f}", "--max-turns", str(max_turns), "--append-system-prompt-file", system_append_file, "--no-chrome", "--disable-slash-commands"] # session persistence is required for the A4 --resume (defect A-012) if disallowed: argv += ["--disallowedTools", *disallowed] if effort: argv += ["--effort", effort] argv += (["--resume", session_id] if resume else ["--session-id", session_id]) return argv def remote_shell_line(workspace: str, argv: list[str], extra_env: dict | None = None) -> str: parts = [] for a in argv: parts.append(f'"{a}"' if a.startswith("$(cat ") else shlex.quote(a)) # prompt expansion must stay double-quoted (WP29 A-006) env = " ".join(f"{k}={shlex.quote(str(v))}" for k, v in (extra_env or {}).items()) return (f"cd {shlex.quote(workspace)} && export PATH=/opt/homebrew/bin:/usr/local/bin:$PATH WP42_DEPTH=1 " f"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 DISABLE_AUTOUPDATER=1 {env} && set -a && . {NODE_ENV_FILE} && set +a && exec {' '.join(parts)} < /dev/null") def run_session(node: str, workspace: str, argv: list[str], raw_path: str, run_id: str, session_id: str, store, wall_seconds: int, seq_start: int = 0, extra_env: dict | None = None, on_event=None) -> dict: if os.environ.get("WP42_DEPTH"): raise LaunchError("refusing to launch: WP42_DEPTH is set (nested controller)") line = remote_shell_line(workspace, argv, extra_env) ssh = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", "-o", "ServerAliveInterval=30", node, line] started = utcnow(); t0 = time.monotonic() os.makedirs(os.path.dirname(raw_path), exist_ok=True) proc = subprocess.Popen(ssh, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1) result = {}; seq = seq_start; timed_out = False; err_buf = [] threading.Thread(target=lambda: err_buf.extend(proc.stderr), daemon=True).start() with open(raw_path, "a", encoding="utf-8") as raw: while True: if time.monotonic() - t0 > wall_seconds: timed_out = True; kill_remote(node, session_id) try: proc.wait(timeout=15) except subprocess.TimeoutExpired: proc.kill() break l = proc.stdout.readline() if not l: if proc.poll() is not None: break time.sleep(0.05); continue raw.write(l); raw.flush(); seq += 1 ev = normalise(l, run_id, seq, time.monotonic() - t0) if ev: ev["session_id"] = ev.get("session_id") or session_id; store.add_event(ev) if on_event: try: on_event(ev) except Exception: pass # noqa: BLE001 r = parse_result_line(l) if r: result = r if not timed_out: for l in proc.stdout: raw.write(l); seq += 1 ev = normalise(l, run_id, seq, time.monotonic() - t0) if ev: ev["session_id"] = ev.get("session_id") or session_id; store.add_event(ev) r = parse_result_line(l) if r: result = r return {"exit_code": proc.poll(), "result": result, "timed_out": timed_out, "seq_end": seq, "started_utc": started, "ended_utc": utcnow(), "wall_s": round(time.monotonic() - t0, 1), "stderr_tail": "".join(err_buf)[-2000:]} def kill_remote(node: str, pattern: str) -> int: cmd = f"pkill -f {shlex.quote(pattern)}; sleep 1; pkill -9 -f {shlex.quote(pattern)}; true" return subprocess.run(["ssh", "-o", "BatchMode=yes", node, cmd], capture_output=True, timeout=60).returncode def ssh(node: str, cmd: str, timeout=600, check=False, attempts: int = 4) -> subprocess.CompletedProcess: for i in range(attempts): r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=30", node, cmd], capture_output=True, text=True, timeout=timeout) if r.returncode != 255: break # 255 = ssh-level failure (connection refused/reset), retry; other codes are the remote command's time.sleep(3 * (i + 1) + 2) if check and r.returncode != 0: raise LaunchError(f"ssh {node} failed ({r.returncode}): {cmd[:200]}\n{r.stderr[-800:]}") return r def rsync_to(node: str, src_dir: str, dst_dir: str, excludes=(), delete=True) -> None: cmd = ["rsync", "-a", "-e", "ssh -o BatchMode=yes"] + (["--delete"] if delete else []) for e in excludes: cmd += ["--exclude", e] cmd += [src_dir.rstrip("/") + "/", f"{node}:{dst_dir.rstrip('/')}/"] for i in range(5): r = subprocess.run(cmd, capture_output=True, text=True, timeout=900) if r.returncode == 0: return time.sleep(3 * (i + 1) + 2) raise LaunchError(f"rsync to {node} failed: {r.stderr[-500:]}") def rsync_from(node: str, src_dir: str, dst_dir: str, excludes=(), attempts: int = 5) -> bool: """Copy a remote directory back. Retries with backoff: under 12 concurrent sessions the nodes' sshd MaxStartups throttle occasionally refuses a connection (pilot defect A-010: three runs lost their deliverables silently). Raises after the last attempt.""" os.makedirs(dst_dir, exist_ok=True) cmd = ["rsync", "-a", "-e", "ssh -o BatchMode=yes -o ConnectTimeout=30"] for e in excludes: cmd += ["--exclude", e] cmd += [f"{node}:{src_dir.rstrip('/')}/", dst_dir.rstrip("/") + "/"] last = "" for i in range(attempts): r = subprocess.run(cmd, capture_output=True, text=True, timeout=900) if r.returncode == 0: return True last = r.stderr[-400:]; time.sleep(3 * (i + 1) + 2) raise LaunchError(f"rsync from {node}:{src_dir} failed after {attempts} attempts: {last}")