From 31850c4a81c483b25f25201f67cd38c8c851a834 Mon Sep 17 00:00:00 2001
From: "jj.wang" <jj.wang@alphabets.cn>
Date: Wed, 21 Oct 2026 07:04:32 +0000
Subject: [PATCH] Scheduling core, state store, CLI and unit tests

One tick polls in-flight runs, posts results, classifies the first line and
moves state/assignee; failures get a structured note with transient retry and
backoff, timeouts are killed and handed over, resume failures and long threads
rebuild with a handoff summary. Then the lowest Todo items are dispatched within
terminal concurrency and serialize groups, and a heartbeat is written.
CLI: tick, new, show, comment, state, assign, smoke, check-config.
---
 agentd/cli.py              | 210 +++++++++++++++
 agentd/core.py             | 512 +++++++++++++++++++++++++++++++++++++
 agentd/store.py            |  61 +++++
 cli/agentd                 |   9 +
 tests/__init__.py          |   0
 tests/fakes.py             | 129 ++++++++++
 tests/test_config.py       |  71 +++++
 tests/test_core.py         | 223 ++++++++++++++++
 tests/test_local_runner.py |  81 ++++++
 tests/test_prompt.py       |  51 ++++
 tests/test_text.py         |  79 ++++++
 11 files changed, 1426 insertions(+)
 create mode 100644 agentd/cli.py
 create mode 100644 agentd/core.py
 create mode 100644 agentd/store.py
 create mode 100755 cli/agentd
 create mode 100644 tests/__init__.py
 create mode 100644 tests/fakes.py
 create mode 100644 tests/test_config.py
 create mode 100644 tests/test_core.py
 create mode 100644 tests/test_local_runner.py
 create mode 100644 tests/test_prompt.py
 create mode 100644 tests/test_text.py

diff --git a/agentd/cli.py b/agentd/cli.py
new file mode 100644
index 0000000..ede2331
--- /dev/null
+++ b/agentd/cli.py
@@ -0,0 +1,210 @@
+"""agentd command line. Tokens come from the environment (agent-access.py exec --secret-env ...)."""
+import argparse
+import os
+import re
+import sys
+import time
+
+from . import config as config_mod
+from .core import Agentd
+from .plane import Plane, brief_to_html, html_to_text, text_to_html
+from .store import Busy
+
+
+def identity(cfg, wanted=None):
+    """Terminal whose token the CLI uses: --as, then agentd.cli_terminal, then the first with a token."""
+    for term in [wanted, cfg.agentd.get("cli_terminal")] + list(cfg.terminals):
+        if term and term in cfg.terminals and cfg.token(term):
+            return term
+    sys.exit("no Plane token in the environment; run under agent-access.py exec %s" % " ".join(cfg.secret_env_args()))
+
+
+def client(cfg, term):
+    return Plane(cfg.plane["base_url"], cfg.plane["workspace"], cfg.token(term), cfg.plane.get("timeout", 60))
+
+
+def seq_of(ref):
+    m = re.search(r"(\d+)$", ref)
+    if not m:
+        sys.exit("bad work item reference %r (use 12 or GW-12)" % ref)
+    return int(m.group(1))
+
+
+def cmd_tick(cfg, a):
+    d = Agentd(cfg)
+    if a.loop:
+        d.loop(a.interval or cfg.agentd.get("interval", 120))
+    try:
+        d.tick(project=a.project)
+    except Busy as exc:
+        print("skip: %s" % exc)
+
+
+def cmd_new(cfg, a, pl=None, quiet=False):
+    p = cfg.project(a.project)
+    pl = pl or client(cfg, identity(cfg, a.as_))
+    text = sys.stdin.read() if a.file == "-" else open(a.file, encoding="utf-8").read()
+    body = {"name": a.title, "description_html": brief_to_html(text), "state": pl.states(p["id"])[a.state]}
+    if a.labels:
+        labels = pl.labels(p["id"])
+        body["labels"] = [labels[x] for x in a.labels.split(",")]
+    if a.assign:
+        body["assignees"] = [pl.members()[a.assign]]
+    issue = pl.create_issue(p["id"], body)
+    if a.module:
+        mods = pl.modules(p["id"])
+        mid = mods.get(a.module) or pl.create_module(p["id"], a.module)["id"]
+        pl.add_to_module(p["id"], mid, issue["id"])
+    if not quiet:
+        print("created %s-%s %s" % (p.get("key_prefix", a.project), issue["sequence_id"], issue["id"]))
+    return issue
+
+
+def cmd_show(cfg, a):
+    p = cfg.project(a.project)
+    pl = client(cfg, identity(cfg, a.as_))
+    i = pl.find_issue(p["id"], seq_of(a.ref))
+    sn = {v: k for k, v in pl.states(p["id"]).items()}
+    ln = {v: k for k, v in pl.labels(p["id"]).items()}
+    un = {v: k for k, v in pl.members().items()}
+    print("%s-%s %s | state=%s | assignees=%s | labels=%s" % (
+        p.get("key_prefix", a.project), i["sequence_id"], i["name"], sn.get(i.get("state"), i.get("state")),
+        [un.get(x, x) for x in i.get("assignees") or []], [ln.get(x, x) for x in i.get("labels") or []]))
+    cs = pl.list_comments(p["id"], i["id"])
+    for c in cs[-a.n:] if a.n else []:
+        who = c.get("actor") or c.get("created_by")
+        print("--- %s (%s)" % (c.get("created_at"), un.get(who, who)))
+        print(html_to_text(c.get("comment_html"))[:2500])
+
+
+def cmd_comment(cfg, a):
+    p = cfg.project(a.project)
+    pl = client(cfg, identity(cfg, a.as_))
+    i = pl.find_issue(p["id"], seq_of(a.ref))
+    text = sys.stdin.read() if a.file == "-" else open(a.file, encoding="utf-8").read()
+    pl.post_comment(p["id"], i["id"], text_to_html(text))
+    print("commented %s" % a.ref)
+
+
+def cmd_state(cfg, a):
+    p = cfg.project(a.project)
+    pl = client(cfg, identity(cfg, a.as_))
+    i = pl.find_issue(p["id"], seq_of(a.ref))
+    name = p["states"].get(a.state, a.state)  # accepts keys (todo) or Plane names (Todo)
+    pl.set_state(p["id"], i["id"], pl.states(p["id"])[name])
+    print("%s -> %s" % (a.ref, name))
+
+
+def cmd_assign(cfg, a):
+    p = cfg.project(a.project)
+    pl = client(cfg, identity(cfg, a.as_))
+    i = pl.find_issue(p["id"], seq_of(a.ref))
+    pl.set_assignees(p["id"], i["id"], [pl.members()[a.user]])
+    print("%s assigned %s" % (a.ref, a.user))
+
+
+SMOKE_BRIEF = ("<p>agentd 冒烟测试。不要读写任何文件，不要运行任何命令。</p>"
+               "<p>你的最后一条消息必须恰好是下面这一行（不要加任何别的内容）：</p><pre>[review] ok</pre>")
+
+
+def cmd_smoke(cfg, a):
+    p = cfg.project(a.project)
+    if a.terminal not in cfg.terminals:
+        sys.exit("unknown terminal %s" % a.terminal)
+    pl = client(cfg, identity(cfg, a.as_))
+    user = cfg.terminals[a.terminal]["plane_user"]
+    body = {"name": "agentd smoke %s %s" % (a.terminal, time.strftime("%Y-%m-%d %H:%M")),
+            "description_html": SMOKE_BRIEF, "state": pl.states(p["id"])[p["states"]["todo"]],
+            "assignees": [pl.members()[user]]}
+    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)
+    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:
+            d.tick(project=a.project, only_issue=issue["id"])
+        except Busy as exc:
+            print("skip: %s" % exc)
+        state = pl.get_issue(p["id"], issue["id"]).get("state")
+        if state in done_states:
+            break
+        time.sleep(a.interval)
+    last = ""
+    for c in reversed(pl.list_comments(p["id"], issue["id"])):
+        text = html_to_text(c.get("comment_html"))
+        if not text.startswith("[worker]"):
+            last = text
+            break
+    ok = state == states[p["states"]["review"]] and re.search(r"(?m)^\[review\] ok\s*$", last)
+    print("%s %s: state=%s\n%s" % ("PASS" if ok else "FAIL", key, {v: k for k, v in states.items()}.get(state, state), last[-800:]))
+    if ok:
+        pl.set_state(p["id"], issue["id"], states[p["states"]["done"]])
+    sys.exit(0 if ok else 1)
+
+
+def cmd_check_config(cfg, a):
+    print("config %s is valid" % cfg.path)
+    print("projects: %s" % ", ".join(cfg.projects))
+    for name, t in cfg.terminals.items():
+        env = cfg.token_env(name)
+        print("terminal %-10s %-13s %-8s executors=%s token %s: %s" % (
+            name, t["transport"], t["os"], ",".join(t["executors"]), env, "set" if os.environ.get(env) else "MISSING"))
+    print("cron/exec secret args: %s" % " ".join(cfg.secret_env_args()))
+
+
+def parser():
+    ap = argparse.ArgumentParser(prog="agentd", description=__doc__)
+    ap.add_argument("--config", help="YAML config (default: $AGENTD_CONFIG, config/agentd.local.yaml, config/agentd.yaml)")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+    s = sub.add_parser("tick", help="run the dispatcher once or in a loop")
+    g = s.add_mutually_exclusive_group()
+    g.add_argument("--once", action="store_true")
+    g.add_argument("--loop", action="store_true")
+    s.add_argument("--interval", type=int)
+    s.add_argument("--project")
+    s.set_defaults(fn=cmd_tick)
+
+    def item(name, fn, help_):
+        s = sub.add_parser(name, help=help_)
+        s.add_argument("project")
+        s.add_argument("ref", help="sequence number or KEY-n")
+        s.add_argument("--as", dest="as_", help="terminal identity to act as")
+        s.set_defaults(fn=fn)
+        return s
+
+    s = sub.add_parser("new", help="create a work item from a brief file")
+    s.add_argument("project")
+    s.add_argument("title")
+    s.add_argument("--file", required=True, help="brief file (HTML passes through, text goes in <pre>); - = stdin")
+    s.add_argument("--assign")
+    s.add_argument("--labels", default="")
+    s.add_argument("--module")
+    s.add_argument("--state", default="Backlog")
+    s.add_argument("--as", dest="as_")
+    s.set_defaults(fn=cmd_new)
+    item("show", cmd_show, "show a work item and its last n comments").add_argument("n", nargs="?", type=int, default=1)
+    item("comment", cmd_comment, "post a comment from a file (- = stdin)").add_argument("file")
+    item("state", cmd_state, "set state (Plane name or config key)").add_argument("state")
+    item("assign", cmd_assign, "set the assignee").add_argument("user")
+    s = sub.add_parser("smoke", help="end-to-end round on a terminal: expects '[review] ok'")
+    s.add_argument("project")
+    s.add_argument("terminal")
+    s.add_argument("--interval", type=int, default=30)
+    s.add_argument("--timeout", type=int, default=30, help="minutes")
+    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)
+    return ap
+
+
+def main(argv=None):
+    a = parser().parse_args(argv)
+    try:
+        cfg = config_mod.load(a.config)
+    except config_mod.ConfigError as exc:
+        sys.exit(str(exc))
+    a.fn(cfg, a)
diff --git a/agentd/core.py b/agentd/core.py
new file mode 100644
index 0000000..0572f75
--- /dev/null
+++ b/agentd/core.py
@@ -0,0 +1,512 @@
+"""Scheduling core. One tick per project:
+  (a) poll in-flight runs; on finish post the result, classify its first line and move state/assignee;
+      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."""
+import os
+import re
+import time
+from collections import Counter
+from pathlib import Path
+
+from . import ROOT
+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
+
+TAGS = ("review", "question", "blocked", "progress")
+WORKER = "[worker]"
+DEFAULT_TRANSIENT = [
+    r"time[sd]?[ -]?out",
+    r"connection (refused|reset)",
+    r"banner exchange",
+    r"\b429\b|too many requests|rate.?limit",
+    r"\b50[0-4]\b|internal server error|bad gateway|service unavailable",
+    r"stream (disconnected|error)",
+]
+DEFAULT_INTRO = ("你是 {terminal} 上的 {executor}，仓库 {repo}。先读仓库根目录 AGENTS.md。下面是看板协作协议"
+                 "（看板是 Plane，工作项编号 {prefix}-k），然后是分派给你的工作项。"
+                 "你的最后一条消息会被原样贴回工作项评论，首行必须是 [review]/[question]/[blocked]/[progress] 之一。")
+OUTPUT_RULE = "输出要求：最后一条消息首行为 [review]/[question]/[blocked]/[progress] 之一，内容按协议；不要等待人工回复。"
+
+
+# ---------------------------------------------------------------- pure helpers (unit-tested)
+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
+
+
+def matches_any(text, patterns):
+    return any(re.search(p, text or "", re.I) for p in patterns)
+
+
+def classify_failure(result, timed_out, retry_cfg):
+    """-> (kind, transient). kind in timeout | env | crash | logic (design §4)."""
+    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", [])):
+        return "env", False
+    kind = "logic" if result.exit_code == 0 else "crash"
+    return kind, matches_any(text, retry_cfg.get("transient_patterns", DEFAULT_TRANSIENT))
+
+
+def parse_budget(label):
+    """`budget:2h` / `budget:90m` / `budget:1.5h` -> hours."""
+    m = re.fullmatch(r"budget:(\d+(?:\.\d+)?)([hm]?)", label)
+    if not m:
+        return None
+    value = float(m.group(1))
+    return value / 60 if m.group(2) == "m" else value
+
+
+def handoff_summary(rec, reason):
+    s = "本工作项此前已进行 %d 轮，执行器线程已重建（原因：%s）。各轮首行标签：%s。" % (
+        rec.get("runs", 0), reason or "上一线程不可续", ", ".join(rec.get("tags", [])) or "无")
+    if rec.get("last_message"):
+        s += "\n\n上一轮最后消息：\n" + rec["last_message"]
+    return s
+
+
+def build_prompt(key, issue, comments, seen, resumed, protocol="", intro="", handoff=None, retry_note=None):
+    """Fresh thread: intro + protocol + work item + handoff + all comments.
+    Resumed thread: only comments not seen before. `[worker]` comments are always skipped."""
+    parts = []
+    if resumed:
+        parts.append("工作项 %s 有新消息，请在同一线程里继续：" % key)
+        new = [c for c in comments if c.get("id") not in seen]
+    else:
+        if intro:
+            parts.append(intro)
+        if protocol:
+            parts.append("## 协议\n" + protocol.strip())
+        parts.append("## 工作项 %s: %s\n\n%s" % (key, issue.get("name", ""), html_to_text(issue.get("description_html"))))
+        if handoff:
+            parts.append("## 交接摘要\n" + handoff)
+        new = comments
+    shown = 0
+    for c in new:
+        body = html_to_text(c.get("comment_html"))
+        if not body or body.startswith(WORKER):
+            continue
+        parts.append("### 评论（%s）\n%s" % (c.get("created_at", ""), body))
+        shown += 1
+    if retry_note:
+        parts.append(retry_note)
+    elif resumed and not shown:
+        parts.append("（没有新评论；请从上次停下的地方继续推进。）")
+    parts.append(OUTPUT_RULE)
+    return "\n\n".join(parts)
+
+
+def new_rec(seq):
+    return {"seq": seq, "runs": 0, "retries": 0, "tags": [], "seen_comments": [], "thread_rounds": 0}
+
+
+# ---------------------------------------------------------------- per-project view for one tick
+class Ctx:
+    def __init__(self, name, pcfg, reader):
+        self.name, self.cfg, self.reader = name, pcfg, reader
+        self.pid = pcfg["id"]
+        self.prefix = pcfg.get("key_prefix", name)
+        self.states = reader.states(self.pid)
+        self.labels = reader.labels(self.pid)
+        self.members = reader.members()
+        self.label_names = {v: k for k, v in self.labels.items()}
+        self.member_names = {v: k for k, v in self.members.items()}
+        self.issues = {i["id"]: i for i in reader.list_issues(self.pid)}
+
+    def state_id(self, key):
+        return self.states[self.cfg["states"][key]]
+
+    def key(self, seq):
+        return "%s-%s" % (self.prefix, seq)
+
+
+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.now = now
+        self.log = self.store.log
+        self.plane_factory = plane_factory or (lambda term: Plane(
+            cfg.plane["base_url"], cfg.plane["workspace"], cfg.token(term), cfg.plane.get("timeout", 60)))
+        self.transport_factory = transport_factory or (lambda term: make_transport(cfg, term))
+        self.executor_factory = executor_factory or (lambda term, name: make_executor(cfg, term, name, self.transport(term)))
+        self._reset_caches()
+
+    def _reset_caches(self):
+        self._clients, self._transports, self._executors, self._ready = {}, {}, {}, {}
+
+    # ---------------------------------------------------------------- caches
+    def client(self, terminal):
+        if terminal not in self._clients:
+            self._clients[terminal] = self.plane_factory(terminal)
+        return self._clients[terminal]
+
+    def transport(self, terminal):
+        if terminal not in self._transports:
+            self._transports[terminal] = self.transport_factory(terminal)
+        return self._transports[terminal]
+
+    def executor(self, terminal, name):
+        if (terminal, name) not in self._executors:
+            self._executors[(terminal, name)] = self.executor_factory(terminal, name)
+        return self._executors[(terminal, name)]
+
+    def reader(self, pcfg):
+        """Any identity with a token can read; prefer the project's own terminals."""
+        order = list(dict.fromkeys(list(pcfg["routing"]["by_assignee"].values()) + list(self.cfg.terminals)))
+        for term in order:
+            try:
+                return self.client(term)
+            except Exception:
+                continue
+        raise RuntimeError("no Plane token available for any terminal")
+
+    # ---------------------------------------------------------------- tick
+    def tick(self, project=None, only_issue=None):
+        with self.store.lock():
+            self._reset_caches()
+            data = self.store.load()
+            errors = []
+            for name in ([project] if project else list(self.cfg.projects)):
+                try:
+                    self.tick_project(data, name, only_issue)
+                except Exception as exc:
+                    errors.append("%s: %r" % (name, exc))
+                    self.log("%s tick error: %r" % (name, exc))
+                finally:
+                    self.store.save(data)
+            self.store.heartbeat({"errors": errors, "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")]})
+            return data
+
+    def loop(self, interval):
+        from .store import Busy
+        while True:
+            try:
+                self.tick()
+            except Busy as exc:
+                self.log("skip: %s" % exc)
+            except Exception as exc:
+                self.log("tick error: %r" % (exc,))
+            time.sleep(interval)
+
+    def tick_project(self, data, name, only_issue=None):
+        pcfg = self.cfg.project(name)
+        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"):
+                try:
+                    self.poll(ctx, data, iid, rec)
+                except Exception as exc:
+                    self.log("%s poll error: %r" % (ctx.key(rec.get("seq")), exc))
+        self.pick(ctx, data, pstate, only_issue)
+
+    # ---------------------------------------------------------------- (a) poll / finish
+    def poll(self, ctx, data, iid, rec):
+        fl = rec["inflight"]
+        ex = self.executor(fl["terminal"], fl["executor"])
+        if ex.poll(fl["handle"]) == "running":
+            if self.now() < fl["deadline"]:
+                return
+            self.log("%s run %s exceeded its budget; killing" % (ctx.key(rec["seq"]), fl["run_id"]))
+            try:
+                ex.kill(fl["handle"])
+            except Exception as exc:
+                self.log("kill %s failed: %r" % (fl["run_id"], exc))
+            hours = (fl["deadline"] - fl["started"]) / 3600
+            result = ExecResult("", None, fl["handle"].get("thread_id"), [fl["handle"].get("run_dir", "")],
+                                "超过时限 %.1f h 未结束，已结束任务进程。" % hours)
+            return self.finish(ctx, data, iid, rec, result, timed_out=True)
+        result = ex.collect(fl["handle"])
+        if hasattr(ex, "forget"):
+            ex.forget(fl["handle"])
+        self.finish(ctx, data, iid, rec, result)
+
+    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"]
+        term, exname = fl["terminal"], fl["executor"]
+        client = self.client(term)
+        key = ctx.key(rec["seq"])
+        issue = ctx.issues.get(iid)
+        still_ours = issue is None or issue.get("state") == ctx.state_id("running")
+        message = (result.last_message or "").strip()
+        tag = classify_first_line(message)
+        ok = not timed_out and message and (tag or result.exit_code == 0)
+        if not ok:
+            return self.fail(ctx, data, iid, rec, client, result, timed_out, still_ours)
+        if tag is None:
+            message, tag = "[review]\n（未按协议标注首行，按 review 处理）\n\n" + message, "review"
+        body = message
+        limit = self.cfg.agentd.get("comment_max_chars", 6000)
+        if len(body) > limit:
+            where = result.log_paths[-1] if result.log_paths else fl["handle"].get("run_dir", "")
+            body = body[:limit] + "\n…（已截断，全文见 %s 上 %s）" % (term, where)
+        header = "[%s] 第 %d 轮结束（exit=%s，run=%s，终端 %s，日志 %s）" % (
+            exname, fl["round"], result.exit_code, fl["run_id"], term, fl["handle"].get("run_dir", ""))
+        self.post(ctx, client, iid, rec, header, body)
+        self.close_run(rec, result)
+        rec["tags"].append(tag)
+        rec.update(last_message=message[:4000], last_tag=tag, retries=0)
+        if 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)
+        self.log("%s run %s finished exit=%s tag=%s" % (key, fl["run_id"], result.exit_code, tag))
+
+    def close_run(self, rec, result):
+        fl = rec.pop("inflight")
+        rec["last_exit"] = result.exit_code
+        if result.thread_id and self.executor(fl["terminal"], fl["executor"]).resume:
+            rec["thread_id"], rec["thread_on"] = result.thread_id, [fl["terminal"], fl["executor"]]
+        return fl
+
+    def guarded(self, key, fn, *args):
+        try:
+            fn(*args)
+        except Exception as exc:
+            self.log("%s %s failed: %r" % (key, fn.__name__, exc))
+
+    def fail(self, ctx, data, iid, rec, client, result, timed_out, still_ours):
+        fl, pcfg = rec["inflight"], ctx.cfg
+        key = ctx.key(rec["seq"])
+        retry_cfg = pcfg.get("retry", {})
+        max_n = retry_cfg.get("transient_max", 3)
+        backoff = retry_cfg.get("backoff_minutes", [2, 5, 10])
+        kind, transient = classify_failure(result, timed_out, retry_cfg)
+        n = rec.get("retries", 0)
+        who = pcfg.get("answerer") or pcfg["reviewer"]
+        if not still_ours:
+            plan, action = "none", "工作项已被移出 %s，不重试。" % pcfg["states"]["running"]
+        elif fl.get("resumed") and kind in ("crash", "logic") and not transient:
+            plan, action = "rebuild", "续线程失败，下一轮在新线程重跑（附交接摘要）。"
+        elif transient and n < max_n:
+            delay = backoff[min(n, len(backoff) - 1)]
+            n += 1
+            plan, action = "retry", "临时性错误，%g 分钟后自动重试（续原线程）。" % delay
+        else:
+            plan, action = "waiting", "不再自动重试，转 %s 并指派给 %s。" % (pcfg["states"]["waiting"], who)
+            if kind == "env" and fl.get("group"):
+                action += "串行组 %s 暂停派发，直到本项离开 %s。" % (fl["group"], pcfg["states"]["waiting"])
+        note = "%s 第 %d 轮失败：类型=%s%s，run=%s，exit=%s，重试 %d/%d。%s\n日志：%s" % (
+            WORKER, fl["round"], kind, "（临时性）" if transient else "", fl["run_id"], result.exit_code, n, max_n,
+            action, ", ".join(result.log_paths) or fl["handle"].get("run_dir", ""))
+        frame = result.tail or "（无输出）"
+        if result.last_message:
+            frame += "\n--- 最后消息\n" + result.last_message[:1500]
+        self.post(ctx, client, iid, rec, note, frame)
+        self.close_run(rec, result)
+        rec["last_tag"] = "failed:" + kind
+        if plan == "rebuild":
+            rec["rebuild"] = "续线程失败（%s）" % kind
+            rec.pop("thread_id", None)
+            rec.pop("thread_on", None)
+            rec["retry"] = {"at": self.now(), "reason": "rebuild"}
+        elif plan == "retry":
+            rec["retries"] = n
+            rec["retry"] = {"at": self.now() + delay * 60, "reason": kind}
+        elif plan == "waiting":
+            rec["retries"] = 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)
+        self.log("%s run %s failed kind=%s transient=%s -> %s" % (key, fl["run_id"], kind, transient, plan))
+
+    def transition(self, ctx, client, iid, tag, terminal_user):
+        pcfg = ctx.cfg
+        if tag == "progress":
+            self.move(ctx, client, iid, "todo", terminal_user)
+        elif tag in ("question", "blocked"):
+            self.move(ctx, client, iid, "waiting", pcfg.get("answerer") or pcfg["reviewer"])
+        else:
+            self.move(ctx, client, iid, "review", pcfg["reviewer"])
+
+    def move(self, ctx, client, iid, state_key, user):
+        state, users = ctx.state_id(state_key), [ctx.members[user]]
+        client.set_state(ctx.pid, iid, state, users)
+        if iid in ctx.issues:  # keep this tick's view consistent
+            ctx.issues[iid].update(state=state, assignees=users)
+
+    def post(self, ctx, client, iid, rec, text, pre=None):
+        c = client.post_comment(ctx.pid, iid, text_to_html(text, pre))
+        if isinstance(c, dict) and c.get("id"):
+            rec.setdefault("seen_comments", []).append(c["id"])
+
+    # ---------------------------------------------------------------- (b) pick / start
+    def pick(self, ctx, data, pstate, only_issue=None):
+        load, groups = Counter(), set()
+        for p in data["projects"].values():
+            for r in p["issues"].values():
+                fl = r.get("inflight")
+                if fl:
+                    load[fl["terminal"]] += 1
+                    if fl.get("group"):
+                        groups.add(fl["group"])
+        for g, info in list(data["paused_groups"].items()):
+            if info["project"] == ctx.name:
+                issue = ctx.issues.get(info["issue"])
+                if issue is None or issue.get("state") != ctx.state_id("waiting"):
+                    del data["paused_groups"][g]
+                    self.log("serialize group %s resumed (%s left Waiting)" % (g, info.get("key")))
+        groups |= set(data["paused_groups"])
+        todo, running, now = ctx.state_id("todo"), ctx.state_id("running"), self.now()
+        cands = []
+        for iid, issue in ctx.issues.items():
+            if only_issue and iid != only_issue:
+                continue
+            rec = pstate["issues"].get(iid, {})
+            if rec.get("inflight"):
+                continue
+            if issue.get("state") == todo:
+                cands.append(issue)
+            elif rec.get("retry"):
+                if issue.get("state") != running:
+                    rec.pop("retry")
+                elif rec["retry"]["at"] <= now:
+                    cands.append(issue)
+        for issue in sorted(cands, key=lambda i: i.get("sequence_id", 0)):
+            route = self.route(ctx, issue)
+            if isinstance(route, str):
+                if route:
+                    self.log("%s not dispatched: %s" % (ctx.key(issue.get("sequence_id")), route))
+                continue
+            term, group = route["terminal"], route["group"]
+            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)
+            except Exception as exc:
+                self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
+                continue
+            load[term] += 1
+            if group:
+                groups.add(group)
+
+    def terminal_ready(self, term, our_load):
+        """Optional guard against processes we did not start (e.g. a manual codex session)."""
+        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)
+                    ready = False
+                elif count > 0 and our_load == 0:
+                    self.log("%s busy (%d %s running that agentd did not start)" % (term, count, name))
+                    ready = False
+            self._ready[term] = ready
+        return self._ready[term]
+
+    def route(self, ctx, issue):
+        """Work item -> {terminal, executor, model, effort, tier, group, timeout_hours, script}; a str = not routable."""
+        pcfg, routing = ctx.cfg, ctx.cfg["routing"]
+        users = [ctx.member_names.get(a, a) for a in issue.get("assignees") or []]
+        term = next((routing["by_assignee"][u] for u in users if u in routing["by_assignee"]), None)
+        if term is None:
+            return ""  # not assigned to a terminal: silently ignored
+        tcfg = self.cfg.terminals[term]
+        r = {"terminal": term, "executor": routing.get("default_executor", {}).get(term) or tcfg["executors"][0],
+             "model": None, "effort": None, "tier": None, "group": None, "script": None,
+             "timeout_hours": pcfg.get("timeout_hours", 4)}
+        caps = []
+        for label in [ctx.label_names.get(l, l) for l in issue.get("labels") or []]:
+            spec = routing.get("by_label", {}).get(label)
+            if spec:
+                for k, v in spec.items():
+                    r["group" if k == "serialize_group" else k] = v
+            elif label.startswith("exec:") and label[5:] in self.cfg.executors:
+                r["executor"] = label[5:]
+            elif label.startswith("model:"):
+                r["model"] = label[6:]
+            elif label.startswith("effort:"):
+                r["effort"] = label[7:]
+            elif label.startswith("cap:"):
+                caps.append(label[4:])
+            elif parse_budget(label) is not None:
+                r["timeout_hours"] = parse_budget(label)
+        missing = [c for c in caps if c not in tcfg.get("capabilities", [])]
+        if missing:
+            return "terminal %s lacks capabilities %s" % (term, missing)
+        if r["executor"] not in tcfg["executors"]:
+            return "executor %s not available on %s" % (r["executor"], term)
+        ecfg = self.cfg.executors[r["executor"]]
+        r["model"] = r["model"] or (ecfg.get("models") or [None])[0]
+        r["effort"] = r["effort"] or ecfg.get("default_effort")
+        return r
+
+    def protocol(self, pcfg):
+        path = pcfg.get("protocol_file")
+        if not path:
+            return ""
+        p = Path(os.path.expanduser(path))
+        p = p if p.is_absolute() else ROOT / p
+        if not p.exists():
+            self.log("protocol file %s missing" % p)
+            return ""
+        return p.read_text(encoding="utf-8")
+
+    def start(self, ctx, pstate, issue, route):
+        iid, seq = issue["id"], issue["sequence_id"]
+        key = ctx.key(seq)
+        rec = pstate["issues"].setdefault(iid, new_rec(seq))
+        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
+        k = rec["runs"] + 1
+        run_id = "%s-%d-%d" % (ctx.prefix, seq, k)
+        thread, reason = rec.get("thread_id"), rec.pop("rebuild", None)
+        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:
+            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
+        if is_retry and thread:
+            retry_note = "（上一轮因临时性错误中断，请在同一线程从中断处继续。）"
+        comments = ctx.reader.list_comments(ctx.pid, iid)
+        seen = set(rec.get("seen_comments", []))
+        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)
+        now = self.now()
+        rec.update(runs=k, thread_rounds=rec.get("thread_rounds", 0) + 1 if thread else 1, 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)
+        mode = ("续线程 %s" % thread[:12]) if thread else ("新线程（重建：%s）" % reason if reason 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)
+        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))
+        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))
diff --git a/agentd/store.py b/agentd/store.py
new file mode 100644
index 0000000..2772df8
--- /dev/null
+++ b/agentd/store.py
@@ -0,0 +1,61 @@
+"""JSON state store under state/: one file for all projects, an exclusive tick lock, heartbeat and log.
+
+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 {...}"""
+import contextlib
+import fcntl
+import json
+import os
+import time
+from pathlib import Path
+
+
+class Busy(Exception):
+    pass
+
+
+class Store:
+    def __init__(self, state_dir, echo=True):
+        self.dir = Path(state_dir)
+        self.echo = echo
+        self.path = self.dir / "agentd-state.json"
+
+    def load(self):
+        if self.path.exists():
+            data = json.loads(self.path.read_text(encoding="utf-8"))
+        else:
+            data = {}
+        data.setdefault("projects", {})
+        data.setdefault("paused_groups", {})
+        return data
+
+    def save(self, data):
+        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)
+
+    @contextlib.contextmanager
+    def lock(self):
+        self.dir.mkdir(parents=True, exist_ok=True)
+        with open(self.dir / "agentd.lock", "w") as fh:
+            try:
+                fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            except BlockingIOError:
+                raise Busy("another tick holds %s" % (self.dir / "agentd.lock")) from None
+            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")
+
+    def log(self, msg):
+        self.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:
+            fh.write(line + "\n")
diff --git a/cli/agentd b/cli/agentd
new file mode 100755
index 0000000..948ee48
--- /dev/null
+++ b/cli/agentd
@@ -0,0 +1,9 @@
+#!/usr/bin/env python3
+"""Entry point: agentd tick|new|show|comment|state|assign|smoke|check-config (see agentd/cli.py)."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from agentd.cli import main  # noqa: E402
+
+main()
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/fakes.py b/tests/fakes.py
new file mode 100644
index 0000000..b72b9fe
--- /dev/null
+++ b/tests/fakes.py
@@ -0,0 +1,129 @@
+"""In-memory fakes for Plane and executors (no network, no processes)."""
+import copy
+import tempfile
+
+from agentd.config import Config
+from agentd.core import Agentd
+from agentd.executors import ExecResult
+from agentd.store import Store
+
+STATES = {"Backlog": "s-bl", "Todo": "s-todo", "In Progress": "s-run", "Waiting": "s-wait", "Review": "s-rev", "Done": "s-done"}
+MEMBERS = {"window10": "u-w10", "oregon": "u-or", "kappiee": "u-k"}
+LABELS = {"model:astra": "l-astra", "needs-game": "l-game", "exec:claude": "l-claude", "budget:30m": "l-b30", "cap:gpu": "l-gpu"}
+
+
+def config_data():
+    return {
+        "plane": {"base_url": "https://plane.example", "workspace": "ws"},
+        "agentd": {"max_rounds_per_thread": 3},
+        "terminals": {
+            "win10": {"transport": "warpgate-ssh", "target": "win10", "os": "windows", "launch": "schtasks", "repo": "C:\\repo",
+                      "plane_user": "window10", "token_key": "TOK_W10", "concurrency": 1, "capabilities": ["game"],
+                      "executors": ["codex"]},
+            "oregon": {"transport": "local", "os": "linux", "launch": "subprocess", "repo": "/repo", "plane_user": "oregon",
+                       "token_key": "TOK_OR", "concurrency": 2, "executors": ["claude-code", "shell"]},
+        },
+        "executors": {
+            "codex": {"run": "codex exec -", "resume": "codex exec resume {thread} -", "models": ["sol", "astra"], "default_effort": "medium"},
+            "claude-code": {"run": "claude -p", "resume": "claude -p --resume {thread}", "models": ["sonnet"]},
+            "shell": {"run": "{script}", "resume": False},
+        },
+        "projects": {"GWENT": {
+            "id": "proj", "key_prefix": "GW", "reviewer": "oregon", "answerer": "kappiee",
+            "states": {"todo": "Todo", "running": "In Progress", "waiting": "Waiting", "review": "Review", "done": "Done"},
+            "routing": {"by_assignee": {"window10": "win10", "oregon": "oregon"},
+                        "by_label": {"model:astra": {"model": "astra", "effort": "high"}, "needs-game": {"serialize_group": "game"},
+                                     "exec:claude": {"executor": "claude-code"}},
+                        "default_executor": {"win10": "codex", "oregon": "claude-code"}},
+            "retry": {"transient_max": 2, "backoff_minutes": [2, 5]},
+            "timeout_hours": 4,
+        }},
+    }
+
+
+class FakePlane:
+    def __init__(self):
+        self.issues, self.comments, self.patches, self.n = {}, {}, [], 0
+
+    def add_issue(self, seq, state="Todo", assignee="window10", labels=(), desc="<p>do <b>it</b></p>"):
+        iid = "i%d" % seq
+        self.issues[iid] = {"id": iid, "sequence_id": seq, "name": "T-%d" % seq, "description_html": desc,
+                            "state": STATES[state], "assignees": [MEMBERS[assignee]], "labels": [LABELS[l] for l in labels]}
+        self.comments[iid] = []
+        return iid
+
+    def add_comment(self, iid, html):
+        self.n += 1
+        c = {"id": "c%d" % self.n, "created_at": "2026-10-21T00:00:%02d" % self.n, "comment_html": html}
+        self.comments[iid].append(c)
+        return c
+
+    def texts(self, iid):
+        from agentd.plane import html_to_text
+        return [html_to_text(c["comment_html"]) for c in self.comments[iid]]
+
+    def state_of(self, iid):
+        return {v: k for k, v in STATES.items()}[self.issues[iid]["state"]]
+
+    def assignee_of(self, iid):
+        return {v: k for k, v in MEMBERS.items()}[self.issues[iid]["assignees"][0]]
+
+    # client API used by core
+    def states(self, pid): return dict(STATES)
+    def labels(self, pid): return dict(LABELS)
+    def members(self): return dict(MEMBERS)
+    def list_issues(self, pid): return copy.deepcopy(list(self.issues.values()))
+    def list_comments(self, pid, iid): return copy.deepcopy(self.comments[iid])
+    def post_comment(self, pid, iid, html): return copy.deepcopy(self.add_comment(iid, html))
+
+    def set_state(self, pid, iid, state, assignees=None):
+        self.patches.append((iid, state, assignees))
+        self.issues[iid]["state"] = state
+        if assignees is not None:
+            self.issues[iid]["assignees"] = assignees
+
+
+class FakeExecutor:
+    def __init__(self, resume=True):
+        self.resume = resume
+        self.started, self.killed, self.done = [], [], {}
+
+    def start(self, run):
+        self.started.append(run)
+        return {"run_dir": "/runs/" + run.run_id, "run_id": run.run_id, "thread_id": run.thread_id}
+
+    def poll(self, handle):
+        return "finished" if handle["run_id"] in self.done else "running"
+
+    def collect(self, handle):
+        return self.done[handle["run_id"]]
+
+    def kill(self, handle):
+        self.killed.append(handle["run_id"])
+
+    def finish(self, message, exit_code=0, thread="th-1", tail=""):
+        run = self.started[-1]
+        self.done[run.run_id] = ExecResult(message, exit_code, thread, ["/runs/%s/last.md" % run.run_id], tail)
+
+
+class Harness:
+    def __init__(self, data=None):
+        self.cfg = Config(data or config_data())
+        self.plane = FakePlane()
+        self.execs = {}
+        self.clock = [1_000_000.0]
+        self.tmp = tempfile.TemporaryDirectory()
+        self.d = Agentd(self.cfg, store=Store(self.tmp.name, echo=False), plane_factory=lambda term: self.plane,
+                        executor_factory=self.executor, now=lambda: self.clock[0])
+
+    def executor(self, term, name):
+        return self.execs.setdefault((term, name), FakeExecutor(resume=name != "shell"))
+
+    def tick(self):
+        return self.d.tick()
+
+    def rec(self, iid):
+        return self.d.store.load()["projects"]["GWENT"]["issues"][iid]
+
+    def close(self):
+        self.tmp.cleanup()
diff --git a/tests/test_config.py b/tests/test_config.py
new file mode 100644
index 0000000..9c15cc6
--- /dev/null
+++ b/tests/test_config.py
@@ -0,0 +1,71 @@
+import copy
+import json
+import os
+import unittest
+from unittest import mock
+
+from agentd import ROOT, config
+from tests.fakes import config_data
+
+SCHEMA = json.loads((ROOT / "config" / "schema.json").read_text())
+
+
+class ConfigTest(unittest.TestCase):
+    def test_example_config_is_valid(self):
+        cfg = config.load(ROOT / "config" / "agentd.example.yaml")
+        self.assertIn("GWENT", cfg.projects)
+        self.assertEqual(cfg.token_env("win10"), "PLANE_WINDOW10_API_TOKEN")
+
+    def test_fixture_is_valid_with_both_validators(self):
+        self.assertEqual(config.validate(config_data()), [])
+        self.assertEqual(config.mini_validate(config_data(), SCHEMA), [])
+
+    def check_both(self, data, fragment):
+        for errs in (config.validate(data), config.mini_validate(data, SCHEMA) or config.semantic_errors(data)):
+            self.assertTrue(errs, "expected an error mentioning %r" % fragment)
+            self.assertTrue(any(fragment in e for e in errs), errs)
+
+    def test_missing_required(self):
+        d = config_data()
+        del d["terminals"]["win10"]["token_key"]
+        self.check_both(d, "token_key")
+
+    def test_bad_enum(self):
+        d = config_data()
+        d["terminals"]["oregon"]["transport"] = "telnet"
+        self.check_both(d, "telnet")
+
+    def test_unknown_key(self):
+        d = config_data()
+        d["projects"]["GWENT"]["routing"]["by_label"]["needs-game"]["serialise"] = "x"
+        self.check_both(d, "serialise")
+
+    def test_wrong_type(self):
+        d = config_data()
+        d["terminals"]["oregon"]["concurrency"] = "two"
+        self.check_both(d, "concurrency")
+
+    def test_semantic_routing_to_unknown_terminal(self):
+        d = config_data()
+        d["projects"]["GWENT"]["routing"]["by_assignee"]["mac"] = "mac"
+        self.assertTrue(any("unknown terminal 'mac'" in e for e in config.validate(d)))
+
+    def test_semantic_default_executor_not_on_terminal(self):
+        d = config_data()
+        d["projects"]["GWENT"]["routing"]["default_executor"]["win10"] = "shell"
+        self.assertTrue(any("not in that terminal's executors" in e for e in config.validate(d)))
+
+    def test_semantic_unknown_executor_on_terminal(self):
+        d = copy.deepcopy(config_data())
+        d["terminals"]["oregon"]["executors"].append("gemini")
+        self.assertTrue(any("unknown executor 'gemini'" in e for e in config.validate(d)))
+
+    def test_tokens_come_from_env_named_by_key(self):
+        cfg = config.Config(config_data())
+        with mock.patch.dict(os.environ, {"TOK_W10": "secret"}, clear=False):
+            self.assertEqual(cfg.token("win10"), "secret")
+        self.assertEqual(cfg.secret_env_args(), ["--secret-env TOK_OR=TOK_OR", "--secret-env TOK_W10=TOK_W10"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_core.py b/tests/test_core.py
new file mode 100644
index 0000000..f7262dd
--- /dev/null
+++ b/tests/test_core.py
@@ -0,0 +1,223 @@
+import unittest
+
+from tests.fakes import Harness, config_data
+
+
+class CoreTest(unittest.TestCase):
+    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_pick_starts_lowest_todo_and_marks_in_progress(self):
+        a, b = self.p.add_issue(7), self.p.add_issue(3)
+        self.p.add_issue(1, state="Backlog")
+        self.p.add_issue(2, assignee="kappiee")  # not routed
+        self.h.tick()
+        runs = self.codex().started
+        self.assertEqual([r.run_id for r in runs], ["GW-3-1"])  # concurrency 1 on win10
+        self.assertEqual(self.p.state_of(b), "In Progress")
+        self.assertEqual(self.p.state_of(a), "Todo")
+        self.assertTrue(self.p.texts(b)[-1].startswith("[worker] 第 1 轮开始：codex / sol / medium @ win10，新线程"))
+        self.assertIsNone(runs[0].thread_id)
+        self.assertIn("do it", runs[0].prompt)
+
+    def test_review_question_progress_transitions(self):
+        for tag, state, who in (("review", "Review", "oregon"), ("question", "Waiting", "kappiee"),
+                                ("blocked", "Waiting", "kappiee"), ("progress", "Todo", "window10")):
+            with self.subTest(tag=tag):
+                h = Harness()
+                iid = h.plane.add_issue(1)
+                h.tick()
+                ex = h.execs[("win10", "codex")]
+                ex.finish("[%s] text" % tag)
+                h.tick()
+                if tag == "progress":  # picked again in the same tick, resuming the thread
+                    self.assertEqual(len(ex.started), 2)
+                    self.assertEqual(ex.started[1].thread_id, "th-1")
+                    self.assertEqual(h.plane.state_of(iid), "In Progress")
+                    self.assertIn(("i1", "s-todo", ["u-w10"]), h.plane.patches)
+                else:
+                    self.assertEqual((h.plane.state_of(iid), h.plane.assignee_of(iid)), (state, who))
+                texts = h.plane.texts(iid)
+                result = [t for t in texts if t.startswith("[codex]")][0]
+                self.assertIn("第 1 轮结束（exit=0，run=GW-1-1，终端 win10", result)
+                self.assertIn("[%s] text" % tag, result)
+                self.assertEqual(h.rec(iid)["tags"], [tag])
+                h.close()
+
+    def test_untagged_message_becomes_review(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("all done")
+        self.h.tick()
+        self.assertEqual(self.p.state_of(iid), "Review")
+        self.assertIn("未按协议标注首行", self.p.texts(iid)[-1])
+
+    def test_resume_sends_only_new_comments(self):
+        iid = self.p.add_issue(1)
+        self.p.add_comment(iid, "<p>old context</p>")
+        self.h.tick()
+        self.codex().finish("[question] A or B?")
+        self.h.tick()
+        self.p.add_comment(iid, "<p>B please</p>")
+        self.p.issues[iid].update(state="s-todo", assignees=["u-w10"])
+        self.h.tick()
+        run = self.codex().started[-1]
+        self.assertEqual(run.thread_id, "th-1")
+        self.assertIn("B please", run.prompt)
+        self.assertNotIn("old context", run.prompt)
+        self.assertNotIn("A or B?", run.prompt)
+        self.assertIn("续线程 th-1", self.p.texts(iid)[-1])
+
+    def test_transient_failure_retries_with_backoff_then_waits(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("", exit_code=1, tail="Connection reset by peer")
+        self.h.tick()
+        note = self.p.texts(iid)[-1]
+        self.assertIn("类型=crash（临时性）", note)
+        self.assertIn("重试 1/2", note)
+        self.assertEqual(self.p.state_of(iid), "In Progress")
+        self.assertEqual(len(self.codex().started), 1)          # backoff 2 min not elapsed
+        self.h.clock[0] += 121
+        self.h.tick()
+        self.assertEqual(len(self.codex().started), 2)
+        self.assertEqual(self.codex().started[1].thread_id, "th-1")  # resumes the original thread
+        self.assertIn("临时性错误中断", self.codex().started[1].prompt)
+        self.codex().finish("", exit_code=1, tail="HTTP 503 Service Unavailable")
+        self.h.tick()
+        self.h.clock[0] += 301
+        self.h.tick()
+        self.assertEqual(len(self.codex().started), 3)
+        self.codex().finish("", exit_code=1, tail="banner exchange failed")
+        self.h.tick()
+        self.assertEqual((self.p.state_of(iid), self.p.assignee_of(iid)), ("Waiting", "kappiee"))
+        self.assertIn("不再自动重试", self.p.texts(iid)[-1])
+        self.assertEqual(self.h.rec(iid)["retries"], 0)
+
+    def test_logic_failure_on_fresh_thread_goes_waiting(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("", exit_code=2, thread=None, tail="panic: bad config")
+        self.h.tick()
+        self.assertEqual(self.p.state_of(iid), "Waiting")
+        self.assertIn("类型=crash，run=GW-1-1，exit=2，重试 0/2", self.p.texts(iid)[-1])
+        self.assertIn("panic: bad config", self.p.texts(iid)[-1])
+
+    def test_resume_failure_rebuilds_thread_with_handoff(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("[progress] half way")
+        self.h.tick()  # finishes round 1, starts round 2 resuming th-1
+        self.assertEqual(self.codex().started[1].thread_id, "th-1")
+        self.codex().finish("", exit_code=1, thread=None, tail="thread not found")
+        self.h.tick()  # round 2 fails -> rebuild immediately
+        run = self.codex().started[2]
+        self.assertIsNone(run.thread_id)
+        self.assertIn("## 交接摘要", run.prompt)
+        self.assertIn("续线程失败", run.prompt)
+        self.assertIn("[progress] half way", run.prompt)
+        self.assertIn("新线程（重建：续线程失败（crash））", self.p.texts(iid)[-1])
+
+    def test_thread_rebuilt_after_max_rounds(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        for _ in range(3):
+            self.codex().finish("[progress] more")
+            self.h.tick()
+        runs = self.codex().started
+        self.assertEqual([r.thread_id for r in runs], [None, "th-1", "th-1", None])
+        self.assertIn("线程已续 3 轮（上限 3）", runs[3].prompt)
+        self.assertIn("progress, progress, progress", runs[3].prompt)
+        self.assertEqual(self.h.rec(iid)["thread_rounds"], 1)
+
+    def test_timeout_kills_and_waits(self):
+        iid = self.p.add_issue(1, labels=["budget:30m"])
+        self.h.tick()
+        self.h.clock[0] += 29 * 60
+        self.h.tick()
+        self.assertEqual(self.codex().killed, [])
+        self.h.clock[0] += 2 * 60
+        self.h.tick()
+        self.assertEqual(self.codex().killed, ["GW-1-1"])
+        self.assertEqual(self.p.state_of(iid), "Waiting")
+        self.assertIn("类型=timeout", self.p.texts(iid)[-1])
+        self.assertIn("0.5 h", self.p.texts(iid)[-1])
+
+    def test_serialize_group_and_concurrency(self):
+        a = self.p.add_issue(1, assignee="oregon", labels=["needs-game"])
+        b = self.p.add_issue(2, assignee="oregon", labels=["needs-game"])
+        c = self.p.add_issue(3, assignee="oregon")
+        d = self.p.add_issue(4, assignee="oregon")
+        self.h.tick()
+        started = [r.run_id for r in self.h.execs[("oregon", "claude-code")].started]
+        self.assertEqual(started, ["GW-1-1", "GW-3-1"])  # 2 slots; GW-2 waits for the game group
+        self.assertEqual([self.p.state_of(x) for x in (a, b, c, d)], ["In Progress", "Todo", "In Progress", "Todo"])
+
+    def test_label_routing(self):
+        self.p.add_issue(1, labels=["model:astra"])
+        self.p.add_issue(2, assignee="oregon", labels=["cap:gpu"])  # oregon lacks the capability
+        self.h.tick()
+        run = self.codex().started[0]
+        self.assertEqual((run.model, run.effort), ("astra", "high"))
+        self.assertNotIn(("oregon", "claude-code"), self.h.execs)
+
+    def test_human_moving_item_during_run_is_respected(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.p.issues[iid]["state"] = "s-done"
+        self.codex().finish("[review] ok")
+        self.h.tick()
+        self.assertEqual(self.p.state_of(iid), "Done")
+        self.assertIn("[review] ok", self.p.texts(iid)[-1])
+
+    def test_post_failure_keeps_run_in_flight(self):
+        iid = self.p.add_issue(1)
+        self.h.tick()
+        self.codex().finish("[review] ok")
+        orig = self.p.post_comment
+        self.p.post_comment = lambda *a: (_ for _ in ()).throw(RuntimeError("plane down"))
+        self.h.tick()
+        self.assertIn("inflight", self.h.rec(iid))
+        self.p.post_comment = orig
+        self.h.tick()
+        self.assertNotIn("inflight", self.h.rec(iid))
+        self.assertEqual(self.p.state_of(iid), "Review")
+
+    def test_heartbeat_written(self):
+        self.p.add_issue(1)
+        self.h.tick()
+        import json, pathlib
+        hb = json.loads((pathlib.Path(self.h.tmp.name) / "heartbeat.json").read_text())
+        self.assertEqual(hb["inflight"][0]["run"], "GW-1-1")
+        self.assertEqual(hb["errors"], [])
+
+
+class EnvFailureTest(unittest.TestCase):
+    def test_env_failure_pauses_group_until_item_leaves_waiting(self):
+        data = config_data()
+        data["projects"]["GWENT"]["retry"]["env_patterns"] = ["foreground window lost"]
+        h = Harness(data)
+        a = h.plane.add_issue(1, labels=["needs-game"])
+        b = h.plane.add_issue(2, labels=["needs-game"])
+        h.tick()
+        ex = h.execs[("win10", "codex")]
+        ex.finish("", exit_code=1, tail="foreground window lost")
+        h.tick()
+        self.assertEqual(h.plane.state_of(a), "Waiting")
+        self.assertEqual(h.plane.state_of(b), "Todo")      # group paused
+        self.assertIn("暂停派发", h.plane.texts(a)[-1])
+        h.plane.issues[a]["state"] = "s-done"
+        h.tick()
+        self.assertEqual(h.plane.state_of(b), "In Progress")
+        h.close()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_local_runner.py b/tests/test_local_runner.py
new file mode 100644
index 0000000..1d81d4a
--- /dev/null
+++ b/tests/test_local_runner.py
@@ -0,0 +1,81 @@
+"""Real LocalTransport + runner executors against throwaway scripts (local processes only, no network)."""
+import os
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+from agentd.config import Config
+from agentd.executors import Run, make_executor
+from agentd.transports.local import LocalTransport
+from tests.fakes import config_data
+
+FAKE_CLAUDE = """#!/usr/bin/env bash
+prompt="$(cat)"
+echo "args: $*" >&2
+printf '{"type":"result","subtype":"success","is_error":false,"result":"[review] got %s","session_id":"sess-42"}' "${#prompt}"
+"""
+
+
+def wait(ex, handle, seconds=10):
+    end = time.time() + seconds
+    while time.time() < end:
+        if ex.poll(handle) == "finished":
+            return True
+        time.sleep(0.1)
+    return False
+
+
+class LocalRunnerTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        root = Path(self.tmp.name)
+        fake = root / "fake-claude"
+        fake.write_text(FAKE_CLAUDE)
+        fake.chmod(0o755)
+        data = config_data()
+        data["agentd"]["state_dir"] = str(root / "state")
+        data["terminals"]["oregon"]["bins"] = {"claude": str(fake)}
+        data["terminals"]["oregon"]["repo"] = str(root)
+        self.cfg = Config(data)
+        self.t = LocalTransport(self.cfg, "oregon")
+        self.root = root
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_shell_executor_round(self):
+        ex = make_executor(self.cfg, "oregon", "shell", self.t)
+        self.assertFalse(ex.resume)
+        run = Run("GW-1-1", str(self.root), "hello prompt", script="read line; echo \"[review] echo: $line\"")
+        h = ex.start(run)
+        self.assertTrue(wait(ex, h))
+        res = ex.collect(h)
+        self.assertEqual((res.last_message, res.exit_code, res.thread_id), ("[review] echo: hello prompt", 0, None))
+
+    def test_claude_executor_round_and_resume(self):
+        ex = make_executor(self.cfg, "oregon", "claude-code", self.t)
+        h = ex.start(Run("GW-2-1", str(self.root), "abc", model="sonnet"))
+        self.assertTrue(wait(ex, h))
+        res = ex.collect(h)
+        self.assertEqual((res.last_message, res.thread_id, res.exit_code), ("[review] got 3", "sess-42", 0))
+        h2 = ex.start(Run("GW-2-2", str(self.root), "abcd", model="sonnet", thread_id="sess-42"))
+        self.assertTrue(wait(ex, h2))
+        self.assertIn("--resume sess-42", self.t.read_text(os.path.join(h2["run_dir"], "stderr.txt")))
+
+    def test_kill_and_exit_code(self):
+        ex = make_executor(self.cfg, "oregon", "shell", self.t)
+        h = ex.start(Run("GW-3-1", str(self.root), "", script="sleep 30"))
+        time.sleep(0.5)
+        self.assertEqual(ex.poll(h), "running")
+        ex.kill(h)
+        time.sleep(0.3)
+        self.assertEqual(ex.poll(h), "running")  # killed runner never writes exit.txt; core handles it as timeout
+        h = ex.start(Run("GW-3-2", str(self.root), "", script="echo partial; exit 3"))
+        self.assertTrue(wait(ex, h))
+        res = ex.collect(h)
+        self.assertEqual((res.exit_code, res.last_message), (3, "partial"))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_prompt.py b/tests/test_prompt.py
new file mode 100644
index 0000000..ce24692
--- /dev/null
+++ b/tests/test_prompt.py
@@ -0,0 +1,51 @@
+import unittest
+
+from agentd.core import build_prompt, handoff_summary
+
+ISSUE = {"name": "T-1 do it", "description_html": "<p>goal &amp; scope</p>"}
+COMMENTS = [
+    {"id": "c1", "created_at": "t1", "comment_html": "<p>please also check X</p>"},
+    {"id": "c2", "created_at": "t2", "comment_html": "<p>[worker] 第 1 轮开始</p>"},
+    {"id": "c3", "created_at": "t3", "comment_html": "<p>[codex] 第 1 轮结束</p><pre>[question] A or B?</pre>"},
+    {"id": "c4", "created_at": "t4", "comment_html": "<p>B, please</p>"},
+]
+
+
+class PromptTest(unittest.TestCase):
+    def test_fresh_prompt_has_everything_but_worker_comments(self):
+        p = build_prompt("GW-1", ISSUE, COMMENTS, set(), False, protocol="PROTO", intro="INTRO")
+        self.assertTrue(p.startswith("INTRO"))
+        self.assertIn("## 协议\nPROTO", p)
+        self.assertIn("## 工作项 GW-1: T-1 do it\n\ngoal & scope", p)
+        self.assertIn("please also check X", p)
+        self.assertIn("[question] A or B?", p)
+        self.assertNotIn("[worker]", p)
+        self.assertTrue(p.rstrip().endswith("不要等待人工回复。"))
+
+    def test_resumed_prompt_only_new_comments(self):
+        p = build_prompt("GW-1", ISSUE, COMMENTS, {"c1", "c2", "c3"}, True, protocol="PROTO", intro="INTRO")
+        self.assertNotIn("PROTO", p)
+        self.assertNotIn("INTRO", p)
+        self.assertNotIn("please also check X", p)
+        self.assertIn("B, please", p)
+        self.assertIn("同一线程", p)
+
+    def test_resumed_without_new_comments_asks_to_continue(self):
+        p = build_prompt("GW-1", ISSUE, COMMENTS, {"c1", "c2", "c3", "c4"}, True)
+        self.assertIn("没有新评论", p)
+        p = build_prompt("GW-1", ISSUE, COMMENTS, {"c1", "c2", "c3", "c4"}, True, retry_note="RETRY")
+        self.assertIn("RETRY", p)
+        self.assertNotIn("没有新评论", p)
+
+    def test_handoff(self):
+        rec = {"runs": 3, "tags": ["progress", "progress", "question"], "last_message": "[question] A or B?"}
+        h = handoff_summary(rec, "线程已续 3 轮")
+        self.assertIn("3 轮", h)
+        self.assertIn("progress, progress, question", h)
+        self.assertIn("[question] A or B?", h)
+        p = build_prompt("GW-1", ISSUE, COMMENTS, set(), False, handoff=h)
+        self.assertIn("## 交接摘要\n" + h, p)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_text.py b/tests/test_text.py
new file mode 100644
index 0000000..f4cbffb
--- /dev/null
+++ b/tests/test_text.py
@@ -0,0 +1,79 @@
+import unittest
+
+from agentd.core import classify_failure, classify_first_line, parse_budget
+from agentd.executors import ExecResult
+from agentd.executors.claude_code import parse_output
+from agentd.plane import brief_to_html, html_to_text, text_to_html
+
+
+class FirstLineTest(unittest.TestCase):
+    def test_tags(self):
+        self.assertEqual(classify_first_line("[review] done\nmore"), "review")
+        self.assertEqual(classify_first_line("\n\n  [Question] which one?"), "question")
+        self.assertEqual(classify_first_line("**[blocked]** no device"), "blocked")
+        self.assertEqual(classify_first_line("[progress]"), "progress")
+
+    def test_untagged(self):
+        self.assertIsNone(classify_first_line("done, see [review] below"))
+        self.assertIsNone(classify_first_line("[done] finished"))
+        self.assertIsNone(classify_first_line(""))
+        self.assertIsNone(classify_first_line(None))
+
+
+class HtmlTest(unittest.TestCase):
+    def test_html_to_text(self):
+        self.assertEqual(html_to_text("<p>a &amp; b</p><p>c<br/>d</p>"), "a & b\nc\nd")
+        self.assertEqual(html_to_text("<ul><li>one</li><li>two</li></ul>"), "- one\n- two")
+        self.assertEqual(html_to_text("<pre>x &lt; y\nz</pre>"), "x < y\nz")
+        self.assertEqual(html_to_text(None), "")
+        self.assertEqual(html_to_text("<p>a</p>\n\n\n\n<p>b</p>"), "a\n\nb")
+
+    def test_text_to_html_escapes(self):
+        self.assertEqual(text_to_html("a<b>\nc", pre="<x>"), "<p>a&lt;b&gt;<br>c</p><pre>&lt;x&gt;</pre>")
+        self.assertEqual(html_to_text(text_to_html("[worker] hi & bye")), "[worker] hi & bye")
+
+    def test_brief(self):
+        self.assertEqual(brief_to_html("<p>x</p>"), "<p>x</p>")
+        self.assertEqual(brief_to_html("a < b"), "<pre>a &lt; b</pre>")
+
+
+class FailureTest(unittest.TestCase):
+    def kind(self, tail, exit_code=1, timed_out=False, cfg=None):
+        return classify_failure(ExecResult("", exit_code, None, [], tail), timed_out, cfg or {})
+
+    def test_transient_patterns(self):
+        for tail in ("request timed out", "Connection reset by peer", "connection refused",
+                     "kex_exchange_identification: banner exchange: Connection to UNKNOWN port 65535: invalid format",
+                     "HTTP 429 Too Many Requests", "unexpected status 502 Bad Gateway", "503 Service Unavailable",
+                     "stream disconnected before completion"):
+            self.assertEqual(self.kind(tail), ("crash", True), tail)
+
+    def test_not_transient(self):
+        self.assertEqual(self.kind("SyntaxError: invalid syntax"), ("crash", False))
+        self.assertEqual(self.kind("", exit_code=0), ("logic", False))
+
+    def test_timeout_and_env(self):
+        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_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))
+
+    def test_budget(self):
+        self.assertEqual(parse_budget("budget:2h"), 2)
+        self.assertEqual(parse_budget("budget:30m"), 0.5)
+        self.assertEqual(parse_budget("budget:1.5"), 1.5)
+        self.assertIsNone(parse_budget("budget:soon"))
+
+
+class ClaudeOutputTest(unittest.TestCase):
+    def test_parse(self):
+        obj = parse_output('{"type":"result","subtype":"success","is_error":false,"result":"[review] ok","session_id":"s1"}')
+        self.assertEqual((obj["result"], obj["session_id"]), ("[review] ok", "s1"))
+        self.assertEqual(parse_output('noise\n{"result":"x","session_id":"s2"}')["session_id"], "s2")
+        self.assertEqual(parse_output(""), {})
+
+
+if __name__ == "__main__":
+    unittest.main()