Core: cancel moved runs, retry bookkeeping, progress cap, safer classification
- A run whose item leaves In Progress is killed, gets a [worker] cancel comment, frees its terminal slot and serialize group and keeps its thread id; if the kill fails the slot stays taken and the cancel is retried next tick. - Transient retries are the same logical round: they do not count toward max_rounds_per_thread and do not trigger a max-rounds rebuild. - New project setting max_progress_rounds (default 12): after that many consecutive [progress] rounds the item goes to Waiting (answerer) with a note. - retry/rebuild markers are consumed only after the run launched, so a failed start no longer leaves an In Progress item that is never picked again; a rebuild is no longer reported as a retry. - State is saved after every finished round and every launched run. - Failure classification: transient patterns only look at the tail, and an exit-0 failure is logic (never retried) unless the tail shows a transient error. - First-line tag: tolerate BOM and a leading code fence / rule line. - Start comment notes when a handoff summary was attached.
This commit is contained in:
jj.wang
+87
-17
@@ -34,11 +34,19 @@ OUTPUT_RULE = "输出要求：最后一条消息首行为 [review]/[question]/[b
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- pure helpers (unit-tested)
|
||||
FENCE = re.compile(r"^(```|~~~|-{3,}$|\*{3,}$)")
|
||||
|
||||
|
||||
def classify_first_line(text):
|
||||
"""Tag of the first non-empty line (`[review]` etc.), tolerating markdown emphasis; None if untagged."""
|
||||
first = next((l.strip() for l in (text or "").splitlines() if l.strip()), "")
|
||||
m = re.match(r"\[(\w+)\]", first.lstrip("*_#> `").lower())
|
||||
return m.group(1) if m and m.group(1) in TAGS else None
|
||||
"""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):
|
||||
@@ -46,14 +54,18 @@ def matches_any(text, patterns):
|
||||
|
||||
|
||||
def classify_failure(result, timed_out, retry_cfg):
|
||||
"""-> (kind, transient). kind in timeout | env | crash | logic (design §4)."""
|
||||
"""-> (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
|
||||
text = "%s\n%s" % (result.tail or "", result.last_message or "")
|
||||
if matches_any(text, retry_cfg.get("env_patterns", [])):
|
||||
if matches_any("%s\n%s" % (result.tail or "", result.last_message or ""), retry_cfg.get("env_patterns", [])):
|
||||
return "env", False
|
||||
kind = "logic" if result.exit_code == 0 else "crash"
|
||||
return kind, matches_any(text, retry_cfg.get("transient_patterns", DEFAULT_TRANSIENT))
|
||||
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):
|
||||
@@ -209,6 +221,7 @@ class Agentd:
|
||||
self.poll(ctx, data, iid, rec)
|
||||
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)
|
||||
|
||||
# ---------------------------------------------------------------- (a) poll / finish
|
||||
@@ -216,6 +229,9 @@ class Agentd:
|
||||
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"]))
|
||||
@@ -232,6 +248,37 @@ class Agentd:
|
||||
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"]
|
||||
@@ -258,7 +305,16 @@ class Agentd:
|
||||
self.close_run(rec, result)
|
||||
rec["tags"].append(tag)
|
||||
rec.update(last_message=message[:4000], last_tag=tag, retries=0)
|
||||
if still_ours:
|
||||
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)
|
||||
@@ -317,6 +373,7 @@ class Agentd:
|
||||
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)
|
||||
@@ -390,6 +447,7 @@ class Agentd:
|
||||
except Exception as exc:
|
||||
self.log("%s start failed on %s: %r" % (ctx.key(issue.get("sequence_id")), term, exc))
|
||||
continue
|
||||
self.store.save(data) # a launched run must be on disk before anything else can fail
|
||||
load[term] += 1
|
||||
if group:
|
||||
groups.add(group)
|
||||
@@ -465,18 +523,19 @@ class Agentd:
|
||||
term, exname = route["terminal"], route["executor"]
|
||||
tcfg, pcfg = self.cfg.terminals[term], ctx.cfg
|
||||
ex = self.executor(term, exname)
|
||||
is_retry = rec.pop("retry", None) is not None and issue.get("state") == ctx.state_id("running")
|
||||
if not is_retry:
|
||||
rec["retries"] = 0
|
||||
# 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.pop("rebuild", None)
|
||||
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 rec.get("thread_rounds", 0) >= max_rounds:
|
||||
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
|
||||
@@ -489,8 +548,14 @@ class Agentd:
|
||||
prompt = build_prompt(key, issue, comments, seen, bool(thread), self.protocol(pcfg), intro, handoff, retry_note)
|
||||
run = Run(run_id, tcfg["repo"], prompt, route["model"], route["effort"], route["tier"], thread, route["script"])
|
||||
handle = ex.start(run)
|
||||
rec.pop("retry", None)
|
||||
rec.pop("rebuild", None)
|
||||
if not is_retry:
|
||||
rec["retries"] = 0
|
||||
now = self.now()
|
||||
rec.update(runs=k, thread_rounds=rec.get("thread_rounds", 0) + 1 if thread else 1, inflight={
|
||||
# transient retries continue the same logical round: they do not use up max_rounds_per_thread
|
||||
rounds = (rec.get("thread_rounds", 0) + (0 if is_retry else 1)) if thread else 1
|
||||
rec.update(runs=k, thread_rounds=rounds, inflight={
|
||||
"run_id": run_id, "terminal": term, "executor": exname, "handle": handle, "round": k,
|
||||
"started": now, "deadline": now + route["timeout_hours"] * 3600, "resumed": bool(thread),
|
||||
"group": route["group"], "model": route["model"], "effort": route["effort"]})
|
||||
@@ -500,7 +565,12 @@ class Agentd:
|
||||
rec["seen_comments"] = sorted(seen | {c["id"] for c in comments if c.get("id")})
|
||||
rec["last_comment_at"] = max([c.get("created_at", "") for c in comments] + [rec.get("last_comment_at", "")])
|
||||
client = self.client(term)
|
||||
mode = ("续线程 %s" % thread[:12]) if thread else ("新线程（重建：%s）" % reason if reason else "新线程")
|
||||
if 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)
|
||||
detail = " / ".join(x for x in (exname, route["model"], route["effort"], route["tier"]) if x)
|
||||
|
||||
Reference in New Issue
Block a user