Docs: binary resolution, cancellation, progress cap, phase-1 deviations

- runbook: how executor binaries are resolved (bins or PATH) and what that
  assumes on oregon (no codex installed) and win10; cancellation and
  max_progress_rounds in the item lifecycle; secrets scrubbed from children.
- design: blocked-by ordering is a TODO (not read by the scheduler); failure
  classification, success-when-tagged, one-time immediate rebuild, retries
  outside the thread-round budget, ID-based new-comment filter.
- executor contract: how core reads ExecResult, kill used for cancel, Windows
  quoting assumptions, codex exec resume without -C/-s.

agentd/config.py

@@ -157,6 +157,17 @@ class Config:
     def token(self, terminal):
         return os.environ.get(self.token_env(terminal))
 
+    def secret_keys(self):
+        """Every env var name that carries a secret (Plane tokens, webhook secrets): never handed to executors."""
+        keys = {t["token_key"] for t in self.terminals.values()}
+        keys |= {p["webhook_secret_key"] for p in self.projects.values() if p.get("webhook_secret_key")}
+        return keys
+
+    def scrubbed_env(self):
+        """os.environ without the secrets above, for every child process agentd starts."""
+        drop = self.secret_keys()
+        return {k: v for k, v in os.environ.items() if k not in drop}
+
     def secret_env_args(self):
         """`--secret-env` arguments for agent-access.py exec, one per distinct token key."""
         keys = sorted({t["token_key"] for t in self.terminals.values()})

agentd/core.py

@@ -34,11 +34,19 @@ OUTPUT_RULE = "输出要求：最后一条消息首行为 [review]/[question]/[b
 
 
 # ---------------------------------------------------------------- pure helpers (unit-tested)
+FENCE = re.compile(r"^(```|~~~|-{3,}$|\*{3,}$)")
+
+
 def classify_first_line(text):
-    """Tag of the first non-empty line (`[review]` etc.), tolerating markdown emphasis; None if untagged."""
-    first = next((l.strip() for l in (text or "").splitlines() if l.strip()), "")
-    m = re.match(r"\[(\w+)\]", first.lstrip("*_#> `").lower())
-    return m.group(1) if m and m.group(1) in TAGS else None
+    """Tag of the first meaningful line (`[review]` etc.), tolerating a BOM, markdown emphasis and a leading code
+    fence / rule line; None if untagged. Trailing text after the tag is fine (`[progress] half way`)."""
+    for line in (text or "").splitlines():
+        first = line.strip().lstrip("\ufeff").strip()
+        if not first or FENCE.match(first):
+            continue
+        m = re.match(r"\[(\w+)\]", first.lstrip("*_#> `").lower())
+        return m.group(1) if m and m.group(1) in TAGS else None
+    return None
 
 
 def matches_any(text, patterns):
@@ -46,14 +54,18 @@ def matches_any(text, patterns):
 
 
 def classify_failure(result, timed_out, retry_cfg):
-    """-> (kind, transient). kind in timeout | env | crash | logic (design §4)."""
+    """-> (kind, transient). kind in timeout | env | crash | logic (design §4).
+    env patterns look at the tail and the last message; transient patterns only at the tail (stderr / last event),
+    so an agent's own prose ("tests timed out") never triggers a retry. logic (exit 0, no message) is never
+    transient: an exit-0 run whose tail shows a transient error is a crash."""
     if timed_out:
         return "timeout", False
-    text = "%s\n%s" % (result.tail or "", result.last_message or "")
-    if matches_any(text, retry_cfg.get("env_patterns", [])):
+    if matches_any("%s\n%s" % (result.tail or "", result.last_message or ""), retry_cfg.get("env_patterns", [])):
         return "env", False
-    kind = "logic" if result.exit_code == 0 else "crash"
-    return kind, matches_any(text, retry_cfg.get("transient_patterns", DEFAULT_TRANSIENT))
+    transient = matches_any(result.tail or "", retry_cfg.get("transient_patterns", DEFAULT_TRANSIENT))
+    if result.exit_code == 0 and not transient:
+        return "logic", False
+    return "crash", transient
 
 
 def parse_budget(label):
@@ -209,6 +221,7 @@ class Agentd:
                     self.poll(ctx, data, iid, rec)
                 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)
 
     # ---------------------------------------------------------------- (a) poll / finish
@@ -216,6 +229,9 @@ class Agentd:
         fl = rec["inflight"]
         ex = self.executor(fl["terminal"], fl["executor"])
         if ex.poll(fl["handle"]) == "running":
+            issue = ctx.issues.get(iid)
+            if issue is not None and issue.get("state") != ctx.state_id("running"):
+                return self.cancel(ctx, iid, rec, ex, issue)
             if self.now() < fl["deadline"]:
                 return
             self.log("%s run %s exceeded its budget; killing" % (ctx.key(rec["seq"]), fl["run_id"]))
@@ -232,6 +248,37 @@ class Agentd:
             ex.forget(fl["handle"])
         self.finish(ctx, data, iid, rec, result)
 
+    def cancel(self, ctx, iid, rec, ex, issue):
+        """The item left In Progress while its run was still going (moved by a person): kill the run, say so, free
+        the terminal slot and serialize group. The thread id is kept so a later round can resume it."""
+        fl = rec["inflight"]
+        key = ctx.key(rec["seq"])
+        try:
+            ex.kill(fl["handle"])
+        except Exception as exc:  # keep the slot until the kill succeeds; a live run must not share the terminal
+            self.log("%s cancel of run %s: kill failed, retrying next tick: %r" % (key, fl["run_id"], exc))
+            return
+        if hasattr(ex, "forget"):
+            self.guarded(key, ex.forget, fl["handle"])
+        thread = fl["handle"].get("thread_id")
+        if not thread and ex.resume:
+            try:  # a fresh run may already have announced its thread (codex thread.started)
+                thread = ex.collect(fl["handle"]).thread_id
+            except Exception:
+                thread = None
+        state = {v: k for k, v in ctx.states.items()}.get(issue.get("state"), issue.get("state"))
+        try:
+            self.post(ctx, self.client(fl["terminal"]), iid, rec,
+                      "%s 运行已取消：工作项状态被改为 %s（run=%s）。" % (WORKER, state, fl["run_id"]))
+        except Exception as exc:
+            self.log("%s cancel comment failed: %r" % (key, exc))
+        rec.pop("inflight")
+        rec.pop("retry", None)
+        if thread:
+            rec["thread_id"], rec["thread_on"] = thread, [fl["terminal"], fl["executor"]]
+        rec["last_tag"] = "cancelled"
+        self.log("%s run %s cancelled: item moved to %s" % (key, fl["run_id"], state))
+
     def finish(self, ctx, data, iid, rec, result, timed_out=False):
         """Post first, then forget the in-flight run: if Plane fails, the next tick collects again."""
         fl = rec["inflight"]
@@ -258,7 +305,16 @@ class Agentd:
         self.close_run(rec, result)
         rec["tags"].append(tag)
         rec.update(last_message=message[:4000], last_tag=tag, retries=0)
-        if still_ours:
+        rec["progress_streak"] = rec.get("progress_streak", 0) + 1 if tag == "progress" else 0
+        cap = ctx.cfg.get("max_progress_rounds", 12)
+        if still_ours and tag == "progress" and rec["progress_streak"] >= cap:
+            who = ctx.cfg.get("answerer") or ctx.cfg["reviewer"]
+            rec["progress_streak"] = 0
+            self.guarded(key, self.post, ctx, client, iid, rec, "%s 已连续 %d 轮 [progress]（上限 max_progress_rounds=%d），"
+                         "暂停自动续跑，转 %s 并指派给 %s；移回 %s 即继续。" % (
+                             WORKER, cap, cap, ctx.cfg["states"]["waiting"], who, ctx.cfg["states"]["todo"]))
+            self.guarded(key, self.move, ctx, client, iid, "waiting", who)
+        elif still_ours:
             self.guarded(key, self.transition, ctx, client, iid, tag, self.cfg.terminals[term]["plane_user"])
         else:
             self.log("%s was moved during the run; not changing its state" % key)
@@ -317,6 +373,7 @@ class Agentd:
             rec["retry"] = {"at": self.now() + delay * 60, "reason": kind}
         elif plan == "waiting":
             rec["retries"] = 0
+            rec["progress_streak"] = 0
             if kind == "env" and fl.get("group"):
                 data["paused_groups"][fl["group"]] = {"project": ctx.name, "issue": iid, "key": key}
             self.guarded(key, self.move, ctx, client, iid, "waiting", who)
@@ -390,6 +447,7 @@ class Agentd:
             except Exception as exc:
                 self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
                 continue
+            self.store.save(data)  # a launched run must be on disk before anything else can fail
             load[term] += 1
             if group:
                 groups.add(group)
@@ -465,18 +523,19 @@ class Agentd:
         term, exname = route["terminal"], route["executor"]
         tcfg, pcfg = self.cfg.terminals[term], ctx.cfg
         ex = self.executor(term, exname)
-        is_retry = rec.pop("retry", None) is not None and issue.get("state") == ctx.state_id("running")
-        if not is_retry:
-            rec["retries"] = 0
+        # retry / rebuild markers are only consumed once the run has launched: if start fails they must survive,
+        # otherwise an In Progress item without a run would never be picked again.
+        retry = rec.get("retry")
+        is_retry = bool(retry) and retry.get("reason") != "rebuild" and issue.get("state") == ctx.state_id("running")
         k = rec["runs"] + 1
         run_id = "%s-%d-%d" % (ctx.prefix, seq, k)
-        thread, reason = rec.get("thread_id"), rec.pop("rebuild", None)
+        thread, reason = rec.get("thread_id"), rec.get("rebuild")
         max_rounds = pcfg.get("max_rounds_per_thread") or self.cfg.agentd.get("max_rounds_per_thread", 8)
         if thread and not ex.resume:
             thread = None
         elif thread and rec.get("thread_on") != [term, exname]:
             thread, reason = None, "换了终端/执行器"
-        elif thread and rec.get("thread_rounds", 0) >= max_rounds:
+        elif thread and not is_retry and rec.get("thread_rounds", 0) >= max_rounds:  # a retry is the same round
             thread, reason = None, "线程已续 %d 轮（上限 %d）" % (rec["thread_rounds"], max_rounds)
         handoff = handoff_summary(rec, reason) if not thread and rec["runs"] > 0 else None
         retry_note = None
@@ -489,8 +548,14 @@ class Agentd:
         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()
-        rec.update(runs=k, thread_rounds=rec.get("thread_rounds", 0) + 1 if thread else 1, inflight={
+        # 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"]})
@@ -500,7 +565,12 @@ class Agentd:
         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)
-        mode = ("续线程 %s" % thread[:12]) if thread else ("新线程（重建：%s）" % reason if reason else "新线程")
+        if thread:
+            mode = "续线程 %s" % thread[:12]
+        elif reason:
+            mode = "新线程（重建：%s，附交接摘要）" % reason if handoff else "新线程（重建：%s）" % reason
+        else:
+            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)

agentd/store.py

@@ -3,7 +3,7 @@
 state/agentd-state.json:
   {"projects": {PROJECT: {"issues": {issue_uuid: rec}}}, "paused_groups": {group: {...}}}
 rec: seq, runs, thread_id, thread_on [terminal, executor], thread_rounds, tags [...], last_message,
-     seen_comments [...], last_comment_at, retries, retry {at, reason}, rebuild, inflight {...}"""
+     seen_comments [...], last_comment_at, retries, retry {at, reason}, rebuild, progress_streak, inflight {...}"""
 import contextlib
 import fcntl
 import json
@@ -31,11 +31,18 @@ class Store:
         data.setdefault("paused_groups", {})
         return data
 
-    def save(self, data):
+    def _atomic_write(self, path, obj):
+        """Write temp + fsync + rename, so a reader (or a crash) never sees half a file."""
         self.dir.mkdir(parents=True, exist_ok=True)
-        tmp = self.path.with_suffix(".tmp")
-        tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
-        os.replace(tmp, self.path)
+        tmp = path.with_suffix(".tmp")
+        with open(tmp, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(obj, ensure_ascii=False, indent=1))
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp, path)
+
+    def save(self, data):
+        self._atomic_write(self.path, data)
 
     @contextlib.contextmanager
     def lock(self):
@@ -48,9 +55,8 @@ class Store:
             yield
 
     def heartbeat(self, info):
-        self.dir.mkdir(parents=True, exist_ok=True)
         info = dict(info, ts=time.time(), at=time.strftime("%Y-%m-%dT%H:%M:%S"))
-        (self.dir / "heartbeat.json").write_text(json.dumps(info, ensure_ascii=False, indent=1), encoding="utf-8")
+        self._atomic_write(self.dir / "heartbeat.json", info)
 
     def log(self, msg):
         self.dir.mkdir(parents=True, exist_ok=True)

agentd/transports/local.py

@@ -59,7 +59,9 @@ class LocalTransport:
     def launch(self, task, script_path, run_dir):
         # `setsid -f` double-forks: the runner leads its own session (pgid == pid, see kill) and is reparented
         # to init, so a long-running `tick --loop` never accumulates zombies.
+        # The executor must not hold Plane credentials (design §8): drop every configured secret from its env.
         subprocess.run(["setsid", "-f", "bash", script_path], cwd=run_dir, check=True, close_fds=True,
+                       env=self.cfg.scrubbed_env(),
                        stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 
     def kill(self, task, run_dir):

agentd/transports/warpgate_ssh.py

@@ -15,7 +15,8 @@ from . import TransportError
 RUNNER = r"""$ErrorActionPreference = 'Continue'
 [Console]::OutputEncoding = [Text.Encoding]::UTF8
 $OutputEncoding = [Text.Encoding]::UTF8
-Set-Location {cwd}
+try {{ Set-Location -LiteralPath {cwd} -ErrorAction Stop }} catch {{
+  Set-Content -Path '{d}\exit.txt' -Value ('exit=97 at ' + (Get-Date).ToString('s') + ' (cd failed)'); exit 97 }}
 Set-Content -Path '{d}\pid.txt' -Value $PID
 Set-Content -Path '{d}\started.txt' -Value (Get-Date).ToString('s')
 $prompt = Get-Content -Raw -Encoding UTF8 '{d}\prompt.md'
@@ -53,7 +54,7 @@ class WarpgateSshTransport:
 
     # ---------------------------------------------------------------- relay primitives
     def ps(self, script, timeout=150):
-        env = dict(os.environ, WIN10_TIMEOUT=str(timeout))
+        env = dict(self.cfg.scrubbed_env(), WIN10_TIMEOUT=str(timeout))
         try:
             out = subprocess.run([self.ps_script, script], capture_output=True, text=True, timeout=timeout + 60, env=env)
         except subprocess.TimeoutExpired:
@@ -64,7 +65,8 @@ class WarpgateSshTransport:
 
     def put(self, local, remote):
         sha = hashlib.sha256(open(local, "rb").read()).hexdigest()
-        out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600)
+        out = subprocess.run([self.put_script, local, remote], capture_output=True, text=True, timeout=600,
+                             env=self.cfg.scrubbed_env())
         if ("remote sha256 " + sha) not in out.stdout:
             raise TransportError("upload of %s failed: %s" % (remote, (out.stdout + out.stderr).strip()[-300:]))
 
@@ -113,12 +115,18 @@ class WarpgateSshTransport:
     def runner(self, run_dir, cwd, command, stdout_name):
         return "run.ps1", RUNNER.format(cwd=self.quote(cwd), d=run_dir.replace("'", "''"), command=command, stdout=stdout_name)
 
+    def launch_script(self, task, script_path):
+        """PowerShell that (re)creates and runs the one-shot task. Values sit in '...' literals (`'` doubled). The
+        script path is wrapped in \\" so that Windows PowerShell 5.1 (legacy native quoting) hands schtasks
+        /TR `... -File "<path>"` intact even when the path contains spaces."""
+        return ("schtasks /Delete /F /TN '%(t)s' 2>$null | Out-Null\n"
+                "$tr = 'powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File \\\"%(s)s\\\"'\n"
+                "schtasks /Create /F /TN '%(t)s' /SC ONCE /ST 00:00 /TR $tr | Out-Null\n"
+                "schtasks /Run /TN '%(t)s' | Out-Null\n"
+                "'LAUNCHED=' + $LASTEXITCODE" % dict(t=task.replace("'", "''"), s=script_path.replace("'", "''")))
+
     def launch(self, task, script_path, run_dir):
-        out = self.ps("schtasks /Delete /F /TN '%(t)s' 2>$null | Out-Null\n"
-                      "$tr = 'powershell.exe -NoProfile -NonInteractive -ExecutionPolicy Bypass -WindowStyle Hidden -File \"%(s)s\"'\n"
-                      "schtasks /Create /F /TN '%(t)s' /SC ONCE /ST 00:00 /TR $tr | Out-Null\n"
-                      "schtasks /Run /TN '%(t)s' | Out-Null\n"
-                      "'LAUNCHED=' + $LASTEXITCODE" % dict(t=task, s=script_path))
+        out = self.ps(self.launch_script(task, script_path))
         if "LAUNCHED=0" not in out:
             raise TransportError("schtasks launch of %s failed: %s" % (task, out.strip()[-300:]))
 

config/agentd.example.yaml

@@ -86,4 +86,5 @@ projects:
       # transient_patterns / env_patterns default to the built-in lists in agentd/core.py
     timeout_hours: 4
     max_rounds_per_thread: 8
+    max_progress_rounds: 12             # after this many consecutive [progress] rounds: note + Waiting (answerer)
     notify: {ntfy_topic: gwent, 'on': [waiting, review, failed]}   # phase 3, not used yet; 'on' must be quoted (YAML 1.1 boolean)

config/schema.json

@@ -131,6 +131,7 @@
         },
         "timeout_hours": {"type": "number", "minimum": 0},
         "max_rounds_per_thread": {"type": "integer", "minimum": 1},
+        "max_progress_rounds": {"type": "integer", "minimum": 1},
         "notify": {
           "type": "object",
           "additionalProperties": false,

docs/design.md

@@ -4,7 +4,7 @@ Plane 看板驱动的通用 agent 调度服务：任何人/agent 在 Plane 建
 
 ## 1. 目标与非目标
 - 目标：多项目隔离（一份配置一段）、多终端（本机/经 Warpgate 的远程机）、多 CLI（执行器适配器）、多轮（线程持久、进程不常驻）、失败可解释可重试、凭据只经 Vault 注入内存、任何会话可接手。
-- 非目标：不是工作流引擎（无 DAG/分支，顺序交给 Plane "blocked by"）；不替代 Plane/Temporal 的界面；不做账号系统；不存密钥。
+- 非目标：不是工作流引擎（无 DAG/分支，顺序交给 Plane "blocked by"——**TODO：阶段 1 尚未实现**，调度不读 blocked-by 关系，被阻塞的 Todo 项照样会被拾起）；不替代 Plane/Temporal 的界面；不做账号系统；不存密钥。
 
 ## 2. 架构
 ```
@@ -32,9 +32,18 @@ Plane ──webhook v2（HMAC）/ 轮询兜底──► agentd（接收端 + 调
 | 退出码≠0 / 无最后消息 | 结构化失败说明（类型 crash/timeout/env/logic、run、exit、重试 n/N、日志、最后一帧） | 临时性（网络/中继/429/5xx/超时）自动重试 ≤3，退避 2/5/10 min，续原线程；否则 Waiting |
 | 超时（默认 4 h，`budget:*` 覆盖） | 结束任务进程，失败说明 | Waiting |
 | 环境类（焦点丢失/弹窗/断网） | 当 blocked 处理，暂停同串行组的派发 | Waiting |
+| 运行中工作项被移出 In Progress | 结束任务进程，回帖"运行已取消：工作项状态被改为 X"，释放终端槽位/串行组，保留线程 | 不改 |
+| 连续 `max_progress_rounds`（默认 12）轮 `[progress]` | 回帖说明，不再续跑 | Waiting → 应答者 |
 | agentd 自身 | cron 拉起；心跳 10 min 无更新 → ntfy 报警 | — |
 回到 Todo 的工作项无条件重新拾起；逻辑性失败永不自动重试。
 
+阶段 1 的具体取舍（均有单测）：
+- **分类**：超时 → timeout；env 模式匹配（stderr 尾部 + 最后消息）→ env；临时性模式**只看 stderr 尾部/最后事件**（不看 agent 自己写的文字）；退出码 0 且无最后消息且尾部不像临时错误 → logic（永不重试），其余 → crash（尾部像临时错误才重试）。
+- **成功判定**：最后消息首行带合法标签即算成功，**即使退出码≠0**（失败的收尾钩子不应吞掉已完成的工作；回帖头里照写 exit）；无标签则退出码 0 才算成功（按 review 处理并注明），否则走失败流程。
+- **续线程失败**：非临时性的 crash/logic 发生在续线程的一轮上 → **立即**在新线程重跑一次（附交接摘要），不等退避；重建出来的那一轮若再失败就转 Waiting，不会二次重建。
+- **重试与线程轮数**：临时性重试是同一逻辑轮，不计入 `max_rounds_per_thread`，run id 仍递增（`GW-12-4` → `GW-12-5`）。
+- **新评论**：续线程时按评论 ID（`seen_comments`）过滤，而不是按时间戳，避免时钟偏差/同秒评论漏发或重发。
+
 ## 5. 配置（示例，值用 Vault 键名）
 ```yaml
 plane: {base_url: https://plane.kappiee.com, workspace: kappiee}

docs/executor-contract.md

@@ -15,7 +15,13 @@ agentd 不常驻：一轮运行比起它的那次 tick 活得久，所以执行
 `Run(run_id, repo, prompt, model, effort, tier, thread_id, script)`：`thread_id` 非空即续该线程。
 `ExecResult(last_message, exit_code, thread_id, log_paths, tail)`：`last_message` 原样回帖，首行应为 `[review]/[question]/[blocked]/[progress]`；`thread_id` 供下一轮续接；`tail` 是失败说明里的"最后一帧"（stderr 尾部/最后事件）。
 
-执行器**不接触 Plane**：不持有 token、不改工作项。回帖、状态流转、重试、超时、线程重建都由 core 统一处理。
+执行器**不接触 Plane**：不持有 token、不改工作项。回帖、状态流转、重试、超时、线程重建都由 core 统一处理。transport 启动 runner/中继脚本时传入去掉所有 `token_key`/`webhook_secret_key` 变量的环境（`Config.scrubbed_env()`）；新 transport 必须同样处理。
+
+## core 如何解读 `ExecResult`
+- 成功 = 最后消息首行带合法标签（**不看退出码**），或无标签但退出码 0（按 review 处理并注明）。首行判定跳过空行、BOM、开头的 ```/~~~ 代码围栏行和 `---` 分隔线，容忍 `**`、`>`、`#`、反引号等修饰，标签后可跟任意文字（`[progress] 第 2 步`）。
+- 失败分类：`tail` 决定是否临时性（`retry.transient_patterns`），`tail` + `last_message` 决定 env（`retry.env_patterns`）。所以适配器应把 stderr 尾部/最后一个事件放进 `tail`，不要把 agent 正文塞进去。
+- 续线程（`thread_id` 非空）的一轮非临时性失败 → core 立即在新线程重跑一次（附交接摘要），只一次。
+- `kill(handle)` 既用于超时，也用于取消（工作项在运行中被移出 In Progress）；必须能在运行未结束时调用。取消后 core 会尝试 `collect(handle)` 取线程 ID，所以 `collect` 在未结束的运行上也不能抛出致命错误（抛异常会被忽略）。
 
 ## RunnerExecutor（现有三个适配器的共同形态）
 1. 在终端上建运行目录 `<run_root>/<run_id>`（清空旧文件），写 `prompt.md`。
@@ -35,7 +41,8 @@ agentd 不常驻：一轮运行比起它的那次 tick 活得久，所以执行
 ## Transport 接口（`agentd/transports/`）
 `os`、`join`、`quote`、`default_run_root`、`mkdir(path, clean)`、`write_text`、`read_text`（不存在返回 None）、`grep_first(path, needle)`、`tail(path, n)`、`runner(run_dir, cwd, command, stdout_name)`、`launch(task, script, run_dir)`、`kill(task, run_dir)`、`forget(task)`、`process_count(name)`（None = 不可达）。
 - `local`：Linux 本机文件 + `setsid -f`；kill = 对 `pid.txt` 的进程组发 SIGTERM。
-- `warpgate-ssh`：Windows，经 `win10-ps.sh`（PowerShell `-EncodedCommand`）和 `win10-put.sh`（分块上传 + SHA-256 校验）；文件读取用 base64 标记回传；kill = `taskkill /T /F /PID <pid.txt>` + `schtasks /End /Delete`。脚本目录由 `agentd.helpers_dir`（或终端 `helpers_dir`）配置；脚本目前写死 `--target win10`。
+- `warpgate-ssh`：Windows，经 `win10-ps.sh`（PowerShell `-EncodedCommand`）和 `win10-put.sh`（分块上传 + SHA-256 校验）；文件读取用 base64 标记回传；kill = `taskkill /T /F /PID <pid.txt>` + `schtasks /End /Delete`。脚本目录由 `agentd.helpers_dir`（或终端 `helpers_dir`）配置；脚本目前写死 `--target win10`。路径都放进 PowerShell 单引号字面量（`'` 双写）；`schtasks /TR` 里脚本路径用 `\"…\"` 包裹，依赖 Windows PowerShell 5.1 的旧式原生参数传递（PowerShell 7.3+ 的 Standard 模式下需改写）。runner 进不了 repo 时写 `exit=97` 结束。
+- Codex 续线程：`codex exec resume <thread>` 不接受 `-C`/`-s`，`resume` 模板里不要写；工作目录由 runner 先进入 repo 保证，沙箱沿用 Codex 对该会话/全局配置的默认值（若需强制，用 `-c 'sandbox_mode="workspace-write"'`，上线前在 win10 上验证）。
 
 ## 新增一个执行器
 1. `agentd/executors/<name>.py`：继承 `RunnerExecutor`，设 `stdout_name`，实现 `collect`（必要时覆盖 `resume`/`command`）。完全不同的形态（如 HTTP API）可直接继承 `Executor`，只要 handle 可序列化。

docs/runbook.md

@@ -8,7 +8,7 @@ cp config/agentd.example.yaml config/agentd.local.yaml   # git-ignored；或用
 cli/agentd check-config                       # 校验 + 列出各终端 token 环境变量是否已注入 + cron 所需 --secret-env 参数
 python3 -m unittest                           # 离线单测
 ```
-配置里只写 Vault 键名（`token_key`）。agentd 读同名环境变量，token 只由 `agent-access.py exec --secret-env KEY=KEY` 注入进程内存；不要 export 到 shell、不要写文件。
+配置里只写 Vault 键名（`token_key`）。agentd 读同名环境变量，token 只由 `agent-access.py exec --secret-env KEY=KEY` 注入进程内存；不要 export 到 shell、不要写文件。agentd 起的所有子进程（本机 runner/执行器、win10 中继脚本）拿到的环境都去掉了全部 `token_key` 与 `webhook_secret_key` 变量。
 
 ## 运行
 单次 tick（cron 推荐）：
@@ -26,6 +26,12 @@ crontab（每 2 分钟；tick 有文件锁，重叠的 tick 直接跳过）：
 常驻替代：`... exec ... -- cli/agentd tick --loop --interval 120`（vault 会话超时要够长，cron 更稳）。
 cron 的 PATH 很短：本机执行器的二进制用终端 `bins` 写绝对路径（如 `claude: /home/ubuntu/.local/bin/claude`）。
 
+### 执行器二进制的解析（`bins` 或 PATH）
+命令模板的首词（`codex`、`claude`…）若在该终端的 `bins` 里，替换为那里的绝对路径；否则**原样交给终端的 shell 按 PATH 查找**。没有任何其他探测或回退：
+- oregon（local）：runner 是 `bash run.sh`，环境继承 agentd 进程（cron 下 PATH 只有 `/usr/bin:/bin`），所以 `claude` 必须写进 `bins`。**示例配置的 oregon 列了 `codex` 执行器但没有 `bins.codex`，即假定 `codex` 在 PATH 上——目前 oregon 没装 codex**（`which codex` 为空），带 `exec:codex` 的 oregon 工作项会以 exit 127（`command not found`，类型=crash，非临时性）转 Waiting。要用就装 codex 并加 `bins: {codex: <绝对路径>}`，否则从 oregon 的 `executors` 去掉 codex。
+- win10（warpgate-ssh）：runner 是计划任务里的 `powershell.exe -File run.ps1`，以创建任务的用户（Administrator，交互登录）运行，PATH 取该用户的环境。示例配置用 `bins.codex` 写死 `codex.exe` 绝对路径；去掉它就等于假定 Administrator 的 PATH 里有 `codex`。
+- `check-config` 不检查二进制是否存在；首次接入一个终端用 `agentd smoke <PROJECT> <terminal>` 验证。
+
 人工操作（同样在 `exec --secret-env` 下）：
 ```
 cli/agentd new GWENT "T-010 标题" --file brief.md --assign window10 --labels model:astra,needs-game --module M1 --state Todo
@@ -50,8 +56,11 @@ CLI 以 `--as <terminal>` / `agentd.cli_terminal` / 第一个有 token 的终端
 ## 工作项生命周期
 Todo + 指派给终端用户 → agentd 选最小编号、终端有空位、串行组空闲的项 → In Progress，评论 `[worker] 第 k 轮开始…` → 执行器结束后回帖 `[<executor>] 第 k 轮结束（exit, run, 终端, 日志）` + 最后消息：
 - `[review]` → Review，指派 `reviewer`；`[question]`/`[blocked]` → Waiting，指派 `answerer`（默认 reviewer）；`[progress]` → Todo，留在原终端，下一轮续线程；首行无标签 → 按 review 处理并注明。
-- 回到 Todo 的项总会被重新拾起；续线程时只发送上次之后的新评论（`[worker]` 评论永远跳过）。
-- 线程重建：换终端/执行器、续线程失败、线程轮数 ≥ `max_rounds_per_thread` → 新线程，首轮输入附交接摘要（各轮标签 + 上一轮最后消息），开始评论里注明原因。
+- 回到 Todo 的项总会被重新拾起；续线程时只发送上次之后的新评论（按评论 ID 判断，`[worker]` 评论永远跳过）。
+- 线程重建：换终端/执行器、续线程失败、线程轮数 ≥ `max_rounds_per_thread` → 新线程，首轮输入附交接摘要（各轮标签 + 上一轮最后消息），开始评论里注明原因与"附交接摘要"。临时性错误的自动重试算同一轮，不计入线程轮数（run id 照常递增）。
+- 连续 `max_progress_rounds`（项目配置，默认 12）轮 `[progress]` → 不再续跑，回帖说明并转 Waiting 指派 answerer；人工移回 Todo 后重新计数。
+- 运行中工作项被移出 In Progress（任何状态）→ 下一 tick 结束该运行（kill），回帖 `[worker] 运行已取消：工作项状态被改为 <状态>`，释放终端槽位与串行组，保留线程 ID（移回 Todo 时续原线程）。kill 失败则保留槽位，下一 tick 再试。
+- **未实现（TODO）**：Plane 的 "blocked by" 关系不参与调度——被阻塞的项只要在 Todo 且指派给终端就会被拾起。需要先后顺序时，先只把前一项放进 Todo。
 
 ## 常见故障
 | 现象 | 原因 / 处理 |
@@ -64,6 +73,7 @@ Todo + 指派给终端用户 → agentd 选最小编号、终端有空位、串
 | `类型=env` | 匹配 `retry.env_patterns`（焦点丢失/弹窗/断网等），转 Waiting，同串行组暂停到该项离开 Waiting |
 | `win10 unreachable; skipping` / `busy (N codex running…)` | 中继不通，或 win10 上有不是 agentd 起的 codex（`busy_processes`）；等它结束 |
 | `schtasks launch … failed` / `upload … failed` | 中继或 Windows 计划任务问题；手工 `win10-ps.sh "schtasks /Query /TN agentd-<run>"` 查看 |
-| 工作项在运行中被人改了状态 | 结果照常回帖，但不再改状态/重试 |
+| 工作项在运行中被人改了状态 | 运行还没结束：下一 tick 取消（kill + `[worker] 运行已取消…` 评论），不改状态。已经结束（`exit.txt` 已写）：结果照常回帖，但不再改状态/重试 |
 | Plane 回帖失败 | 在途记录保留，下一 tick 重新 collect 并回帖（可能重复一条评论） |
-| 想手工终止一轮 | 把工作项移出 In Progress（如 Waiting），再在终端上结束任务：本机 `kill -TERM -<pid.txt>`，win10 `taskkill /T /F /PID <pid.txt>`。被杀的 runner 不写 `exit.txt`，该运行占着终端槽位直到 deadline 才被收尾（不改状态）；急用时停 cron 后从 `state/agentd-state.json` 删掉该项的 `inflight` |
+| 想手工终止一轮 | 把工作项移出 In Progress（如 Waiting）即可，下一 tick agentd 自己 kill 并释放槽位。中继不通、kill 一直失败时：在终端上手工结束（本机 `kill -TERM -<pid.txt>`，win10 `taskkill /T /F /PID <pid.txt>`），停 cron 后从 `state/agentd-state.json` 删掉该项的 `inflight` |
+| `已连续 N 轮 [progress]` | 达到 `max_progress_rounds`：看最近几轮是否在原地打转；拆分任务或补充说明后移回 Todo |

tests/fakes.py

@@ -87,8 +87,12 @@ class FakeExecutor:
     def __init__(self, resume=True):
         self.resume = resume
         self.started, self.killed, self.done = [], [], {}
+        self.fail_starts, self.fail_kills = 0, 0
 
     def start(self, run):
+        if self.fail_starts:
+            self.fail_starts -= 1
+            raise RuntimeError("relay down")
         self.started.append(run)
         return {"run_dir": "/runs/" + run.run_id, "run_id": run.run_id, "thread_id": run.thread_id}
 
@@ -99,6 +103,9 @@ class FakeExecutor:
         return self.done[handle["run_id"]]
 
     def kill(self, handle):
+        if self.fail_kills:
+            self.fail_kills -= 1
+            raise RuntimeError("relay down")
         self.killed.append(handle["run_id"])
 
     def finish(self, message, exit_code=0, thread="th-1", tail=""):

tests/test_core.py

@@ -123,7 +123,7 @@ class CoreTest(unittest.TestCase):
         self.assertIn("## 交接摘要", run.prompt)
         self.assertIn("续线程失败", run.prompt)
         self.assertIn("[progress] half way", run.prompt)
-        self.assertIn("新线程（重建：续线程失败（crash））", self.p.texts(iid)[-1])
+        self.assertIn("新线程（重建：续线程失败（crash），附交接摘要）", self.p.texts(iid)[-1])
 
     def test_thread_rebuilt_after_max_rounds(self):
         iid = self.p.add_issue(1)
@@ -198,6 +198,141 @@ class CoreTest(unittest.TestCase):
         self.assertEqual(hb["inflight"][0]["run"], "GW-1-1")
         self.assertEqual(hb["errors"], [])
 
+    # ---------------------------------------------------------------- reviewer decisions / fixed defects
+    def test_item_moved_out_of_in_progress_cancels_run(self):
+        a = self.p.add_issue(1, labels=["needs-game"])
+        b = self.p.add_issue(2, labels=["needs-game"])
+        self.h.tick()
+        self.codex().finish("[progress] step 1")
+        self.h.tick()                                   # round 2 resumes th-1
+        self.assertEqual(self.codex().started[-1].run_id, "GW-1-2")
+        self.p.issues[a]["state"] = "s-wait"            # a person moves it while round 2 runs
+        self.h.tick()
+        self.assertEqual(self.codex().killed, ["GW-1-2"])
+        self.assertIn("[worker] 运行已取消：工作项状态被改为 Waiting（run=GW-1-2）", self.p.texts(a)[-1])
+        rec = self.h.rec(a)
+        self.assertNotIn("inflight", rec)
+        self.assertEqual((rec["thread_id"], rec["thread_on"]), ("th-1", ["win10", "codex"]))
+        self.assertEqual(self.p.state_of(a), "Waiting")  # agentd does not touch the state
+        # terminal slot (concurrency 1) and serialize group were released in the same tick
+        self.assertEqual(self.codex().started[-1].run_id, "GW-2-1")
+        self.assertEqual(self.p.state_of(b), "In Progress")
+        # back to Todo later: resumes the kept thread
+        self.codex().finish("[review] b done")
+        self.p.issues[a]["state"] = "s-todo"
+        self.h.tick()
+        self.assertEqual((self.codex().started[-1].run_id, self.codex().started[-1].thread_id), ("GW-1-3", "th-1"))
+
+    def test_cancel_keeps_slot_when_kill_fails(self):
+        a = self.p.add_issue(1)
+        self.p.add_issue(2)
+        self.h.tick()
+        self.codex().fail_kills = 1
+        self.p.issues[a]["state"] = "s-bl"
+        self.h.tick()
+        self.assertIn("inflight", self.h.rec(a))
+        self.assertEqual(len(self.codex().started), 1)   # GW-2 must not share win10 with a live run
+        self.h.tick()
+        self.assertNotIn("inflight", self.h.rec(a))
+        self.assertIn("被改为 Backlog", self.p.texts(a)[-1])
+        self.assertEqual(self.codex().started[-1].run_id, "GW-2-1")
+
+    def test_transient_retries_do_not_use_up_thread_rounds(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("[progress] 1")
+        self.h.tick()
+        self.codex().finish("[progress] 2")
+        self.h.tick()                                   # round 3: thread_rounds == 3 == max
+        self.assertEqual(self.h.rec(iid)["thread_rounds"], 3)
+        self.codex().finish("", exit_code=1, tail="stream disconnected before completion")
+        self.h.tick()
+        self.h.clock[0] += 121
+        self.h.tick()
+        retry = self.codex().started[-1]
+        self.assertEqual((retry.run_id, retry.thread_id), ("GW-1-4", "th-1"))   # same thread, new run id
+        self.assertEqual(self.h.rec(iid)["thread_rounds"], 3)
+        self.codex().finish("[progress] 3")
+        self.h.tick()
+        self.assertIsNone(self.codex().started[-1].thread_id)                    # next real round rebuilds
+        self.assertIn("线程已续 3 轮（上限 3）", self.p.texts(iid)[-1])
+
+    def test_failed_start_keeps_retry_marker(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("", exit_code=1, tail="Connection reset by peer")
+        self.h.tick()
+        self.h.clock[0] += 121
+        self.codex().fail_starts = 1
+        self.h.tick()                                   # relay down while relaunching
+        self.assertIn("retry", self.h.rec(iid))
+        self.assertEqual(self.p.state_of(iid), "In Progress")
+        self.h.tick()
+        self.assertEqual(self.codex().started[-1].thread_id, "th-1")
+        self.assertIn("inflight", self.h.rec(iid))
+
+    def test_progress_cap_moves_to_waiting(self):
+        data = config_data()
+        data["projects"]["GWENT"]["max_progress_rounds"] = 2
+        h = Harness(data)
+        iid = h.plane.add_issue(1)
+        h.tick()
+        ex = h.execs[("win10", "codex")]
+        ex.finish("[progress] 1")
+        h.tick()
+        self.assertEqual(len(ex.started), 2)
+        ex.finish("[progress] 2")
+        h.tick()
+        self.assertEqual(len(ex.started), 2)
+        self.assertEqual((h.plane.state_of(iid), h.plane.assignee_of(iid)), ("Waiting", "kappiee"))
+        self.assertIn("已连续 2 轮 [progress]", h.plane.texts(iid)[-1])
+        h.plane.issues[iid].update(state="s-todo", assignees=["u-w10"])   # a person lets it continue: fresh allowance
+        h.tick()
+        ex.finish("[progress] 3")
+        h.tick()
+        self.assertEqual(len(ex.started), 4)
+        h.close()
+
+    def test_tagged_message_counts_as_success_even_with_nonzero_exit(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("[review] done, but a hook failed", exit_code=1)
+        self.h.tick()
+        self.assertEqual(self.p.state_of(iid), "Review")
+        self.assertIn("第 1 轮结束（exit=1，", self.p.texts(iid)[-1])
+
+    def test_untagged_message_with_nonzero_exit_is_a_failure(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("I gave up, tests timed out", exit_code=1, tail="")
+        self.h.tick()
+        self.assertEqual(self.p.state_of(iid), "Waiting")   # agent prose never makes it transient
+        self.assertIn("类型=crash，run=GW-1-1", self.p.texts(iid)[-1])
+
+    def test_rebuild_after_resume_failure_happens_once(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("[progress] half way")
+        self.h.tick()
+        self.codex().finish("", exit_code=1, thread=None, tail="thread not found")
+        self.h.tick()                                   # immediate rebuild (round 3, fresh thread)
+        self.assertIsNone(self.codex().started[2].thread_id)
+        self.assertNotIn("重试", self.p.texts(iid)[-1])
+        self.codex().finish("", exit_code=1, thread=None, tail="thread not found")
+        self.h.tick()
+        self.assertEqual(len(self.codex().started), 3)  # no second rebuild
+        self.assertEqual((self.p.state_of(iid), self.p.assignee_of(iid)), ("Waiting", "kappiee"))
+
+    def test_state_saved_right_after_each_start(self):
+        self.p.add_issue(1, assignee="oregon")
+        self.p.add_issue(2, assignee="oregon")
+        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))
+        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"]])
 
 class EnvFailureTest(unittest.TestCase):
     def test_env_failure_pauses_group_until_item_leaves_waiting(self):

tests/test_prompt.py

@@ -37,6 +37,13 @@ class PromptTest(unittest.TestCase):
         self.assertIn("RETRY", p)
         self.assertNotIn("没有新评论", p)
 
+    def test_new_comment_filter_is_by_id_not_timestamp(self):
+        late_but_old = [{"id": "c9", "created_at": "t0", "comment_html": "<p>clock-skewed reply</p>"},
+                        {"id": "c1", "created_at": "t9", "comment_html": "<p>edited old one</p>"}]
+        p = build_prompt("GW-1", ISSUE, late_but_old, {"c1"}, True)
+        self.assertIn("clock-skewed reply", p)
+        self.assertNotIn("edited old one", p)
+
     def test_handoff(self):
         rec = {"runs": 3, "tags": ["progress", "progress", "question"], "last_message": "[question] A or B?"}
         h = handoff_summary(rec, "线程已续 3 轮")

tests/test_text.py

@@ -13,6 +13,18 @@ class FirstLineTest(unittest.TestCase):
         self.assertEqual(classify_first_line("**[blocked]** no device"), "blocked")
         self.assertEqual(classify_first_line("[progress]"), "progress")
 
+    def test_edge_cases(self):
+        self.assertEqual(classify_first_line("\ufeff[review] ok"), "review")
+        self.assertEqual(classify_first_line("```\n[progress] still going\n```"), "progress")
+        self.assertEqual(classify_first_line("```text\n[blocked] no device\n```"), "blocked")
+        self.assertEqual(classify_first_line("---\n[question] which?"), "question")
+        self.assertEqual(classify_first_line("[progress]: step 2 of 5, next: tests"), "progress")
+        self.assertEqual(classify_first_line("> `[review]` merged"), "review")
+        self.assertIsNone(classify_first_line("note first\n[review] later"))   # tag must lead
+        msg = html_to_text(text_to_html("[codex] 第 1 轮结束", pre="\n  [review] ok\n"))
+        self.assertEqual(msg, "[codex] 第 1 轮结束\n\n  [review] ok")       # <pre>-wrapped body keeps its lines
+        self.assertEqual(classify_first_line(html_to_text("<pre>\n[progress] x</pre>")), "progress")
+
     def test_untagged(self):
         self.assertIsNone(classify_first_line("done, see [review] below"))
         self.assertIsNone(classify_first_line("[done] finished"))
@@ -56,6 +68,13 @@ class FailureTest(unittest.TestCase):
         self.assertEqual(self.kind("whatever", timed_out=True), ("timeout", False))
         self.assertEqual(self.kind("window lost focus", cfg={"env_patterns": ["lost focus"]}), ("env", False))
 
+    def test_logic_is_never_transient_and_prose_is_ignored(self):
+        self.assertEqual(self.kind("HTTP 429 Too Many Requests", exit_code=0), ("crash", True))
+        r = ExecResult("tests timed out", 1, None, [], "")
+        self.assertEqual(classify_failure(r, False, {}), ("crash", False))
+        r = ExecResult("foreground window lost", 1, None, [], "")
+        self.assertEqual(classify_failure(r, False, {"env_patterns": ["window lost"]}), ("env", False))
+
     def test_custom_transient_patterns_replace_defaults(self):
         self.assertEqual(self.kind("timed out", cfg={"transient_patterns": ["flaky"]}), ("crash", False))
         self.assertEqual(self.kind("flaky relay", cfg={"transient_patterns": ["flaky"]}), ("crash", True))

tests/test_transports.py

@@ -0,0 +1,122 @@
+"""Transport hardening: secrets never reach executors, Windows script quoting, command shapes, atomic state files."""
+import json
+import os
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from agentd import ROOT, config
+from agentd.config import Config
+from agentd.executors import Run, make_executor
+from agentd.store import Store
+from agentd.transports.local import LocalTransport
+from agentd.transports.warpgate_ssh import WarpgateSshTransport
+from tests.fakes import config_data
+
+
+class RecordingWin(WarpgateSshTransport):
+    """Windows transport with the relay replaced by an in-memory recorder."""
+    def __init__(self, cfg, terminal):
+        super().__init__(cfg, terminal)
+        self.files, self.launched = {}, []
+
+    def mkdir(self, path, clean=False):
+        pass
+
+    def write_text(self, path, text):
+        self.files[path] = text
+
+    def launch(self, task, script_path, run_dir):
+        self.launched.append((task, script_path, run_dir))
+
+
+class LocalSecretsTest(unittest.TestCase):
+    def test_executor_env_has_no_plane_tokens(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            data = config_data()
+            data["agentd"]["state_dir"] = str(Path(tmp) / "state")
+            data["terminals"]["oregon"]["repo"] = tmp
+            data["projects"]["GWENT"]["webhook_secret_key"] = "HOOK_GW"
+            cfg = Config(data)
+            self.assertEqual(cfg.secret_keys(), {"TOK_OR", "TOK_W10", "HOOK_GW"})
+            t = LocalTransport(cfg, "oregon")
+            ex = make_executor(cfg, "oregon", "shell", t)
+            script = 'echo "or=${TOK_OR:-none} w10=${TOK_W10:-none} hook=${HOOK_GW:-none} keep=${KEEP_ME:-none}"'
+            with mock.patch.dict(os.environ, {"TOK_OR": "s1", "TOK_W10": "s2", "HOOK_GW": "s3", "KEEP_ME": "yes"}):
+                h = ex.start(Run("GW-9-1", tmp, "", script=script))
+            end = time.time() + 10
+            while ex.poll(h) != "finished" and time.time() < end:
+                time.sleep(0.05)
+            self.assertEqual(ex.collect(h).last_message, "or=none w10=none hook=none keep=yes")
+
+
+class WindowsScriptTest(unittest.TestCase):
+    def setUp(self):
+        self.cfg = config.load(ROOT / "config" / "agentd.example.yaml")
+        self.t = RecordingWin(self.cfg, "win10")
+        self.ex = make_executor(self.cfg, "win10", "codex", self.t)
+
+    def test_codex_run_and_resume_commands(self):
+        run_dir = self.t.join(self.ex.run_root(), "GW-5-1")
+        fresh = self.ex.command(Run("GW-5-1", self.t.repo, "p", "gpt-6-astra", "high"), run_dir)
+        self.assertTrue(fresh.startswith("& 'C:\\Users\\Administrator\\AppData\\Local\\Programs\\OpenAI\\Codex\\bin\\codex.exe' exec --json"))
+        self.assertIn("-C '%s' -s workspace-write" % self.t.repo, fresh)
+        self.assertIn("-o '%s\\last.md'" % run_dir, fresh)
+        resumed = self.ex.command(Run("GW-5-2", self.t.repo, "p", "gpt-6-astra", "high", thread_id="th-9"), run_dir)
+        self.assertIn(" exec resume th-9 --json -m gpt-6-astra", resumed)
+        self.assertNotIn(" -C ", resumed)   # `codex exec resume` rejects -C/-s; the runner already cd's into the repo
+        self.assertNotIn(" -s ", resumed)
+        tiered = self.ex.command(Run("GW-5-3", self.t.repo, "p", "gpt-6.1-sol", "medium", tier="fast"), run_dir)
+        self.assertIn("-c 'service_tier=\"fast\"'", tiered)
+
+    def test_run_dirs_are_unique_per_run(self):
+        a = self.ex.start(Run("GW-5-1", self.t.repo, "p", "m", "e"))
+        b = self.ex.start(Run("GW-5-2", self.t.repo, "p", "m", "e"))
+        self.assertNotEqual(a["run_dir"], b["run_dir"])
+        self.assertEqual([x[0] for x in self.t.launched], ["agentd-GW-5-1", "agentd-GW-5-2"])
+        self.assertTrue(a["run_dir"].endswith("\\runtime\\agentd\\GW-5-1"))
+
+    def test_runner_quotes_paths_with_spaces_and_apostrophes(self):
+        _, script = self.t.runner(r"C:\a b\O'Neil\run", r"C:\my repo\O'Neil", "& 'C:\\x\\codex.exe' exec -", "events.jsonl")
+        self.assertIn(r"Set-Location -LiteralPath 'C:\my repo\O''Neil' -ErrorAction Stop", script)
+        self.assertIn(r"'C:\a b\O''Neil\run\prompt.md'", script)
+        self.assertIn(r"'C:\a b\O''Neil\run\exit.txt' -Value ('exit=97", script)   # cd failure still ends the run
+        self.assertIn(r"Add-Content -Path 'C:\a b\O''Neil\run\events.jsonl'", script)
+
+    def test_launch_script_quoting(self):
+        s = self.t.launch_script("agentd-GW-1-1", r"C:\My Docs\O'Neil\run.ps1")
+        self.assertIn(r"-File \"C:\My Docs\O''Neil\run.ps1\"'", s)
+        self.assertIn("/TN 'agentd-GW-1-1'", s)
+
+    def test_relay_env_has_no_plane_tokens(self):
+        seen = {}
+
+        def fake_run(args, **kw):
+            seen.update(kw["env"])
+            return mock.Mock(returncode=0, stdout="X=ok", stderr="")
+        with mock.patch.dict(os.environ, {"PLANE_WINDOW10_API_TOKEN": "s"}), \
+                mock.patch("agentd.transports.warpgate_ssh.subprocess.run", fake_run):
+            WarpgateSshTransport(self.cfg, "win10").ps("'X=ok'")
+        self.assertNotIn("PLANE_WINDOW10_API_TOKEN", seen)
+        self.assertIn("WIN10_TIMEOUT", seen)
+
+
+class StoreTest(unittest.TestCase):
+    def test_save_is_atomic_and_heartbeat_is_complete_json(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            st = Store(tmp, echo=False)
+            st.save({"projects": {"A": {"issues": {}}}})
+            with mock.patch("agentd.store.os.replace", side_effect=OSError("disk full")):
+                with self.assertRaises(OSError):
+                    st.save({"projects": {"B": {}}})
+            self.assertIn("A", st.load()["projects"])          # old state survives a failed write
+            st.heartbeat({"errors": [], "inflight": []})
+            hb = json.loads((Path(tmp) / "heartbeat.json").read_text())
+            self.assertEqual(hb["errors"], [])
+            self.assertIn("ts", hb)
+
+
+if __name__ == "__main__":
+    unittest.main()