Heartbeat: record why each Todo item was not dispatched
Each tick lists skipped items with terminal and reason (concurrency n/m with the holding runs, serialize group holder, busy_process not started by agentd, relay down, no route for assignee, start failed) in heartbeat.json "skipped". A reason is logged (agentd.log / cron.log) only when it is new for that item; the last set is kept in the state file as "skips".
This commit is contained in:
jj.wang
+44
-10
@@ -161,6 +161,8 @@ class Agentd:
|
||||
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:
|
||||
@@ -207,13 +209,33 @@ class Agentd:
|
||||
self.log("%s tick error: %r" % (name, exc))
|
||||
finally:
|
||||
self.store.save(data)
|
||||
self.store.heartbeat({"errors": errors, "down": sorted(self._down), "inflight": [
|
||||
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:
|
||||
@@ -444,21 +466,24 @@ class Agentd:
|
||||
|
||||
# ---------------------------------------------------------------- (b) pick / start
|
||||
def pick(self, ctx, data, pstate, only_issue=None):
|
||||
load, groups = Counter(), set()
|
||||
for p in data["projects"].values():
|
||||
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.add(fl["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")))
|
||||
groups |= set(data["paused_groups"])
|
||||
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():
|
||||
@@ -477,20 +502,28 @@ class Agentd:
|
||||
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))
|
||||
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] >= self.cfg.terminals[term].get("concurrency", 1) or (group and group in groups):
|
||||
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"):
|
||||
@@ -498,8 +531,9 @@ class Agentd:
|
||||
# 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.add(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)."""
|
||||
@@ -515,7 +549,7 @@ class Agentd:
|
||||
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))
|
||||
self._busy[term] = "busy_process %s (%d running, not started by agentd)" % (name, count)
|
||||
ready = False
|
||||
self._ready[term] = ready
|
||||
return self._ready[term]
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from tests.fakes import Harness
|
||||
|
||||
|
||||
class FakeProbe:
|
||||
def __init__(self, counts):
|
||||
self.counts = counts
|
||||
|
||||
def process_count(self, name):
|
||||
return self.counts.get(name, 0)
|
||||
|
||||
|
||||
class SkipReasonTest(unittest.TestCase):
|
||||
"""Why a Todo item was not dispatched: heartbeat "skipped", logged once per new reason."""
|
||||
|
||||
def setUp(self):
|
||||
self.h = Harness()
|
||||
self.p = self.h.plane
|
||||
self.lines = []
|
||||
self.h.d.log = lambda msg: self.lines.append(msg)
|
||||
|
||||
def tearDown(self):
|
||||
self.h.close()
|
||||
|
||||
def skipped(self):
|
||||
hb = json.loads((self.h.d.store.dir / "heartbeat.json").read_text())
|
||||
return {s["item"]: (s["terminal"], s["reason"]) for s in hb["skipped"]}
|
||||
|
||||
def not_dispatched(self):
|
||||
return [l for l in self.lines if "not dispatched" in l]
|
||||
|
||||
def test_reasons_in_heartbeat_and_logged_once(self):
|
||||
self.p.add_issue(1) # starts on win10
|
||||
self.p.add_issue(2) # win10 full
|
||||
self.p.add_issue(3, assignee="kappiee") # no terminal for kappiee
|
||||
self.p.add_issue(4, assignee="oregon", labels=["needs-game"]) # starts on oregon, holds the game group
|
||||
self.p.add_issue(5, assignee="oregon", labels=["needs-game"]) # group held
|
||||
self.h.tick()
|
||||
self.assertEqual(self.skipped(), {
|
||||
"GW-2": ("win10", "concurrency 1/1 (GW-1)"),
|
||||
"GW-3": (None, "no route for assignee kappiee"),
|
||||
"GW-5": ("oregon", "serialize_group game held by GW-4"),
|
||||
})
|
||||
self.assertEqual(len(self.not_dispatched()), 3)
|
||||
self.h.tick() # same reasons: heartbeat still has them, the log does not repeat them
|
||||
self.assertEqual(len(self.skipped()), 3)
|
||||
self.assertEqual(len(self.not_dispatched()), 3)
|
||||
|
||||
def test_busy_process_and_relay_down(self):
|
||||
self.h.cfg.terminals["win10"]["busy_processes"] = ["codex"]
|
||||
probe = FakeProbe({"codex": 1})
|
||||
self.h.d.transport_factory = lambda term: probe
|
||||
self.p.add_issue(1)
|
||||
self.h.tick()
|
||||
self.assertEqual(self.skipped(), {"GW-1": ("win10", "busy_process codex (1 running, not started by agentd)")})
|
||||
probe.counts = {}
|
||||
self.h.tick() # leftover codex.exe gone: dispatched, nothing skipped
|
||||
self.assertEqual(self.skipped(), {})
|
||||
self.assertEqual(self.p.state_of("i1"), "In Progress")
|
||||
self.p.add_issue(2, assignee="oregon")
|
||||
self.h.cfg.terminals["oregon"]["busy_processes"] = ["claude"]
|
||||
probe.counts = {"claude": None} # probe fails: relay down
|
||||
probe.process_count = lambda name: None
|
||||
self.h.tick()
|
||||
self.assertEqual(self.skipped()["GW-2"], ("oregon", "relay down"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user