Compare commits
3 Commits
34b2ba1cc7
...
20e4e4ed73
| Author | SHA1 | Date | |
|---|---|---|---|
| 20e4e4ed73 | |||
| dc8271a06a | |||
| f2f7e4308a |
@@ -7,6 +7,6 @@ Plane 看板驱动的通用 agent 调度服务:在 Plane 建工作项并指派
|
|||||||
- 凭据全部经 Agent Vault 注入内存,仓库和配置里只有键名。
|
- 凭据全部经 Agent Vault 注入内存,仓库和配置里只有键名。
|
||||||
|
|
||||||
## 使用
|
## 使用
|
||||||
- 契约与扩展:`docs/executor-contract.md`;安装、cron、日志、故障:`docs/runbook.md`
|
- 契约与扩展:`docs/executor-contract.md`;安装、cron、日志、故障:`docs/runbook.md`;迁到无 agent CLI 的 LXC:`docs/deploy-lxc.md`
|
||||||
- 入口:`cli/agentd tick|new|show|comment|state|assign|smoke|check-config`(在 `agent-access.py exec --secret-env …` 下运行)
|
- 入口:`cli/agentd tick|new|show|comment|state|assign|smoke|check-config`(在 `agent-access.py exec --secret-env …` 下运行)
|
||||||
- 测试:`python3 -m unittest`(离线,不访问 Plane/win10)
|
- 测试:`python3 -m unittest`(离线,不访问 Plane/win10)
|
||||||
|
|||||||
+13
-5
@@ -91,11 +91,19 @@ def semantic_errors(data):
|
|||||||
for ex in t.get("executors", []):
|
for ex in t.get("executors", []):
|
||||||
if ex not in execs:
|
if ex not in execs:
|
||||||
errs.append("terminals.%s: unknown executor '%s'" % (tname, ex))
|
errs.append("terminals.%s: unknown executor '%s'" % (tname, ex))
|
||||||
if t.get("transport") == "warpgate-ssh":
|
if t.get("transport") == "warpgate-ssh" and t.get("os") == "windows":
|
||||||
if t.get("os") != "windows" or t.get("launch") != "schtasks":
|
if t.get("launch") != "schtasks":
|
||||||
errs.append("terminals.%s: warpgate-ssh currently supports only os=windows, launch=schtasks" % tname)
|
errs.append("terminals.%s: warpgate-ssh on windows needs launch=schtasks" % tname)
|
||||||
if t.get("target") != "win10":
|
if t.get("target", tname) != "win10":
|
||||||
errs.append("terminals.%s: warpgate-ssh helper scripts only reach target 'win10' for now" % tname)
|
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":
|
if t.get("transport") == "local" and t.get("os") != "linux":
|
||||||
errs.append("terminals.%s: local transport supports only os=linux" % tname)
|
errs.append("terminals.%s: local transport supports only os=linux" % tname)
|
||||||
for ename, e in execs.items():
|
for ename, e in execs.items():
|
||||||
|
|||||||
@@ -491,6 +491,8 @@ class Agentd:
|
|||||||
r["model"] = label[6:]
|
r["model"] = label[6:]
|
||||||
elif label.startswith("effort:"):
|
elif label.startswith("effort:"):
|
||||||
r["effort"] = label[7:]
|
r["effort"] = label[7:]
|
||||||
|
elif label.startswith("tier:"):
|
||||||
|
r["tier"] = label[5:]
|
||||||
elif label.startswith("cap:"):
|
elif label.startswith("cap:"):
|
||||||
caps.append(label[4:])
|
caps.append(label[4:])
|
||||||
elif parse_budget(label) is not None:
|
elif parse_budget(label) is not None:
|
||||||
@@ -503,6 +505,10 @@ class Agentd:
|
|||||||
ecfg = self.cfg.executors[r["executor"]]
|
ecfg = self.cfg.executors[r["executor"]]
|
||||||
r["model"] = r["model"] or (ecfg.get("models") or [None])[0]
|
r["model"] = r["model"] or (ecfg.get("models") or [None])[0]
|
||||||
r["effort"] = r["effort"] or ecfg.get("default_effort")
|
r["effort"] = r["effort"] or ecfg.get("default_effort")
|
||||||
|
if r["tier"] is None: # per-model default unless a label set one; `tier:none` = no tier argument
|
||||||
|
r["tier"] = (ecfg.get("tier_defaults") or {}).get(r["model"])
|
||||||
|
if r["tier"] in ("none", ""):
|
||||||
|
r["tier"] = None
|
||||||
return r
|
return r
|
||||||
|
|
||||||
def protocol(self, pcfg):
|
def protocol(self, pcfg):
|
||||||
|
|||||||
@@ -87,13 +87,16 @@ class RunnerExecutor(Executor):
|
|||||||
def start(self, run):
|
def start(self, run):
|
||||||
t = self.transport
|
t = self.transport
|
||||||
run_dir = t.join(self.run_root(), run.run_id)
|
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_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
|
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}
|
return {"run_dir": run_dir, "task": task, "thread_id": run.thread_id}
|
||||||
|
|
||||||
def poll(self, handle):
|
def poll(self, handle):
|
||||||
|
|||||||
@@ -4,17 +4,23 @@ import signal
|
|||||||
import subprocess
|
import subprocess
|
||||||
from pathlib import Path
|
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
|
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
|
echo $$ > {d}/pid.txt
|
||||||
date -Is > {d}/started.txt
|
date +%Y-%m-%dT%H:%M:%S%z > {d}/started.txt
|
||||||
(
|
(
|
||||||
{command}
|
{command}
|
||||||
) < {d}/prompt.md > {d}/{stdout} 2> {d}/stderr.txt
|
) < {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:
|
class LocalTransport:
|
||||||
os = "linux"
|
os = "linux"
|
||||||
|
|
||||||
@@ -28,7 +34,7 @@ class LocalTransport:
|
|||||||
return os.path.join(*parts)
|
return os.path.join(*parts)
|
||||||
|
|
||||||
def quote(self, s):
|
def quote(self, s):
|
||||||
return "'" + str(s).replace("'", "'\\''") + "'"
|
return sh_quote(s)
|
||||||
|
|
||||||
def mkdir(self, path, clean=False):
|
def mkdir(self, path, clean=False):
|
||||||
p = Path(path)
|
p = Path(path)
|
||||||
|
|||||||
@@ -1,16 +1,59 @@
|
|||||||
"""Windows terminal behind Warpgate, reached through the server-access relay helper scripts:
|
"""Terminals behind Warpgate, reached through the server-access relay client:
|
||||||
win10-ps.sh <script> run PowerShell (sent as -EncodedCommand)
|
|
||||||
win10-put.sh <local> <remote> upload a file (gzip+base64 chunks, SHA-256 verified)
|
agent-access.py --config <identity> run --target <target> --timeout N -- <argv...>
|
||||||
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."""
|
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 base64
|
||||||
import hashlib
|
import hashlib
|
||||||
import os
|
import os
|
||||||
|
import posixpath
|
||||||
import re
|
import re
|
||||||
import subprocess
|
import subprocess
|
||||||
|
import sys
|
||||||
import tempfile
|
import tempfile
|
||||||
|
|
||||||
from . import TransportError
|
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'
|
RUNNER = r"""$ErrorActionPreference = 'Continue'
|
||||||
[Console]::OutputEncoding = [Text.Encoding]::UTF8
|
[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 ""
|
return base64.b64decode(m.group(1)).decode("utf-8", "replace") if m.group(1) else ""
|
||||||
|
|
||||||
|
|
||||||
class WarpgateSshTransport:
|
class WindowsWarpgateTransport:
|
||||||
os = "windows"
|
os = "windows"
|
||||||
|
|
||||||
def __init__(self, cfg, terminal):
|
def __init__(self, cfg, terminal):
|
||||||
@@ -54,21 +97,17 @@ class WarpgateSshTransport:
|
|||||||
|
|
||||||
# ---------------------------------------------------------------- relay primitives
|
# ---------------------------------------------------------------- relay primitives
|
||||||
def ps(self, script, timeout=150):
|
def ps(self, script, timeout=150):
|
||||||
env = dict(self.cfg.scrubbed_env(), WIN10_TIMEOUT=str(timeout))
|
rc, out, err = call(self.cfg, [self.ps_script, script], timeout, {"WIN10_TIMEOUT": str(timeout)})
|
||||||
try:
|
if rc != 0 and not out.strip():
|
||||||
out = subprocess.run([self.ps_script, script], capture_output=True, text=True, timeout=timeout + 60, env=env)
|
raise TransportError("relay exit %s: %s" % (rc, err.strip()[-300:]))
|
||||||
except subprocess.TimeoutExpired:
|
return out
|
||||||
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", "")
|
|
||||||
|
|
||||||
def put(self, local, remote):
|
def put(self, local, remote):
|
||||||
sha = hashlib.sha256(open(local, "rb").read()).hexdigest()
|
with open(local, "rb") as fh:
|
||||||
out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600,
|
sha = hashlib.sha256(fh.read()).hexdigest()
|
||||||
env=self.cfg.scrubbed_env())
|
rc, out, err = call(self.cfg, [self.put_script, local, remote], 540)
|
||||||
if ("remote sha256 " + sha) not in out.stdout:
|
if ("remote sha256 " + sha) not in out:
|
||||||
raise TransportError("upload of %s failed: %s" % (remote, (out.stdout + out.stderr).strip()[-300:]))
|
raise TransportError("upload of %s failed: %s" % (remote, (out + err).strip()[-300:]))
|
||||||
|
|
||||||
# ---------------------------------------------------------------- file ops
|
# ---------------------------------------------------------------- file ops
|
||||||
def mkdir(self, path, clean=False):
|
def mkdir(self, path, clean=False):
|
||||||
@@ -146,3 +185,177 @@ class WarpgateSshTransport:
|
|||||||
return None
|
return None
|
||||||
m = re.search(r"COUNT=(\d+)", out)
|
m = re.search(r"COUNT=(\d+)", out)
|
||||||
return int(m.group(1)) if m else None
|
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)
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ vault: {client: ~/.local/share/server-access-client/agent-access.py, config: htt
|
|||||||
|
|
||||||
agentd:
|
agentd:
|
||||||
state_dir: state # relative to the repo root; JSON state, logs, heartbeat, local run dirs
|
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
|
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
|
max_rounds_per_thread: 8 # rebuild the executor thread (with a handoff summary) after this many rounds
|
||||||
comment_max_chars: 6000
|
comment_max_chars: 6000
|
||||||
@@ -29,14 +29,19 @@ terminals:
|
|||||||
capabilities: [android, game]
|
capabilities: [android, game]
|
||||||
executors: [codex]
|
executors: [codex]
|
||||||
oregon:
|
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
|
os: linux
|
||||||
launch: subprocess
|
launch: nohup
|
||||||
repo: /home/ubuntu/Documents/wjj/project/gwent
|
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
|
plane_user: oregon
|
||||||
token_key: PLANE_OREGON_API_TOKEN
|
token_key: PLANE_OREGON_API_TOKEN
|
||||||
concurrency: 2
|
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: [claude-code, codex, shell]
|
||||||
|
|
||||||
executors:
|
executors:
|
||||||
@@ -45,6 +50,7 @@ executors:
|
|||||||
run: "codex exec --json -C '{repo}' -s workspace-write -m {model} -c 'model_reasoning_effort=\"{effort}\"' {tier} -o '{out}' -"
|
run: "codex exec --json -C '{repo}' -s workspace-write -m {model} -c 'model_reasoning_effort=\"{effort}\"' {tier} -o '{out}' -"
|
||||||
resume: "codex exec resume {thread} --json -m {model} -c 'model_reasoning_effort=\"{effort}\"' {tier} -o '{out}' -"
|
resume: "codex exec resume {thread} --json -m {model} -c 'model_reasoning_effort=\"{effort}\"' {tier} -o '{out}' -"
|
||||||
tier_arg: "-c 'service_tier=\"{tier}\"'"
|
tier_arg: "-c 'service_tier=\"{tier}\"'"
|
||||||
|
tier_defaults: {gpt-6.1-sol: fast} # tier when no label sets one (`tier:<x>`, `tier:none` = no tier)
|
||||||
thread_from: events.jsonl:thread.started.thread_id
|
thread_from: events.jsonl:thread.started.thread_id
|
||||||
last_message: '{out}'
|
last_message: '{out}'
|
||||||
models: [gpt-6.1-sol, gpt-6-astra]
|
models: [gpt-6.1-sol, gpt-6-astra]
|
||||||
|
|||||||
+10
-4
@@ -5,7 +5,7 @@ vault: {client: ~/.local/share/server-access-client/agent-access.py, config: htt
|
|||||||
|
|
||||||
agentd:
|
agentd:
|
||||||
state_dir: ~/.gwent-board/agentd # JSON state, agentd.log, heartbeat.json, local run dirs (runs/)
|
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
|
interval: 120
|
||||||
max_rounds_per_thread: 8
|
max_rounds_per_thread: 8
|
||||||
comment_max_chars: 6000
|
comment_max_chars: 6000
|
||||||
@@ -27,14 +27,19 @@ terminals:
|
|||||||
capabilities: [android, game]
|
capabilities: [android, game]
|
||||||
executors: [codex]
|
executors: [codex]
|
||||||
oregon:
|
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
|
os: linux
|
||||||
launch: subprocess
|
launch: nohup
|
||||||
repo: /home/ubuntu/Documents/wjj/project/gwent
|
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
|
plane_user: oregon
|
||||||
token_key: PLANE_OREGON_API_TOKEN
|
token_key: PLANE_OREGON_API_TOKEN
|
||||||
concurrency: 1 # one shared worktree: no parallel edits
|
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: [claude-code]
|
||||||
|
|
||||||
executors:
|
executors:
|
||||||
@@ -44,6 +49,7 @@ executors:
|
|||||||
# `exec resume` takes no -C/-s: the runner already cd's into the repo, the sandbox goes in as a -c override
|
# `exec resume` takes no -C/-s: the runner already cd's into the repo, the sandbox goes in as a -c override
|
||||||
resume: "codex exec resume {thread} --json -m {model} -c 'model_reasoning_effort=\"{effort}\"' -c 'sandbox_mode=\"workspace-write\"' {tier} -o '{out}' -"
|
resume: "codex exec resume {thread} --json -m {model} -c 'model_reasoning_effort=\"{effort}\"' -c 'sandbox_mode=\"workspace-write\"' {tier} -o '{out}' -"
|
||||||
tier_arg: "-c 'service_tier=\"{tier}\"'"
|
tier_arg: "-c 'service_tier=\"{tier}\"'"
|
||||||
|
tier_defaults: {gpt-6.1-sol: fast} # tier when no label sets one (`tier:<x>`, `tier:none` = no tier)
|
||||||
thread_from: events.jsonl:thread.started.thread_id
|
thread_from: events.jsonl:thread.started.thread_id
|
||||||
last_message: '{out}'
|
last_message: '{out}'
|
||||||
models: [gpt-6.1-sol, gpt-6-astra]
|
models: [gpt-6.1-sol, gpt-6-astra]
|
||||||
|
|||||||
+2
-1
@@ -48,7 +48,7 @@
|
|||||||
"properties": {
|
"properties": {
|
||||||
"transport": {"enum": ["local", "warpgate-ssh"]},
|
"transport": {"enum": ["local", "warpgate-ssh"]},
|
||||||
"target": {"type": "string"},
|
"target": {"type": "string"},
|
||||||
"os": {"enum": ["linux", "windows"]},
|
"os": {"enum": ["linux", "darwin", "windows"]},
|
||||||
"launch": {"enum": ["subprocess", "nohup", "schtasks"]},
|
"launch": {"enum": ["subprocess", "nohup", "schtasks"]},
|
||||||
"repo": {"type": "string"},
|
"repo": {"type": "string"},
|
||||||
"plane_user": {"type": "string"},
|
"plane_user": {"type": "string"},
|
||||||
@@ -76,6 +76,7 @@
|
|||||||
"default_effort": {"type": "string"},
|
"default_effort": {"type": "string"},
|
||||||
"permission_mode": {"type": "string"},
|
"permission_mode": {"type": "string"},
|
||||||
"tier_arg": {"type": "string"},
|
"tier_arg": {"type": "string"},
|
||||||
|
"tier_defaults": {"type": "object", "additionalProperties": {"type": "string"}},
|
||||||
"script": {"type": "string"}
|
"script": {"type": "string"}
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -0,0 +1,97 @@
|
|||||||
|
# 在精简 LXC 上部署 agentd(design §9 阶段 3)
|
||||||
|
|
||||||
|
目标:调度器跑在一台**不装任何 agent CLI** 的 Debian 12 LXC(`agentd-01`)上;所有终端(win10、oregon、mac…)都经 Warpgate SSH 远程启动(`transport: warpgate-ssh`)。LXC 只做调度、回帖、通知。
|
||||||
|
|
||||||
|
## 1. 系统包
|
||||||
|
```bash
|
||||||
|
apt-get update
|
||||||
|
apt-get install -y python3 python3-yaml openssh-client git cron ca-certificates curl
|
||||||
|
# win10 的 helper 脚本另需(Debian 12 默认已有):bash coreutils(base64/sha256sum) gzip libc-bin(iconv)
|
||||||
|
adduser --disabled-password --gecos '' agentd # 普通用户运行 agentd,不用 root
|
||||||
|
```
|
||||||
|
Python ≥ 3.10(Debian 12 是 3.11)。`jsonschema` 可选,没有就用内置子集校验器。
|
||||||
|
|
||||||
|
## 2. 网络
|
||||||
|
| 去向 | 用途 |
|
||||||
|
|---|---|
|
||||||
|
| `nas.tail9531b.ts.net:8443`(tailnet) | Agent Vault 身份配置 `identity/agent.json`(含 vault token 与 Warpgate 主机公钥)→ LXC 必须加入 tailnet(`tailscale up`,LXC 需 `/dev/net/tun`) |
|
||||||
|
| Vault(身份配置里的地址) | 取 Plane token、Warpgate 密码 |
|
||||||
|
| Warpgate SSH `115.28.210.101:2222` | 所有终端的中继 |
|
||||||
|
| `https://plane.kappiee.com` | Plane API |
|
||||||
|
|
||||||
|
## 3. Agent Vault 客户端与身份
|
||||||
|
```bash
|
||||||
|
sudo -u agentd mkdir -p ~agentd/.local/share/server-access-client
|
||||||
|
# 从 oregon 复制(权限 700,只给 agentd 用户):
|
||||||
|
scp oregon:~/.local/share/server-access-client/{agent-access.py,agent-access-askpass.py} \
|
||||||
|
agentd-01:~agentd/.local/share/server-access-client/
|
||||||
|
chmod 700 ~agentd/.local/share/server-access-client/*
|
||||||
|
cd ~agentd/.local/share/server-access-client && sudo -u agentd python3 -I agent-access.py \
|
||||||
|
--config https://nas.tail9531b.ts.net:8443/identity/agent.json list # authenticated: true,能看到 win10 / oregon 等 target
|
||||||
|
```
|
||||||
|
- 身份是 tailnet 上的 URL(`--config https://…/identity/agent.json`),不落盘;也可以换成本地私有 JSON 文件(`chmod 600`),把 `vault.config` 指过去。
|
||||||
|
- **Warpgate known_hosts 不用配**:客户端每次用身份配置里的 `warpgate.host_key` 生成临时 known_hosts(`StrictHostKeyChecking=yes`),不读 `~/.ssh/known_hosts`。只要求有 `ssh` 可执行文件。
|
||||||
|
- `askpass` 必须与 `agent-access.py` 同目录且可执行。
|
||||||
|
- Warpgate 侧:该身份(`agent` 用户)要有每个终端 target 的权限(现有:`win10`、`oregon`)。
|
||||||
|
|
||||||
|
## 4. 代码与配置
|
||||||
|
```bash
|
||||||
|
sudo -u agentd git clone https://git.kappiee.com/kappiee/plane-agentd.git ~agentd/plane-agentd
|
||||||
|
# win10 仍用 helper 脚本(见 design §9 TODO),复制这两个:
|
||||||
|
sudo -u agentd mkdir -p ~agentd/helpers
|
||||||
|
scp oregon:/home/ubuntu/Documents/wjj/project/gwent-design/tools/{win10-ps.sh,win10-put.sh} agentd-01:~agentd/helpers/
|
||||||
|
# 看板协作协议(拼进每轮 prompt,由 agentd 在本机读取):
|
||||||
|
scp oregon:/home/ubuntu/Documents/wjj/project/gwent/design/board-protocol.md agentd-01:~agentd/board-protocol.md
|
||||||
|
cp ~agentd/plane-agentd/config/agentd.yaml ~agentd/plane-agentd/config/agentd.local.yaml # git-ignored,优先于 agentd.yaml
|
||||||
|
```
|
||||||
|
`config/agentd.local.yaml` 里改:
|
||||||
|
- `vault.client: ~/.local/share/server-access-client/agent-access.py`(`~` 在本机展开)
|
||||||
|
- `agentd.state_dir: ~/state`(JSON 状态、日志、心跳);`agentd.helpers_dir: ~/helpers`
|
||||||
|
- `projects.GWENT.protocol_file: ~/board-protocol.md`(**它不在终端上读,是 LXC 本机路径**;协议更新时重新复制)
|
||||||
|
- 所有终端都是 `transport: warpgate-ssh`:
|
||||||
|
- win10:`target: win10, os: windows, launch: schtasks`(不变)
|
||||||
|
- oregon:`target: oregon, os: linux, launch: nohup, repo: /home/ubuntu/Documents/wjj/project/gwent, run_root: /home/ubuntu/.gwent-board/agentd-runs, bins: {claude: /home/ubuntu/.local/bin/claude}`
|
||||||
|
- 远程 POSIX 终端的 `repo`、`run_root` 必须是绝对路径(在远程展开,`~` 不行);`bins` 写远程的绝对路径(SSH 非登录 shell 只有系统 PATH)。
|
||||||
|
```bash
|
||||||
|
cd ~agentd/plane-agentd && python3 -m unittest && python3 cli/agentd check-config
|
||||||
|
```
|
||||||
|
|
||||||
|
## 5. cron
|
||||||
|
`cron/agentd-tick.sh` 里写死了 oregon 的路径;在 LXC 上用同样的形状(单行 crontab 或复制脚本改路径):
|
||||||
|
```
|
||||||
|
*/2 * * * * flock -n $HOME/state/cron.lock timeout 900 python3 -I $HOME/.local/share/server-access-client/agent-access.py --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec --secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN --secret-env PLANE_WINDOW10_API_TOKEN=PLANE_WINDOW10_API_TOKEN --timeout 850 -- python3 $HOME/plane-agentd/cli/agentd tick --once >> $HOME/state/agentd-cron.log 2>&1
|
||||||
|
```
|
||||||
|
- token 只经 `exec --secret-env` 进 agentd 进程内存;agentd 起的中继子进程环境里去掉了所有 `token_key`(`Config.scrubbed_env()`),远程执行器在终端上由 sshd 起,本来就拿不到 LXC 的环境。
|
||||||
|
- `--secret-env` 的完整列表:`cli/agentd check-config` 最后一行。
|
||||||
|
- **切换时先停 oregon 上的 cron**(`crontab -e` 注释掉 `agentd-tick.sh`),再在 LXC 上启用;两个调度器同时跑会重复派单。状态文件 `~/.gwent-board/agentd/agentd-state.json` 从 oregon 复制到 LXC 的 `state_dir`,在途运行(handle 里是终端上的绝对路径)照常被新调度器 poll/collect。
|
||||||
|
|
||||||
|
## 6. 日志与心跳
|
||||||
|
| 路径(LXC) | 内容 |
|
||||||
|
|---|---|
|
||||||
|
| `~/state/agentd.log` | 每 tick 的动作、错误 |
|
||||||
|
| `~/state/heartbeat.json` | `at` 超过 10 分钟不更新 = 调度器没在跑 |
|
||||||
|
| `~/state/agentd-state.json` | 全部工作项状态 |
|
||||||
|
| `~/state/agentd-cron.log` | cron 输出(空闲时为空;`skip: another tick holds …` 无害) |
|
||||||
|
| 终端上 `<run_root>/<run_id>/` | 每轮的 `prompt.md run.sh/run.ps1 pid.txt launch.pid exit.txt stderr.txt` + 执行器输出 |
|
||||||
|
|
||||||
|
## 7. 加一个终端(例:mac)
|
||||||
|
1. Warpgate 加 SSH target(如 `mac`),给 `agent` 身份授权;`agent-access.py … run --target mac --timeout 30 -- bash -c 'echo ok; uname'` 能通。
|
||||||
|
2. 在终端上装好 agent CLI 并登录(Claude Code / Codex),clone 工作仓库;记下二进制绝对路径。
|
||||||
|
3. 配置:
|
||||||
|
```yaml
|
||||||
|
mac:
|
||||||
|
transport: warpgate-ssh
|
||||||
|
target: mac
|
||||||
|
os: darwin # linux | darwin;darwin 用 perl POSIX::setsid 脱离会话(无 setsid 命令)
|
||||||
|
launch: nohup
|
||||||
|
repo: /Users/me/src/gwent
|
||||||
|
run_root: /Users/me/.agentd-runs
|
||||||
|
bins: {claude: /Users/me/.local/bin/claude}
|
||||||
|
plane_user: mac
|
||||||
|
token_key: PLANE_MAC_API_TOKEN
|
||||||
|
concurrency: 1
|
||||||
|
executors: [claude-code]
|
||||||
|
```
|
||||||
|
远程 shell 是目标账户的登录 shell(bash/zsh 均可,agentd 发的是 `bash -c '<script>'`);终端需有 `bash base64 nohup pgrep pkill`(Linux 另需 `setsid`,macOS 需 `perl`)。
|
||||||
|
4. Plane:建用户 `mac`,token 存 Vault 键 `PLANE_MAC_API_TOKEN`;项目 `routing.by_assignee` 加 `mac: mac`,`default_executor` 加 `mac: claude-code`;cron 加 `--secret-env PLANE_MAC_API_TOKEN=PLANE_MAC_API_TOKEN`。
|
||||||
|
5. `cli/agentd check-config`,然后 `cli/agentd smoke GWENT mac`。
|
||||||
+4
-3
@@ -50,7 +50,7 @@ plane: {base_url: https://plane.kappiee.com, workspace: kappiee}
|
|||||||
vault: {client: ~/.local/share/server-access-client/agent-access.py, config: https://nas.tail9531b.ts.net:8443/identity/agent.json}
|
vault: {client: ~/.local/share/server-access-client/agent-access.py, config: https://nas.tail9531b.ts.net:8443/identity/agent.json}
|
||||||
terminals:
|
terminals:
|
||||||
win10: {transport: warpgate-ssh, target: win10, os: windows, launch: schtasks, repo: 'C:\Users\Administrator\Documents\GwentAssistantProject\gwent', plane_user: window10, token_key: PLANE_WINDOW10_API_TOKEN, concurrency: 1, capabilities: [android, game], executors: [codex, powershell]}
|
win10: {transport: warpgate-ssh, target: win10, os: windows, launch: schtasks, repo: 'C:\Users\Administrator\Documents\GwentAssistantProject\gwent', plane_user: window10, token_key: PLANE_WINDOW10_API_TOKEN, concurrency: 1, capabilities: [android, game], executors: [codex, powershell]}
|
||||||
oregon: {transport: local, os: linux, launch: subprocess, repo: /home/ubuntu/Documents/wjj/project/gwent, plane_user: oregon, token_key: PLANE_OREGON_API_TOKEN, concurrency: 2, executors: [claude-code, codex, shell]}
|
oregon: {transport: warpgate-ssh, target: oregon, os: linux, launch: nohup, repo: /home/ubuntu/Documents/wjj/project/gwent, run_root: /home/ubuntu/.gwent-board/agentd-runs, plane_user: oregon, token_key: PLANE_OREGON_API_TOKEN, concurrency: 1, executors: [claude-code]}
|
||||||
executors:
|
executors:
|
||||||
codex: {run: 'codex exec --json -C {repo} -s workspace-write -m {model} -c model_reasoning_effort="{effort}" {tier} -o {out} -', resume: 'codex exec resume {thread} --json -m {model} -c model_reasoning_effort="{effort}" {tier} -o {out} -', thread_from: events.jsonl:thread.started.thread_id, last_message: '{out}', models: [gpt-6-astra, gpt-6.1-sol], default_effort: medium}
|
codex: {run: 'codex exec --json -C {repo} -s workspace-write -m {model} -c model_reasoning_effort="{effort}" {tier} -o {out} -', resume: 'codex exec resume {thread} --json -m {model} -c model_reasoning_effort="{effort}" {tier} -o {out} -', thread_from: events.jsonl:thread.started.thread_id, last_message: '{out}', models: [gpt-6-astra, gpt-6.1-sol], default_effort: medium}
|
||||||
claude-code: {run: 'claude -p --model {model} --output-format json --permission-mode {perm} -', resume: 'claude -p --resume {thread} --model {model} --output-format json -', thread_from: json:session_id, models: [opus, sonnet, haiku]}
|
claude-code: {run: 'claude -p --model {model} --output-format json --permission-mode {perm} -', resume: 'claude -p --resume {thread} --model {model} --output-format json -', thread_from: json:session_id, models: [opus, sonnet, haiku]}
|
||||||
@@ -74,7 +74,7 @@ projects:
|
|||||||
```
|
```
|
||||||
agentd/ 调度核心、plane 客户端、vault 注入、状态存储(json/sqlite)、心跳
|
agentd/ 调度核心、plane 客户端、vault 注入、状态存储(json/sqlite)、心跳
|
||||||
agentd/executors/ codex.py claude_code.py shell.py(适配器契约见 docs/executor-contract.md)
|
agentd/executors/ codex.py claude_code.py shell.py(适配器契约见 docs/executor-contract.md)
|
||||||
agentd/transports/ local.py warpgate_ssh.py(含 Windows schtasks 启动)
|
agentd/transports/ local.py warpgate_ssh.py(远程 Linux/macOS:bash -c 经中继;Windows:helper 脚本 + schtasks)
|
||||||
agentd/web/ 配置页面(阶段 2):读写 YAML + schema 校验 + 连通/冒烟测试 + 最近运行
|
agentd/web/ 配置页面(阶段 2):读写 YAML + schema 校验 + 连通/冒烟测试 + 最近运行
|
||||||
cli/ agentd new|show|comment|state|assign|smoke
|
cli/ agentd new|show|comment|state|assign|smoke
|
||||||
config/ agentd.yaml(示例 + 实际)、schema.json
|
config/ agentd.yaml(示例 + 实际)、schema.json
|
||||||
@@ -95,4 +95,5 @@ docs/ design.md executor-contract.md runbook.md
|
|||||||
- Plane VM(plane-01)只跑 Plane:它是设备化节点(固定版本、每日备份停写路径),不放 agentd。
|
- Plane VM(plane-01)只跑 Plane:它是设备化节点(固定版本、每日备份停写路径),不放 agentd。
|
||||||
- 阶段 1–2:agentd 跑在 oregon(已有 Vault 客户端、Warpgate 中继脚本、Claude Code、现有 worker 状态),cron 拉起;配置页面同进程,只监听 tailnet 地址。
|
- 阶段 1–2:agentd 跑在 oregon(已有 Vault 客户端、Warpgate 中继脚本、Claude Code、现有 worker 状态),cron 拉起;配置页面同进程,只监听 tailnet 地址。
|
||||||
- 阶段 3 起:迁到 PVE 上的精简 LXC `agentd-01`(1 vCPU / 1–2 GB,tailnet + 内网 10.77.10.0/24,与 Plane/Temporal 同网段,webhook 走内网)。**LXC 上不装任何 agent CLI**,只有 Python 3.10+、PyYAML、Agent Vault 客户端、ssh 客户端;所有终端(win10、oregon、mac…)一律远程启动(`transport: warpgate-ssh`),LXC 只负责调度、回帖、通知。对外访问经 Warpgate HTTP target,复用既有登录。
|
- 阶段 3 起:迁到 PVE 上的精简 LXC `agentd-01`(1 vCPU / 1–2 GB,tailnet + 内网 10.77.10.0/24,与 Plane/Temporal 同网段,webhook 走内网)。**LXC 上不装任何 agent CLI**,只有 Python 3.10+、PyYAML、Agent Vault 客户端、ssh 客户端;所有终端(win10、oregon、mac…)一律远程启动(`transport: warpgate-ssh`),LXC 只负责调度、回帖、通知。对外访问经 Warpgate HTTP target,复用既有登录。
|
||||||
- 远程启动的通用实现(P1-6):Windows 用计划任务(已实现);Linux/macOS 经 Warpgate SSH 上传 `run.sh`,`setsid nohup` 脱离会话,轮询 `exit` 文件取回 `last.md`/线程 ID;中继目标名由终端配置给出,不写死。
|
- 远程启动的通用实现(P1-6,已实现):Linux/macOS 经 Warpgate SSH(`agent-access.py run --target <终端 target> -- bash -c …`)上传 `prompt.md`/`run.sh` 并一次调用内启动,`setsid nohup` 脱离会话(macOS 用 perl `POSIX::setsid`),轮询 `exit.txt`,从终端上的输出文件取回最后消息/线程 ID,kill 按进程组 + 会话;中继目标名由终端配置给出。oregon 已切到这条路径(调度器暂仍在 oregon,启动经 Warpgate 回到本机)。部署步骤:`docs/deploy-lxc.md`。
|
||||||
|
- Windows 仍用 helper 脚本 `win10-ps.sh` / `win10-put.sh`(写死 `--target win10`)+ 计划任务。**TODO(统一)**:Windows 也改走 `relay()`(argv = `powershell.exe -EncodedCommand …`,文件同样 base64 分块),删掉对 helper 脚本与 `helpers_dir` 的依赖,LXC 上就不必再复制这两个脚本。
|
||||||
|
|||||||
@@ -24,11 +24,11 @@ agentd 不常驻:一轮运行比起它的那次 tick 活得久,所以执行
|
|||||||
- `kill(handle)` 既用于超时,也用于取消(工作项在运行中被移出 In Progress);必须能在运行未结束时调用。取消后 core 会尝试 `collect(handle)` 取线程 ID,所以 `collect` 在未结束的运行上也不能抛出致命错误(抛异常会被忽略)。
|
- `kill(handle)` 既用于超时,也用于取消(工作项在运行中被移出 In Progress);必须能在运行未结束时调用。取消后 core 会尝试 `collect(handle)` 取线程 ID,所以 `collect` 在未结束的运行上也不能抛出致命错误(抛异常会被忽略)。
|
||||||
|
|
||||||
## RunnerExecutor(现有三个适配器的共同形态)
|
## RunnerExecutor(现有三个适配器的共同形态)
|
||||||
1. 在终端上建运行目录 `<run_root>/<run_id>`(清空旧文件),写 `prompt.md`。
|
1. 在终端上建运行目录 `<run_root>/<run_id>`(清空旧文件),写 `prompt.md`。transport 有 `start_run` 时(远程 POSIX),建目录、写 `prompt.md` 与 runner、启动合并成一次中继调用。
|
||||||
2. 用配置里的命令模板生成命令:`run`(新线程)或 `resume`(续线程)。占位符:`{repo} {model} {effort} {tier} {out} {thread} {perm} {script} {run_dir}`;未知占位符原样保留。`{tier}` 展开为 `tier_arg`(如 `-c 'service_tier="fast"'`),无 tier 时为空。模板首词若在终端 `bins` 里,替换为绝对路径(Windows 前加 `&`)。参数请用单引号,bash 和 PowerShell 都认。
|
2. 用配置里的命令模板生成命令:`run`(新线程)或 `resume`(续线程)。占位符:`{repo} {model} {effort} {tier} {out} {thread} {perm} {script} {run_dir}`;未知占位符原样保留。`{tier}` 展开为 `tier_arg`(如 `-c 'service_tier="fast"'`),无 tier 时为空。tier 来源:`tier:<x>` 标签或 `routing.by_label` 的 `tier`,否则执行器的 `tier_defaults[<model>]`;`tier:none` 表示不加档位参数。模板首词若在终端 `bins` 里,替换为绝对路径(Windows 前加 `&`)。参数请用单引号,bash 和 PowerShell 都认。
|
||||||
3. transport 生成 runner 脚本(Linux `run.sh`,Windows `run.ps1`):进入 repo,写 `pid.txt`/`started.txt`,把 `prompt.md` 从 stdin 喂给命令,stdout → `stdout_name`,stderr → `stderr.txt`,结束写 `exit.txt`(`exit=N at …`)。
|
3. transport 生成 runner 脚本(Linux `run.sh`,Windows `run.ps1`):进入 repo,写 `pid.txt`/`started.txt`,把 `prompt.md` 从 stdin 喂给命令,stdout → `stdout_name`,stderr → `stderr.txt`,结束写 `exit.txt`(`exit=N at …`)。
|
||||||
4. transport 分离启动(Linux `setsid -f bash run.sh`,Windows 一次性计划任务 `agentd-<run_id>`)。
|
4. transport 分离启动(本机 `setsid -f bash run.sh`;远程 Linux `setsid nohup bash run.sh &`,macOS 用 `perl POSIX::setsid`,`$!` 记入 `launch.pid`;Windows 一次性计划任务 `agentd-<run_id>`)。
|
||||||
5. `poll` = `exit.txt` 是否存在;`collect` 由各适配器解析:
|
5. `poll` = `exit.txt` 是否存在;`collect` 由各适配器从终端上保存的文件解析(远程时经中继读回,不依赖调度器本机有任何 CLI):
|
||||||
|
|
||||||
| 适配器 | stdout | 最后消息 | 线程 ID |
|
| 适配器 | stdout | 最后消息 | 线程 ID |
|
||||||
|---|---|---|---|
|
|---|---|---|---|
|
||||||
@@ -39,9 +39,10 @@ agentd 不常驻:一轮运行比起它的那次 tick 活得久,所以执行
|
|||||||
配置里的 `thread_from` / `last_message` 目前只是说明性字段,解析逻辑在适配器里。
|
配置里的 `thread_from` / `last_message` 目前只是说明性字段,解析逻辑在适配器里。
|
||||||
|
|
||||||
## Transport 接口(`agentd/transports/`)
|
## Transport 接口(`agentd/transports/`)
|
||||||
`os`、`join`、`quote`、`default_run_root`、`mkdir(path, clean)`、`write_text`、`read_text`(不存在返回 None)、`grep_first(path, needle)`、`tail(path, n)`、`runner(run_dir, cwd, command, stdout_name)`、`launch(task, script, run_dir)`、`kill(task, run_dir)`、`forget(task)`、`process_count(name)`(None = 不可达)。
|
`os`、`join`、`quote`、`default_run_root`、`mkdir(path, clean)`、`write_text`、`read_text`(不存在返回 None)、`grep_first(path, needle)`、`tail(path, n)`、`runner(run_dir, cwd, command, stdout_name)`、`launch(task, script, run_dir)`、`kill(task, run_dir)`、`forget(task)`、`process_count(name)`(None = 不可达);可选 `start_run(task, run_dir, files, script_name)`(一次完成建目录 + 写文件 + 启动)。
|
||||||
- `local`:Linux 本机文件 + `setsid -f`;kill = 对 `pid.txt` 的进程组发 SIGTERM。
|
- `local`:Linux 本机文件 + `setsid -f`;kill = 对 `pid.txt` 的进程组发 SIGTERM。用于调度器与执行器同机的部署和测试。
|
||||||
- `warpgate-ssh`:Windows,经 `win10-ps.sh`(PowerShell `-EncodedCommand`)和 `win10-put.sh`(分块上传 + SHA-256 校验);文件读取用 base64 标记回传;kill = `taskkill /T /F /PID <pid.txt>` + `schtasks /End /Delete`。脚本目录由 `agentd.helpers_dir`(或终端 `helpers_dir`)配置;脚本目前写死 `--target win10`。路径都放进 PowerShell 单引号字面量(`'` 双写);`schtasks /TR` 里脚本路径用 `\"…\"` 包裹,依赖 Windows PowerShell 5.1 的旧式原生参数传递(PowerShell 7.3+ 的 Standard 模式下需改写)。runner 进不了 repo 时写 `exit=97` 结束。
|
- `warpgate-ssh` + `os: linux|darwin`(`PosixWarpgateTransport`):每个操作是一次 `agent-access.py --config <vault.config> run --target <target> -- bash -c '<script>'`(`relay()`,目标名取终端 `target`,缺省为终端名;客户端把 argv `shlex.join` 后交给目标账户的登录 shell,stdin 为 /dev/null)。文件:base64 放进 `printf %s '<b64>' | base64 -d`,超过 60000 字符分块追加再解码,回显字节数校验;读文件/`grep_first`/`tail` 的结果以 `B64=` 标记回传。runner 与本机相同(`run.sh`),`setsid nohup` 脱离 SSH 会话(pid = pgid = sid);kill = 对该进程组 `kill -TERM -- -<pid>` + `pkill -TERM -s <pid>`,最多等 5 s 后 KILL,`exit.txt` 已存在时不发信号(防 pid 复用);`process_count` = `pgrep -x`。`repo`/`run_root` 必须是远程绝对路径。每次调用约 10–14 s,所以启动只用一次调用。
|
||||||
|
- `warpgate-ssh` + `os: windows`(`WindowsWarpgateTransport`):经 `win10-ps.sh`(PowerShell `-EncodedCommand`)和 `win10-put.sh`(分块上传 + SHA-256 校验);文件读取用 base64 标记回传;kill = `taskkill /T /F /PID <pid.txt>` + `schtasks /End /Delete`。脚本目录由 `agentd.helpers_dir`(或终端 `helpers_dir`)配置;脚本目前写死 `--target win10`(TODO:改走同一个 `relay()`,argv 为 `powershell.exe -NoProfile -NonInteractive -EncodedCommand <b64>`,去掉 helper 脚本)。路径都放进 PowerShell 单引号字面量(`'` 双写);`schtasks /TR` 里脚本路径用 `\"…\"` 包裹,依赖 Windows PowerShell 5.1 的旧式原生参数传递(PowerShell 7.3+ 的 Standard 模式下需改写)。runner 进不了 repo 时写 `exit=97` 结束。
|
||||||
- Codex 续线程:`codex exec resume <thread>` 不接受 `-C`/`-s`,`resume` 模板里不要写;工作目录由 runner 先进入 repo 保证,沙箱沿用 Codex 对该会话/全局配置的默认值(若需强制,用 `-c 'sandbox_mode="workspace-write"'`,上线前在 win10 上验证)。
|
- Codex 续线程:`codex exec resume <thread>` 不接受 `-C`/`-s`,`resume` 模板里不要写;工作目录由 runner 先进入 repo 保证,沙箱沿用 Codex 对该会话/全局配置的默认值(若需强制,用 `-c 'sandbox_mode="workspace-write"'`,上线前在 win10 上验证)。
|
||||||
|
|
||||||
## 新增一个执行器
|
## 新增一个执行器
|
||||||
@@ -55,4 +56,4 @@ agentd 不常驻:一轮运行比起它的那次 tick 活得久,所以执行
|
|||||||
2. 项目 `routing.by_assignee` 加 `<plane_user>: <name>`,`default_executor` 加默认执行器。
|
2. 项目 `routing.by_assignee` 加 `<plane_user>: <name>`,`default_executor` 加默认执行器。
|
||||||
3. cron 的 `agent-access.py exec` 加 `--secret-env <token_key>=<token_key>`(`agentd check-config` 会打印完整参数)。
|
3. cron 的 `agent-access.py exec` 加 `--secret-env <token_key>=<token_key>`(`agentd check-config` 会打印完整参数)。
|
||||||
4. `agentd smoke <PROJECT> <name>` 验证一轮往返。
|
4. `agentd smoke <PROJECT> <name>` 验证一轮往返。
|
||||||
新的 transport 类型(如 Linux 远程机)需在 `agentd/transports/` 实现上面的接口并在 `make_transport` 登记。
|
远程 Linux/macOS 机器不需要新代码:`transport: warpgate-ssh, target: <Warpgate target>, os: linux|darwin, launch: nohup`,`repo`/`run_root`/`bins` 写远程绝对路径(步骤见 `docs/deploy-lxc.md` §7)。全新的 transport 类型需在 `agentd/transports/` 实现上面的接口并在 `make_transport` 登记。
|
||||||
|
|||||||
+2
-1
@@ -9,7 +9,8 @@ from agentd.store import Store
|
|||||||
|
|
||||||
STATES = {"Backlog": "s-bl", "Todo": "s-todo", "In Progress": "s-run", "Waiting": "s-wait", "Review": "s-rev", "Done": "s-done"}
|
STATES = {"Backlog": "s-bl", "Todo": "s-todo", "In Progress": "s-run", "Waiting": "s-wait", "Review": "s-rev", "Done": "s-done"}
|
||||||
MEMBERS = {"window10": "u-w10", "oregon": "u-or", "kappiee": "u-k"}
|
MEMBERS = {"window10": "u-w10", "oregon": "u-or", "kappiee": "u-k"}
|
||||||
LABELS = {"model:astra": "l-astra", "needs-game": "l-game", "exec:claude": "l-claude", "budget:30m": "l-b30", "cap:gpu": "l-gpu"}
|
LABELS = {"model:astra": "l-astra", "needs-game": "l-game", "exec:claude": "l-claude", "budget:30m": "l-b30", "cap:gpu": "l-gpu",
|
||||||
|
"tier:none": "l-tnone", "tier:flex": "l-tflex"}
|
||||||
|
|
||||||
|
|
||||||
def config_data():
|
def config_data():
|
||||||
|
|||||||
@@ -362,5 +362,32 @@ class EnvFailureTest(unittest.TestCase):
|
|||||||
h.close()
|
h.close()
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
class TierDefaultTest(unittest.TestCase):
|
||||||
|
"""executors.<x>.tier_defaults: tier per model unless a label (`tier:<x>`, `tier:none`) sets one."""
|
||||||
|
def run_with(self, labels):
|
||||||
|
data = config_data()
|
||||||
|
data["executors"]["codex"]["tier_defaults"] = {"sol": "fast"}
|
||||||
|
h = Harness(data)
|
||||||
|
try:
|
||||||
|
iid = h.plane.add_issue(1, labels=labels)
|
||||||
|
h.tick()
|
||||||
|
return h.execs[("win10", "codex")].started[0].tier, h.plane.texts(iid)[-1]
|
||||||
|
finally:
|
||||||
|
h.close()
|
||||||
|
|
||||||
|
def test_default_model_gets_its_tier(self):
|
||||||
|
tier, text = self.run_with(())
|
||||||
|
self.assertEqual(tier, "fast")
|
||||||
|
self.assertIn("codex / sol / medium / fast @ win10", text)
|
||||||
|
|
||||||
|
def test_other_model_has_no_default(self):
|
||||||
|
self.assertIsNone(self.run_with(("model:astra",))[0])
|
||||||
|
|
||||||
|
def test_labels_override(self):
|
||||||
|
self.assertEqual(self.run_with(("tier:flex",))[0], "flex")
|
||||||
|
self.assertIsNone(self.run_with(("tier:none",))[0])
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
@@ -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()
|
||||||
@@ -12,11 +12,11 @@ from agentd.config import Config
|
|||||||
from agentd.executors import Run, make_executor
|
from agentd.executors import Run, make_executor
|
||||||
from agentd.store import Store
|
from agentd.store import Store
|
||||||
from agentd.transports.local import LocalTransport
|
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
|
from tests.fakes import config_data
|
||||||
|
|
||||||
|
|
||||||
class RecordingWin(WarpgateSshTransport):
|
class RecordingWin(WindowsWarpgateTransport):
|
||||||
"""Windows transport with the relay replaced by an in-memory recorder."""
|
"""Windows transport with the relay replaced by an in-memory recorder."""
|
||||||
def __init__(self, cfg, terminal):
|
def __init__(self, cfg, terminal):
|
||||||
super().__init__(cfg, terminal)
|
super().__init__(cfg, terminal)
|
||||||
@@ -95,7 +95,7 @@ class WindowsScriptTest(unittest.TestCase):
|
|||||||
|
|
||||||
def fake_run(args, **kw):
|
def fake_run(args, **kw):
|
||||||
seen.update(kw["env"])
|
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"}), \
|
with mock.patch.dict(os.environ, {"PLANE_WINDOW10_API_TOKEN": "s"}), \
|
||||||
mock.patch("agentd.transports.warpgate_ssh.subprocess.run", fake_run):
|
mock.patch("agentd.transports.warpgate_ssh.subprocess.run", fake_run):
|
||||||
WarpgateSshTransport(self.cfg, "win10").ps("'X=ok'")
|
WarpgateSshTransport(self.cfg, "win10").ps("'X=ok'")
|
||||||
|
|||||||
Reference in New Issue
Block a user