1"""Bounded session launcher: runs `claude --bare -p` on an executor node over SSH, streams stream-json to a local raw file,2normalises events into the state store, enforces a wall-clock limit by killing the remote process (matched by session id),3and supports resumption of an existing session. Never logs the API key.45Nested-invocation guard: the executor is given no Agent tool and the launcher refuses to start if the environment variable6PGAR_DEPTH is already set (the controller runs at depth 0; executors run on another host at depth 1; nothing goes deeper).7"""8from __future__ import annotations9import json, os, shlex, subprocess, threading, time10from .contracts import utcnow, new_session_id11from .events import normalise, parse_result_line1213DEFAULT_TOOLS = ["Bash", "Read", "Write", "Edit", "Glob", "Grep"]14EXECUTOR_ENTRYPOINT = "claude" # allow-listed executor entry point1516class LaunchError(RuntimeError):17 pass1819def build_command(workspace: str, ctl_dir: str, prompt_file: str, system_append_file: str, model: str, session_id: str,20 max_budget_usd: float, max_turns: int, resume: bool = False, tools=None, python_env="~/wp29/.env", effort=None) -> list[str]:21 """Return the remote argv (list) for the executor process. The API key is sourced from the node-side env file."""22 tools = tools or DEFAULT_TOOLS23 argv = [EXECUTOR_ENTRYPOINT, "--bare", "-p", f"$(cat {shlex.quote(prompt_file)})",24 "--output-format", "stream-json", "--verbose",25 "--model", model,26 "--tools", ",".join(tools), "--allowedTools", *tools,27 "--permission-prompts", "none",28 "--max-budget-usd", f"{max_budget_usd:.2f}", "--max-turns", str(max_turns),29 "--append-system-prompt-file", system_append_file,30 "--no-chrome", "--disable-slash-commands"]31 if effort:32 argv += ["--effort", effort]33 if resume:34 argv += ["--resume", session_id]35 else:36 argv += ["--session-id", session_id]37 return argv3839def remote_shell_line(workspace: str, argv: list[str], env_file="~/wp29/.env") -> str:40 """Compose the remote shell line. `"$(cat prompt)"` is expanded by the remote shell inside double quotes so the whole file becomes one argument."""41 parts = []42 for a in argv:43 parts.append(f'"{a}"' if a.startswith("$(cat ") else shlex.quote(a)) # the prompt expansion MUST be double-quoted: unquoted, the shell word-splits it and only the first word reaches the client (defect found 2026-10-30, amendment A-006)44 cmd = " ".join(parts)45 return (f"cd {shlex.quote(workspace)} && export PATH=/opt/homebrew/bin:/usr/local/bin:$PATH PGAR_DEPTH=1 "46 f"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC=1 DISABLE_AUTOUPDATER=1 && set -a && . {env_file} && set +a && exec {cmd}")4748def run_session(node: str, workspace: str, argv: list[str], raw_path: str, run_id: str, session_id: str, store, wall_seconds: int,49 seq_start: int = 0, on_event=None, env_file="~/wp29/.env") -> dict:50 """Execute one session; return {'exit_code','result','timed_out','n_events','seq_end','started_utc','ended_utc'}."""51 if os.environ.get("PGAR_DEPTH"):52 raise LaunchError("refusing to launch: PGAR_DEPTH is set (nested controller)")53 line = remote_shell_line(workspace, argv, env_file)54 ssh = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", "-o", "ServerAliveInterval=30", node, line]55 started = utcnow(); t0 = time.monotonic()56 os.makedirs(os.path.dirname(raw_path), exist_ok=True)57 proc = subprocess.Popen(ssh, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)58 result = {}; seq = seq_start; timed_out = False; stderr_buf = []59 def _drain_err():60 for l in proc.stderr:61 stderr_buf.append(l)62 threading.Thread(target=_drain_err, daemon=True).start()63 with open(raw_path, "a", encoding="utf-8") as raw:64 while True:65 if time.monotonic() - t0 > wall_seconds:66 timed_out = True67 kill_remote(node, session_id)68 try: proc.wait(timeout=15)69 except subprocess.TimeoutExpired: proc.kill()70 break71 l = proc.stdout.readline()72 if not l:73 if proc.poll() is not None:74 break75 time.sleep(0.05); continue76 raw.write(l); raw.flush()77 seq += 178 ev = normalise(l, run_id, seq, time.monotonic() - t0)79 if ev:80 ev["session_id"] = ev.get("session_id") or session_id81 store.add_event(ev)82 if on_event:83 try: on_event(ev)84 except Exception: pass # noqa: BLE00185 r = parse_result_line(l)86 if r: result = r87 # drain remainder after process end88 if not timed_out:89 for l in proc.stdout:90 raw.write(l); seq += 191 ev = normalise(l, run_id, seq, time.monotonic() - t0)92 if ev:93 ev["session_id"] = ev.get("session_id") or session_id; store.add_event(ev)94 r = parse_result_line(l)95 if r: result = r96 code = proc.poll()97 ended = utcnow()98 err = "".join(stderr_buf)[-2000:]99 return {"exit_code": code, "result": result, "timed_out": timed_out, "seq_end": seq, "started_utc": started, "ended_utc": ended,100 "wall_s": round(time.monotonic() - t0, 1), "stderr_tail": err}101102def kill_remote(node: str, session_id: str) -> int:103 """Kill the executor process on the node by matching its unique session id on the command line (and children)."""104 cmd = f"pkill -f {shlex.quote(session_id)}; sleep 1; pkill -9 -f {shlex.quote(session_id)}; true"105 return subprocess.run(["ssh", "-o", "BatchMode=yes", node, cmd], capture_output=True, timeout=60).returncode106107def ssh(node: str, cmd: str, timeout=600, check=False) -> subprocess.CompletedProcess:108 r = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20", node, cmd], capture_output=True, text=True, timeout=timeout)109 if check and r.returncode != 0:110 raise LaunchError(f"ssh {node} failed ({r.returncode}): {cmd[:200]}\n{r.stderr[-800:]}")111 return r112113def rsync_to(node: str, src_dir: str, dst_dir: str, excludes=()) -> None:114 cmd = ["rsync", "-a", "--delete", "-e", "ssh -o BatchMode=yes"]115 for e in excludes: cmd += ["--exclude", e]116 cmd += [src_dir.rstrip("/") + "/", f"{node}:{dst_dir.rstrip('/')}/"]117 r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)118 if r.returncode != 0:119 raise LaunchError(f"rsync to {node} failed: {r.stderr[-500:]}")120121def rsync_from(node: str, src_dir: str, dst_dir: str, excludes=()) -> bool:122 os.makedirs(dst_dir, exist_ok=True)123 cmd = ["rsync", "-a", "-e", "ssh -o BatchMode=yes"]124 for e in excludes: cmd += ["--exclude", e]125 cmd += [f"{node}:{src_dir.rstrip('/')}/", dst_dir.rstrip("/") + "/"]126 r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)127 return r.returncode == 0128