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

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

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

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