warpgate-ssh: remote Linux/macOS terminals through the relay; oregon launched remotely

The dispatcher moves to an LXC without agent CLIs (design 9), so every terminal is
reached through Warpgate. The relay call (agent-access.py run --target <t>) lives in
one helper; the target comes from the terminal config.

os linux/darwin (PosixWarpgateTransport): each primitive is one bash -c call. Files
go up as base64 (chunked above 60k), start_run stages prompt.md + run.sh and launches
in a single round trip; the runner is detached with setsid nohup (pid == pgid == sid,
$! kept in launch.pid); kill signals the process group and session (TERM, then KILL),
skipped once exit.txt exists. macOS detaches with perl POSIX::setsid.

os windows keeps the win10-ps.sh / win10-put.sh helper scripts (target win10) and
schtasks, now through the shared call() helper.

oregon is now warpgate-ssh (target oregon, run_root ~/.gwent-board/agentd-runs);
the local transport stays for co-located setups and tests.

agentd/config.py

@@ -91,11 +91,19 @@ def semantic_errors(data):
         for ex in t.get("executors", []):
             if ex not in execs:
                 errs.append("terminals.%s: unknown executor '%s'" % (tname, ex))
-        if t.get("transport") == "warpgate-ssh":
-            if t.get("os") != "windows" or t.get("launch") != "schtasks":
-                errs.append("terminals.%s: warpgate-ssh currently supports only os=windows, launch=schtasks" % tname)
-            if t.get("target") != "win10":
-                errs.append("terminals.%s: warpgate-ssh helper scripts only reach target 'win10' for now" % tname)
+        if t.get("transport") == "warpgate-ssh" and t.get("os") == "windows":
+            if t.get("launch") != "schtasks":
+                errs.append("terminals.%s: warpgate-ssh on windows needs launch=schtasks" % tname)
+            if t.get("target", tname) != "win10":
+                errs.append("terminals.%s: the Windows helper scripts only reach target 'win10' for now" % tname)
+        elif t.get("transport") == "warpgate-ssh":
+            if t.get("launch") != "nohup":
+                errs.append("terminals.%s: warpgate-ssh on %s needs launch=nohup" % (tname, t.get("os")))
+            for key in ("repo", "run_root"):
+                if key in t and not str(t[key]).startswith("/"):
+                    errs.append("terminals.%s.%s: must be an absolute path on the remote (no ~)" % (tname, key))
+            if not data.get("vault", {}).get("config"):
+                errs.append("terminals.%s: remote POSIX terminals need vault.config (relay identity)" % tname)
         if t.get("transport") == "local" and t.get("os") != "linux":
             errs.append("terminals.%s: local transport supports only os=linux" % tname)
     for ename, e in execs.items():

agentd/executors/base.py

@@ -87,13 +87,16 @@ class RunnerExecutor(Executor):
     def start(self, run):
         t = self.transport
         run_dir = t.join(self.run_root(), run.run_id)
-        t.mkdir(run_dir, clean=True)
-        t.write_text(t.join(run_dir, "prompt.md"), run.prompt)
         script_name, script = t.runner(run_dir, run.repo, self.command(run, run_dir), self.stdout_name)
-        script_path = t.join(run_dir, script_name)
-        t.write_text(script_path, script)
         task = "agentd-" + run.run_id
-        t.launch(task, script_path, run_dir)
+        files = {"prompt.md": run.prompt, script_name: script}
+        if hasattr(t, "start_run"):  # remote POSIX: stage + launch in one relay round trip
+            t.start_run(task, run_dir, files, script_name)
+        else:
+            t.mkdir(run_dir, clean=True)
+            for name, text in files.items():
+                t.write_text(t.join(run_dir, name), text)
+            t.launch(task, t.join(run_dir, script_name), run_dir)
         return {"run_dir": run_dir, "task": task, "thread_id": run.thread_id}
 
     def poll(self, handle):

agentd/transports/local.py

@@ -4,17 +4,23 @@ import signal
 import subprocess
 from pathlib import Path
 
+# Shared with the remote POSIX transport (warpgate_ssh): plain sh + `date +FMT`, so it also runs on macOS.
 RUNNER = """#!/usr/bin/env bash
-cd {cwd} || {{ echo "exit=97 at $(date -Is) (cd failed)" > {d}/exit.txt; exit 97; }}
+cd {cwd} || {{ echo "exit=97 at $(date +%Y-%m-%dT%H:%M:%S%z) (cd failed)" > {d}/exit.txt; exit 97; }}
 echo $$ > {d}/pid.txt
-date -Is > {d}/started.txt
+date +%Y-%m-%dT%H:%M:%S%z > {d}/started.txt
 (
 {command}
 ) < {d}/prompt.md > {d}/{stdout} 2> {d}/stderr.txt
-echo "exit=$? at $(date -Is)" > {d}/exit.txt
+echo "exit=$? at $(date +%Y-%m-%dT%H:%M:%S%z)" > {d}/exit.txt
 """
 
 
+def sh_quote(s):
+    """POSIX shell single-quote literal (always quoted, `'` -> `'\\''`)."""
+    return "'" + str(s).replace("'", "'\\''") + "'"
+
+
 class LocalTransport:
     os = "linux"
 
@@ -28,7 +34,7 @@ class LocalTransport:
         return os.path.join(*parts)
 
     def quote(self, s):
-        return "'" + str(s).replace("'", "'\\''") + "'"
+        return sh_quote(s)
 
     def mkdir(self, path, clean=False):
         p = Path(path)

agentd/transports/warpgate_ssh.py

@@ -1,16 +1,59 @@
-"""Windows terminal behind Warpgate, reached through the server-access relay helper scripts:
-  win10-ps.sh  <script>          run PowerShell (sent as -EncodedCommand)
-  win10-put.sh <local> <remote>  upload a file (gzip+base64 chunks, SHA-256 verified)
-Runs are launched as one-shot scheduled tasks (schtasks) so they outlive the SSH session.
-Note: the helper scripts currently hard-code `--target win10`; `target` is checked against that."""
+"""Terminals behind Warpgate, reached through the server-access relay client:
+
+    agent-access.py --config <identity> run --target <target> --timeout N -- <argv...>
+
+The client shlex-joins argv and the target account's login shell runs it (cmd.exe on Windows); stdin is
+/dev/null, so everything travels in argv. `relay()` is the single place that builds that call.
+
+- os linux / darwin -> PosixWarpgateTransport: every operation is one `bash -c '<script>'` relay call. Files go up
+  as base64 chunks (`printf %s '<b64>' | base64 -d`), the runner starts detached with `setsid nohup bash run.sh`
+  (its own session: pid == pgid == sid), `poll` reads exit.txt, `kill` signals the process group and session.
+- os windows -> WindowsWarpgateTransport: still drives the helper scripts win10-ps.sh (PowerShell
+  -EncodedCommand) and win10-put.sh (chunked upload, SHA-256 verified), which hard-code `--target win10`;
+  runs are one-shot scheduled tasks (schtasks) so they outlive the SSH session.
+  TODO: move Windows onto relay() (`powershell -EncodedCommand` as argv) and drop the helper scripts."""
 import base64
 import hashlib
 import os
+import posixpath
 import re
 import subprocess
+import sys
 import tempfile
 
 from . import TransportError
+from .local import RUNNER as POSIX_RUNNER, sh_quote
+
+DEFAULT_CLIENT = "~/.local/share/server-access-client/agent-access.py"
+
+
+def relay_argv(cfg, target, argv, timeout):
+    """argv for one `agent-access.py run` call (the vault client and identity come from the `vault` config)."""
+    v = cfg.data.get("vault") or {}
+    cmd = [sys.executable or "python3", "-I", os.path.expanduser(v.get("client") or DEFAULT_CLIENT)]
+    if v.get("config"):
+        cmd += ["--config", v["config"]]
+    return cmd + ["run", "--target", target, "--timeout", str(int(timeout)), "--"] + list(argv)
+
+
+def call(cfg, argv, timeout, extra_env=None):
+    """Run a local relay process (client or helper script) with Plane secrets scrubbed from its environment.
+    -> (returncode, stdout, stderr); raises TransportError on timeout / launch failure."""
+    env = dict(cfg.scrubbed_env(), **(extra_env or {}))
+    try:
+        out = subprocess.run(argv, capture_output=True, timeout=timeout + 60, env=env, stdin=subprocess.DEVNULL)
+    except subprocess.TimeoutExpired:
+        raise TransportError("relay timeout after %ss" % timeout) from None
+    except OSError as exc:
+        raise TransportError("relay launch failed: %s" % exc) from None
+    dec = lambda b: (b or b"").decode("utf-8", "replace").replace("\r", "")  # noqa: E731
+    return out.returncode, dec(out.stdout), dec(out.stderr)
+
+
+def relay(cfg, target, argv, timeout=150):
+    """Run argv on a Warpgate target. -> (returncode, stdout, stderr)."""
+    return call(cfg, relay_argv(cfg, target, argv, timeout), timeout)
+
 
 RUNNER = r"""$ErrorActionPreference = 'Continue'
 [Console]::OutputEncoding = [Text.Encoding]::UTF8
@@ -32,7 +75,7 @@ def _b64_payload(out, marker):
     return base64.b64decode(m.group(1)).decode("utf-8", "replace") if m.group(1) else ""
 
 
-class WarpgateSshTransport:
+class WindowsWarpgateTransport:
     os = "windows"
 
     def __init__(self, cfg, terminal):
@@ -54,21 +97,17 @@ class WarpgateSshTransport:
 
     # ---------------------------------------------------------------- relay primitives
     def ps(self, script, timeout=150):
-        env = dict(self.cfg.scrubbed_env(), WIN10_TIMEOUT=str(timeout))
-        try:
-            out = subprocess.run([self.ps_script, script], capture_output=True, text=True, timeout=timeout + 60, env=env)
-        except subprocess.TimeoutExpired:
-            raise TransportError("relay timeout after %ss" % timeout) from None
-        if out.returncode != 0 and not out.stdout.strip():
-            raise TransportError("relay exit %s: %s" % (out.returncode, out.stderr.strip()[-300:]))
-        return out.stdout.replace("\r", "")
+        rc, out, err = call(self.cfg, [self.ps_script, script], timeout, {"WIN10_TIMEOUT": str(timeout)})
+        if rc != 0 and not out.strip():
+            raise TransportError("relay exit %s: %s" % (rc, err.strip()[-300:]))
+        return out
 
     def put(self, local, remote):
-        sha = hashlib.sha256(open(local, "rb").read()).hexdigest()
-        out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600,
-                             env=self.cfg.scrubbed_env())
-        if ("remote sha256 " + sha) not in out.stdout:
-            raise TransportError("upload of %s failed: %s" % (remote, (out.stdout + out.stderr).strip()[-300:]))
+        with open(local, "rb") as fh:
+            sha = hashlib.sha256(fh.read()).hexdigest()
+        rc, out, err = call(self.cfg, [self.put_script, local, remote], 540)
+        if ("remote sha256 " + sha) not in out:
+            raise TransportError("upload of %s failed: %s" % (remote, (out + err).strip()[-300:]))
 
     # ---------------------------------------------------------------- file ops
     def mkdir(self, path, clean=False):
@@ -146,3 +185,177 @@ class WarpgateSshTransport:
             return None
         m = re.search(r"COUNT=(\d+)", out)
         return int(m.group(1)) if m else None
+
+
+# ==================================================================== Linux / macOS
+DETACH = {
+    # bash puts no background job in its own process group (no job control), so setsid(1) does not fork:
+    # $! is the runner itself and becomes pid == pgid == sid.
+    "linux": "setsid nohup bash %s",
+    # macOS has no setsid(1); perl's POSIX::setsid does the same in-process before exec.
+    "darwin": "nohup perl -MPOSIX -e 'POSIX::setsid() or die \"setsid: $!\"; exec @ARGV' bash %s",
+}
+
+
+class PosixWarpgateTransport:
+    """Linux/macOS terminal through the relay. Each primitive is one `bash -c` round trip (~10 s through
+    Vault + Warpgate), so `start_run` stages the run directory and launches it in a single call."""
+    CHUNK = 60000    # base64 chars per call: one argv element must stay below 128 KiB (MAX_ARG_STRLEN) on both ends
+
+    def __init__(self, cfg, terminal):
+        self.cfg, self.terminal = cfg, terminal
+        t = cfg.terminals[terminal]
+        self.os = t["os"]
+        self.target = t.get("target", terminal)
+        self.repo = t["repo"]
+
+    def default_run_root(self):
+        return posixpath.join(self.repo, "runtime", "agentd")
+
+    def join(self, *parts):
+        return posixpath.join(*parts)
+
+    def quote(self, s):
+        return sh_quote(s)
+
+    # ---------------------------------------------------------------- relay primitive
+    def sh(self, script, timeout=150, marker=None):
+        """Run a bash script on the target. With `marker`, its absence in stdout is an error."""
+        rc, out, err = relay(self.cfg, self.target, ["bash", "-c", script], timeout)
+        if (marker and marker not in out) or (not marker and rc != 0 and not out.strip()):
+            raise TransportError("relay %s exit %s: %s" % (self.target, rc, (out + err).strip()[-300:]))
+        return out
+
+    # ---------------------------------------------------------------- file ops
+    def mkdir_script(self, path, clean=False):
+        q = self.quote(path)
+        s = "mkdir -p %s || exit 1\n" % q
+        if clean:
+            s += "find %s -maxdepth 1 -type f -exec rm -f {} + || exit 1\n" % q
+        return s
+
+    def mkdir(self, path, clean=False):
+        self.sh(self.mkdir_script(path, clean) + "echo MKDIR=ok\n", marker="MKDIR=ok")
+
+    @staticmethod
+    def b64(text):
+        return base64.b64encode(text.encode("utf-8")).decode("ascii")
+
+    def put_script(self, path, b64):
+        """Write a base64 payload (alphabet has no quote characters) to path atomically, echo its byte size."""
+        q, part = self.quote(path), self.quote(path + ".part")
+        return ("printf %%s '%s' | base64 -d > %s && mv -f %s %s || exit 1\n"
+                "echo \"WROTE=$(wc -c < %s | tr -d ' ')\"\n" % (b64, part, part, q, q))
+
+    def write_text(self, path, text):
+        data = self.b64(text)
+        size = len(text.encode("utf-8"))
+        if len(data) <= self.CHUNK:
+            out = self.sh(self.put_script(path, data), marker="WROTE=")
+        else:  # append chunks to <path>.b64, decode on the last call
+            q, qb = self.quote(path), self.quote(path + ".b64")
+            chunks = [data[i:i + self.CHUNK] for i in range(0, len(data), self.CHUNK)]
+            for i, c in enumerate(chunks):
+                s = "printf %%s '%s' %s %s || exit 1\n" % (c, ">" if i == 0 else ">>", qb)
+                if i < len(chunks) - 1:
+                    self.sh(s + "echo CHUNK=ok\n", marker="CHUNK=ok")
+                else:
+                    out = self.sh(s + "base64 -d < %s > %s && rm -f %s || exit 1\necho \"WROTE=$(wc -c < %s | tr -d ' ')\"\n"
+                                  % (qb, q, qb, q), marker="WROTE=")
+        if re.findall(r"WROTE=(\d+)", out) != [str(size)]:
+            raise TransportError("upload of %s: size mismatch (%s)" % (path, out.strip()[-200:]))
+
+    def read_text(self, path):
+        q = self.quote(path)
+        out = self.sh("if [ -f %s ]; then printf B64=; base64 < %s | tr -d '\\n'; echo; else echo MISSING; fi" % (q, q),
+                      timeout=200)
+        if "B64=" not in out:
+            if "MISSING" in out:
+                return None
+            raise TransportError("read %s: unexpected relay output" % path)
+        return _b64_payload(out, "B64")
+
+    def _b64_cmd(self, pipeline):
+        out = self.sh("printf B64=; { %s; } 2>/dev/null | base64 | tr -d '\\n'; echo" % pipeline)
+        text = _b64_payload(out, "B64")
+        if text is None:
+            raise TransportError("relay %s: unexpected output" % self.target)
+        return text
+
+    def grep_first(self, path, needle):
+        line = self._b64_cmd("grep -F -m1 -e %s %s" % (self.quote(needle), self.quote(path)))
+        return line.rstrip("\n") or None
+
+    def tail(self, path, n):
+        return self._b64_cmd("tail -n %d %s" % (int(n), self.quote(path))).rstrip("\n")
+
+    # ---------------------------------------------------------------- processes
+    def runner(self, run_dir, cwd, command, stdout_name):
+        return "run.sh", POSIX_RUNNER.format(cwd=self.quote(cwd), d=self.quote(run_dir), command=command,
+                                             stdout=stdout_name)
+
+    def launch_script(self, run_dir, script_name):
+        """Detach the runner from the SSH session; stdio closed so the channel ends at once. $! -> launch.pid
+        (kill falls back to it until run.sh has written pid.txt)."""
+        d = self.quote(run_dir)
+        return ("cd %s || exit 1\n"
+                "%s </dev/null >/dev/null 2>&1 &\n"
+                "echo $! > launch.pid\n"
+                "echo LAUNCHED=$!\n" % (d, DETACH[self.os] % self.quote(script_name)))
+
+    def launch(self, task, script_path, run_dir):
+        self.sh(self.launch_script(run_dir, posixpath.basename(script_path)), marker="LAUNCHED=")
+
+    def start_run(self, task, run_dir, files, script_name):
+        """mkdir + clean, write `files` {name: text}, launch `script_name` -- one relay call unless a file is
+        larger than CHUNK (those go up first through write_text)."""
+        big = {n: t for n, t in files.items() if len(self.b64(t)) > self.CHUNK}
+        script = self.mkdir_script(run_dir, clean=True)
+        if big:
+            self.mkdir(run_dir, clean=True)
+            for name, text in big.items():
+                self.write_text(self.join(run_dir, name), text)
+            script = ""
+        sizes = {}
+        for name, text in files.items():
+            if name not in big:
+                script += self.put_script(self.join(run_dir, name), self.b64(text))
+                sizes[name] = len(text.encode("utf-8"))
+        out = self.sh(script + self.launch_script(run_dir, script_name), marker="LAUNCHED=")
+        wrote = re.findall(r"WROTE=(\d+)", out)
+        if wrote != [str(v) for v in sizes.values()]:
+            raise TransportError("staging %s: size mismatch %s != %s" % (run_dir, wrote, list(sizes.values())))
+
+    def kill_script(self, run_dir):
+        """TERM then KILL the runner's process group and session. Skipped once exit.txt exists (the runner is
+        done and its pid may be reused)."""
+        d = self.quote(run_dir)
+        return ("d=%s\n"
+                "if [ -f \"$d/exit.txt\" ]; then echo KILLED=finished; exit 0; fi\n"
+                "p=$(head -n1 \"$d/pid.txt\" 2>/dev/null); [ -n \"$p\" ] || p=$(head -n1 \"$d/launch.pid\" 2>/dev/null)\n"
+                "case \"$p\" in ''|*[!0-9]*) echo KILLED=nopid; exit 0;; esac\n"
+                "kill -TERM -- \"-$p\" 2>/dev/null; pkill -TERM -s \"$p\" 2>/dev/null\n"
+                "for i in 1 2 3 4 5; do pgrep -s \"$p\" >/dev/null 2>&1 || break; sleep 1; done\n"
+                "kill -KILL -- \"-$p\" 2>/dev/null; pkill -KILL -s \"$p\" 2>/dev/null\n"
+                "sleep 1; echo \"KILLED=ok left=$(pgrep -s \"$p\" 2>/dev/null | wc -l | tr -d ' ')\"\n" % d)
+
+    def kill(self, task, run_dir):
+        self.sh(self.kill_script(run_dir), marker="KILLED=")
+
+    def forget(self, task):
+        pass
+
+    def process_count(self, name):
+        try:
+            out = self.sh("echo COUNT=$(pgrep -x %s | wc -l | tr -d ' ')" % self.quote(name), timeout=60)
+        except TransportError:
+            return None
+        m = re.search(r"COUNT=(\d+)", out)
+        return int(m.group(1)) if m else None
+
+
+def WarpgateSshTransport(cfg, terminal):
+    """Factory: the transport class for a warpgate-ssh terminal follows its `os`."""
+    if cfg.terminals[terminal]["os"] == "windows":
+        return WindowsWarpgateTransport(cfg, terminal)
+    return PosixWarpgateTransport(cfg, terminal)

config/agentd.example.yaml

@@ -7,7 +7,7 @@ vault: {client: ~/.local/share/server-access-client/agent-access.py, config: htt
 
 agentd:
   state_dir: state                      # relative to the repo root; JSON state, logs, heartbeat, local run dirs
-  helpers_dir: /home/ubuntu/Documents/wjj/project/gwent-design/tools   # win10-ps.sh / win10-put.sh
+  helpers_dir: /home/ubuntu/Documents/wjj/project/gwent-design/tools   # win10-ps.sh / win10-put.sh (Windows terminals only)
   interval: 120                         # seconds between ticks in --loop mode
   max_rounds_per_thread: 8              # rebuild the executor thread (with a handoff summary) after this many rounds
   comment_max_chars: 6000
@@ -29,14 +29,19 @@ terminals:
     capabilities: [android, game]
     executors: [codex]
   oregon:
-    transport: local
+    # remote even though agentd runs on oregon today: the dispatcher is moving to an LXC without agent CLIs
+    # (design §9), so this goes through Warpgate like any other host. `transport: local` + `launch: subprocess`
+    # still works for a dispatcher co-located with its executors.
+    transport: warpgate-ssh
+    target: oregon                      # agent-access.py target name
     os: linux
-    launch: subprocess
+    launch: nohup
     repo: /home/ubuntu/Documents/wjj/project/gwent
+    run_root: /home/ubuntu/.gwent-board/agentd-runs   # absolute: the remote shell does not expand ~ inside quotes
     plane_user: oregon
     token_key: PLANE_OREGON_API_TOKEN
     concurrency: 2
-    bins: {claude: /home/ubuntu/.local/bin/claude}   # cron has a minimal PATH
+    bins: {claude: /home/ubuntu/.local/bin/claude}   # the SSH shell has the system PATH only
     executors: [claude-code, codex, shell]
 
 executors:

config/agentd.yaml

@@ -5,7 +5,7 @@ vault: {client: ~/.local/share/server-access-client/agent-access.py, config: htt
 
 agentd:
   state_dir: ~/.gwent-board/agentd     # JSON state, agentd.log, heartbeat.json, local run dirs (runs/)
-  helpers_dir: /home/ubuntu/Documents/wjj/project/gwent-design/tools   # win10-ps.sh / win10-put.sh
+  helpers_dir: /home/ubuntu/Documents/wjj/project/gwent-design/tools   # win10-ps.sh / win10-put.sh (Windows terminals only)
   interval: 120
   max_rounds_per_thread: 8
   comment_max_chars: 6000
@@ -27,14 +27,19 @@ terminals:
     capabilities: [android, game]
     executors: [codex]
   oregon:
-    transport: local
+    # remote even though agentd runs on oregon today: the dispatcher is moving to an LXC without agent CLIs
+    # (design §9), so this goes through Warpgate like any other host. `transport: local` + `launch: subprocess`
+    # still works for a dispatcher co-located with its executors.
+    transport: warpgate-ssh
+    target: oregon                      # agent-access.py target name
     os: linux
-    launch: subprocess
+    launch: nohup
     repo: /home/ubuntu/Documents/wjj/project/gwent
+    run_root: /home/ubuntu/.gwent-board/agentd-runs   # absolute: the remote shell does not expand ~ inside quotes
     plane_user: oregon
     token_key: PLANE_OREGON_API_TOKEN
     concurrency: 1                      # one shared worktree: no parallel edits
-    bins: {claude: /home/ubuntu/.local/bin/claude}   # cron has a minimal PATH; codex is not installed on oregon
+    bins: {claude: /home/ubuntu/.local/bin/claude}   # the SSH non-login shell has the system PATH only; no codex on oregon
     executors: [claude-code]
 
 executors:

config/schema.json

@@ -48,7 +48,7 @@
       "properties": {
         "transport": {"enum": ["local", "warpgate-ssh"]},
         "target": {"type": "string"},
-        "os": {"enum": ["linux", "windows"]},
+        "os": {"enum": ["linux", "darwin", "windows"]},
         "launch": {"enum": ["subprocess", "nohup", "schtasks"]},
         "repo": {"type": "string"},
         "plane_user": {"type": "string"},

tests/test_posix_remote.py

@@ -0,0 +1,205 @@
+"""Remote POSIX transport (warpgate-ssh, os linux/darwin): the generated bash is executed by a local bash in
+place of the relay, so quoting, chunked uploads, detached launch and process-group kill are exercised for real."""
+import base64
+import os
+import shutil
+import subprocess
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from agentd import ROOT, config
+from agentd.config import Config
+from agentd.executors import Run, make_executor
+from agentd.transports import make_transport
+from agentd.transports.warpgate_ssh import PosixWarpgateTransport, relay_argv
+from tests.fakes import config_data
+
+FAKE_CLAUDE = """#!/usr/bin/env bash
+prompt="$(cat)"
+echo "args: $*" >&2
+printf '{"type":"result","subtype":"success","is_error":false,"result":"[review] got %s","session_id":"sess-7"}' "${#prompt}"
+"""
+
+
+def remote_data(root):
+    d = config_data()
+    d["vault"] = {"client": "/opt/vault/agent-access.py", "config": "https://vault.example/identity.json"}
+    d["terminals"]["mac"] = {"transport": "warpgate-ssh", "target": "mac-target", "os": "linux", "launch": "nohup",
+                             "repo": str(root / "my repo's"), "run_root": str(root / "run root"), "plane_user": "oregon",
+                             "token_key": "TOK_MAC", "concurrency": 1, "executors": ["claude-code", "shell"],
+                             "bins": {"claude": str(root / "bin dir" / "claude")}}
+    return d
+
+
+class LocalBash(PosixWarpgateTransport):
+    """The relay replaced by `bash -c` on this machine; records every script (one entry = one relay call)."""
+    def __init__(self, cfg, terminal):
+        super().__init__(cfg, terminal)
+        self.calls = []
+
+    def sh(self, script, timeout=150, marker=None):
+        self.calls.append(script)
+        out = subprocess.run(["bash", "-c", script], capture_output=True, text=True, timeout=timeout,
+                             env=self.cfg.scrubbed_env())
+        if marker and marker not in out.stdout:
+            raise AssertionError("marker %s missing: %s %s" % (marker, out.stdout, out.stderr))
+        return out.stdout
+
+
+def wait(ex, h, seconds=10):
+    end = time.time() + seconds
+    while time.time() < end:
+        if ex.poll(h) == "finished":
+            return True
+        time.sleep(0.1)
+    return False
+
+
+class PosixRemoteTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self.tmp.name)
+        (self.root / "my repo's").mkdir()
+        (self.root / "bin dir").mkdir()
+        fake = self.root / "bin dir" / "claude"
+        fake.write_text(FAKE_CLAUDE)
+        fake.chmod(0o755)
+        self.cfg = Config(remote_data(self.root))
+        self.t = LocalBash(self.cfg, "mac")
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_factory_picks_posix_for_linux_and_config_is_valid(self):
+        self.assertEqual(config.validate(remote_data(self.root)), [])
+        self.assertIsInstance(make_transport(self.cfg, "mac"), PosixWarpgateTransport)
+        self.assertEqual(make_transport(self.cfg, "win10").os, "windows")
+
+    def test_live_config_runs_oregon_remotely(self):
+        cfg = config.load(ROOT / "config" / "agentd.yaml")
+        t = make_transport(cfg, "oregon")
+        self.assertIsInstance(t, PosixWarpgateTransport)
+        self.assertEqual((t.target, t.os), ("oregon", "linux"))
+        self.assertEqual(cfg.executors["codex"]["tier_defaults"], {"gpt-6.1-sol": "fast"})
+
+    def test_relay_argv_and_scrubbed_env(self):
+        argv = relay_argv(self.cfg, "mac-target", ["bash", "-c", "echo hi"], 30)
+        self.assertEqual(argv[1:], ["-I", "/opt/vault/agent-access.py", "--config", "https://vault.example/identity.json",
+                                    "run", "--target", "mac-target", "--timeout", "30", "--", "bash", "-c", "echo hi"])
+        seen = {}
+
+        def fake_run(args, **kw):
+            seen.update(args=args, env=kw["env"])
+            return mock.Mock(returncode=0, stdout=b"MKDIR=ok\n", stderr=b"")
+        with mock.patch.dict(os.environ, {"TOK_MAC": "s", "TOK_W10": "s"}), \
+                mock.patch("agentd.transports.warpgate_ssh.subprocess.run", fake_run):
+            PosixWarpgateTransport(self.cfg, "mac").mkdir("/x")
+        self.assertNotIn("TOK_MAC", seen["env"])
+        self.assertNotIn("TOK_W10", seen["env"])
+        self.assertEqual(seen["args"][-3:-1], ["bash", "-c"])
+
+    def test_claude_round_is_one_call_and_resume_uses_session_id(self):
+        ex = make_executor(self.cfg, "mac", "claude-code", self.t)
+        prompt = "it's \"quoted\" $HOME `x` \\ 中文\n"
+        h = ex.start(Run("GW-4-1", self.cfg.terminals["mac"]["repo"], prompt, model="sonnet"))
+        self.assertEqual(len(self.t.calls), 1)        # mkdir + prompt + run.sh + launch in one relay call
+        self.assertIn("setsid nohup bash 'run.sh' </dev/null >/dev/null 2>&1 &", self.t.calls[0])
+        self.assertTrue(wait(ex, h))
+        rd = Path(h["run_dir"])
+        self.assertEqual((rd / "prompt.md").read_text(encoding="utf-8"), prompt)
+        self.assertEqual((rd / "pid.txt").read_text().strip(), (rd / "launch.pid").read_text().strip())
+        res = ex.collect(h)
+        self.assertEqual((res.last_message, res.exit_code, res.thread_id),
+                         ("[review] got %d" % len(prompt.rstrip("\n")), 0, "sess-7"))
+        cmd = ex.command(Run("GW-4-2", "r", "p", model="sonnet", thread_id="sess-7"), h["run_dir"])
+        self.assertEqual(cmd, "'%s' -p --resume sess-7" % (self.root / "bin dir" / "claude"))
+
+    def test_runner_quotes_paths_and_cd_failure_writes_exit(self):
+        _, script = self.t.runner("/a b/O'Neil/run", "/no such/repo's", "cmd", "out.json")
+        self.assertIn("cd '/no such/repo'\\''s' ||", script)
+        self.assertIn("< '/a b/O'\\''Neil/run'/prompt.md > '/a b/O'\\''Neil/run'/out.json", script)
+        ex = make_executor(self.cfg, "mac", "shell", self.t)
+        h = ex.start(Run("GW-4-3", str(self.root / "missing"), "", script="true"))
+        self.assertTrue(wait(ex, h))
+        self.assertEqual(ex.exit_code(h), 97)
+
+    def test_chunked_upload_and_reads(self):
+        self.t.CHUNK = 64
+        path = str(self.root / "run root" / "x y" / "f'.txt")
+        text = "line1 'q' \"d\" $(nope)\n" * 20 + "last ünïcode"
+        self.t.mkdir(os.path.dirname(path))
+        n = len(self.t.calls)
+        self.t.write_text(path, text)
+        self.assertGreater(len(self.t.calls) - n, 2)
+        self.assertEqual(Path(path).read_text(encoding="utf-8"), text)
+        self.assertFalse(os.path.exists(path + ".b64"))
+        self.assertEqual(self.t.read_text(path), text)
+        self.assertIsNone(self.t.read_text(path + ".missing"))
+        self.assertEqual(self.t.grep_first(path, "$(nope)"), "line1 'q' \"d\" $(nope)")
+        self.assertIsNone(self.t.grep_first(path, "absent"))
+        self.assertEqual(self.t.tail(path, 2), "line1 'q' \"d\" $(nope)\nlast ünïcode")
+        self.assertEqual(self.t.tail(path + ".missing", 2), "")
+
+    def test_big_prompt_goes_up_before_the_launch_call(self):
+        self.t.CHUNK = 100
+        ex = make_executor(self.cfg, "mac", "shell", self.t)
+        h = ex.start(Run("GW-4-4", self.cfg.terminals["mac"]["repo"], "z" * 500, script="wc -c"))
+        self.assertGreater(len(self.t.calls), 2)
+        self.assertTrue(wait(ex, h))
+        self.assertEqual(ex.collect(h).last_message, "500")
+
+    @unittest.skipUnless(shutil.which("setsid") and shutil.which("pgrep"), "needs setsid/pgrep")
+    def test_kill_ends_the_whole_process_group(self):
+        ex = make_executor(self.cfg, "mac", "shell", self.t)
+        marker = "agentd-test-%d" % os.getpid()
+        h = ex.start(Run("GW-4-5", self.cfg.terminals["mac"]["repo"], "",
+                         script="bash -c 'sleep 300; : %s' & sleep 300; : %s" % (marker, marker)))
+        end = time.time() + 5
+        while time.time() < end and not subprocess.run(["pgrep", "-f", marker], capture_output=True).stdout:
+            time.sleep(0.1)
+        self.assertTrue(subprocess.run(["pgrep", "-f", marker], capture_output=True).stdout)
+        ex.kill(h)
+        self.assertIn("KILLED=ok left=0", subprocess.run(["bash", "-c", self.t.kill_script(h["run_dir"])],
+                                                         capture_output=True, text=True).stdout)
+        self.assertFalse(subprocess.run(["pgrep", "-f", marker], capture_output=True).stdout)
+        self.assertEqual(ex.poll(h), "running")       # killed runners write no exit.txt; core treats it as cancelled
+
+    def test_kill_after_finish_does_not_signal_a_reused_pid(self):
+        ex = make_executor(self.cfg, "mac", "shell", self.t)
+        h = ex.start(Run("GW-4-6", self.cfg.terminals["mac"]["repo"], "", script="true"))
+        self.assertTrue(wait(ex, h))
+        (Path(h["run_dir"]) / "pid.txt").write_text("1\n")
+        self.assertIn("KILLED=finished", subprocess.run(["bash", "-c", self.t.kill_script(h["run_dir"])],
+                                                        capture_output=True, text=True).stdout)
+
+    def test_darwin_detaches_without_setsid_binary(self):
+        d = remote_data(self.root)
+        d["terminals"]["mac"]["os"] = "darwin"
+        t = PosixWarpgateTransport(Config(d), "mac")
+        s = t.launch_script("/r", "run.sh")
+        self.assertIn("nohup perl -MPOSIX -e 'POSIX::setsid() or die", s)
+        self.assertNotIn("setsid nohup", s)
+        self.assertIn("echo $! > launch.pid", s)
+
+    def test_semantic_checks(self):
+        d = remote_data(self.root)
+        d["terminals"]["mac"]["launch"] = "schtasks"
+        d["terminals"]["mac"]["run_root"] = "~/runs"
+        del d["vault"]["config"]
+        errs = config.validate(d)
+        for frag in ("needs launch=nohup", "mac.run_root: must be an absolute path", "need vault.config"):
+            self.assertTrue(any(frag in e for e in errs), (frag, errs))
+        d = remote_data(self.root)
+        d["terminals"]["win10"]["target"] = "win11"
+        self.assertTrue(any("only reach target 'win10'" in e for e in config.validate(d)))
+
+    def test_upload_payload_is_plain_base64(self):
+        s = self.t.put_script("/r/p.md", base64.b64encode("x'y".encode()).decode())
+        self.assertIn("printf %s 'eCd5' | base64 -d > '/r/p.md.part' && mv -f '/r/p.md.part' '/r/p.md'", s)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_transports.py

@@ -12,11 +12,11 @@ from agentd.config import Config
 from agentd.executors import Run, make_executor
 from agentd.store import Store
 from agentd.transports.local import LocalTransport
-from agentd.transports.warpgate_ssh import WarpgateSshTransport
+from agentd.transports.warpgate_ssh import PosixWarpgateTransport, WarpgateSshTransport, WindowsWarpgateTransport
 from tests.fakes import config_data
 
 
-class RecordingWin(WarpgateSshTransport):
+class RecordingWin(WindowsWarpgateTransport):
     """Windows transport with the relay replaced by an in-memory recorder."""
     def __init__(self, cfg, terminal):
         super().__init__(cfg, terminal)
@@ -95,7 +95,7 @@ class WindowsScriptTest(unittest.TestCase):
 
         def fake_run(args, **kw):
             seen.update(kw["env"])
-            return mock.Mock(returncode=0, stdout="X=ok", stderr="")
+            return mock.Mock(returncode=0, stdout=b"X=ok", stderr=b"")
         with mock.patch.dict(os.environ, {"PLANE_WINDOW10_API_TOKEN": "s"}), \
                 mock.patch("agentd.transports.warpgate_ssh.subprocess.run", fake_run):
             WarpgateSshTransport(self.cfg, "win10").ps("'X=ok'")