1"""Bounded session launcher: runs `claude --bare -p` on an executor node over SSH, streams stream-json to an immutable raw file,2normalises events into the store, enforces a wall-clock limit by killing the remote process (matched by its unique session id),3and supports one bounded resume (arm A4). The API key is read on the node from ~/wp42/.env; it is never logged.45Recursion guard: workers receive no Agent/Task tool; the launcher refuses to start if WP42_DEPTH is set in its own environment."""6from __future__ import annotations7import os, shlex, subprocess, threading, time8from .contracts import utcnow9from .events import normalise, parse_result_line10from .config import NODE_ENV_FILE1112EXECUTOR = "claude"1314class LaunchError(RuntimeError):15 pass1617def build_argv(prompt_file: str, system_append_file: str, model: str, session_id: str, max_budget_usd: float, max_turns: int,18 tools: list[str], disallowed: list[str], resume: bool = False, effort: str | None = None) -> list[str]:19 argv = [EXECUTOR, "--bare", "-p", f"$(cat {shlex.quote(prompt_file)})", "--output-format", "stream-json", "--verbose",20 "--model", model, "--tools", ",".join(tools), "--allowedTools", *tools, "--permission-prompts", "none",21 "--max-budget-usd", f"{max_budget_usd:.2f}", "--max-turns", str(max_turns),22 "--append-system-prompt-file", system_append_file, "--no-chrome", "--disable-slash-commands"] # session persistence is required for the A4 --resume (defect A-012)23 if disallowed: argv += ["--disallowedTools", *disallowed]24 if effort: argv += ["--effort", effort]25 argv += (["--resume", session_id] if resume else ["--session-id", session_id])26 return argv2728def remote_shell_line(workspace: str, argv: list[str], extra_env: dict | None = None) -> str:29 parts = []30 for a in argv:31 parts.append(f'"{a}"' if a.startswith("$(cat ") else shlex.quote(a)) # prompt expansion must stay double-quoted (WP29 A-006)32 env = " ".join(f"{k}={shlex.quote(str(v))}" for k, v in (extra_env or {}).items())33 return (f"cd {shlex.quote(workspace)} && export PATH=/opt/homebrew/bin:/usr/local/bin:$PATH WP42_DEPTH=1 "34 f"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 DISABLE_AUTOUPDATER=1 {env} && set -a && . {NODE_ENV_FILE} && set +a && exec {' '.join(parts)} < /dev/null")3536def run_session(node: str, workspace: str, argv: list[str], raw_path: str, run_id: str, session_id: str, store, wall_seconds: int,37 seq_start: int = 0, extra_env: dict | None = None, on_event=None) -> dict:38 if os.environ.get("WP42_DEPTH"):39 raise LaunchError("refusing to launch: WP42_DEPTH is set (nested controller)")40 line = remote_shell_line(workspace, argv, extra_env)41 ssh = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", "-o", "ServerAliveInterval=30", node, line]42 started = utcnow(); t0 = time.monotonic()43 os.makedirs(os.path.dirname(raw_path), exist_ok=True)44 proc = subprocess.Popen(ssh, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)45 result = {}; seq = seq_start; timed_out = False; err_buf = []46 threading.Thread(target=lambda: err_buf.extend(proc.stderr), daemon=True).start()47 with open(raw_path, "a", encoding="utf-8") as raw:48 while True:49 if time.monotonic() - t0 > wall_seconds:50 timed_out = True; kill_remote(node, session_id)51 try: proc.wait(timeout=15)52 except subprocess.TimeoutExpired: proc.kill()53 break54 l = proc.stdout.readline()55 if not l:56 if proc.poll() is not None: break57 time.sleep(0.05); continue58 raw.write(l); raw.flush(); seq += 159 ev = normalise(l, run_id, seq, time.monotonic() - t0)60 if ev:61 ev["session_id"] = ev.get("session_id") or session_id; store.add_event(ev)62 if on_event:63 try: on_event(ev)64 except Exception: pass # noqa: BLE00165 r = parse_result_line(l)66 if r: result = r67 if not timed_out:68 for l in proc.stdout:69 raw.write(l); seq += 170 ev = normalise(l, run_id, seq, time.monotonic() - t0)71 if ev: ev["session_id"] = ev.get("session_id") or session_id; store.add_event(ev)72 r = parse_result_line(l)73 if r: result = r74 return {"exit_code": proc.poll(), "result": result, "timed_out": timed_out, "seq_end": seq, "started_utc": started, "ended_utc": utcnow(),75 "wall_s": round(time.monotonic() - t0, 1), "stderr_tail": "".join(err_buf)[-2000:]}7677def kill_remote(node: str, pattern: str) -> int:78 cmd = f"pkill -f {shlex.quote(pattern)}; sleep 1; pkill -9 -f {shlex.quote(pattern)}; true"79 return subprocess.run(["ssh", "-o", "BatchMode=yes", node, cmd], capture_output=True, timeout=60).returncode8081def ssh(node: str, cmd: str, timeout=600, check=False, attempts: int = 4) -> subprocess.CompletedProcess:82 for i in range(attempts):83 r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=30", node, cmd], capture_output=True, text=True, timeout=timeout)84 if r.returncode != 255: break # 255 = ssh-level failure (connection refused/reset), retry; other codes are the remote command's85 time.sleep(3 * (i + 1) + 2)86 if check and r.returncode != 0:87 raise LaunchError(f"ssh {node} failed ({r.returncode}): {cmd[:200]}\n{r.stderr[-800:]}")88 return r8990def rsync_to(node: str, src_dir: str, dst_dir: str, excludes=(), delete=True) -> None:91 cmd = ["rsync", "-a", "-e", "ssh -o BatchMode=yes"] + (["--delete"] if delete else [])92 for e in excludes: cmd += ["--exclude", e]93 cmd += [src_dir.rstrip("/") + "/", f"{node}:{dst_dir.rstrip('/')}/"]94 for i in range(5):95 r = subprocess.run(cmd, capture_output=True, text=True, timeout=900)96 if r.returncode == 0: return97 time.sleep(3 * (i + 1) + 2)98 raise LaunchError(f"rsync to {node} failed: {r.stderr[-500:]}")99100def rsync_from(node: str, src_dir: str, dst_dir: str, excludes=(), attempts: int = 5) -> bool:101 """Copy a remote directory back. Retries with backoff: under 12 concurrent sessions the nodes' sshd MaxStartups throttle102 occasionally refuses a connection (pilot defect A-010: three runs lost their deliverables silently). Raises after the last attempt."""103 os.makedirs(dst_dir, exist_ok=True)104 cmd = ["rsync", "-a", "-e", "ssh -o BatchMode=yes -o ConnectTimeout=30"]105 for e in excludes: cmd += ["--exclude", e]106 cmd += [f"{node}:{src_dir.rstrip('/')}/", dst_dir.rstrip("/") + "/"]107 last = ""108 for i in range(attempts):109 r = subprocess.run(cmd, capture_output=True, text=True, timeout=900)110 if r.returncode == 0: return True111 last = r.stderr[-400:]; time.sleep(3 * (i + 1) + 2)112 raise LaunchError(f"rsync from {node}:{src_dir} failed after {attempts} attempts: {last}")113