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.

agentd/cli.py

@@ -121,14 +121,16 @@ def cmd_smoke(cfg, a):
         body["labels"] = [labels[x] for x in a.labels.split(",")]
     issue = pl.create_issue(p["id"], body)
     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))
-    d = Agentd(cfg)
+    print("created %s; %s every %ss (timeout %s min)" % (
+        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"])
     done_states = {states[p["states"][k]] for k in ("review", "waiting")}
     deadline = time.time() + a.timeout * 60
     state = None
     while time.time() < deadline:
         try:
+            if d:
                 d.tick(project=a.project, only_issue=issue["id"])
         except Busy as exc:
             print("skip: %s" % exc)
@@ -199,6 +201,8 @@ def parser():
     s.add_argument("--interval", type=int, default=30)
     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("--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.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)

agentd/config.py

@@ -155,8 +155,18 @@ class Config:
         return p if p.is_absolute() else ROOT / p
 
     def helpers_dir(self, terminal):
-        d = self.terminals[terminal].get("helpers_dir") or self.agentd.get("helpers_dir") or str(ROOT / "tools")
-        return Path(os.path.expanduser(d))
+        d = self.terminals[terminal].get("helpers_dir") or self.agentd.get("helpers_dir") or "tools/win10"
+        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):
         """Environment variable that carries the terminal's Plane token (== the Vault key name)."""

agentd/core.py

@@ -3,7 +3,13 @@
       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;
   (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 re
 import time
@@ -15,7 +21,7 @@ from .executors import ExecResult, Run, make_executor
 from .executors.base import fill
 from .plane import Plane, html_to_text, text_to_html
 from .store import Store
-from .transports import make_transport
+from .transports import TransportError, make_transport
 
 TAGS = ("review", "question", "blocked", "progress")
 WORKER = "[worker]"
@@ -143,7 +149,7 @@ class Ctx:
 class Agentd:
     def __init__(self, cfg, store=None, plane_factory=None, executor_factory=None, transport_factory=None, now=time.time):
         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.log = self.store.log
         self.plane_factory = plane_factory or (lambda term: Plane(
@@ -154,6 +160,12 @@ class Agentd:
 
     def _reset_caches(self):
         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
     def client(self, terminal):
@@ -195,9 +207,11 @@ class Agentd:
                     self.log("%s tick error: %r" % (name, exc))
                 finally:
                     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"]}
-                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
 
     def loop(self, interval):
@@ -216,14 +230,43 @@ class Agentd:
         pstate = data["projects"].setdefault(name, {"issues": {}})
         ctx = Ctx(name, pcfg, self.reader(pcfg))
         for iid, rec in list(pstate["issues"].items()):
-            if rec.get("inflight"):
+            run = rec.get("inflight") or rec.get("launching")
+            if not run or run["terminal"] in self._down:
+                continue
             try:
+                if rec.get("inflight"):
                     self.poll(ctx, data, iid, rec)
+                else:
+                    self.resolve_launch(ctx, iid, rec)
+            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)
 
+    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
     def poll(self, ctx, data, iid, rec):
         fl = rec["inflight"]
@@ -404,7 +447,7 @@ class Agentd:
         load, groups = Counter(), set()
         for p in data["projects"].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:
                     load[fl["terminal"]] += 1
                     if fl.get("group"):
@@ -422,7 +465,7 @@ class Agentd:
             if only_issue and iid != only_issue:
                 continue
             rec = pstate["issues"].get(iid, {})
-            if rec.get("inflight"):
+            if rec.get("inflight") or rec.get("launching"):
                 continue
             if issue.get("state") == todo:
                 cands.append(issue)
@@ -438,15 +481,21 @@ class Agentd:
                     self.log("%s not dispatched: %s" % (ctx.key(issue.get("sequence_id")), route))
                 continue
             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):
                 continue
             if not self.terminal_ready(term, load[term]):
                 continue
             try:
-                self.start(ctx, pstate, issue, route)
+                self.start(ctx, data, pstate, issue, route)
             except Exception as exc:
                 self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
+                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
             load[term] += 1
             if group:
@@ -454,14 +503,17 @@ class Agentd:
 
     def terminal_ready(self, term, our_load):
         """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:
             names = self.cfg.terminals[term].get("busy_processes", [])
             ready = True
             for name in names:
                 count = self.transport(term).process_count(name)
                 if count is None:
-                    self.log("%s unreachable; skipping" % term)
+                    self.mark_down(term, "busy-process probe failed")
                     ready = False
+                    break
                 elif count > 0 and our_load == 0:
                     self.log("%s busy (%d %s running that agentd did not start)" % (term, count, name))
                     ready = False
@@ -522,7 +574,7 @@ class Agentd:
             return ""
         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"]
         key = ctx.key(seq)
         rec = pstate["issues"].setdefault(iid, new_rec(seq))
@@ -552,25 +604,11 @@ class Agentd:
         intro = fill(ctx.cfg.get("intro", DEFAULT_INTRO),
                      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)
-        run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"])
-        handle = ex.start(run)
-        rec.pop("retry", None)
-        rec.pop("rebuild", None)
-        if not is_retry:
-            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)
+        relaunch = rec.get("relaunch") or {}
+        attempt = relaunch.get("attempt", 1) if relaunch.get("run_id") == run_id else 1
+        dir_name = run_id if attempt == 1 else "%s-r%d" % (run_id, attempt)
+        run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"],
+                  dir_name=dir_name)
         if thread:
             mode = "续线程 %s" % thread[:12]
         elif reason:
@@ -579,10 +617,53 @@ class Agentd:
             mode = "新线程（附交接摘要）" if handoff else "新线程"
         if is_retry:
             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:
-            self.move(ctx, client, iid, "running", tcfg["plane_user"])
-            self.post(ctx, client, iid, rec, "%s 第 %d 轮开始：%s @ %s，%s，run=%s。" % (WORKER, k, detail, term, mode, run_id))
+            # an adopted run whose item a person moved meanwhile is left alone: poll cancels it next tick
+            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:
             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))

agentd/executors/base.py

@@ -17,6 +17,7 @@ class Run:
     tier: Optional[str] = None
     thread_id: Optional[str] = None   # set => resume this thread
     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
@@ -35,9 +36,15 @@ class Executor:
     def __init__(self, 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)
         raise NotImplementedError
 
+    def probe(self, handle):         # -> names of launch evidence present on the terminal ([] = never launched)
+        raise NotImplementedError
+
     def poll(self, handle):          # -> "running" | "finished"
         raise NotImplementedError
 
@@ -84,11 +91,15 @@ class RunnerExecutor(Executor):
             program = "& " + program
         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):
         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)
-        task = "agentd-" + run.run_id
         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)
@@ -97,7 +108,13 @@ class RunnerExecutor(Executor):
             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 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):
         text = self.transport.read_text(self.transport.join(handle["run_dir"], "exit.txt"))

agentd/store.py

@@ -17,8 +17,9 @@ class Busy(Exception):
 
 
 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.log_dir = Path(log_dir) if log_dir else self.dir
         self.echo = echo
         self.path = self.dir / "agentd-state.json"
 
@@ -59,9 +60,9 @@ class Store:
         self._atomic_write(self.dir / "heartbeat.json", info)
 
     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
         if self.echo:
             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")

agentd/transports/local.py

@@ -4,6 +4,10 @@ import signal
 import subprocess
 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.
 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; }}
@@ -40,6 +44,8 @@ class LocalTransport:
         p = Path(path)
         p.mkdir(parents=True, exist_ok=True)
         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():
                 if f.is_file():
                     f.unlink()
@@ -51,6 +57,9 @@ class LocalTransport:
         p = Path(path)
         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):
         text = self.read_text(path) or ""
         return next((line for line in text.splitlines() if needle in line), None)

agentd/transports/warpgate_ssh.py

@@ -113,9 +113,21 @@ class WindowsWarpgateTransport:
     def mkdir(self, path, clean=False):
         q = self.quote(path)
         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
-        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):
         fd, tmp = tempfile.mkstemp(prefix="agentd-", suffix=".txt")
@@ -230,7 +242,9 @@ class PosixWarpgateTransport:
     def mkdir_script(self, path, clean=False):
         q = self.quote(path)
         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
         return s
 
@@ -275,6 +289,12 @@ class PosixWarpgateTransport:
             raise TransportError("read %s: unexpected relay output" % path)
         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):
         out = self.sh("printf B64=; { %s; } 2>/dev/null | base64 | tr -d '\\n'; echo" % pipeline)
         text = _b64_payload(out, "B64")

config/agentd.example.yaml

@@ -7,7 +7,7 @@ vault: {client: ~/.local/share/server-access-client/agent-access.py, config: htt
 
 agentd:
   state_dir: state                      # relative to the repo root; JSON state, logs, heartbeat, local run dirs
-  helpers_dir: /home/ubuntu/Documents/wjj/project/gwent-design/tools   # win10-ps.sh / win10-put.sh (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
   max_rounds_per_thread: 8              # rebuild the executor thread (with a handoff summary) after this many rounds
   comment_max_chars: 6000

config/agentd.oregon.yaml

@@ -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

config/agentd.yaml

@@ -1,11 +1,13 @@
-# plane-agentd live configuration on oregon (phase 1). No secrets: token_key values are Vault key names; the tokens
-# are injected by `agent-access.py exec --secret-env KEY=KEY` (see cron/agentd-tick.sh and docs/runbook.md).
+# plane-agentd live configuration on the LXC agentd-01 (/opt/plane-agentd, runs as root; docs/deploy-lxc.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}
 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: /home/ubuntu/Documents/wjj/project/gwent-design/tools   # win10-ps.sh / win10-put.sh (Windows terminals only)
+  state_dir: /var/lib/plane-agentd     # JSON state, heartbeat.json, lock
+  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
   max_rounds_per_thread: 8
   comment_max_chars: 6000
@@ -27,9 +29,7 @@ terminals:
     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.
+    # the dispatcher (LXC, no agent CLIs) reaches oregon through Warpgate like any other host (design §9)
     transport: warpgate-ssh
     target: oregon                      # agent-access.py target name
     os: linux
@@ -70,7 +70,7 @@ projects:
     states: {backlog: Backlog, todo: Todo, running: In Progress, waiting: Waiting, review: Review, done: Done}
     reviewer: 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:
       by_assignee: {window10: win10, oregon: oregon}
       by_label:

config/protocols/gwent-board-protocol.md

@@ -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、密码、完整日志。
+- 有疑问优先按任务书里的默认值继续并在评论里说明，而不是停下等人。

config/schema.json

@@ -28,6 +28,7 @@
       "additionalProperties": false,
       "properties": {
         "state_dir": {"type": "string"},
+        "log_dir": {"type": "string"},
         "helpers_dir": {"type": "string"},
         "interval": {"type": "integer", "minimum": 10},
         "max_rounds_per_thread": {"type": "integer", "minimum": 1},

cron/agentd-tick-lxc.sh

@@ -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

cron/agentd-tick.sh

@@ -1,9 +1,12 @@
 #!/usr/bin/env bash
-# cron entry (*/2): one agentd tick under server-access exec (Plane tokens in the process env only), with a lock.
-# The outer flock skips a tick while the previous one still runs; agentd also holds its own state lock.
+# oregon cron entry (*/2) -- ROLLBACK ONLY since PAD-11: the dispatcher runs on the LXC agentd-01
+# (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 \
   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 \
   --secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN \
   --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

tests/fakes.py

@@ -6,6 +6,7 @@ from agentd.config import Config
 from agentd.core import Agentd
 from agentd.executors import ExecResult
 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"}
 MEMBERS = {"window10": "u-w10", "oregon": "u-or", "kappiee": "u-k"}
@@ -89,15 +90,39 @@ class FakeExecutor:
         self.resume = resume
         self.started, self.killed, self.done = [], [], {}
         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):
         if self.fail_starts:
             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)
-        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):
+        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"
 
     def collect(self, handle):

tests/test_core.py

@@ -1,3 +1,4 @@
+import json
 import unittest
 
 from tests.fakes import Harness, config_data
@@ -337,10 +338,13 @@ class CoreTest(unittest.TestCase):
         saves = []
         orig = self.h.d.store.save
         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()
-        # one save per launched run (a crash mid-tick cannot orphan it), then the end-of-tick save
-        self.assertEqual(saves, [["GW-1-1"], ["GW-1-1", "GW-2-1"], ["GW-1-1", "GW-2-1"]])
+        # the launching marker is on disk before each launch call, the run right after it (a crash mid-tick cannot
+        # 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):
     def test_env_failure_pauses_group_until_item_leaves_waiting(self):
@@ -391,3 +395,113 @@ class TierDefaultTest(unittest.TestCase):
 
 if __name__ == "__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()

tests/test_posix_remote.py

@@ -203,3 +203,17 @@ class PosixRemoteTest(unittest.TestCase):
 
 if __name__ == "__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

tests/test_transports.py

@@ -120,3 +120,30 @@ class StoreTest(unittest.TestCase):
 
 if __name__ == "__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])

tools/win10/win10-ps.sh

@@ -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)

tools/win10/win10-put.sh

@@ -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'"