From 72681db46b90227e12d5afe2d1fe5459e75a1d36 Mon Sep 17 00:00:00 2001 From: "jj.wang" Date: Wed, 21 Oct 2026 13:21:16 +0000 Subject: [PATCH] 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". --- agentd/core.py | 54 +++++++++++++++++++++++++++------- tests/test_skips.py | 71 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 115 insertions(+), 10 deletions(-) create mode 100644 tests/test_skips.py diff --git a/agentd/core.py b/agentd/core.py index 2ca243d..56bb83d 100644 --- a/agentd/core.py +++ b/agentd/core.py @@ -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] diff --git a/tests/test_skips.py b/tests/test_skips.py new file mode 100644 index 0000000..6df184b --- /dev/null +++ b/tests/test_skips.py @@ -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()