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

Double-start guard: before the launch call a `launching` marker (the planned run and its handle) is saved. If the
launch call fails or the process dies before the run is recorded, the next tick probes the run directory: launch
evidence (exit.txt / launch.pid / pid.txt) -> adopt the run; nothing -> relaunch with the same run id in a fresh
directory (`<run_id>-r2`, -r3 ...); probe unreachable -> keep the marker (it holds the terminal slot).
A terminal whose relay fails is skipped for the rest of the tick; other terminals carry on."""
import os
import re
import time
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 TransportError, 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)
FENCE = re.compile(r"^(```|~~~|-{3,}$|\*{3,}$)")


def classify_first_line(text):
    """Tag of the first meaningful line (`[review]` etc.), tolerating a BOM, markdown emphasis and a leading code
    fence / rule line; None if untagged. Trailing text after the tag is fine (`[progress] half way`)."""
    for line in (text or "").splitlines():
        first = line.strip().lstrip("\ufeff").strip()
        if not first or FENCE.match(first):
            continue
        m = re.match(r"\[(\w+)\]", first.lstrip("*_#> `").lower())
        return m.group(1) if m and m.group(1) in TAGS else None
    return None


def matches_any(text, patterns):
    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).
    env patterns look at the tail and the last message; transient patterns only at the tail (stderr / last event),
    so an agent's own prose ("tests timed out") never triggers a retry. logic (exit 0, no message) is never
    transient: an exit-0 run whose tail shows a transient error is a crash."""
    if timed_out:
        return "timeout", False
    if matches_any("%s\n%s" % (result.tail or "", result.last_message or ""), retry_cfg.get("env_patterns", [])):
        return "env", False
    transient = matches_any(result.tail or "", retry_cfg.get("transient_patterns", DEFAULT_TRANSIENT))
    if result.exit_code == 0 and not transient:
        return "logic", False
    return "crash", transient


def parse_budget(label):
    """`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, log_dir=cfg.log_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 = {}, {}, {}, {}
        self._down = set()   # terminals whose relay failed during this tick
        self._busy = {}      # terminal -> why terminal_ready() said no
        self._skipped = []   # Todo items not dispatched this tick, with the reason (heartbeat "skipped")

    def mark_down(self, term, exc):
        if term not in self._down:
            self.log("%s unreachable, skipping it for the rest of this tick: %s" % (term, exc))
        self._down.add(term)

    # ---------------------------------------------------------------- caches
    def client(self, terminal):
        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)
            if not only_issue and self.note_skips(data, [project] if project else list(self.cfg.projects)):
                self.store.save(data)
            self.store.heartbeat({"errors": errors, "down": sorted(self._down), "skipped": self._skipped, "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")],
                "launching": [{"project": p, "run": r["launching"]["run_id"], "terminal": r["launching"]["terminal"]}
                              for p, ps in data["projects"].items() for r in ps["issues"].values() if r.get("launching")]})
            return data

    def note_skips(self, data, projects):
        """Log a skip reason only when it is new for that item (the cron log stays quiet on repeat ticks).
        Persisted as data["skips"] {item: {project, terminal, reason}}; returns True when that changed."""
        prev = data.get("skips", {})
        now = {k: v for k, v in prev.items() if v.get("project") not in projects}
        for s in self._skipped:
            entry = {"project": s["project"], "terminal": s["terminal"], "reason": s["reason"]}
            if prev.get(s["item"]) != entry:
                self.log("%s not dispatched (%s): %s" % (s["item"], s["terminal"] or "-", s["reason"]))
            now[s["item"]] = entry
        changed = now != prev
        data["skips"] = now
        return changed

    def skip(self, ctx, issue, term, reason):
        self._skipped.append({"project": ctx.name, "item": ctx.key(issue.get("sequence_id")), "terminal": term,
                              "reason": reason})

    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()):
            run = rec.get("inflight") or rec.get("launching")
            if not run or run["terminal"] in self._down:
                continue
            try:
                if rec.get("inflight"):
                    self.poll(ctx, data, iid, rec)
                else:
                    self.resolve_launch(ctx, iid, rec)
            except TransportError as exc:
                self.mark_down(run["terminal"], exc)
            except Exception as exc:
                self.log("%s poll error: %r" % (ctx.key(rec.get("seq")), exc))
            self.store.save(data)  # persist each round's outcome (thread id, retry) before touching the next
        self.pick(ctx, data, pstate, only_issue)

    def resolve_launch(self, ctx, iid, rec):
        """A launch that was never confirmed (the launch call failed or agentd died mid-start): probe the run
        directory and adopt the run if it exists; otherwise relaunch in a fresh directory, never wiping one."""
        m = rec["launching"]
        key = ctx.key(rec["seq"])
        present = self.executor(m["terminal"], m["executor"]).probe(m["handle"])  # TransportError -> marker kept
        if present:
            self.log("%s run %s: unconfirmed launch found on %s (%s); adopting it" % (
                key, m["run_id"], m["terminal"], ",".join(present)))
            return self.launched(ctx, iid, rec, m, m["handle"], adopted=True)
        rec.pop("launching")
        issue = ctx.issues.get(iid)
        state = issue and issue.get("state")
        if state == ctx.state_id("todo") or (state == ctx.state_id("running") and rec.get("retry")):
            rec["relaunch"] = {"run_id": m["run_id"], "attempt": m.get("attempt", 1) + 1}
            self.log("%s run %s: nothing at %s; relaunching in %s-r%d" % (
                key, m["run_id"], m["handle"].get("run_dir"), m["run_id"], rec["relaunch"]["attempt"]))
        else:
            rec.pop("relaunch", None)
            self.log("%s run %s: nothing at %s and the item is no longer dispatchable; marker dropped" % (
                key, m["run_id"], m["handle"].get("run_dir")))

    # ---------------------------------------------------------------- (a) poll / finish
    def poll(self, ctx, data, iid, rec):
        fl = rec["inflight"]
        ex = self.executor(fl["terminal"], fl["executor"])
        if ex.poll(fl["handle"]) == "running":
            issue = ctx.issues.get(iid)
            if issue is not None and issue.get("state") != ctx.state_id("running"):
                return self.cancel(ctx, iid, rec, ex, issue)
            if self.now() < fl["deadline"]:
                return
            self.log("%s run %s exceeded its budget; killing" % (ctx.key(rec["seq"]), fl["run_id"]))
            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 cancel(self, ctx, iid, rec, ex, issue):
        """The item left In Progress while its run was still going (moved by a person): kill the run, say so, free
        the terminal slot and serialize group. The thread id is kept so a later round can resume it."""
        fl = rec["inflight"]
        key = ctx.key(rec["seq"])
        try:
            ex.kill(fl["handle"])
        except Exception as exc:  # keep the slot until the kill succeeds; a live run must not share the terminal
            self.log("%s cancel of run %s: kill failed, retrying next tick: %r" % (key, fl["run_id"], exc))
            return
        if hasattr(ex, "forget"):
            self.guarded(key, ex.forget, fl["handle"])
        thread = fl["handle"].get("thread_id")
        if not thread and ex.resume:
            try:  # a fresh run may already have announced its thread (codex thread.started)
                thread = ex.collect(fl["handle"]).thread_id
            except Exception:
                thread = None
        state = {v: k for k, v in ctx.states.items()}.get(issue.get("state"), issue.get("state"))
        try:
            self.post(ctx, self.client(fl["terminal"]), iid, rec,
                      "%s 运行已取消：工作项状态被改为 %s（run=%s）。" % (WORKER, state, fl["run_id"]))
        except Exception as exc:
            self.log("%s cancel comment failed: %r" % (key, exc))
        rec.pop("inflight")
        rec.pop("retry", None)
        if thread:
            rec["thread_id"], rec["thread_on"] = thread, [fl["terminal"], fl["executor"]]
        rec["last_tag"] = "cancelled"
        self.log("%s run %s cancelled: item moved to %s" % (key, fl["run_id"], state))

    def finish(self, ctx, data, iid, rec, result, timed_out=False):
        """Post first, then forget the in-flight run: if Plane fails, the next tick collects again."""
        fl = rec["inflight"]
        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)
        rec["progress_streak"] = rec.get("progress_streak", 0) + 1 if tag == "progress" else 0
        cap = ctx.cfg.get("max_progress_rounds", 12)
        if still_ours and tag == "progress" and rec["progress_streak"] >= cap:
            who = ctx.cfg.get("answerer") or ctx.cfg["reviewer"]
            rec["progress_streak"] = 0
            self.guarded(key, self.post, ctx, client, iid, rec, "%s 已连续 %d 轮 [progress]（上限 max_progress_rounds=%d），"
                         "暂停自动续跑，转 %s 并指派给 %s；移回 %s 即继续。" % (
                             WORKER, cap, cap, ctx.cfg["states"]["waiting"], who, ctx.cfg["states"]["todo"]))
            self.guarded(key, self.move, ctx, client, iid, "waiting", who)
        elif still_ours:
            self.guarded(key, self.transition, ctx, client, iid, tag, self.cfg.terminals[term]["plane_user"])
        else:
            self.log("%s was moved during the run; not changing its state" % key)
        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
            rec["progress_streak"] = 0
            if kind == "env" and fl.get("group"):
                data["paused_groups"][fl["group"]] = {"project": ctx.name, "issue": iid, "key": key}
            self.guarded(key, self.move, ctx, client, iid, "waiting", who)
        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)
        self.notify(ctx, iid, state_key, user)

    def notify(self, ctx, iid, event, user):
        """Notification hook (design §7 phase 3), called on every state move (event = todo/waiting/review/...).
        Config `notify: {kind: none}` is the only kind today: a no-op. An ntfy sender would go here."""
        if self.cfg.notify.get("kind", "none") == "none":
            return

    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, holders = Counter(), {}, {}   # groups / holders: group / terminal -> keys of the runs holding it
        for pname, p in data["projects"].items():
            prefix = self.cfg.projects.get(pname, {}).get("key_prefix", pname)
            for r in p["issues"].values():
                fl = r.get("inflight") or r.get("launching")   # an unconfirmed launch holds its slot
                if fl:
                    load[fl["terminal"]] += 1
                    holders.setdefault(fl["terminal"], []).append("%s-%s" % (prefix, r.get("seq")))
                    if fl.get("group"):
                        groups.setdefault(fl["group"], []).append("%s-%s" % (prefix, r.get("seq")))
        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")))
        for g, info in data["paused_groups"].items():
            groups.setdefault(g, []).append("%s in %s (paused)" % (info.get("key"), ctx.cfg["states"]["waiting"]))
        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") or rec.get("launching"):
                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):
                users = [ctx.member_names.get(a, a) for a in issue.get("assignees") or []]
                self.skip(ctx, issue, None, route or "no route for assignee %s" % (",".join(users) or "(none)"))
                continue
            term, group = route["terminal"], route["group"]
            cap = self.cfg.terminals[term].get("concurrency", 1)
            if term in self._down:
                self.skip(ctx, issue, term, "relay down")
                continue
            if load[term] >= cap:
                self.skip(ctx, issue, term, "concurrency %d/%d (%s)" % (load[term], cap, ", ".join(holders.get(term, [])) or "-"))
                continue
            if group and group in groups:
                self.skip(ctx, issue, term, "serialize_group %s held by %s" % (group, ", ".join(groups[group])))
                continue
            if not self.terminal_ready(term, load[term]):
                self.skip(ctx, issue, term, self._busy.get(term) or "relay down")
                continue
            try:
                self.start(ctx, data, pstate, issue, route)
            except Exception as exc:
                self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
                self.skip(ctx, issue, term, "start failed: %s" % (str(exc)[:200] or type(exc).__name__))
                if isinstance(exc, TransportError):
                    self.mark_down(term, exc)
                if not pstate["issues"].get(issue["id"], {}).get("launching"):
                    continue
                # the launch call failed but may have started the run: its marker keeps the slot until probed
            self.store.save(data)  # a launched run must be on disk before anything else can fail
            load[term] += 1
            holders.setdefault(term, []).append(ctx.key(issue.get("sequence_id")))
            if group:
                groups.setdefault(group, []).append(ctx.key(issue.get("sequence_id")))

    def terminal_ready(self, term, our_load):
        """Optional guard against processes we did not start (e.g. a manual codex session)."""
        if term in self._down:
            return False
        if term not in self._ready:
            names = self.cfg.terminals[term].get("busy_processes", [])
            ready = True
            for name in names:
                count = self.transport(term).process_count(name)
                if count is None:
                    self.mark_down(term, "busy-process probe failed")
                    ready = False
                    break
                elif count > 0 and our_load == 0:
                    self._busy[term] = "busy_process %s (%d running, not started by agentd)" % (name, count)
                    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("tier:"):
                r["tier"] = label[5:]
            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")
        if r["tier"] is None:  # per-model default unless a label set one; `tier:none` = no tier argument
            r["tier"] = (ecfg.get("tier_defaults") or {}).get(r["model"])
        if r["tier"] in ("none", ""):
            r["tier"] = None
        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, data, 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)
        # retry / rebuild markers are only consumed once the run has launched: if start fails they must survive,
        # otherwise an In Progress item without a run would never be picked again.
        retry = rec.get("retry")
        is_retry = bool(retry) and retry.get("reason") != "rebuild" and issue.get("state") == ctx.state_id("running")
        k = rec["runs"] + 1
        run_id = "%s-%d-%d" % (ctx.prefix, seq, k)
        thread, reason = rec.get("thread_id"), rec.get("rebuild")
        max_rounds = pcfg.get("max_rounds_per_thread") or self.cfg.agentd.get("max_rounds_per_thread", 8)
        if thread and not ex.resume:
            thread = None
        elif thread and rec.get("thread_on") != [term, exname]:
            thread, reason = None, "换了终端/执行器"
        elif thread and not is_retry and rec.get("thread_rounds", 0) >= max_rounds:  # a retry is the same round
            thread, reason = None, "线程已续 %d 轮（上限 %d）" % (rec["thread_rounds"], max_rounds)
        handoff = handoff_summary(rec, reason) if not thread and rec["runs"] > 0 else None
        retry_note = None
        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)
        relaunch = rec.get("relaunch") or {}
        attempt = relaunch.get("attempt", 1) if relaunch.get("run_id") == run_id else 1
        dir_name = run_id if attempt == 1 else "%s-r%d" % (run_id, attempt)
        run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"],
                  dir_name=dir_name)
        if thread:
            mode = "续线程 %s" % thread[:12]
        elif reason:
            mode = "新线程（重建：%s，附交接摘要）" % reason if handoff else "新线程（重建：%s）" % reason
        else:
            mode = "新线程（附交接摘要）" if handoff else "新线程"
        if is_retry:
            mode += "，重试 %d" % rec.get("retries", 0)
        if attempt > 1:
            mode += "，重新启动（目录 %s）" % dir_name
        # Everything the bookkeeping needs once the run is known to exist; saved before the launch call so that a
        # launch whose outcome is unknown (relay timeout, crash) is probed and adopted instead of started twice.
        plan = {"run_id": run_id, "terminal": term, "executor": exname, "attempt": attempt, "at": self.now(),
                "handle": ex.handle(run), "round": k, "budget_s": route["timeout_hours"] * 3600,
                "resumed": bool(thread), "group": route["group"], "model": route["model"], "effort": route["effort"],
                "is_retry": is_retry,
                # transient retries continue the same logical round: they do not use up max_rounds_per_thread
                "thread_rounds": (rec.get("thread_rounds", 0) + (0 if is_retry else 1)) if thread else 1,
                "seen": sorted(seen | {c["id"] for c in comments if c.get("id")}),
                "last_comment_at": max([c.get("created_at", "") for c in comments] + [rec.get("last_comment_at", "")]),
                "mode": mode,
                "detail": " / ".join(x for x in (exname, route["model"], route["effort"], route["tier"]) if x)}
        rec["launching"] = plan
        self.store.save(data)
        handle = ex.start(run)   # raises -> the marker stays and the next tick probes the run directory
        self.launched(ctx, iid, rec, plan, handle)

    def launched(self, ctx, iid, rec, plan, handle, adopted=False):
        """The run exists on the terminal: record it as in flight, then move the item and post the start note."""
        key, term = ctx.key(rec["seq"]), plan["terminal"]
        rec.pop("launching", None)
        rec.pop("relaunch", None)
        rec.pop("retry", None)
        rec.pop("rebuild", None)
        if not plan["is_retry"]:
            rec["retries"] = 0
        rec.update(runs=plan["round"], thread_rounds=plan["thread_rounds"], inflight={
            "run_id": plan["run_id"], "terminal": term, "executor": plan["executor"], "handle": handle,
            "round": plan["round"], "started": plan["at"], "deadline": plan["at"] + plan["budget_s"],
            "resumed": plan["resumed"], "group": plan["group"], "model": plan["model"], "effort": plan["effort"]})
        if not plan["resumed"]:
            rec.pop("thread_id", None)
            rec.pop("thread_on", None)
        rec["seen_comments"] = plan["seen"]
        rec["last_comment_at"] = plan["last_comment_at"]
        mode = plan["mode"] + ("，启动未确认、已接管" if adopted else "")
        issue = ctx.issues.get(iid)
        client = self.client(term)
        try:
            # an adopted run whose item a person moved meanwhile is left alone: poll cancels it next tick
            if issue is not None and issue.get("state") in (ctx.state_id("todo"), ctx.state_id("running")):
                self.move(ctx, client, iid, "running", self.cfg.terminals[term]["plane_user"])
            self.post(ctx, client, iid, rec, "%s 第 %d 轮开始：%s @ %s，%s，run=%s。" % (
                WORKER, plan["round"], plan["detail"], term, mode, plan["run_id"]))
        except Exception as exc:
            self.log("%s bookkeeping after start failed: %r" % (key, exc))
        self.log("%s run %s %s on %s (%s, %s)" % (key, plan["run_id"], "adopted" if adopted else "started", term,
                                                  plan["detail"], mode))