agentd 冒烟测试。不要读写任何文件，不要运行任何命令。
" + "你的最后一条消息必须恰好是下面这一行（不要加任何别的内容）：
[review] ok") + + +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
); - = 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="do it
"):
+ 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, "old context
")
+ self.h.tick()
+ self.codex().finish("[question] A or B?")
+ self.h.tick()
+ self.p.add_comment(iid, "B please
")
+ 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": "goal & scope
"}
+COMMENTS = [
+ {"id": "c1", "created_at": "t1", "comment_html": "please also check X
"},
+ {"id": "c2", "created_at": "t2", "comment_html": "[worker] 第 1 轮开始
"},
+ {"id": "c3", "created_at": "t3", "comment_html": "[codex] 第 1 轮结束
[question] A or B?
"},
+ {"id": "c4", "created_at": "t4", "comment_html": "B, please
"},
+]
+
+
+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("a & b
c
d
"), "a & b\nc\nd")
+ self.assertEqual(html_to_text("x < y\nz"), "x < y\nz") + self.assertEqual(html_to_text(None), "") + self.assertEqual(html_to_text("
a
\n\n\n\nb
"), "a\n\nb") + + def test_text_to_html_escapes(self): + self.assertEqual(text_to_html("a\nc", pre="a<b>
c
<x>") + self.assertEqual(html_to_text(text_to_html("[worker] hi & bye")), "[worker] hi & bye") + + def test_brief(self): + self.assertEqual(brief_to_html("
x
"), "x
") + self.assertEqual(brief_to_html("a < b"), "a < b") + + +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()