Compare commits

...

2 Commits

Author SHA1 Message Date
jj.wang 27ea9119a5 LXC deployment: live config for agentd-01, oregon rollback config, win10 helpers and board protocol in the repo
- config/agentd.yaml is now the agentd-01 live config (state /var/lib/plane-agentd, logs /var/log/plane-agentd);
  config/agentd.oregon.yaml keeps the oregon dispatcher settings for rollback (cron/agentd-tick.sh uses it).
- cron/agentd-tick-lxc.sh: the LXC cron entry.
- tools/win10/: win10-ps.sh / win10-put.sh (vault client dir via AGENT_ACCESS_HOME); relative helpers_dir
  resolves against the repo, default tools/win10.
- config/protocols/gwent-board-protocol.md: copy of gwent/design/board-protocol.md used in prompts.
- agentd.log_dir (default state_dir).
- smoke --watch: create the item and wait for the scheduled dispatcher instead of ticking locally.
2026-10-21 11:56:15 +00:00
jj.wang 4f45d368ad Double-start guard: launching marker, probe-and-adopt, fresh-dir relaunch; skip unreachable terminals per tick
Before the launch call the planned run (run id, terminal, handle, bookkeeping) is saved as a launching
marker. If the launch call fails or agentd dies before recording the run, the next tick probes the run
directory (exit.txt / launch.pid / pid.txt): evidence -> adopt the run; nothing -> relaunch with the same
run id in <run_id>-r2 (-r3 ...); probe unreachable -> keep the marker, which holds the terminal slot.
Transports refuse to clean a run directory holding a pid file. A terminal whose relay fails is skipped
for the rest of the tick (heartbeat.down) while other terminals carry on.
2026-10-21 11:54:40 +00:00
20 changed files with 587 additions and 70 deletions
+7 -3
View File
@@ -121,15 +121,17 @@ def cmd_smoke(cfg, a):
body["labels"] = [labels[x] for x in a.labels.split(",")] body["labels"] = [labels[x] for x in a.labels.split(",")]
issue = pl.create_issue(p["id"], body) issue = pl.create_issue(p["id"], body)
key = "%s-%s" % (p.get("key_prefix", a.project), issue["sequence_id"]) key = "%s-%s" % (p.get("key_prefix", a.project), issue["sequence_id"])
print("created %s; ticking every %ss (timeout %s min)" % (key, a.interval, a.timeout)) print("created %s; %s every %ss (timeout %s min)" % (
d = Agentd(cfg) key, "watching (the scheduled dispatcher runs it)" if a.watch else "ticking", a.interval, a.timeout), flush=True)
d = None if a.watch else Agentd(cfg)
states = pl.states(p["id"]) states = pl.states(p["id"])
done_states = {states[p["states"][k]] for k in ("review", "waiting")} done_states = {states[p["states"][k]] for k in ("review", "waiting")}
deadline = time.time() + a.timeout * 60 deadline = time.time() + a.timeout * 60
state = None state = None
while time.time() < deadline: while time.time() < deadline:
try: try:
d.tick(project=a.project, only_issue=issue["id"]) if d:
d.tick(project=a.project, only_issue=issue["id"])
except Busy as exc: except Busy as exc:
print("skip: %s" % exc) print("skip: %s" % exc)
state = pl.get_issue(p["id"], issue["id"]).get("state") state = pl.get_issue(p["id"], issue["id"]).get("state")
@@ -199,6 +201,8 @@ def parser():
s.add_argument("--interval", type=int, default=30) s.add_argument("--interval", type=int, default=30)
s.add_argument("--timeout", type=int, default=30, help="minutes") s.add_argument("--timeout", type=int, default=30, help="minutes")
s.add_argument("--labels", default="", help="comma-separated Plane labels, e.g. model:sol") s.add_argument("--labels", default="", help="comma-separated Plane labels, e.g. model:sol")
s.add_argument("--watch", action="store_true",
help="do not tick here; only wait for the scheduled dispatcher (use when it runs on another host)")
s.add_argument("--as", dest="as_") s.add_argument("--as", dest="as_")
s.set_defaults(fn=cmd_smoke) s.set_defaults(fn=cmd_smoke)
sub.add_parser("check-config", help="validate the config and list token env vars").set_defaults(fn=cmd_check_config) sub.add_parser("check-config", help="validate the config and list token env vars").set_defaults(fn=cmd_check_config)
+12 -2
View File
@@ -155,8 +155,18 @@ class Config:
return p if p.is_absolute() else ROOT / p return p if p.is_absolute() else ROOT / p
def helpers_dir(self, terminal): def helpers_dir(self, terminal):
d = self.terminals[terminal].get("helpers_dir") or self.agentd.get("helpers_dir") or str(ROOT / "tools") d = self.terminals[terminal].get("helpers_dir") or self.agentd.get("helpers_dir") or "tools/win10"
return Path(os.path.expanduser(d)) p = Path(os.path.expanduser(d))
return p if p.is_absolute() else ROOT / p
@property
def log_dir(self):
"""agentd.log lives here (default: state_dir)."""
d = self.agentd.get("log_dir")
if not d:
return self.state_dir
p = Path(os.path.expanduser(d))
return p if p.is_absolute() else ROOT / p
def token_env(self, terminal): def token_env(self, terminal):
"""Environment variable that carries the terminal's Plane token (== the Vault key name).""" """Environment variable that carries the terminal's Plane token (== the Vault key name)."""
+120 -39
View File
@@ -3,7 +3,13 @@
on failure post a structured note and retry (transient) or hand over (Waiting); kill on timeout; on failure post a structured note and retry (transient) or hand over (Waiting); kill on timeout;
(b) start the lowest-sequence Todo items whose terminal has a free slot and whose serialize group is idle; (b) start the lowest-sequence Todo items whose terminal has a free slot and whose serialize group is idle;
(c) write the heartbeat. (c) write the heartbeat.
Everything is deterministic; models only run inside executors.""" Everything is deterministic; models only run inside executors.
Double-start guard: before the launch call a `launching` marker (the planned run and its handle) is saved. If the
launch call fails or the process dies before the run is recorded, the next tick probes the run directory: launch
evidence (exit.txt / launch.pid / pid.txt) -> adopt the run; nothing -> relaunch with the same run id in a fresh
directory (`<run_id>-r2`, -r3 ...); probe unreachable -> keep the marker (it holds the terminal slot).
A terminal whose relay fails is skipped for the rest of the tick; other terminals carry on."""
import os import os
import re import re
import time import time
@@ -15,7 +21,7 @@ from .executors import ExecResult, Run, make_executor
from .executors.base import fill from .executors.base import fill
from .plane import Plane, html_to_text, text_to_html from .plane import Plane, html_to_text, text_to_html
from .store import Store from .store import Store
from .transports import make_transport from .transports import TransportError, make_transport
TAGS = ("review", "question", "blocked", "progress") TAGS = ("review", "question", "blocked", "progress")
WORKER = "[worker]" WORKER = "[worker]"
@@ -143,7 +149,7 @@ class Ctx:
class Agentd: class Agentd:
def __init__(self, cfg, store=None, plane_factory=None, executor_factory=None, transport_factory=None, now=time.time): def __init__(self, cfg, store=None, plane_factory=None, executor_factory=None, transport_factory=None, now=time.time):
self.cfg = cfg self.cfg = cfg
self.store = store or Store(cfg.state_dir) self.store = store or Store(cfg.state_dir, log_dir=cfg.log_dir)
self.now = now self.now = now
self.log = self.store.log self.log = self.store.log
self.plane_factory = plane_factory or (lambda term: Plane( self.plane_factory = plane_factory or (lambda term: Plane(
@@ -154,6 +160,12 @@ class Agentd:
def _reset_caches(self): def _reset_caches(self):
self._clients, self._transports, self._executors, self._ready = {}, {}, {}, {} self._clients, self._transports, self._executors, self._ready = {}, {}, {}, {}
self._down = set() # terminals whose relay failed during this tick
def mark_down(self, term, exc):
if term not in self._down:
self.log("%s unreachable, skipping it for the rest of this tick: %s" % (term, exc))
self._down.add(term)
# ---------------------------------------------------------------- caches # ---------------------------------------------------------------- caches
def client(self, terminal): def client(self, terminal):
@@ -195,9 +207,11 @@ class Agentd:
self.log("%s tick error: %r" % (name, exc)) self.log("%s tick error: %r" % (name, exc))
finally: finally:
self.store.save(data) self.store.save(data)
self.store.heartbeat({"errors": errors, "inflight": [ self.store.heartbeat({"errors": errors, "down": sorted(self._down), "inflight": [
{"project": p, "run": r["inflight"]["run_id"], "terminal": r["inflight"]["terminal"]} {"project": p, "run": r["inflight"]["run_id"], "terminal": r["inflight"]["terminal"]}
for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("inflight")]}) for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("inflight")],
"launching": [{"project": p, "run": r["launching"]["run_id"], "terminal": r["launching"]["terminal"]}
for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("launching")]})
return data return data
def loop(self, interval): def loop(self, interval):
@@ -216,14 +230,43 @@ class Agentd:
pstate = data["projects"].setdefault(name, {"issues": {}}) pstate = data["projects"].setdefault(name, {"issues": {}})
ctx = Ctx(name, pcfg, self.reader(pcfg)) ctx = Ctx(name, pcfg, self.reader(pcfg))
for iid, rec in list(pstate["issues"].items()): for iid, rec in list(pstate["issues"].items()):
if rec.get("inflight"): run = rec.get("inflight") or rec.get("launching")
try: if not run or run["terminal"] in self._down:
continue
try:
if rec.get("inflight"):
self.poll(ctx, data, iid, rec) self.poll(ctx, data, iid, rec)
except Exception as exc: else:
self.log("%s poll error: %r" % (ctx.key(rec.get("seq")), exc)) self.resolve_launch(ctx, iid, rec)
self.store.save(data) # persist each round's outcome (thread id, retry) before touching the next except TransportError as exc:
self.mark_down(run["terminal"], exc)
except Exception as exc:
self.log("%s poll error: %r" % (ctx.key(rec.get("seq")), exc))
self.store.save(data) # persist each round's outcome (thread id, retry) before touching the next
self.pick(ctx, data, pstate, only_issue) self.pick(ctx, data, pstate, only_issue)
def resolve_launch(self, ctx, iid, rec):
"""A launch that was never confirmed (the launch call failed or agentd died mid-start): probe the run
directory and adopt the run if it exists; otherwise relaunch in a fresh directory, never wiping one."""
m = rec["launching"]
key = ctx.key(rec["seq"])
present = self.executor(m["terminal"], m["executor"]).probe(m["handle"]) # TransportError -> marker kept
if present:
self.log("%s run %s: unconfirmed launch found on %s (%s); adopting it" % (
key, m["run_id"], m["terminal"], ",".join(present)))
return self.launched(ctx, iid, rec, m, m["handle"], adopted=True)
rec.pop("launching")
issue = ctx.issues.get(iid)
state = issue and issue.get("state")
if state == ctx.state_id("todo") or (state == ctx.state_id("running") and rec.get("retry")):
rec["relaunch"] = {"run_id": m["run_id"], "attempt": m.get("attempt", 1) + 1}
self.log("%s run %s: nothing at %s; relaunching in %s-r%d" % (
key, m["run_id"], m["handle"].get("run_dir"), m["run_id"], rec["relaunch"]["attempt"]))
else:
rec.pop("relaunch", None)
self.log("%s run %s: nothing at %s and the item is no longer dispatchable; marker dropped" % (
key, m["run_id"], m["handle"].get("run_dir")))
# ---------------------------------------------------------------- (a) poll / finish # ---------------------------------------------------------------- (a) poll / finish
def poll(self, ctx, data, iid, rec): def poll(self, ctx, data, iid, rec):
fl = rec["inflight"] fl = rec["inflight"]
@@ -404,7 +447,7 @@ class Agentd:
load, groups = Counter(), set() load, groups = Counter(), set()
for p in data["projects"].values(): for p in data["projects"].values():
for r in p["issues"].values(): for r in p["issues"].values():
fl = r.get("inflight") fl = r.get("inflight") or r.get("launching") # an unconfirmed launch holds its slot
if fl: if fl:
load[fl["terminal"]] += 1 load[fl["terminal"]] += 1
if fl.get("group"): if fl.get("group"):
@@ -422,7 +465,7 @@ class Agentd:
if only_issue and iid != only_issue: if only_issue and iid != only_issue:
continue continue
rec = pstate["issues"].get(iid, {}) rec = pstate["issues"].get(iid, {})
if rec.get("inflight"): if rec.get("inflight") or rec.get("launching"):
continue continue
if issue.get("state") == todo: if issue.get("state") == todo:
cands.append(issue) cands.append(issue)
@@ -438,15 +481,21 @@ class Agentd:
self.log("%s not dispatched: %s" % (ctx.key(issue.get("sequence_id")), route)) self.log("%s not dispatched: %s" % (ctx.key(issue.get("sequence_id")), route))
continue continue
term, group = route["terminal"], route["group"] term, group = route["terminal"], route["group"]
if term in self._down:
continue
if load[term] >= self.cfg.terminals[term].get("concurrency", 1) or (group and group in groups): if load[term] >= self.cfg.terminals[term].get("concurrency", 1) or (group and group in groups):
continue continue
if not self.terminal_ready(term, load[term]): if not self.terminal_ready(term, load[term]):
continue continue
try: try:
self.start(ctx, pstate, issue, route) self.start(ctx, data, pstate, issue, route)
except Exception as exc: except Exception as exc:
self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc)) self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
continue if isinstance(exc, TransportError):
self.mark_down(term, exc)
if not pstate["issues"].get(issue["id"], {}).get("launching"):
continue
# the launch call failed but may have started the run: its marker keeps the slot until probed
self.store.save(data) # a launched run must be on disk before anything else can fail self.store.save(data) # a launched run must be on disk before anything else can fail
load[term] += 1 load[term] += 1
if group: if group:
@@ -454,14 +503,17 @@ class Agentd:
def terminal_ready(self, term, our_load): def terminal_ready(self, term, our_load):
"""Optional guard against processes we did not start (e.g. a manual codex session).""" """Optional guard against processes we did not start (e.g. a manual codex session)."""
if term in self._down:
return False
if term not in self._ready: if term not in self._ready:
names = self.cfg.terminals[term].get("busy_processes", []) names = self.cfg.terminals[term].get("busy_processes", [])
ready = True ready = True
for name in names: for name in names:
count = self.transport(term).process_count(name) count = self.transport(term).process_count(name)
if count is None: if count is None:
self.log("%s unreachable; skipping" % term) self.mark_down(term, "busy-process probe failed")
ready = False ready = False
break
elif count > 0 and our_load == 0: elif count > 0 and our_load == 0:
self.log("%s busy (%d %s running that agentd did not start)" % (term, count, name)) self.log("%s busy (%d %s running that agentd did not start)" % (term, count, name))
ready = False ready = False
@@ -522,7 +574,7 @@ class Agentd:
return "" return ""
return p.read_text(encoding="utf-8") return p.read_text(encoding="utf-8")
def start(self, ctx, pstate, issue, route): def start(self, ctx, data, pstate, issue, route):
iid, seq = issue["id"], issue["sequence_id"] iid, seq = issue["id"], issue["sequence_id"]
key = ctx.key(seq) key = ctx.key(seq)
rec = pstate["issues"].setdefault(iid, new_rec(seq)) rec = pstate["issues"].setdefault(iid, new_rec(seq))
@@ -552,25 +604,11 @@ class Agentd:
intro = fill(ctx.cfg.get("intro", DEFAULT_INTRO), intro = fill(ctx.cfg.get("intro", DEFAULT_INTRO),
dict(terminal=term, executor=exname, repo=tcfg["repo"], prefix=ctx.prefix, key=key)) dict(terminal=term, executor=exname, repo=tcfg["repo"], prefix=ctx.prefix, key=key))
prompt = build_prompt(key, issue, comments, seen, bool(thread), self.protocol(pcfg), intro, handoff, retry_note) prompt = build_prompt(key, issue, comments, seen, bool(thread), self.protocol(pcfg), intro, handoff, retry_note)
run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"]) relaunch = rec.get("relaunch") or {}
handle = ex.start(run) attempt = relaunch.get("attempt", 1) if relaunch.get("run_id") == run_id else 1
rec.pop("retry", None) dir_name = run_id if attempt == 1 else "%s-r%d" % (run_id, attempt)
rec.pop("rebuild", None) run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"],
if not is_retry: dir_name=dir_name)
rec["retries"] = 0
now = self.now()
# transient retries continue the same logical round: they do not use up max_rounds_per_thread
rounds = (rec.get("thread_rounds", 0) + (0 if is_retry else 1)) if thread else 1
rec.update(runs=k, thread_rounds=rounds, inflight={
"run_id": run_id, "terminal": term, "executor": exname, "handle": handle, "round": k,
"started": now, "deadline": now + route["timeout_hours"] * 3600, "resumed": bool(thread),
"group": route["group"], "model": route["model"], "effort": route["effort"]})
if not thread:
rec.pop("thread_id", None)
rec.pop("thread_on", None)
rec["seen_comments"] = sorted(seen | {c["id"] for c in comments if c.get("id")})
rec["last_comment_at"] = max([c.get("created_at", "") for c in comments] + [rec.get("last_comment_at", "")])
client = self.client(term)
if thread: if thread:
mode = "续线程 %s" % thread[:12] mode = "续线程 %s" % thread[:12]
elif reason: elif reason:
@@ -579,10 +617,53 @@ class Agentd:
mode = "新线程(附交接摘要)" if handoff else "新线程" mode = "新线程(附交接摘要)" if handoff else "新线程"
if is_retry: if is_retry:
mode += ",重试 %d" % rec.get("retries", 0) mode += ",重试 %d" % rec.get("retries", 0)
detail = " / ".join(x for x in (exname, route["model"], route["effort"], route["tier"]) if x) if attempt > 1:
mode += ",重新启动(目录 %s)" % dir_name
# Everything the bookkeeping needs once the run is known to exist; saved before the launch call so that a
# launch whose outcome is unknown (relay timeout, crash) is probed and adopted instead of started twice.
plan = {"run_id": run_id, "terminal": term, "executor": exname, "attempt": attempt, "at": self.now(),
"handle": ex.handle(run), "round": k, "budget_s": route["timeout_hours"] * 3600,
"resumed": bool(thread), "group": route["group"], "model": route["model"], "effort": route["effort"],
"is_retry": is_retry,
# transient retries continue the same logical round: they do not use up max_rounds_per_thread
"thread_rounds": (rec.get("thread_rounds", 0) + (0 if is_retry else 1)) if thread else 1,
"seen": sorted(seen | {c["id"] for c in comments if c.get("id")}),
"last_comment_at": max([c.get("created_at", "") for c in comments] + [rec.get("last_comment_at", "")]),
"mode": mode,
"detail": " / ".join(x for x in (exname, route["model"], route["effort"], route["tier"]) if x)}
rec["launching"] = plan
self.store.save(data)
handle = ex.start(run) # raises -> the marker stays and the next tick probes the run directory
self.launched(ctx, iid, rec, plan, handle)
def launched(self, ctx, iid, rec, plan, handle, adopted=False):
"""The run exists on the terminal: record it as in flight, then move the item and post the start note."""
key, term = ctx.key(rec["seq"]), plan["terminal"]
rec.pop("launching", None)
rec.pop("relaunch", None)
rec.pop("retry", None)
rec.pop("rebuild", None)
if not plan["is_retry"]:
rec["retries"] = 0
rec.update(runs=plan["round"], thread_rounds=plan["thread_rounds"], inflight={
"run_id": plan["run_id"], "terminal": term, "executor": plan["executor"], "handle": handle,
"round": plan["round"], "started": plan["at"], "deadline": plan["at"] + plan["budget_s"],
"resumed": plan["resumed"], "group": plan["group"], "model": plan["model"], "effort": plan["effort"]})
if not plan["resumed"]:
rec.pop("thread_id", None)
rec.pop("thread_on", None)
rec["seen_comments"] = plan["seen"]
rec["last_comment_at"] = plan["last_comment_at"]
mode = plan["mode"] + (",启动未确认、已接管" if adopted else "")
issue = ctx.issues.get(iid)
client = self.client(term)
try: try:
self.move(ctx, client, iid, "running", tcfg["plane_user"]) # an adopted run whose item a person moved meanwhile is left alone: poll cancels it next tick
self.post(ctx, client, iid, rec, "%s 第 %d 轮开始:%s @ %s,%s,run=%s。" % (WORKER, k, detail, term, mode, run_id)) if issue is not None and issue.get("state") in (ctx.state_id("todo"), ctx.state_id("running")):
self.move(ctx, client, iid, "running", self.cfg.terminals[term]["plane_user"])
self.post(ctx, client, iid, rec, "%s 第 %d 轮开始:%s @ %s,%s,run=%s。" % (
WORKER, plan["round"], plan["detail"], term, mode, plan["run_id"]))
except Exception as exc: except Exception as exc:
self.log("%s bookkeeping after start failed: %r" % (key, exc)) self.log("%s bookkeeping after start failed: %r" % (key, exc))
self.log("%s run %s started on %s (%s, %s)" % (key, run_id, term, detail, mode)) self.log("%s run %s %s on %s (%s, %s)" % (key, plan["run_id"], "adopted" if adopted else "started", term,
plan["detail"], mode))
+20 -3
View File
@@ -17,6 +17,7 @@ class Run:
tier: Optional[str] = None tier: Optional[str] = None
thread_id: Optional[str] = None # set => resume this thread thread_id: Optional[str] = None # set => resume this thread
script: Optional[str] = None # shell executor script: Optional[str] = None # shell executor
dir_name: Optional[str] = None # run directory / task name; default run_id (`<run_id>-r2` on a relaunch)
@dataclass @dataclass
@@ -35,9 +36,15 @@ class Executor:
def __init__(self, name, cfg, terminal, tcfg, transport): def __init__(self, name, cfg, terminal, tcfg, transport):
self.name, self.cfg, self.terminal, self.tcfg, self.transport = name, cfg, terminal, tcfg, transport self.name, self.cfg, self.terminal, self.tcfg, self.transport = name, cfg, terminal, tcfg, transport
def handle(self, run): # -> the handle `start(run)` will return, computed without touching the terminal
raise NotImplementedError
def start(self, run): # -> handle (dict) def start(self, run): # -> handle (dict)
raise NotImplementedError raise NotImplementedError
def probe(self, handle): # -> names of launch evidence present on the terminal ([] = never launched)
raise NotImplementedError
def poll(self, handle): # -> "running" | "finished" def poll(self, handle): # -> "running" | "finished"
raise NotImplementedError raise NotImplementedError
@@ -84,11 +91,15 @@ class RunnerExecutor(Executor):
program = "& " + program program = "& " + program
return (program + " " + rest).strip() return (program + " " + rest).strip()
def handle(self, run):
name = run.dir_name or run.run_id
return {"run_dir": self.transport.join(self.run_root(), name), "task": "agentd-" + name, "thread_id": run.thread_id}
def start(self, run): def start(self, run):
t = self.transport t = self.transport
run_dir = t.join(self.run_root(), run.run_id) h = self.handle(run)
run_dir, task = h["run_dir"], h["task"]
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)
task = "agentd-" + run.run_id
files = {"prompt.md": run.prompt, script_name: script} files = {"prompt.md": run.prompt, script_name: script}
if hasattr(t, "start_run"): # remote POSIX: stage + launch in one relay round trip if hasattr(t, "start_run"): # remote POSIX: stage + launch in one relay round trip
t.start_run(task, run_dir, files, script_name) t.start_run(task, run_dir, files, script_name)
@@ -97,7 +108,13 @@ class RunnerExecutor(Executor):
for name, text in files.items(): for name, text in files.items():
t.write_text(t.join(run_dir, name), text) t.write_text(t.join(run_dir, name), text)
t.launch(task, t.join(run_dir, script_name), run_dir) t.launch(task, t.join(run_dir, script_name), run_dir)
return {"run_dir": run_dir, "task": task, "thread_id": run.thread_id} return h
PROBE_FILES = ("exit.txt", "launch.pid", "pid.txt")
def probe(self, handle):
"""Launch evidence in the run directory (one transport call; raises TransportError when unreachable)."""
return self.transport.probe(handle["run_dir"], self.PROBE_FILES)
def poll(self, handle): def poll(self, handle):
text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt")) text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt"))
+4 -3
View File
@@ -17,8 +17,9 @@ class Busy(Exception):
class Store: class Store:
def __init__(self, state_dir, echo=True): def __init__(self, state_dir, echo=True, log_dir=None):
self.dir = Path(state_dir) self.dir = Path(state_dir)
self.log_dir = Path(log_dir) if log_dir else self.dir
self.echo = echo self.echo = echo
self.path = self.dir / "agentd-state.json" self.path = self.dir / "agentd-state.json"
@@ -59,9 +60,9 @@ class Store:
self._atomic_write(self.dir / "heartbeat.json", info) self._atomic_write(self.dir / "heartbeat.json", info)
def log(self, msg): def log(self, msg):
self.dir.mkdir(parents=True, exist_ok=True) self.log_dir.mkdir(parents=True, exist_ok=True)
line = time.strftime("%Y-%m-%dT%H:%M:%S") + " " + msg line = time.strftime("%Y-%m-%dT%H:%M:%S") + " " + msg
if self.echo: if self.echo:
print(line, flush=True) print(line, flush=True)
with open(self.dir / "agentd.log", "a", encoding="utf-8") as fh: with open(self.log_dir / "agentd.log", "a", encoding="utf-8") as fh:
fh.write(line + "\n") fh.write(line + "\n")
+9
View File
@@ -4,6 +4,10 @@ import signal
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from . import TransportError
PID_FILES = ("pid.txt", "launch.pid") # a run directory holding one of these is never cleaned
# Shared with the remote POSIX transport (warpgate_ssh): plain sh + `date +FMT`, so it also runs on macOS. # 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 +%Y-%m-%dT%H:%M:%S%z) (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; }}
@@ -40,6 +44,8 @@ class LocalTransport:
p = Path(path) p = Path(path)
p.mkdir(parents=True, exist_ok=True) p.mkdir(parents=True, exist_ok=True)
if clean: if clean:
if any((p / n).exists() for n in PID_FILES):
raise TransportError("refusing to clean %s: it holds a pid file (HASPID)" % p)
for f in p.iterdir(): for f in p.iterdir():
if f.is_file(): if f.is_file():
f.unlink() f.unlink()
@@ -51,6 +57,9 @@ class LocalTransport:
p = Path(path) p = Path(path)
return p.read_text(encoding="utf-8", errors="replace") if p.exists() else None return p.read_text(encoding="utf-8", errors="replace") if p.exists() else None
def probe(self, run_dir, names):
return [n for n in names if os.path.exists(os.path.join(run_dir, n))]
def grep_first(self, path, needle): def grep_first(self, path, needle):
text = self.read_text(path) or "" text = self.read_text(path) or ""
return next((line for line in text.splitlines() if needle in line), None) return next((line for line in text.splitlines() if needle in line), None)
+23 -3
View File
@@ -113,9 +113,21 @@ class WindowsWarpgateTransport:
def mkdir(self, path, clean=False): def mkdir(self, path, clean=False):
q = self.quote(path) q = self.quote(path)
script = "New-Item -ItemType Directory -Force -Path %s | Out-Null\n" % q script = "New-Item -ItemType Directory -Force -Path %s | Out-Null\n" % q
if clean: if clean: # never wipe a directory a runner has already started in
script += ("if ((Test-Path -LiteralPath (Join-Path %s 'pid.txt')) -or (Test-Path -LiteralPath (Join-Path %s 'launch.pid'))) "
"{ 'HASPID'; return }\n" % (q, q))
script += "Get-ChildItem -LiteralPath %s -File | Remove-Item -Force\n" % q script += "Get-ChildItem -LiteralPath %s -File | Remove-Item -Force\n" % q
self.ps(script + "'MKDIR=ok'") out = self.ps(script + "'MKDIR=ok'")
if "MKDIR=ok" not in out:
raise TransportError("mkdir %s failed: %s" % (path, out.strip()[-300:]))
def probe(self, run_dir, names):
out = self.ps("$d = %s\n'PROBE=' + ((@(%s) | Where-Object { Test-Path -LiteralPath (Join-Path $d $_) }) -join ',')"
% (self.quote(run_dir), ",".join(self.quote(n) for n in names)))
m = re.search(r"PROBE=([^\n]*)", out)
if not m:
raise TransportError("probe %s: unexpected relay output: %s" % (run_dir, out.strip()[-200:]))
return [n for n in m.group(1).strip().split(",") if n]
def write_text(self, path, text): def write_text(self, path, text):
fd, tmp = tempfile.mkstemp(prefix="agentd-", suffix=".txt") fd, tmp = tempfile.mkstemp(prefix="agentd-", suffix=".txt")
@@ -230,7 +242,9 @@ class PosixWarpgateTransport:
def mkdir_script(self, path, clean=False): def mkdir_script(self, path, clean=False):
q = self.quote(path) q = self.quote(path)
s = "mkdir -p %s || exit 1\n" % q s = "mkdir -p %s || exit 1\n" % q
if clean: if clean: # never wipe a directory a runner has already started in
s += "if [ -e %s ] || [ -e %s ]; then echo HASPID; exit 1; fi\n" % (
self.quote(posixpath.join(path, "pid.txt")), self.quote(posixpath.join(path, "launch.pid")))
s += "find %s -maxdepth 1 -type f -exec rm -f {} + || exit 1\n" % q s += "find %s -maxdepth 1 -type f -exec rm -f {} + || exit 1\n" % q
return s return s
@@ -275,6 +289,12 @@ class PosixWarpgateTransport:
raise TransportError("read %s: unexpected relay output" % path) raise TransportError("read %s: unexpected relay output" % path)
return _b64_payload(out, "B64") return _b64_payload(out, "B64")
def probe(self, run_dir, names):
d = self.quote(run_dir)
out = self.sh("printf PROBE=; for f in %s; do [ -e %s/\"$f\" ] && printf '%%s,' \"$f\"; done; echo"
% (" ".join(self.quote(n) for n in names), d), timeout=60, marker="PROBE=")
return [n for n in re.search(r"PROBE=([^\n]*)", out).group(1).split(",") if n]
def _b64_cmd(self, pipeline): def _b64_cmd(self, pipeline):
out = self.sh("printf B64=; { %s; } 2>/dev/null | base64 | tr -d '\\n'; echo" % pipeline) out = self.sh("printf B64=; { %s; } 2>/dev/null | base64 | tr -d '\\n'; echo" % pipeline)
text = _b64_payload(out, "B64") text = _b64_payload(out, "B64")
+1 -1
View File
@@ -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 (Windows terminals only) helpers_dir: tools/win10 # win10-ps.sh / win10-put.sh (Windows terminals only); relative = repo
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
+89
View File
@@ -0,0 +1,89 @@
# plane-agentd dispatcher configuration for oregon -- the ROLLBACK target since the dispatcher moved to the LXC
# agentd-01 (PAD-11; live config there: config/agentd.yaml). Used by cron/agentd-tick.sh (not in oregon's crontab
# while the LXC dispatches). No secrets: token_key values are Vault key names.
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}
agentd:
state_dir: ~/.gwent-board/agentd # JSON state, agentd.log, heartbeat.json, local run dirs (runs/)
helpers_dir: tools/win10 # win10-ps.sh / win10-put.sh (Windows terminals only); relative = repo
interval: 120
max_rounds_per_thread: 8
comment_max_chars: 6000
cli_terminal: oregon
terminals:
win10:
transport: warpgate-ssh
target: win10
os: windows
launch: schtasks
repo: 'C:\Users\Administrator\Documents\GwentAssistantProject\gwent'
run_root: 'C:\Users\Administrator\Documents\GwentAssistantProject\gwent\runtime\agentd'
bins: {codex: 'C:\Users\Administrator\AppData\Local\Programs\OpenAI\Codex\bin\codex.exe'}
plane_user: window10
token_key: PLANE_WINDOW10_API_TOKEN
concurrency: 1
busy_processes: [codex] # do not start while a codex.exe agentd did not start is running
capabilities: [android, game]
executors: [codex]
oregon:
# 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: 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} # the SSH non-login shell has the system PATH only; no codex on oregon
executors: [claude-code]
executors:
codex:
adapter: codex
run: "codex exec --json -C '{repo}' -s workspace-write -m {model} -c 'model_reasoning_effort=\"{effort}\"' {tier} -o '{out}' -"
# `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}' -"
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
last_message: '{out}'
models: [gpt-6.1-sol, gpt-6-astra]
default_effort: medium
claude-code:
adapter: claude-code
run: "claude -p --model {model} --output-format json --permission-mode {perm}"
resume: "claude -p --resume {thread} --model {model} --output-format json --permission-mode {perm}"
permission_mode: acceptEdits
thread_from: json:session_id
last_message: json:result
models: [sonnet, opus, haiku]
projects:
GWENT:
id: 79cb5a60-aa4f-4138-9864-1dd70426e665
key_prefix: GW
states: {backlog: Backlog, todo: Todo, running: In Progress, waiting: Waiting, review: Review, done: Done}
reviewer: oregon
answerer: oregon
protocol_file: config/protocols/gwent-board-protocol.md # relative = repo; refresh from gwent/design/board-protocol.md
routing:
by_assignee: {window10: win10, oregon: oregon}
by_label:
'model:astra': {model: gpt-6-astra, effort: high}
'model:sol': {model: gpt-6.1-sol, tier: fast}
'effort:high': {effort: high}
'needs-game': {serialize_group: game}
default_executor: {win10: codex, oregon: claude-code}
retry:
transient_max: 3
backoff_minutes: [2, 5, 10]
timeout_hours: 4
max_rounds_per_thread: 8
max_progress_rounds: 12
# notify (ntfy) is phase 3: left out on purpose
+8 -8
View File
@@ -1,11 +1,13 @@
# plane-agentd live configuration on oregon (phase 1). No secrets: token_key values are Vault key names; the tokens # plane-agentd live configuration on the LXC agentd-01 (/opt/plane-agentd, runs as root; docs/deploy-lxc.md).
# are injected by `agent-access.py exec --secret-env KEY=KEY` (see cron/agentd-tick.sh and docs/runbook.md). # No secrets: token_key values are Vault key names; the tokens are injected by
# `agent-access.py exec --secret-env KEY=KEY` (cron/agentd-tick-lxc.sh). Rollback to oregon: config/agentd.oregon.yaml.
plane: {base_url: https://plane.kappiee.com, workspace: kappiee} 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}
agentd: agentd:
state_dir: ~/.gwent-board/agentd # JSON state, agentd.log, heartbeat.json, local run dirs (runs/) state_dir: /var/lib/plane-agentd # JSON state, heartbeat.json, lock
helpers_dir: /home/ubuntu/Documents/wjj/project/gwent-design/tools # win10-ps.sh / win10-put.sh (Windows terminals only) log_dir: /var/log/plane-agentd # agentd.log (cron output: cron.log, same dir)
helpers_dir: tools/win10 # win10-ps.sh / win10-put.sh (Windows terminals only); relative = repo
interval: 120 interval: 120
max_rounds_per_thread: 8 max_rounds_per_thread: 8
comment_max_chars: 6000 comment_max_chars: 6000
@@ -27,9 +29,7 @@ terminals:
capabilities: [android, game] capabilities: [android, game]
executors: [codex] executors: [codex]
oregon: oregon:
# remote even though agentd runs on oregon today: the dispatcher is moving to an LXC without agent CLIs # the dispatcher (LXC, no agent CLIs) reaches oregon through Warpgate like any other host (design §9)
# (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 transport: warpgate-ssh
target: oregon # agent-access.py target name target: oregon # agent-access.py target name
os: linux os: linux
@@ -70,7 +70,7 @@ projects:
states: {backlog: Backlog, todo: Todo, running: In Progress, waiting: Waiting, review: Review, done: Done} states: {backlog: Backlog, todo: Todo, running: In Progress, waiting: Waiting, review: Review, done: Done}
reviewer: oregon reviewer: oregon
answerer: oregon answerer: oregon
protocol_file: /home/ubuntu/Documents/wjj/project/gwent/design/board-protocol.md protocol_file: config/protocols/gwent-board-protocol.md # relative = repo; refresh from gwent/design/board-protocol.md
routing: routing:
by_assignee: {window10: win10, oregon: oregon} by_assignee: {window10: win10, oregon: oregon}
by_label: by_label:
+47
View File
@@ -0,0 +1,47 @@
# 看板协作协议(Plane 作为任务板与信箱)
看板 = https://plane.kappiee.com/kappiee/projects/79cb5a60-aa4f-4138-9864-1dd70426e665/issues/ ,工作项编号 `GW-k`。Gitea 只管代码(仓库、提交、PR),不再放任务。
## 账号 = 终端
- `kappiee` 用户(所有者,做决定);`oregon` Claude(设计、派活、验收);`window10` win10 Codex(执行);`mac` 之后接入的 Mac 终端。
- 凭据键(Agent Vault 注入内存,不落盘、不进 win10、不进评论):`PLANE_OREGON_API_TOKEN`、`PLANE_WINDOW10_API_TOKEN`(worker 以 window10 发帖)。API 头 `X-API-Key`。
## 状态机
`Backlog → Todo → In Progress → Waiting → Review → Done / Cancelled`
- Backlog:想法/未排期。Todo + assignee `window10`:交给 Codex,worker 下一个 tick 取走。
- In Progress:worker 已启动 codex 运行(worker 自动设置)。
- Waiting + assignee `oregon`:Codex 有待回答的问题或被阻塞(`[question]`/`[blocked]`),等 Claude 回答或升级给 kappiee。
- Review + assignee `oregon`:产出就绪(`[review]`),Claude 验收。通过 → Done;不通过 → 写意见,改回 Todo + `window10`。
- `[progress]`:worker 自动改回 Todo + `window10`,下一轮续跑。
- 需要用户决定时 Claude 把 assignee 改为 `kappiee`(state 保持 Waiting)。
## 工作项
- 标题 `T-xxx 简述`;描述 = 任务书(目标、允许的工具、不做的事、产出、验收线)。
- 评论 = 消息,作者由账号区分;正文首行保留角色标记 `[claude]` / `[codex]` / `[worker]`,worker 构造 prompt 时跳过 `[worker]` 评论。
- 结果标签(Codex 最后一条消息首行必须是其一):
- `[review]` 产出完成,附:改动落点、产出路径、SHA、自检结果、如何重跑。
- `[question]` 需要回答才能继续,附"若无人回答,我默认会怎么做"。
- `[blocked]` 无法继续,附原因与已尝试的办法。
- `[progress]` 做了一部分,需要下一轮继续。
- 未标注首行按 `[review]` 处理。
## 标签、模块、关系、视图
- 标签:`model:astra`(gpt-6-astra,high)/ `model:sol`(gpt-6.1-sol fast,medium,默认)/ `effort:high` / `needs-game`(需要游戏在线,别与其他占用游戏的任务并行)。
- 模块(Modules)= 任务模块(T-007-0 …);一个模块完成后 Claude 亲自验收再关闭。
- 依赖用关系 "blocked by",被阻塞的项留在 Backlog,前置完成后再转 Todo。
- 视图:`等我决定`(assignee=kappiee 或 Waiting)、`Codex 执行中`(In Progress / Todo + window10)、`待验收`(Review)。
## Codex 侧(worker 自动化)
worker 跑在 Claude 所在的 Linux 机(`gwent-design/tools/board-worker-plane.py`,cron 每 2 分钟一次,token 只由 Vault 注入内存)。
1. 取 state=Todo 且 assignee=`window10` 的工作项,按 `GW-k` 从小到大取一个;win10 上已有 codex 进程则本轮跳过(避免抢游戏/ADB)。
2. 名称 + 描述 + 评论转成文本拼成 prompt 跑 `codex exec`;同一工作项第二次起 `codex exec resume <thread>` 只追加新评论。
3. 结束后把最后一条消息贴回评论,按结果标签改状态与 assignee(见状态机)。
## Claude 侧
- 工具:`gwent-design/tools/plane-claude.sh`(`show` / `comment` / `state` / `assign` / `new` / `done`)。
- 轮询 Waiting 与 Review(assignee=`oregon`):回答问题、验收、改任务书或升级给 kappiee。新任务一律建工作项,不再用临时 brief 文件(紧急修复除外)。
## 通用规则
- 消息要短、要有结论;路径用仓库相对路径;大图只贴缩略拼图(≤1600px)。
- 不在评论里贴 token、密码、完整日志。
- 有疑问优先按任务书里的默认值继续并在评论里说明,而不是停下等人。
+1
View File
@@ -28,6 +28,7 @@
"additionalProperties": false, "additionalProperties": false,
"properties": { "properties": {
"state_dir": {"type": "string"}, "state_dir": {"type": "string"},
"log_dir": {"type": "string"},
"helpers_dir": {"type": "string"}, "helpers_dir": {"type": "string"},
"interval": {"type": "integer", "minimum": 10}, "interval": {"type": "integer", "minimum": 10},
"max_rounds_per_thread": {"type": "integer", "minimum": 1}, "max_rounds_per_thread": {"type": "integer", "minimum": 1},
+11
View File
@@ -0,0 +1,11 @@
#!/usr/bin/env bash
# LXC agentd-01 cron entry (*/2, root): one agentd tick under server-access exec (Plane tokens only in the
# process env), with a lock. The outer flock skips a tick while the previous one still runs; agentd also holds
# its own state lock. Logs: /var/log/plane-agentd/{agentd.log,cron.log}; state: /var/lib/plane-agentd.
exec flock -n /var/lib/plane-agentd/cron.lock \
timeout 900 python3 -I /root/.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 /opt/plane-agentd/cli/agentd --config /opt/plane-agentd/config/agentd.yaml \
tick --once >> /var/log/plane-agentd/cron.log 2>&1
+6 -3
View File
@@ -1,9 +1,12 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# cron entry (*/2): one agentd tick under server-access exec (Plane tokens in the process env only), with a lock. # oregon cron entry (*/2) -- ROLLBACK ONLY since PAD-11: the dispatcher runs on the LXC agentd-01
# The outer flock skips a tick while the previous one still runs; agentd also holds its own state lock. # (cron/agentd-tick-lxc.sh). Never run both: two dispatchers double-dispatch. See docs/runbook.md "回滚到 oregon".
# One agentd tick under server-access exec (Plane tokens in the process env only), with a lock.
exec flock -n /home/ubuntu/.gwent-board/agentd.lock \ exec flock -n /home/ubuntu/.gwent-board/agentd.lock \
timeout 900 python3 -I /home/ubuntu/.local/share/server-access-client/agent-access.py \ timeout 900 python3 -I /home/ubuntu/.local/share/server-access-client/agent-access.py \
--config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \ --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \
--secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN \ --secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN \
--secret-env PLANE_WINDOW10_API_TOKEN=PLANE_WINDOW10_API_TOKEN --timeout 850 -- \ --secret-env PLANE_WINDOW10_API_TOKEN=PLANE_WINDOW10_API_TOKEN --timeout 850 -- \
python3 /home/ubuntu/Documents/wjj/project/plane-agentd/cli/agentd tick --once >> /home/ubuntu/.gwent-board/agentd-cron.log 2>&1 python3 /home/ubuntu/Documents/wjj/project/plane-agentd/cli/agentd \
--config /home/ubuntu/Documents/wjj/project/plane-agentd/config/agentd.oregon.yaml \
tick --once >> /home/ubuntu/.gwent-board/agentd-cron.log 2>&1
+27 -2
View File
@@ -6,6 +6,7 @@ from agentd.config import Config
from agentd.core import Agentd from agentd.core import Agentd
from agentd.executors import ExecResult from agentd.executors import ExecResult
from agentd.store import Store from agentd.store import Store
from agentd.transports import TransportError
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"}
@@ -89,15 +90,39 @@ class FakeExecutor:
self.resume = resume self.resume = resume
self.started, self.killed, self.done = [], [], {} self.started, self.killed, self.done = [], [], {}
self.fail_starts, self.fail_kills = 0, 0 self.fail_starts, self.fail_kills = 0, 0
self.lost_starts = 0 # the run launches but the relay reports a failure (e.g. timeout)
self.fail_probes = self.fail_polls = 0
self.dirs = {} # run_dir -> files on the "terminal"
self.polls = []
def handle(self, run):
return {"run_dir": "/runs/" + (run.dir_name or run.run_id), "run_id": run.run_id, "thread_id": run.thread_id}
def start(self, run): def start(self, run):
if self.fail_starts: if self.fail_starts:
self.fail_starts -= 1 self.fail_starts -= 1
raise RuntimeError("relay down") raise TransportError("relay down")
h = self.handle(run)
if "pid.txt" in self.dirs.get(h["run_dir"], ()):
raise TransportError("HASPID")
self.started.append(run) self.started.append(run)
return {"run_dir": "/runs/" + run.run_id, "run_id": run.run_id, "thread_id": run.thread_id} self.dirs[h["run_dir"]] = {"launch.pid", "pid.txt"}
if self.lost_starts:
self.lost_starts -= 1
raise TransportError("relay timeout after 150s")
return h
def probe(self, handle):
if self.fail_probes:
self.fail_probes -= 1
raise TransportError("relay down")
return sorted(self.dirs.get(handle["run_dir"], ()))
def poll(self, handle): def poll(self, handle):
self.polls.append(handle["run_id"])
if self.fail_polls:
self.fail_polls -= 1
raise TransportError("relay down")
return "finished" if handle["run_id"] in self.done else "running" return "finished" if handle["run_id"] in self.done else "running"
def collect(self, handle): def collect(self, handle):
+117 -3
View File
@@ -1,3 +1,4 @@
import json
import unittest import unittest
from tests.fakes import Harness, config_data from tests.fakes import Harness, config_data
@@ -337,10 +338,13 @@ class CoreTest(unittest.TestCase):
saves = [] saves = []
orig = self.h.d.store.save orig = self.h.d.store.save
self.h.d.store.save = lambda data: (saves.append(sorted( self.h.d.store.save = lambda data: (saves.append(sorted(
r["inflight"]["run_id"] for r in data["projects"]["GWENT"]["issues"].values() if r.get("inflight"))), orig(data)) ("L" if r.get("launching") else "") + (r.get("inflight") or r.get("launching"))["run_id"]
for r in data["projects"]["GWENT"]["issues"].values() if r.get("inflight") or r.get("launching"))), orig(data))
self.h.tick() self.h.tick()
# one save per launched run (a crash mid-tick cannot orphan it), then the end-of-tick save # the launching marker is on disk before each launch call, the run right after it (a crash mid-tick cannot
self.assertEqual(saves, [["GW-1-1"], ["GW-1-1", "GW-2-1"], ["GW-1-1", "GW-2-1"]]) # orphan it or start it twice), then the end-of-tick save
self.assertEqual(saves, [["LGW-1-1"], ["GW-1-1"], ["GW-1-1", "LGW-2-1"], ["GW-1-1", "GW-2-1"],
["GW-1-1", "GW-2-1"]])
class EnvFailureTest(unittest.TestCase): class EnvFailureTest(unittest.TestCase):
def test_env_failure_pauses_group_until_item_leaves_waiting(self): def test_env_failure_pauses_group_until_item_leaves_waiting(self):
@@ -391,3 +395,113 @@ class TierDefaultTest(unittest.TestCase):
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
class DoubleStartGuardTest(unittest.TestCase):
"""A launch whose outcome is unknown is probed on the next tick: adopted if it exists, relaunched in a fresh
directory if not; never started twice in the same directory."""
def setUp(self):
self.h = Harness()
self.p = self.h.plane
def tearDown(self):
self.h.close()
def codex(self):
return self.h.execs[("win10", "codex")]
def test_lost_launch_is_adopted_not_restarted(self):
iid = self.p.add_issue(1)
self.h.d.executor("win10", "codex").lost_starts = 1
self.h.tick() # run starts, but the relay reports a timeout
rec = self.h.rec(iid)
self.assertNotIn("inflight", rec)
self.assertEqual((rec["launching"]["run_id"], rec["launching"]["handle"]["run_dir"]), ("GW-1-1", "/runs/GW-1-1"))
self.assertEqual(self.p.state_of(iid), "Todo")
self.h.tick() # probe finds launch.pid/pid.txt -> adopt
self.assertEqual(len(self.codex().started), 1)
rec = self.h.rec(iid)
self.assertNotIn("launching", rec)
self.assertEqual((rec["inflight"]["run_id"], rec["runs"]), ("GW-1-1", 1))
self.assertEqual(self.p.state_of(iid), "In Progress")
self.assertIn("启动未确认、已接管", self.p.texts(iid)[-1])
self.codex().finish("[review] ok")
self.h.tick()
self.assertEqual(self.p.state_of(iid), "Review")
def test_failed_launch_relaunches_same_run_id_in_fresh_dir(self):
iid = self.p.add_issue(1)
self.codex_ex = self.h.d.executor("win10", "codex")
self.codex_ex.fail_starts = 1
self.h.tick() # nothing launched
self.assertIn("launching", self.h.rec(iid))
self.h.tick() # probe: nothing -> relaunch as GW-1-1 in GW-1-1-r2
run = self.codex().started[-1]
self.assertEqual((run.run_id, run.dir_name), ("GW-1-1", "GW-1-1-r2"))
rec = self.h.rec(iid)
self.assertEqual(rec["inflight"]["handle"]["run_dir"], "/runs/GW-1-1-r2")
self.assertNotIn("relaunch", rec)
self.assertIn("重新启动(目录 GW-1-1-r2)", self.p.texts(iid)[-1])
def test_unprobed_marker_holds_the_terminal_slot(self):
a = self.p.add_issue(1)
b = self.p.add_issue(2)
ex = self.h.d.executor("win10", "codex")
ex.lost_starts = 1
self.h.tick()
self.assertEqual([r.run_id for r in ex.started], ["GW-1-1"]) # GW-2 not started next to a maybe-live run
ex.fail_probes = 1
self.h.tick() # relay down: marker kept, win10 skipped, GW-2 still waits
self.assertIn("launching", self.h.rec(a))
self.assertEqual(len(ex.started), 1)
self.assertEqual(self.h.d.store.load()["projects"]["GWENT"]["issues"].get(b), None)
self.h.tick() # relay back: adopt GW-1-1; concurrency 1 keeps GW-2 waiting
self.assertEqual(self.h.rec(a)["inflight"]["run_id"], "GW-1-1")
self.assertEqual(len(ex.started), 1)
def test_marker_dropped_when_item_moved_away(self):
iid = self.p.add_issue(1)
self.h.d.executor("win10", "codex").fail_starts = 1
self.h.tick()
self.p.issues[iid]["state"] = "s-bl" # a person parks it
self.h.tick()
rec = self.h.rec(iid)
self.assertNotIn("launching", rec)
self.assertNotIn("relaunch", rec)
self.assertEqual(self.codex().started, [])
def test_never_restarts_into_a_dir_with_a_pid_file(self):
iid = self.p.add_issue(1)
ex = self.h.d.executor("win10", "codex")
ex.dirs["/runs/GW-1-1"] = {"pid.txt"} # leftover of a launch the state does not know about
self.h.tick() # HASPID -> marker
self.assertIn("launching", self.h.rec(iid))
self.h.tick() # probe finds pid.txt -> adopt instead of wiping
self.assertEqual(self.h.rec(iid)["inflight"]["handle"]["run_dir"], "/runs/GW-1-1")
self.assertEqual(ex.started, [])
class RelayDownTest(unittest.TestCase):
def test_down_terminal_is_skipped_and_others_continue(self):
data = config_data()
data["terminals"]["win10"]["concurrency"] = 2
h = Harness(data)
w1, w2 = h.plane.add_issue(1), h.plane.add_issue(2)
o = h.plane.add_issue(3, assignee="oregon")
h.tick()
win, ore = h.execs[("win10", "codex")], h.execs[("oregon", "claude-code")]
self.assertEqual(len(win.started), 2)
ore.finish("[review] done")
w3 = h.plane.add_issue(4)
win.fail_polls, win.polls[:] = 1, []
h.tick()
self.assertEqual(len(win.polls), 1) # second win10 run not polled once the relay failed
self.assertEqual(h.plane.state_of(o), "Review") # oregon carried on
self.assertEqual(h.plane.state_of(w3), "Todo") # nothing new started on win10
hb = json.loads((h.d.store.dir / "heartbeat.json").read_text())
self.assertEqual((hb["down"], hb["errors"]), (["win10"], []))
self.assertTrue(all("inflight" in h.rec(i) for i in (w1, w2)))
win.finish("[review] w")
h.tick() # relay back: polled again
self.assertIn("win10", {r["terminal"] for r in json.loads((h.d.store.dir / "heartbeat.json").read_text())["inflight"]})
h.close()
+14
View File
@@ -203,3 +203,17 @@ class PosixRemoteTest(unittest.TestCase):
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
def test_probe_dir_name_and_pid_guard(self):
ex = make_executor(self.cfg, "mac", "shell", self.t)
run = Run("GW-5-1", self.cfg.terminals["mac"]["repo"], "", script="sleep 0.2", dir_name="GW-5-1-r2")
self.assertEqual(ex.probe(ex.handle(run)), [])
h = ex.start(run)
self.assertEqual((os.path.basename(h["run_dir"]), h["task"]), ("GW-5-1-r2", "agentd-GW-5-1-r2"))
self.assertTrue(wait(ex, h))
self.assertEqual(ex.probe(h), ["exit.txt", "launch.pid", "pid.txt"])
n = len(self.t.calls)
with self.assertRaises(AssertionError): # LocalBash: marker LAUNCHED= missing (HASPID)
ex.start(run)
self.assertIn("HASPID", self.t.calls[n])
self.assertTrue(os.path.exists(os.path.join(h["run_dir"], "exit.txt"))) # nothing wiped
+27
View File
@@ -120,3 +120,30 @@ class StoreTest(unittest.TestCase):
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
class GuardTest(unittest.TestCase):
def test_local_clean_refuses_dir_with_pid_file(self):
from agentd.transports import TransportError
with tempfile.TemporaryDirectory() as tmp:
t = LocalTransport(Config(config_data()), "oregon")
(Path(tmp) / "prompt.md").write_text("x")
t.mkdir(tmp, clean=True)
self.assertEqual(t.probe(tmp, ("exit.txt", "pid.txt")), [])
(Path(tmp) / "pid.txt").write_text("1")
with self.assertRaises(TransportError):
t.mkdir(tmp, clean=True)
self.assertEqual(t.probe(tmp, ("exit.txt", "pid.txt")), ["pid.txt"])
def test_windows_probe_and_clean_guard(self):
from agentd.transports import TransportError
cfg = config.load(ROOT / "config" / "agentd.example.yaml")
t = WindowsWarpgateTransport(cfg, "win10")
scripts = []
t.ps = lambda s, timeout=150: (scripts.append(s), "PROBE=launch.pid,pid.txt\n")[1]
self.assertEqual(t.probe("C:\\r\\GW-1-1", ("exit.txt", "launch.pid", "pid.txt")), ["launch.pid", "pid.txt"])
self.assertIn("Join-Path $d $_", scripts[-1])
t.ps = lambda s, timeout=150: (scripts.append(s), "HASPID\n")[1]
with self.assertRaises(TransportError):
t.mkdir("C:\\r\\GW-1-1", clean=True)
self.assertIn("'pid.txt'", scripts[-1])
+18
View File
@@ -0,0 +1,18 @@
#!/usr/bin/env bash
# Run a PowerShell script (passed as $1 text, or a file with -f) on win10 through server-access,
# encoded as -EncodedCommand so quoting never crosses the relay. Output is UTF-8; progress noise off.
# AGENT_ACCESS_HOME = directory of the vault client (default ~/.local/share/server-access-client);
# AGENT_ACCESS_CONFIG = identity URL/file.
set -euo pipefail
CLIENT="${AGENT_ACCESS_HOME:-$HOME/.local/share/server-access-client}/agent-access.py"
CFG="${AGENT_ACCESS_CONFIG:-https://nas.tail9531b.ts.net:8443/identity/agent.json}"
TIMEOUT="${WIN10_TIMEOUT:-120}"
if [ "${1:-}" = "-f" ]; then SCRIPT="$(cat "$2")"; else SCRIPT="$1"; fi
SCRIPT="[Console]::OutputEncoding=[Text.Encoding]::UTF8
\$ErrorActionPreference='Continue'
\$ProgressPreference='SilentlyContinue'
$SCRIPT"
B64="$(printf '%s' "$SCRIPT" | iconv -f UTF-8 -t UTF-16LE | base64 -w0)"
if [ "${#B64}" -gt 7600 ]; then echo "encoded command too long (${#B64} > 7600)" >&2; exit 2; fi
exec python3 -I "$CLIENT" --config "$CFG" run --target win10 --timeout "$TIMEOUT" -- \
powershell.exe -NoProfile -NonInteractive -EncodedCommand "$B64" 2> >(grep -v -E '^#< CLIXML|^<Objs|正在准备首次使用模块' >&2)
+26
View File
@@ -0,0 +1,26 @@
#!/usr/bin/env bash
# Upload a local text/binary file to win10: gzip+base64, appended in chunks through the relay,
# then decoded remotely and verified by SHA-256.
# usage: win10-put.sh <local-file> <remote-windows-path>
set -euo pipefail
HERE="$(cd "$(dirname "$0")" && pwd)"
LOCAL="$1"; REMOTE="$2"
TMP="$REMOTE.b64"
SHA="$(sha256sum "$LOCAL" | cut -d' ' -f1)"
B64="$(gzip -9c "$LOCAL" | base64 -w0)"
CHUNK=2400
"$HERE/win10-ps.sh" "New-Item -ItemType Directory -Force -Path (Split-Path -Parent '$REMOTE') | Out-Null; Set-Content -Path '$TMP' -Value '' -NoNewline -Encoding Ascii; 'reset'" >/dev/null
TOTAL=$(( (${#B64} + CHUNK - 1) / CHUNK ))
for ((i=0; i<${#B64}; i+=CHUNK)); do
PART="${B64:i:CHUNK}"
n=$(( i / CHUNK + 1 ))
"$HERE/win10-ps.sh" "Add-Content -Path '$TMP' -Value '$PART' -NoNewline -Encoding Ascii; 'chunk $n/$TOTAL'" | tail -1
done
"$HERE/win10-ps.sh" "\$b=[Convert]::FromBase64String((Get-Content -Raw -Encoding Ascii '$TMP'))
\$ms=New-Object IO.MemoryStream(,\$b)
\$gz=New-Object IO.Compression.GZipStream(\$ms,[IO.Compression.CompressionMode]::Decompress)
\$out=New-Object IO.MemoryStream; \$gz.CopyTo(\$out)
[IO.File]::WriteAllBytes('$REMOTE',\$out.ToArray())
Remove-Item '$TMP'
'remote sha256 ' + (Get-FileHash '$REMOTE' -Algorithm SHA256).Hash.ToLower()
'expected $SHA'"