jj.wang
713 lines
38 KiB
Python
713 lines
38 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 CallLog, 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
|
||
# one line per remote call (terminal, channel, operation, ms, rc) -> remote.log; fallback/host-key -> agentd.log
|
||
self.calllog = CallLog(self.store.log_dir / "remote.log", self.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, log=self.calllog))
|
||
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))
|