jj.wang
4f45d368ad
Before the launch call the planned run (run id, terminal, handle, bookkeeping) is saved as a launching marker. If the launch call fails or agentd dies before recording the run, the next tick probes the run directory (exit.txt / launch.pid / pid.txt): evidence -> adopt the run; nothing -> relaunch with the same run id in <run_id>-r2 (-r3 ...); probe unreachable -> keep the marker, which holds the terminal slot. Transports refuse to clean a run directory holding a pid file. A terminal whose relay fails is skipped for the rest of the tick (heartbeat.down) while other terminals carry on.
670 lines
35 KiB
Python
670 lines
35 KiB
Python
"""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
|
||
|
||
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)
|
||
self.store.heartbeat({"errors": errors, "down": sorted(self._down), "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 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)
|
||
|
||
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") or r.get("launching") # an unconfirmed launch holds its slot
|
||
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") 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):
|
||
if route:
|
||
self.log("%s not dispatched: %s" % (ctx.key(issue.get("sequence_id")), route))
|
||
continue
|
||
term, group = route["terminal"], route["group"]
|
||
if term in self._down:
|
||
continue
|
||
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, data, pstate, issue, route)
|
||
except Exception as exc:
|
||
self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
|
||
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
|
||
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 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.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("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))
|