#!/usr/bin/env python3 """Run a pi turn on an agent-seat while preserving its local session state.""" from __future__ import annotations import os import shlex import signal import subprocess import sys import time import uuid from pathlib import Path HERE = Path(__file__).resolve() sys.path.insert(0, str(HERE.parents[2] / "systray")) import remote_dispatch # noqa: E402 import containment # noqa: E402 def info_only(argv: list[str]) -> bool: if not argv or any(arg in {"--list-models", "--version", "--help", "-h"} for arg in argv): return True return "-p" not in argv and "--session-id" not in argv and not any( arg.startswith("--session-id=") for arg in argv ) def option_value(argv: list[str], option: str) -> str | None: for index, arg in enumerate(argv): if arg == option and index + 1 < len(argv): return argv[index + 1] if arg.startswith(option + "="): return arg.split("=", 1)[1] return None def replace_option(argv: list[str], option: str, value: str) -> list[str]: result = list(argv) for index, arg in enumerate(result): if arg == option and index + 1 < len(result): result[index + 1] = value return result if arg.startswith(option + "="): result[index] = f"{option}={value}" return result return result def append_record(*, mode: str, exit_code: int, started: float, host: str | None, workspace: str | None, reason: str | None = None) -> None: containment.append_record("pi", "pi", mode, exit_code, started, host=host, workspace=workspace, reason=reason) def main(argv: list[str]) -> int: if info_only(argv): os.execvp("pi", ["pi", *argv]) session_value = option_value(argv, "--session-dir") if not session_value: print("pi-remote: agent invocation requires --session-dir", file=sys.stderr) return remote_dispatch.EXIT_MIRROR session_dir = Path(session_value).expanduser().resolve() session_dir.mkdir(parents=True, exist_ok=True) started = time.monotonic() claim = None child: subprocess.Popen[bytes] | None = None interrupted = False access: dict | None = None pidfile: str | None = None # ssh without a pty cannot propagate a local kill to the remote command (verified: # a SIGTERM'd local ssh client leaves the remote process running), and ssh -tt # merges the remote stdout/stderr into one pty stream, which breaks agent_pi.py's # separated JSON-stdout / provider-failure-stderr protocol. A remote pidfile plus # an explicit `ssh ... kill` on the same signal is the only path that kills the # remote process without touching either stream. def stop(signum: int, _frame: object) -> None: nonlocal interrupted interrupted = True if access is not None and pidfile is not None: try: remote_pid = subprocess.run( [*remote_dispatch.ssh_argv(access), f"cat {shlex.quote(pidfile)} 2>/dev/null"], capture_output=True, text=True, timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ).stdout.strip() if remote_pid.isdigit(): subprocess.run( [*remote_dispatch.ssh_argv(access), f"kill -{signum} {remote_pid} 2>/dev/null"], timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ) except (OSError, subprocess.SubprocessError): pass if child is not None and child.poll() is None: child.send_signal(signum) old_term = signal.signal(signal.SIGTERM, stop) old_int = signal.signal(signal.SIGINT, stop) try: registry = remote_dispatch.load_registry() hosts = remote_dispatch.candidate_hosts(registry, roles=(remote_dispatch.SEAT_ROLE,)) host, access, claim = remote_dispatch.select_host(hosts) host_name = host.get("name") or access["host"] root = remote_dispatch.repo_root(Path.cwd()) rel_dir = remote_dispatch.remote_rel_dir(root) state_rel = remote_dispatch.extra_rel_dir(root, session_dir) mirror_files = remote_dispatch.git_mirror_files(root) made = subprocess.run( [*remote_dispatch.ssh_argv(access), f"mkdir -p {shlex.join([rel_dir, state_rel])} && printf '%s\\n' \"$HOME\""], capture_output=True, text=True, timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ) if made.returncode: raise remote_dispatch.OffloadUnavailable(f"{host_name}: could not prepare mirror: {made.stderr.strip()}", remote_dispatch.EXIT_MIRROR) push = remote_dispatch.sync_extra_dir(access, session_dir, state_rel, pull=False) if push: raise remote_dispatch.OffloadUnavailable(f"{host_name}: {push}", remote_dispatch.EXIT_MIRROR) # Reuse the checked tree mirror machinery after the bare-host selection above. git_mirror = remote_dispatch.git_mirror(root) tracked = subprocess.run(remote_dispatch.push_tracked_argv(access, root, rel_dir), input=mirror_files, capture_output=True, timeout=remote_dispatch.SYNC_TIMEOUT_SEC) pushed = subprocess.run(remote_dispatch.push_argv(access, root, rel_dir, exclude_git=git_mirror is not None), capture_output=True, text=True, timeout=remote_dispatch.SYNC_TIMEOUT_SEC) if tracked.returncode or pushed.returncode: detail = (tracked.stderr if tracked.returncode else pushed.stderr).strip() raise remote_dispatch.OffloadUnavailable(f"{host_name}: mirror push failed: {detail}", remote_dispatch.EXIT_MIRROR) node_home = made.stdout.strip().splitlines()[-1].rstrip("/") if git_mirror is not None: warning = remote_dispatch.prepare_remote_git( access, host_name, rel_dir, f"{node_home}/{rel_dir}", git_mirror ) if warning: print(f"pi-remote: {warning}", file=sys.stderr) remote_session = f"{node_home}/{state_rel}" remote_args = replace_option(argv, "--session-dir", remote_session) pidfile = f"{node_home}/{remote_dispatch.REMOTE_ROOT}/pi-remote/{uuid.uuid4().hex}.pid" # devtools.json installs the fleet's pi binary as `pi-agent` — `pi` on a buildbox's # PATH is reserved for the workstation's own account-routing wrapper of the same # name, which does not exist remotely, so the vendor entry avoids that collision. # # GUARD CHAIN: push_claude_home (modules/buildbox/lib/claude-home.sh) already ships # ~/.claude/bin (the PATH shims: _git-guard-shim.sh and friends) and ~/.claude/lib # (shim-guard.sh, worktree-guard-lib.sh) to every agent-seat box — pi_remote_dispatch # is not a second writer. What was missing is putting that dir FIRST on PATH for the # remote pi-agent process tree and marking it agent-launched, the same two things # _agent-build-scope does for claude/codex/cursor-agent locally. Without # AGENT_BUILD_SCOPE_ACTIVE=1 the shim resolves the real git and does nothing (a # human's own shell must never be gated); without ~/.claude/bin first on PATH, # `git` resolves straight to /usr/bin/git and the shim is never reached at all. # install-git-guard-real is idempotent (re-pins to the same real git each time) and # excludes its own dir when searching, so running it unconditionally before every # dispatch is cheap and self-healing against a box whose pin is stale or missing — # never a second writer for the pin file either, just this process priming its own # prerequisite before pi-agent can rely on it. guard_bin = f"{node_home}/.claude/bin" agent_bin = f"{node_home}/.local/bin" command = ( "mkdir -p {piddir} && cd {cwd} && echo $$ > {pidfile} && " "export PATH={guard_bin}:{agent_bin}:$PATH && export AGENT_BUILD_SCOPE_ACTIVE=1 && " "( {guard_bin}/install-git-guard-real >/dev/null 2>&1 || true ) && " "exec pi-agent {args}" ).format( piddir=shlex.quote(str(Path(pidfile).parent)), cwd=shlex.quote(f"{node_home}/{rel_dir}"), pidfile=shlex.quote(pidfile), args=shlex.join(remote_args), guard_bin=shlex.quote(guard_bin), agent_bin=shlex.quote(agent_bin), ) # ssh joins ALL trailing argv with spaces before sending it to the remote shell — # passing ["bash", "-c", command] as separate elements re-splits `command` on its # own internal spaces instead of handing it to `-c` whole. One pre-quoted string # is the only form ssh preserves (matches prepare_remote_git's git_pointer_command). remote_cmd = f"bash -c {shlex.quote(command)}" child = subprocess.Popen([*remote_dispatch.ssh_argv(access), remote_cmd]) code = child.wait() subprocess.run( [*remote_dispatch.ssh_argv(access), f"rm -f {shlex.quote(pidfile)}"], capture_output=True, timeout=remote_dispatch.PROBE_TIMEOUT_SEC, ) pull_state = remote_dispatch.sync_extra_dir(access, session_dir, state_rel, pull=True) pull_tree = remote_dispatch.Session( host_name, access, root, rel_dir, [], exclude_git=git_mirror is not None, mirror_files=mirror_files, ).pull_back() if pull_state or pull_tree: detail = pull_state or pull_tree print(f"pi-remote: {detail}", file=sys.stderr) append_record(mode="remote", exit_code=remote_dispatch.EXIT_MIRROR, started=started, host=host_name, workspace=f"{node_home}/{rel_dir}", reason=detail) return remote_dispatch.EXIT_MIRROR append_record(mode="remote", exit_code=code, started=started, host=host_name, workspace=f"{node_home}/{rel_dir}", reason="aborted" if interrupted else None) return code except remote_dispatch.OffloadUnavailable as exc: if claim is not None: claim.release() print(f"pi-remote: {exc}", file=sys.stderr) append_record(mode="aborted", exit_code=exc.code, started=started, host=None, workspace=None, reason=str(exc)) return exc.code except (OSError, subprocess.SubprocessError, IndexError) as exc: if claim is not None: claim.release() detail = f"remote dispatch failed: {exc}" print(f"pi-remote: {detail}", file=sys.stderr) append_record(mode="aborted", exit_code=remote_dispatch.EXIT_MIRROR, started=started, host=None, workspace=None, reason=detail) return remote_dispatch.EXIT_MIRROR finally: if claim is not None: claim.release() signal.signal(signal.SIGTERM, old_term) signal.signal(signal.SIGINT, old_int) if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))