From e1ace21e994fba3b9f3b143d132ae034ccede28a Mon Sep 17 00:00:00 2001 From: Josh Mabry Date: Thu, 10 Sep 2026 17:16:35 -0700 Subject: [PATCH] fix(board): the working-state snapshot is current or says it is stale (#401) The PM's injected working state showed two cards in states it had itself just moved them out of (protoEngineer, 2026-09-07: it unblocked bd-p8ft at 07:16:40Z and blocked bd-ezs7 at 07:17:02Z, then planned around "bd-p8ft is confirmed blocked" at 07:17:20Z). The host reads the provider fresh every turn; the plugin's snapshot was refreshed only by the 300s health sweep, at the START of it, and carried nothing that said how old it was. - Every br write that can move a card (create/update/close/reopen/delete) bumps a process-wide revision, in a `finally` around the call so a write that timed out but landed still counts. #431's stall reaches that `finally` only after the br tree is stopped and reaped, so nothing can land after the mark. A live config reload of a knob the hints read bumps it too. - A snapshot records the revision it was read at (read BEFORE the board); while the board has moved past it, or it is older than 120s, the provider leads with a STALE line naming when it was taken. - A refresher of its own (not the claim tick -- a paused loop still refreshes) re-reads within 5s of a change (coalesced), at least every 60s regardless (bounding writers this process can't see), backs off exponentially on failure, and logs a failed or stalled read (BoardError, BoardTimeout included) as one line. One read at a time: a second while one is in flight is skipped. The sweep publishes after its own transitions, and a stall there is the sweep's stall: it goes up to _tick_phase and ends the tick (#431's "first stall ends the tick"), where swallowing it let the preflight and claim scan stall on the same store. - The read is store.live_cards(): one `br list` of the open statuses, plus a `br show` (in _SHOW_BATCH batches) of only the rows whose hint needs more -- blocked cards (reason, edges) and backlog cards that have dependencies -- and, when a card could be stranded, its closed dependencies, so #406's "dependencies closed" hints survive the light read and a CANCELLED dependency is still told from a merged one. Cards newly stranded are logged from this read (once per card). - Hints: a blocked card reads "needs a human (): " / "retries on its own (): ". - Block classes: clear_blocked drops the class whatever it is (noted on its audit comment), still resetting a too-wide park's timeout count (#378); the projection reports blocked_class only while blocked; block_from_review stamps its own class and records a `blocked:` reason -- so an old `transient` can't make the sweep auto-heal an exhausted escalation, and boot's preflight-hold release reads the current block. Tier labels (#339) are never touched. - The snapshot state lives in a process-stable sys.modules slot; stale comments in failures.py and drive.py corrected. - Seam ratchet recomputed on main: MAX_UNCOVERED_STORE 18 -> 16 (clear_blocked, block_from_review now REAL); live_cards and the new _show_by_id are REAL. Includes the adversarial-review round (tests pin the wiring against the mutations review found surviving: no refresher, revision read after the board, write set cut to {"update"}). Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01F2V6GRejF7mNukAoYjj2Av --- changelog.d/433.md | 17 + docs/lifecycle.md | 5 +- failures.py | 10 +- loop/core.py | 8 + loop/drive.py | 29 +- loop/reconcile.py | 136 +++++-- store.py | 197 ++++++++-- tests/test_external_seams.py | 8 +- tests/test_loop.py | 12 +- tests/test_stale_blocked_class_401.py | 80 ++++ tests/test_store.py | 5 +- tests/test_work_snapshot_review_401.py | 426 ++++++++++++++++++++++ tests/test_work_snapshot_staleness_401.py | 318 ++++++++++++++++ work_snapshot.py | 157 ++++++-- 14 files changed, 1294 insertions(+), 114 deletions(-) create mode 100644 changelog.d/433.md create mode 100644 tests/test_stale_blocked_class_401.py create mode 100644 tests/test_work_snapshot_review_401.py create mode 100644 tests/test_work_snapshot_staleness_401.py diff --git a/changelog.d/433.md b/changelog.d/433.md new file mode 100644 index 0000000..89b25b4 --- /dev/null +++ b/changelog.d/433.md @@ -0,0 +1,17 @@ +The agent's working state no longer shows board cards in states they have already left. The +board's snapshot of open cards was refreshed only by the 300s health sweep, and at the start of +it. A PM that unblocked one card and blocked another was then shown both in their old states, +with nothing saying the view was old, and it planned against that. Every board write now marks +the snapshot stale, including a write that timed out and may still have landed. So does a live +change to a setting the hints read. A stale snapshot says so in its first line and names when it +was taken. A refresher of its own, not part of the claim tick, re-reads the board within 5s of a +change, and at least every 60s regardless. So the view is either marked stale or at most about a +minute old, even while the loop is paused or another process writes the board. It reads only the +open cards, not the whole board, and only the detail its hints need, so the cards named as +stranded behind closed dependencies stay in it. A failed or stalled read is one log line, and +the refresher backs off; a stall in the sweep's own read ends that tick like any other stall. A +blocked card's hint names its block reason and whether it retries on its own or needs a human. +Unblocking a card now clears its block class (the class is noted on the unblock's comment), and a +card that is not blocked no longer reports one. An exhausted CI or review escalation now records +a class and reason of its own, so boot no longer mistakes it for an old preflight hold and +releases it. diff --git a/docs/lifecycle.md b/docs/lifecycle.md index 6f32c8e..d7a7790 100644 --- a/docs/lifecycle.md +++ b/docs/lifecycle.md @@ -285,8 +285,9 @@ again once they close. It is not a claim candidate and it shows up in no skip di So the board now names it, wherever a card's next action is shown: the listing, the console chip, the agent's working state (which names a backlog card only when it owes a step, and ranks it after every in-flight card so a pile of stranded cards can't push a PR awaiting -merge out of the capped list), and one sweep log line when the card first becomes stranded -(held in memory, so a restart logs each stranded card once more): +merge out of the capped list), and one log line when the loop first sees the card stranded, +on the working-state snapshot's read (held in memory, so a restart logs each stranded card +once more): - **backlog, every dependency closed** → `dependencies closed — promote`. The step is `board_mark_ready`, and the Ready gate still decides. A `deferred` or `designing` card is diff --git a/failures.py b/failures.py index 058ab23..260487b 100644 --- a/failures.py +++ b/failures.py @@ -97,11 +97,11 @@ class it would otherwise have (a `502 … timeout` is still transient).""" # ── pre-model dispatch / infrastructure failures (#339) ────────────────────────── # The `blocked-class:` a pre-model dispatch/infra failure carries. It is deliberately # NOT one of `classify()`'s categories: it can't be decided from the message alone -# (it needs the loop's dispatch-lifecycle evidence too), and it drives two behaviours -# the message-classes don't — the operator is NOTIFIED rather than auto-healed (it is -# absent from the loop's self-healing set), and an operator unblock RESETS the card's -# escalation tier so a host/adapter incident never leaves a `tier:` label the next -# genuine build inherits. +# (it needs the loop's dispatch-lifecycle evidence too), and the operator is NOTIFIED +# rather than auto-healed (it is absent from the loop's self-healing set). The loop never +# climbs a tier on such a failure, so an unblock (store.clear_blocked) drops the class like +# any other and leaves the card's `tier:` labels alone: they were earned before the +# incident. PRE_MODEL_DISPATCH_CLASS = "dispatch-infra" # The seam SHAPES — how a failure raised BELOW the model call reaches the loop. `coder_seam` diff --git a/loop/core.py b/loop/core.py index f0f4301..41b0807 100644 --- a/loop/core.py +++ b/loop/core.py @@ -244,6 +244,14 @@ def __init__(self, cfg: dict, *, gap_reporter: setup_check.GapReporter | None = # once per DISTINCT reason, and identical repeats collapse to a one-line # "still held (Ns)" WARNING (see _record_preflight_failure). self._preflight_failed_at: dict[str, float] = {} + # The working-state snapshot's refresher (#401): its task, when it last read, how many + # reads in a row failed (its backoff), and whether a read is in flight (one at a time). + # -inf, not 0.0: time.monotonic() starts near zero in a fresh container, and the first + # read must not wait out MIN_INTERVAL_S. + self._snapshot_task: asyncio.Task | None = None + self._snapshot_attempted_at = float("-inf") + self._snapshot_failures = 0 + self._snapshot_reading = False # ── coder.solve() board seam (ADR 0064 P2, opt-in) ───────────────────────── # Route a FRESH build (not a keep-worktree/CI-bounce re-dispatch) through the # `coder` plugin's execution-grounded solve() ladder (greedy → best-of-k → diff --git a/loop/drive.py b/loop/drive.py index be80494..eec13c8 100644 --- a/loop/drive.py +++ b/loop/drive.py @@ -204,6 +204,12 @@ def start(self): log.info("[project_board] loop disabled (project_board.loop_enabled=false) — board API still serves") return None self._task = asyncio.create_task(self._run(), name="project-board-loop") + # The working-state snapshot has a refresher of its own, not a step of the tick: a + # loop paused at its setup gate runs no ticks, and the snapshot must not go stale + # for as long as the pause lasts (#401). + self._snapshot_task = asyncio.create_task( + self._keep_work_snapshot_current(), name="project-board-work-snapshot" + ) # The RUNNING loop's config, in a process-stable slot: after a reload the # routers see the new config while this loop keeps its construction-time # `coders`/`repo`/…; /status compares the two and says "restart to apply". @@ -339,6 +345,9 @@ def reload(self, new_config) -> dict: changed[key] = (cur, new) if changed: setup_check.publish_loop_snapshot(self.cfg) + # The working-state hints read these knobs (auto_merge, review_gate, …), so what + # the snapshot renders changed without a board write: mark it stale (#401). + work_snapshot.mark_stale() log.info( "[project_board] reload applied live: %s (in-flight drives: %d)", ", ".join(f"{k} {o}→{n}" for k, (o, n) in changed.items()), @@ -352,12 +361,13 @@ async def stop(self): _unregister_loop(self) # drop the process-stable handle (ADR 0326) if self._task: setup_check.publish_loop_snapshot(None) # no running loop → nothing to be stale against - if self._task: - self._task.cancel() - try: - await self._task - except (asyncio.CancelledError, Exception): # noqa: BLE001 - pass + for task in (self._task, self._snapshot_task): + if task: + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): # noqa: BLE001 + pass # Cancel any in-flight drives and await them out. A drive cancelled mid-flight # can't run its own cleanup, so its worktree stays in self._inflight — reaped # below. (A completed/blocked drive already popped itself.) @@ -2161,9 +2171,10 @@ async def _drive(self, feature: dict): # called, so no `tier:`/`attempt:` label is added and no ladder # budget is spent. It blocks under the `dispatch-infra` class the # blocked sweep never auto-heals, so the operator is notified with the - # original infra evidence — and an operator unblock resets the tier - # posture (store.clear_blocked) so the next genuine build starts at - # its difficulty-selected tier (#339). + # original infra evidence. An operator unblock (store.clear_blocked) + # drops the class and leaves the card's `tier:` labels alone: the + # incident added none, so the next build resumes the tier the card + # had earned, or its difficulty-selected one (#339). if pre_model: reason = f"pre-model dispatch failure — infra triage, no tier climb: {exc}" log.warning( diff --git a/loop/reconcile.py b/loop/reconcile.py index 424e9be..eefde81 100644 --- a/loop/reconcile.py +++ b/loop/reconcile.py @@ -16,6 +16,10 @@ _loop = sys.modules[__package__] # the loop package, for monkeypatch-visible seams +# How often the working-state refresher checks whether a read is due (an in-memory compare; +# the reads themselves are bounded by work_snapshot.MIN_INTERVAL_S / MAX_AGE_S). +_SNAPSHOT_POLL_S = 1.0 + class ReconcileMixin: # ── merged-verify exhaustion sentinel ↔ operator reset (ADR 0326, #326) ─────── @@ -317,27 +321,12 @@ async def _sweep(self): ``feat-`` worktrees whose feature is gone or already terminal — ``done``/``cancelled`` (a missed reap); (c) label terminal features past the archive window ``archived`` - (#115) — the board's growth valve; archival only, nothing is ever deleted. First, - it publishes the agent's working-state snapshot, and names any card stranded + (#115) — the board's growth valve; archival only, nothing is ever deleted. Last, + it publishes the agent's working-state snapshot, which names any card stranded outside the ready lane with every dependency closed (#406) — surfaced, never moved. - Best-effort; a per-item failure never stops the sweep or the loop.""" + Best-effort; a per-item failure never stops the sweep or the loop. A stalled store + ends the sweep, and the tick with it (#404).""" store = self._store() - # Publish the board's live projection for the host's block (ADR - # 0079 Observe). Done on the SWEEP cadence, not per tick: it is one extra `br` - # call per sweep, and the provider that reads it must never touch the store - # itself — it runs inline on every agent turn. Best-effort, like the rest of the - # sweep. - # - # Annotated first: the snapshot's per-card hint IS the board's `next_action_hint`, - # which only `annotate_next_action` writes — a bare listing published every hint - # empty. It is also what names a card stranded outside the ready lane (#406), and - # the agent this snapshot feeds is the one who can promote it. - try: - feats = store_mod.annotate_next_action(await asyncio.to_thread(store.list_features), self.cfg) - work_snapshot.publish(feats) - self._report_stranded(feats) - except Exception: # noqa: BLE001 — never let a snapshot refresh stop the sweep - log.warning("[project_board] work snapshot refresh failed (ignored)", exc_info=True) for f in await self._list_for_pass(store, "in_progress", "health sweep"): fid = f["id"] if fid in self._inflight_files: @@ -372,14 +361,115 @@ async def _sweep(self): raise except Exception: # noqa: BLE001 log.warning("[project_board] sweep archive pass failed", exc_info=True) + # (d) the host's snapshot, LAST, so it carries this sweep's own + # reconciles and unblocks. Published first, as it used to be, it showed a card the + # sweep had just moved in the state it had left, for a whole interval (#401). The + # refresher would pick those writes up within seconds anyway; publishing here makes + # the sweep's result current the moment it ends. Its stall is the sweep's stall. + await self._publish_work_snapshot(stall_ends_tick=True) + + # ── the host's snapshot (ADR 0079 Observe, #401) ────────── + def _take_work_snapshot(self, store) -> None: + """Read the open cards and publish them for the host's working-state block. + + The revision is read BEFORE the board, so a write that lands mid-read leaves the + snapshot marked stale (and re-read soon) rather than claiming a state its rows may + predate. The read is ``live_cards``, the light one: open statuses only, no + whole-board ``br show``. Rows are annotated as board_list's are, so a card's hint is + the board's own next action. A blocked card, ranked first, gets its block reason and + what moves it, which the posture hints leave blank. A card newly stranded outside + the ready lane is logged (``_report_stranded``).""" + revision = work_snapshot.board_revision() + features = store_mod.annotate_next_action(store.live_cards(), self.cfg) + for f in features: + if f.get("blocked") and not f.get("next_action_hint"): + cls = str(f.get("blocked_class") or "").strip() + reason = str(f.get("blocked_reason") or "").strip() or "no reason recorded" + who = ( + f"retries on its own ({cls})" + if cls in _SELF_HEALING_BLOCKS + else f"needs a human ({cls or 'unclassified'})" + ) + f["next_action_hint"] = f"{who}: {reason}" + work_snapshot.publish(features, revision=revision) + self._report_stranded(features) + + async def _publish_work_snapshot(self, *, stall_ends_tick: bool = False) -> bool: + """``_take_work_snapshot``, off the event loop and best-effort. Returns whether it + published. A failed read backs the refresher off (see ``_work_snapshot_due``), and + the provider keeps marking the old snapshot stale meanwhile. + + A ``BoardError`` (a `br` call that failed, or stalled and was stopped) is an + understood, transient outcome, logged as ONE line like ``_tick_phase`` logs it; + anything else is a bug and keeps its traceback. The exception is a stall inside a + tick (``stall_ends_tick``, the sweep's publish): it goes up to ``_tick_phase``, which + ends the tick on it (#404). Swallowed there, the tick went on to the preflight and the + claim scan, and each stalled again on the same wedged store. + + One read at a time: the refresher and the sweep can both get here, and a second read + while one is in flight is skipped. The one in flight publishes, and a write since it + began leaves that snapshot STALE for the refresher to re-read.""" + if self._snapshot_reading: + return False + self._snapshot_reading = True + self._snapshot_attempted_at = time.monotonic() + try: + await asyncio.to_thread(self._take_work_snapshot, self._store()) + except BoardError as exc: + self._snapshot_failures += 1 + if stall_ends_tick and isinstance(exc, store_mod.BoardTimeout): + raise + log.warning("[project_board] work snapshot refresh failed (retrying with backoff): %s", exc) + return False + except Exception: # noqa: BLE001 — never let a snapshot refresh stop the loop + self._snapshot_failures += 1 + log.warning("[project_board] work snapshot refresh failed (retrying with backoff)", exc_info=True) + return False + finally: + self._snapshot_reading = False + self._snapshot_failures = 0 + return True + + def _work_snapshot_due(self) -> bool: + """Whether the refresher should read now. In memory, no I/O. + + - Never inside ``MIN_INTERVAL_S`` of the last attempt. A burst of writes (a claim, + its labels, a drive's budget stamps) costs ONE read, not one each. + - After failures, wait out an exponential backoff (``MIN_INTERVAL_S`` doubling, up to + ``MAX_AGE_S``). A store that is failing is not hammered. + - Otherwise read when the board changed since the snapshot, when there is none yet, + or when it is older than ``MAX_AGE_S``. The last case is what bounds a writer this + process can't see.""" + since = time.monotonic() - self._snapshot_attempted_at + wait = work_snapshot.MIN_INTERVAL_S + if self._snapshot_failures: + wait = min(work_snapshot.MIN_INTERVAL_S * 2**self._snapshot_failures, work_snapshot.MAX_AGE_S) + if since < wait: + return False + taken = work_snapshot.taken_at() + return work_snapshot.needs_refresh() or taken is None or time.time() - taken >= work_snapshot.MAX_AGE_S + + async def _keep_work_snapshot_current(self) -> None: + """The snapshot's refresher, a task of its own (started with the loop). It is NOT part + of the claim tick. A loop paused at its setup gate (a missing coder, say) runs no + ticks, and a snapshot refreshed only by ticks stayed STALE for as long as the pause + lasted. It polls an in-memory check every ``_SNAPSHOT_POLL_S`` and reads only when + ``_work_snapshot_due`` says so.""" + while not self._stop.is_set() and not self._shutting_down: + if self._work_snapshot_due(): + await self._publish_work_snapshot() + try: + await asyncio.wait_for(self._stop.wait(), timeout=_SNAPSHOT_POLL_S) + except asyncio.TimeoutError: + pass def _report_stranded(self, feats: list[dict]) -> None: """Log each card that has just become stranded outside the ready lane, with every dependency closed (#406). ``store.stranded_posture`` found them, and they read so on every listing and in the agent's working state. This line is the loop's own - record of WHEN it first saw each one: once per card, not once per sweep, and again - only if the card leaves that state and comes back. Nothing is changed on the card. - A backlog card is promoted by the PM, and a block is lifted by whoever set it.""" + record of WHEN it first saw each one: once per card, not once per snapshot read, and + again only if the card leaves that state and comes back. Nothing is changed on the + card. A backlog card is promoted by the PM, and a block is lifted by whoever set it.""" stranded = { f["id"]: f for f in feats @@ -389,7 +479,7 @@ def _report_stranded(self, feats: list[dict]) -> None: seen = getattr(self, "_stranded_seen", set()) for fid in sorted(set(stranded) - seen): f = stranded[fid] - log.info("[project_board] sweep: %s stranded (%s): %s", fid, f["board_state"], f.get("next_action_hint")) + log.info("[project_board] %s stranded (%s): %s", fid, f["board_state"], f.get("next_action_hint")) self._stranded_seen = set(stranded) async def _recover_blocked(self, store) -> None: diff --git a/store.py b/store.py index cb22573..0b5e7a4 100644 --- a/store.py +++ b/store.py @@ -49,7 +49,7 @@ from datetime import datetime, timezone from urllib.parse import urlparse -from . import _TERMINAL_STATES, br_fetch +from . import _TERMINAL_STATES, br_fetch, work_snapshot log = logging.getLogger("protoagent.plugins.project_board") @@ -167,6 +167,12 @@ def locked(self, fid, *args, **kwargs): return locked +# The `br` subcommands that can change what a card PROJECTS as: its status, labels, title +# or external ref. One outdates the agent's working-state snapshot (#401, see `_run`). +# `comments` and `dep` cannot: the working-state line is id + state + title + next-action +# hint, and none of those read either one. +_PROJECTION_WRITES = frozenset({"create", "update", "close", "reopen", "delete"}) + # `br` plain-mode not-found text (stderr). The --json path is matched on the structured # ISSUE_NOT_FOUND code instead, so this only backstops a non-json `show`. _NOT_FOUND_RE = re.compile(r"\bissue not found\b", re.IGNORECASE) @@ -1339,7 +1345,26 @@ def _run(self, *args: str, want_json: bool = False, with_has_more: bool = False, actor). The task-claim path (#356) uses it to run ``--claim`` AS the dispatch target, so the atomic claim stamps that target as owner instead of the board actor — ``br`` REFUSES a duplicate ``--actor``, so a caller that needs a non-default actor - must pass it here rather than in ``args``. Every other caller keeps ``self.actor``.""" + must pass it here rather than in ``args``. Every other caller keeps ``self.actor``. + + A write that can change what a card projects as marks the agent's working-state + snapshot stale (#401). Every board write in this process comes through here. The + mark goes in a ``finally``, AFTER the call and whatever it did. A write that timed + out or errored may still have landed; a spurious mark costs one refresh, while a + missed one lets an outdated snapshot read as current. A stall (``BoardTimeout``) + reaches the ``finally`` only once ``_run_br_process`` has stopped and reaped the + `br` tree (#404), so nothing the write does can land after the mark. Never before + the call either: a snapshot read between an early mark and the write would pair the + new revision with the old rows.""" + try: + return self._shell_br(*args, want_json=want_json, with_has_more=with_has_more, actor=actor) + finally: + if args and args[0] in _PROJECTION_WRITES: + work_snapshot.note_board_write() + + def _shell_br(self, *args: str, want_json: bool = False, with_has_more: bool = False, actor: str = ""): + """``_run``'s body: the ``br`` subprocess, its contention retries, and the + payload normalization. Call ``_run``, never this.""" _warn_blocking_on_event_loop(args[0] if args else "") self._ensure_workspace() # pin to the repo's own .beads/ before any br op (#48) cmd = [BR, *args, "--actor", actor or self.actor] @@ -2660,12 +2685,26 @@ def requeue(self, fid: str) -> dict: return self.get_feature(fid) def block_from_review(self, fid: str, reason: str) -> dict: - """Drop the in-review label and flag Blocked — used when the escalation - ladder is exhausted on a CI failure.""" - self._require(fid) - self._run("update", fid, "--remove-label", LABEL_IN_REVIEW, "--add-label", LABEL_BLOCKED) - if reason: - self.comment(fid, f"escalation exhausted: {reason}") + """Drop the in-review label and flag Blocked — used when the escalation ladder is + exhausted on a CI or review failure. + + A block like any other (#401 review): it stamps its OWN class, ``terminal`` (only a + human moves a card whose automated fixes are spent), replacing any class a + previous block left. It records its reason as the card's latest ``blocked:`` + comment, which is what ``blocked_reason``, the retro and the boot-time preflight + release all read as the CURRENT block. It used to add only the flag and an + ``escalation exhausted:`` note. An older class could resurface under it (a + ``transient`` made the sweep auto-heal a card that needs a human), and an older + ``blocked:`` reason still read as current. A card once held by the preflight was + then released at boot as an "orphaned preflight hold".""" + f = self._require(fid) + want = f"{LABEL_BLOCKED_CLASS_PREFIX}terminal" + args = ["update", fid, "--remove-label", LABEL_IN_REVIEW, "--add-label", LABEL_BLOCKED, "--add-label", want] + for prior in f.get("labels") or []: # replace, never accumulate; never remove what we add + if str(prior).startswith(LABEL_BLOCKED_CLASS_PREFIX) and prior != want: + args += ["--remove-label", prior] + self._run(*args) + self.comment(fid, f"blocked: escalation exhausted: {reason}" if reason else "blocked: escalation exhausted") return self.get_feature(fid) # ── the ONE Done edge for coding features (invariant #2) ────────────────── @@ -2998,43 +3037,50 @@ def flag_blocked(self, fid: str, reason: str, category: str = "") -> dict: return self.get_feature(fid) def clear_blocked(self, fid: str) -> dict: - """Clear the ``blocked`` flag so a feature can be re-dispatched. - - When the block was a NON-MODEL failure — a pre-model dispatch/adapter/infra - incident, ``blocked-class:dispatch-infra`` (#339) — also drop the now-stale - block class so a requeue starts clean. The card's escalation posture is left - UNTOUCHED, and deliberately so: the loop never climbs a tier on a pre-model - failure (``store.escalate`` is not called on that path), so every ``tier:`` - label present at a dispatch-infra block was earned BEFORE the incident by real - model-capability work. Removing them would silently restart a card that had - legitimately escalated on its lower difficulty-selected model and repeat the - work that already failed there — the exact regression the first cut introduced. - A card that never escalated carries no ``tier:`` label, so it still starts at - its difficulty-selected tier on the next build; the ladder stays a - model-capability record, and the infra incident adds nothing to it. A - model-reachable block (or an unclassified one) is untouched, exactly as before. + """Clear the ``blocked`` flag so a feature can be re-dispatched, and the block's + CLASS with it, whatever the class. + + A class describes a block. Only the ``dispatch-infra`` one (#339) and the + ``too-wide`` one (#378) used to be dropped. Every other class outlived its block + and read back as ``blocked_class: terminal`` on a card that was ready, a state that + does not exist; that was the false signal on bd-p8ft in #401. Nothing reads a class + off an unblocked card. The self-heal sweep and its unblock-retry budget read only + blocked cards, a re-block stamps its own class, and the retro mines comments, not + labels. So the class goes, and the one this block carried is kept where history + belongs: on this unblock's own audit comment. + + The card's escalation posture is left UNTOUCHED, and deliberately so (#339): the + loop never climbs a tier on a pre-model failure (``store.escalate`` is not called + on that path), so every ``tier:`` label present at a dispatch-infra block was + earned BEFORE the incident by real model-capability work. Removing them would + silently restart a card that had legitimately escalated on its lower + difficulty-selected model and repeat the work that already failed there — the + exact regression the first cut introduced. A card that never escalated carries no + ``tier:`` label, so it still starts at its difficulty-selected tier on the next + build; the ladder stays a model-capability record, and no block class adds to it. A card the loop PARKED as ``too-wide`` (#378) also gets its timeout count reset: unblocking one is a deliberate retry — after raising ``coder_timeout_s``, or narrowing the card by hand — and with the count still at the threshold its very - next timeout would park it again at once. (The loop's in-process copy of the count - is dropped separately, by the unblock verbs: ``loop.forget_timeout_count``.)""" - from .failures import PRE_MODEL_DISPATCH_CLASS, TOO_WIDE_CLASS + next timeout would park it again at once. ONLY that class resets it: the blocked + sweep's self-heal clears blocks too, and resetting there would stop a one-coder + board's timeouts from ever adding up. (The loop's in-process copy of the count is + dropped separately, by the unblock verbs: ``loop.forget_timeout_count``.)""" + from .failures import TOO_WIDE_CLASS f = self._require(fid) - labels = f.get("labels") or [] + labels = [str(label) for label in f.get("labels") or []] + classes = [label for label in labels if label.startswith(LABEL_BLOCKED_CLASS_PREFIX)] args = ["update", fid, "--remove-label", LABEL_BLOCKED] - cls = next((l.split(":", 1)[1] for l in labels if l.startswith(LABEL_BLOCKED_CLASS_PREFIX)), "") - if cls == PRE_MODEL_DISPATCH_CLASS: - # Drop only the stale infra class — NOT the tier labels, which predate the - # incident and record genuine model-capability escalation (#339). - args += ["--remove-label", f"{LABEL_BLOCKED_CLASS_PREFIX}{cls}"] - elif cls == TOO_WIDE_CLASS: - args += ["--remove-label", f"{LABEL_BLOCKED_CLASS_PREFIX}{cls}"] + for label in classes: + args += ["--remove-label", label] + if f"{LABEL_BLOCKED_CLASS_PREFIX}{TOO_WIDE_CLASS}" in classes: for label in labels: if label.startswith(f"{LABEL_BUDGET_PREFIX}timeout:"): args += ["--remove-label", label] self._run(*args) + if classes: + self.comment(fid, "unblocked — cleared " + ", ".join(classes)) return self.get_feature(fid) # ── escalation ladder (D10) — mechanical; the *policy* (whether to climb at @@ -3254,6 +3300,83 @@ def get_feature(self, fid: str) -> dict | None: return None return self._project(rows[0] if isinstance(rows, list) else rows) + def live_cards(self) -> list[dict]: + """The rows the agent's working-state snapshot reads, from the lightest read that + serves it (#401 review): ONE ``br list`` of the open and in-progress statuses. + ``list_features`` also batches a ``br show`` of every id, closed history included, + plus a ``br ready`` scan. That was the ~350-id call #404 caught stalling, and the + snapshot needs none of it: id, title, state, labels, pr_url, issue_type and assignee + all ride the list rows. ``dag_blocked`` is not computed here. + + ``br show`` is paid only for the rows whose hint reads what a list row lacks: a + blocked card's reason (a comment), and the dependency edges of a blocked or backlog + card, which say whether it is stranded with every dependency closed (#406). A + backlog row that reports no dependencies (``dependency_count`` 0) is skipped. When + such a card's dependencies are all closed they are fetched too, and a CANCELLED one + is returned with the open cards. ``annotate_next_action`` tells a scope cut from a + delivery by the cancelled cards in its listing, and a hint that took a cancelled + dependency for a merged one would tell the agent to promote work whose premise was + dropped. ``publish`` keeps only live states, so that row is never shown.""" + type_args: list[str] = [] + for itype in PULLABLE_ISSUE_TYPES: + type_args += ["--type", itype] + rows, has_more = self._run( + "list", + *type_args, + "--status", + "open", + "--status", + "in_progress", + "--limit", + "0", + want_json=True, + with_has_more=True, + ) + rows = rows or [] + if has_more: + raise BoardError( + "`br list --limit 0` reported has_more=true — the open-card query was truncated, so the " + "working-state snapshot would be incomplete (#114/#138)" + ) + + def wants_detail(row: dict) -> bool: + if LABEL_BLOCKED in (row.get("labels") or []): + return True + # An absent count (a `br` that does not report one) is read as "may have some". + return self.board_state(row) == "backlog" and row.get("dependency_count", 1) != 0 + + detail = self._show_by_id([r["id"] for r in rows if r.get("id") and wants_detail(r)]) + for r in rows: + shown = detail.get(r.get("id")) + if shown is not None: + r["dependencies"] = shown.get("dependencies") + if LABEL_BLOCKED in (r.get("labels") or []): + r["comments"] = shown.get("comments") + cards = [self._project(r) for r in rows] + closed_deps = { + dep + for f in cards + if f["board_state"] in ("backlog", "blocked") and f["depends_on"] and not f["open_depends_on"] + for dep in f["depends_on"] + } - {f["id"] for f in cards} + for row in self._show_by_id(sorted(closed_deps)).values(): + dep = self._project(row) + if dep["board_state"] == "cancelled": + cards.append(dep) + return cards + + def _show_by_id(self, ids: list[str]) -> dict[str, dict]: + """``br show`` rows for ``ids``, keyed by id, in calls of at most ``_SHOW_BATCH`` ids + (#404: no single read grows with the board). No ids, no call: `br show` with no + arguments is an error.""" + shown: dict[str, dict] = {} + for start in range(0, len(ids), _SHOW_BATCH): + batch = self._run("show", *ids[start : start + _SHOW_BATCH], want_json=True) or [] + if isinstance(batch, dict): # 0.1.x bare-dict single-bead path + batch = [batch] + shown.update((r["id"], r) for r in batch if isinstance(r, dict) and r.get("id")) + return shown + def list_features(self, state: str | None = None, include_archived: bool = False) -> list[dict]: """All feature rows for the board projection (every state, incl. the Done column); pass ``state`` to narrow the projection to one board state. @@ -3679,10 +3802,16 @@ def _project(self, bead: dict) -> dict: # off the single `blocked-class:` label, "" when the block predates it or the # caller never classified. Hyphenated on the label, hyphenated here — callers # compare against `failures.Policy.category` with the same normalisation. + # Reported ONLY while the card IS blocked (#401). A class left behind by a path + # that lifted the flag without it (an unblock before clear_blocked dropped every + # class, a merge) described no block at all, and read as `blocked: false, + # blocked_class: terminal`. blocked_class = next( (l[len(LABEL_BLOCKED_CLASS_PREFIX) :] for l in labels if l.startswith(LABEL_BLOCKED_CLASS_PREFIX)), "", ) + if LABEL_BLOCKED not in labels: + blocked_class = "" # A task-type bead's deliverable (#217): the LATEST `deliverable:` comment # (record_delivery's record — only `br show` carries comments, so a `br list` # row projects "") wins over a `deliverable:` label (the fallback for diff --git a/tests/test_external_seams.py b/tests/test_external_seams.py index ddefd8f..36e8120 100644 --- a/tests/test_external_seams.py +++ b/tests/test_external_seams.py @@ -135,16 +135,17 @@ "_find_by_external_ref": "UNCOVERED", "_open_blockers": "REAL", "_prepare_ready": "UNCOVERED", + "_show_by_id": "REAL", # #401: live_cards' detail reads, real `br` (test_work_snapshot_review_401.py) "add_dependency": "REAL", "archive_stale": "UNCOVERED", "attach_pr": "REAL", # #402: every shape through real `br` in tests/test_attach_pr_402.py - "block_from_review": "UNCOVERED", + "block_from_review": "REAL", # #401 review: real `br` in tests/test_work_snapshot_review_401.py "bounce_ci_fail": "UNCOVERED", "cancel_feature": "REAL", "claim": "REAL", "claim_next_ready": "REAL", "claim_task": "REAL", - "clear_blocked": "UNCOVERED", + "clear_blocked": "REAL", # #401: real `br` in tests/test_stale_blocked_class_401.py "clear_budgets": "REAL", "clear_verified_candidate": "UNCOVERED", "comment": "UNCOVERED", @@ -156,6 +157,7 @@ "flag_blocked": "REAL", "get_feature": "REAL", "list_features": "REAL", + "live_cards": "REAL", # #401 review: the snapshot's light read, real `br` (test_work_snapshot_review_401.py) "mark_designing": "UNCOVERED", "mark_done": "UNCOVERED", "mark_ready": "REAL", @@ -187,7 +189,7 @@ # seam is now REAL or an honestly-recorded EXEMPT, so the worktree UNCOVERED floor is 0 and a # newly-added UNCOVERED worktree seam fails this file outright. MAX_UNCOVERED_WORKTREE = 0 -MAX_UNCOVERED_STORE = 18 +MAX_UNCOVERED_STORE = 16 # The EXEMPT ratchet. An EXEMPT drops a seam from MAX_UNCOVERED_WORKTREE, so EXEMPT must itself be # bounded or the label would let real debt vanish (the review finding on the UNCOVERED → EXEMPT diff --git a/tests/test_loop.py b/tests/test_loop.py index ecd7e36..1222f98 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -5774,13 +5774,11 @@ async def test_sweep_publishes_the_work_snapshot_for_the_host_working_state(monk from project_board import work_snapshot class _Store(_SweepStore): - def list_features(self, state=None): - if state is None: # the snapshot read: the whole live board - return [ - {"id": "bd-live", "board_state": "blocked", "title": "Stuck card"}, - {"id": "bd-done", "board_state": "done", "title": "Finished"}, - ] - return super().list_features(state) + def live_cards(self): # the snapshot read: the open cards (#401) + return [ + {"id": "bd-live", "board_state": "blocked", "title": "Stuck card"}, + {"id": "bd-done", "board_state": "done", "title": "Finished"}, + ] work_snapshot.reset() store = _Store() diff --git a/tests/test_stale_blocked_class_401.py b/tests/test_stale_blocked_class_401.py new file mode 100644 index 0000000..33b2762 --- /dev/null +++ b/tests/test_stale_blocked_class_401.py @@ -0,0 +1,80 @@ +"""#401's secondary finding: an unblocked card still reported the class of a block it no longer had. + +bd-p8ft read `blocked: false, blocked_class: terminal` after it was unblocked. That is the +same kind of false working state as the stale snapshot: a card that is ready, described +as dead forever. `store.clear_blocked` dropped only the `dispatch-infra` class (#339), and +every other class outlived its block. The projection reported whatever class label was +left, blocked or not. + +The fix is at both ends. An unblock drops the block's class, whatever it is, and records +it on the unblock's audit comment. The projection reports a class only while the card is +actually blocked, which also covers a card a pre-fix unblock or a merge left carrying one. +The #339 posture is kept: an unblock never touches the card's earned `tier:` labels. All of +this runs through the real `br`, because a label is exactly what a fake `_run` gets wrong. +""" + +from __future__ import annotations + +import shutil + +import pytest + +from project_board import store as store_mod +from project_board.store import BeadsBoard + +requires_br = pytest.mark.skipif( + shutil.which(store_mod.BR) is None, + reason="real `br` (beads) CLI not on PATH — CI installs it and sets PB_REQUIRE_BR=1", +) + +_AC = "- WHEN x THE SYSTEM SHALL y" + + +def _ready(board: BeadsBoard, repo, title: str, path: str) -> str: + (repo / path).write_text("x = 1\n") + fid = board.create_feature(title, spec="s", acceptance_criteria=_AC, files_to_modify=[path])["id"] + board.mark_ready(fid) + return fid + + +def _classes(feature: dict) -> list[str]: + return [label for label in feature["labels"] if label.startswith("blocked-class:")] + + +@requires_br +def test_an_unblock_drops_the_blocks_class_and_keeps_the_earned_tiers(tmp_path): + """Whatever the class, it goes with the flag. The tiers a card climbed stay, because the + ladder is a record of real model-capability work (#339), including for dispatch-infra.""" + board = BeadsBoard(repo=str(tmp_path), actor="test") + for cls, path in (("terminal", "a.py"), ("dispatch-infra", "b.py"), ("transient", "c.py")): + fid = _ready(board, tmp_path, f"blocked {cls}", path) + board._run("update", fid, "--add-label", "tier:reasoning") # a rung it earned before the block + board.flag_blocked(fid, f"a {cls} failure", category=cls) + assert board.get_feature(fid)["blocked_class"] == cls # a blocked card reports its class + + f = board.clear_blocked(fid) + + assert f["board_state"] == "ready" and not f["blocked"], cls + assert _classes(f) == [] and f["blocked_class"] == "", f"{cls}: the class outlived its block" + assert "tier:reasoning" in f["labels"], f"{cls}: an unblock must never touch the earned tiers" + assert f"unblocked — cleared blocked-class:{cls}" in board.feature_comments(fid) + + +@requires_br +def test_a_class_left_on_an_unblocked_card_is_not_reported(tmp_path): + """Cards unblocked before this fix, or closed by a merge, still carry a class label. The + projection, which feeds board_get_feature, GET /features and the board view, must not + describe a block the card does not have. A card still blocked keeps reporting its class.""" + board = BeadsBoard(repo=str(tmp_path), actor="test") + fid = _ready(board, tmp_path, "the bd-p8ft shape", "p.py") + board.flag_blocked(fid, "open_review expects in_progress, got 'ready'", category="terminal") + listed = {f["id"]: f for f in board.list_features()} + assert listed[fid]["blocked_class"] == "terminal" + + board._run("update", fid, "--remove-label", "blocked") # what the pre-fix unblock left behind + f = board.get_feature(fid) + + assert _classes(f) == ["blocked-class:terminal"] # the stale label really is there… + assert f["board_state"] == "ready" and not f["blocked"] + assert f["blocked_class"] == "" # …and is not reported as the card's state + assert {r["id"]: r for r in board.list_features()}[fid]["blocked_class"] == "" diff --git a/tests/test_store.py b/tests/test_store.py index b2533d3..a21b52c 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -1942,7 +1942,8 @@ def test_clear_blocked_dispatch_infra_on_a_never_escalated_card_leaves_difficult def test_clear_blocked_leaves_tier_labels_on_a_model_reachable_block(make_board, monkeypatch): """A model-reachable block (a real capability escalation) keeps its `tier:` labels on unblock — the ladder record is genuine, so the next build resumes at that tier. - Only a pre-model infra block resets the posture.""" + Its block CLASS goes with the flag, like every class (#401): a class describes a block, + and one left on an unblocked card read as `blocked_class: terminal` on a ready card.""" br = Br() b = make_board(br) monkeypatch.setattr( @@ -1955,7 +1956,7 @@ def test_clear_blocked_leaves_tier_labels_on_a_model_reachable_block(make_board, (up,) = br.cmds("update") assert "blocked" in up assert "tier:opus" not in up # untouched — no tier reset for a model-reachable block - assert "blocked-class:terminal" not in up # the class label is left as-is too + assert "blocked-class:terminal" in up # the class goes with the flag (#401) def test_clear_blocked_unclassified_is_unchanged(make_board, monkeypatch): diff --git a/tests/test_work_snapshot_review_401.py b/tests/test_work_snapshot_review_401.py new file mode 100644 index 0000000..5c10c83 --- /dev/null +++ b/tests/test_work_snapshot_review_401.py @@ -0,0 +1,426 @@ +"""Review findings on the working-state snapshot fix (#401, PR #433). + +1. A failed refresh logged a traceback on every attempt and retried with no backoff. +2. The refresh was a full-board read (`br list` of every status + a `br show` of every id + + `br ready`), the ~350-id call #404 caught stalling, run after every write. +3. Key wiring was untested: removing the refresh call, reading the revision after the board, + or cutting the write set to {"update"} all passed the suite. +4. Nothing bounded staleness except in-process writes: a cross-process write was never seen, + and with `health_sweep_interval_s: 0` the snapshot never refreshed. +5. The refresh rode the claim tick, so a loop paused at its setup gate stayed STALE. A live + config reload that changes what the hints say didn't mark it stale. Blocked cards, ranked + first, had blank hints. + +Plus: `block_from_review` stamped no class, so an old one could resurface; the snapshot's +counter lived in module globals a reload could split; and boot released an escalation block +as an "orphaned preflight hold" when the card had once been preflight-held. +""" + +from __future__ import annotations + +import asyncio +import importlib +import logging +import shutil +import subprocess + +import pytest + +from project_board import store as store_mod +from project_board import work_snapshot +from project_board.loop import BoardLoop +from project_board.loop import reconcile as reconcile_mod +from project_board.store import BeadsBoard, BoardError + +requires_br = pytest.mark.skipif( + shutil.which(store_mod.BR) is None, + reason="real `br` (beads) CLI not on PATH — CI installs it and sets PB_REQUIRE_BR=1", +) + +_AC = "- WHEN x THE SYSTEM SHALL y" +_LOG = "protoagent.plugins.project_board" + + +@pytest.fixture(autouse=True) +def _fresh_snapshot(): + work_snapshot.reset() + yield + work_snapshot.reset() + + +def _ready(board: BeadsBoard, repo, title: str, path: str) -> str: + (repo / path).write_text("x = 1\n") + fid = board.create_feature(title, spec="s", acceptance_criteria=_AC, files_to_modify=[path])["id"] + board.mark_ready(fid) + return fid + + +def _stale(items) -> bool: + return bool(items) and isinstance(items[0], str) and items[0].startswith("STALE") + + +class _Cards: + """A store serving `live_cards`, counting the reads.""" + + def __init__(self, rows=None, *, fail: Exception | None = None): + self.rows = rows if rows is not None else [{"id": "bd-1", "title": "t", "board_state": "ready"}] + self.reads = 0 + self.fail = fail + + def live_cards(self): + self.reads += 1 + if self.fail is not None: + raise self.fail + return [dict(r) for r in self.rows] + + +# ── 1: a failed refresh is one line, and backs off ───────────────────────────────────── + + +@pytest.mark.parametrize( + "failure", + [ + store_mod.BoardTimeout("`br list …` timed out after 45s and was stopped — the board store did not answer"), + BoardError("`br list` failed: DATABASE_ERROR: database is locked"), + ], + ids=["stalled", "failed"], +) +async def test_a_failed_refresh_logs_one_line_and_backs_off(monkeypatch, caplog, failure): + """A stalled `br list` used to spew a traceback on every attempt, and the refresh was + retried with no backoff (26 reads in 0.3s). Now a BoardError, #431's BoardTimeout (a + stall, stopped) included, is one line, like `_tick_phase`, and the refresher waits out + an exponential backoff before the next read. Outside a tick a stall ends nothing.""" + store = _Cards(fail=failure) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + monkeypatch.setattr(reconcile_mod, "_SNAPSHOT_POLL_S", 0.01) + loop = BoardLoop({}) + + with caplog.at_level(logging.WARNING, logger=_LOG): + refresher = asyncio.create_task(loop._keep_work_snapshot_current()) + await asyncio.sleep(0.3) + loop._stop.set() + await refresher + + assert store.reads == 1, f"a failing store was read {store.reads} times in 0.3s" + failures = [r for r in caplog.records if "work snapshot refresh failed" in r.message] + assert len(failures) == 1 and failures[0].exc_info is None, "a BoardError must be one line, no traceback" + assert loop._snapshot_failures == 1 and not loop._work_snapshot_due() + + +# ── 2: the snapshot's read is the light one ──────────────────────────────────────────── + + +@requires_br +def test_the_snapshot_reads_only_the_open_cards_and_shows_only_the_blocked_ones(tmp_path, monkeypatch): + """The snapshot needs id, title, state, labels, pr_url, issue_type and assignee, all on + `br list` rows. It used to take the full-board read: every status, a `br show` of EVERY + id, and a `br ready` scan. Now it takes one `br list` of the open statuses, plus a batched + `br show` of only the cards whose hint needs more: here, the blocked one, for its reason.""" + board = BeadsBoard(repo=str(tmp_path), actor="test") + ready = _ready(board, tmp_path, "ready one", "a.py") + stuck = _ready(board, tmp_path, "stuck one", "b.py") + done = _ready(board, tmp_path, "done one", "c.py") + board.flag_blocked(stuck, "open_review expects in_progress, got 'ready'", category="terminal") + board.mark_done(board.claim(done)["id"], reason="shipped by hand") + calls = [] + real = board._shell_br + + def _record(*args, **kwargs): + calls.append(args[0]) + return real(*args, **kwargs) + + monkeypatch.setattr(board, "_shell_br", _record) + rows = {f["id"]: f for f in board.live_cards()} + + assert calls == ["list", "show"], calls # no whole-board show, no `br ready` + assert set(rows) == {ready, stuck} # the closed card is not read at all + assert rows[stuck]["blocked_reason"] == "open_review expects in_progress, got 'ready'" + + +@requires_br +async def test_the_refreshers_snapshot_names_a_stranded_card_and_a_cancelled_dependency(tmp_path, monkeypatch): + """#406 (merged after this PR opened) put a backlog card whose dependencies have all + closed in the working state, with the verb that moves it, off the full-board read. The + light read has no dependency edges (`br list` omits them), so it lost that card, and it + has no closed cards, so a dependency that was CANCELLED read as delivered: the hint told + the agent to promote work whose premise had been cut. The light read now shows the + backlog cards that have dependencies, and fetches the closed ones only when a card could + be stranded. A backlog card with no dependencies costs no `br show`.""" + board = BeadsBoard(repo=str(tmp_path), actor="test") + merged = board.create_feature("Record the origin session", spec="s")["id"] + cut = board.create_feature("Old cleanup path", spec="s")["id"] + stranded = board.create_feature("Wire cleanup through DELETE", spec="s", depends_on=[merged])["id"] + premise_cut = board.create_feature("Extend the old cleanup path", spec="s", depends_on=[cut])["id"] + loose = board.create_feature("Ordinary backlog", spec="s")["id"] + board._run("close", merged, "-r", "merged: https://github.com/o/r/pull/1") + board.cancel_feature(cut, "scope cut") + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: board) + shown_ids: list[str] = [] + real = board._shell_br + + def _record(*args, **kwargs): + if args[0] == "show": + shown_ids.extend(a for a in args[1:] if not a.startswith("-")) + return real(*args, **kwargs) + + monkeypatch.setattr(board, "_shell_br", _record) + + assert await BoardLoop({})._publish_work_snapshot() # the refresher's path, not the sweep + + items = {i["id"]: i for i in work_snapshot.provider() if isinstance(i, dict)} + assert items[stranded]["state"] == "backlog" and f"board_mark_ready({stranded})" in items[stranded]["hint"] + assert "CANCELLED" not in items[stranded]["hint"] + assert f"{cut} was CANCELLED, not merged" in items[premise_cut]["hint"] + assert loose not in items and cut not in items # nothing owed; a cancelled card is never shown + assert loose not in shown_ids, "a backlog card with no dependencies was shown" + + +def test_the_refresh_does_not_take_the_full_board_read(): + class _Store(_Cards): + def list_features(self, *a, **k): + raise AssertionError("the snapshot took the full-board read") + + store = _Store() + BoardLoop({})._take_work_snapshot(store) + assert store.reads == 1 and [c["id"] for c in work_snapshot.provider()] == ["bd-1"] + + +# ── 3: the wiring, pinned against the mutations review found survived ────────────────── + + +async def test_the_refresher_runs_while_the_loop_is_paused_and_picks_up_a_write(monkeypatch): + """Mutation killed: removing the refresher from `start()`. And #5a: the refresh used to + ride the claim tick, and a loop paused at its setup gate runs none.""" + store = _Cards() + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + monkeypatch.setattr(reconcile_mod, "_SNAPSHOT_POLL_S", 0.01) + monkeypatch.setattr(work_snapshot, "MIN_INTERVAL_S", 0.0) + loop = BoardLoop({"loop_enabled": True}) + + async def _paused(): # the setup gate: no ticks at all + await asyncio.Event().wait() + + monkeypatch.setattr(loop, "_run", _paused) + loop.start() + try: + for _ in range(100): + if work_snapshot.taken_at() is not None: + break + await asyncio.sleep(0.01) + store.rows = [{"id": "bd-2", "title": "moved", "board_state": "blocked"}] + work_snapshot.note_board_write() + for _ in range(100): + if not work_snapshot.needs_refresh(): + break + await asyncio.sleep(0.01) + shown = work_snapshot.provider() + finally: + await loop.stop() + + assert not _stale(shown) and [c["id"] for c in shown] == ["bd-2"], shown + + +async def test_one_snapshot_read_at_a_time(monkeypatch): + """The refresher and the sweep can both reach `_publish_work_snapshot`. A second read + while one is in flight is skipped, not queued behind it on the single-flight `br` lock: + the one in flight publishes, and a write since it began leaves that snapshot STALE for + the refresher to re-read.""" + import threading + + release = threading.Event() + + class _Slow(_Cards): + def live_cards(self): + assert release.wait(5), "the first read was never released" + return super().live_cards() + + store = _Slow() + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + loop = BoardLoop({}) + + first = asyncio.create_task(loop._publish_work_snapshot()) + for _ in range(100): + if loop._snapshot_reading: + break + await asyncio.sleep(0.01) + second = await loop._publish_work_snapshot() + release.set() + + assert second is False and await first is True + assert store.reads == 1 and [c["id"] for c in work_snapshot.provider()] == ["bd-1"] + + +def test_a_write_during_the_read_leaves_the_snapshot_stale(): + """Mutation killed: reading the revision AFTER the board. A write that lands mid-read + may or may not be in the rows, so the snapshot must not claim it.""" + + class _Racing(_Cards): + def live_cards(self): + rows = super().live_cards() + work_snapshot.note_board_write() # a tool writes while the board is being read + return rows + + BoardLoop({})._take_work_snapshot(_Racing()) + assert work_snapshot.needs_refresh() and _stale(work_snapshot.provider()) + + +def _stubbed_board(tmp_path, monkeypatch, *, stdout: str) -> BeadsBoard: + """A board whose `br` process (#431's ``_run_br_process``, the one place a `br` runs) + answers ``stdout`` at once: these pin which SUBCOMMANDS mark the snapshot, not `br`.""" + monkeypatch.setattr(store_mod.shutil, "which", lambda *_a, **_k: "/usr/bin/br") + board = BeadsBoard(db=None, repo=str(tmp_path)) + monkeypatch.setattr(board, "_ensure_workspace", lambda: None) + monkeypatch.setattr( + store_mod, + "_run_br_process", + lambda cmd, **_kw: subprocess.CompletedProcess(cmd, 0, stdout=stdout, stderr=""), + ) + return board + + +@pytest.mark.parametrize("subcommand", ["create", "update", "close", "reopen", "delete"]) +def test_every_write_that_can_move_a_card_marks_the_snapshot_stale(tmp_path, monkeypatch, subcommand): + """Mutation killed: cutting `_PROJECTION_WRITES` to {"update"}. A card is created, closed, + reopened and deleted by subcommands other than `update`.""" + board = _stubbed_board(tmp_path, monkeypatch, stdout="bd-1\n") + before = work_snapshot.board_revision() + board._run(subcommand, "bd-1") + assert work_snapshot.board_revision() > before + + +def test_reads_and_comments_do_not_mark_the_snapshot_stale(tmp_path, monkeypatch): + board = _stubbed_board(tmp_path, monkeypatch, stdout="[]") + before = work_snapshot.board_revision() + for args in (("list",), ("show", "bd-1"), ("ready",), ("comments", "add", "bd-1", "x"), ("dep", "add", "a", "b")): + board._run(*args) + assert work_snapshot.board_revision() == before + + +# ── 4: staleness is bounded whatever this process writes ─────────────────────────────── + + +async def test_a_quiet_board_is_still_re_read_on_the_age_bound(monkeypatch): + """A writer outside this process (a hand-run `br`) never bumps the revision, and with + `health_sweep_interval_s: 0` nothing re-read the board at all. The refresher re-reads a + quiet board every MAX_AGE_S.""" + store = _Cards() + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: store) + monkeypatch.setattr(reconcile_mod, "_SNAPSHOT_POLL_S", 0.01) + monkeypatch.setattr(work_snapshot, "MIN_INTERVAL_S", 0.0) + monkeypatch.setattr(work_snapshot, "MAX_AGE_S", 0.1) + loop = BoardLoop({"health_sweep_interval_s": 0}) + + refresher = asyncio.create_task(loop._keep_work_snapshot_current()) + await asyncio.sleep(0.45) + loop._stop.set() + await refresher + + assert store.reads >= 3, f"a quiet board was read {store.reads} time(s) in 0.45s with MAX_AGE_S=0.1" + + +def test_a_snapshot_the_refresher_stopped_renewing_says_it_is_stale(monkeypatch): + work_snapshot.publish([{"id": "bd-1", "title": "t", "board_state": "ready"}]) + assert not _stale(work_snapshot.provider()) + monkeypatch.setattr(work_snapshot, "STALE_AFTER_S", -1.0) # older than the bound + assert _stale(work_snapshot.provider()) + + +# ── 5b / 5c: what the snapshot renders ───────────────────────────────────────────────── + + +def test_a_live_config_change_the_hints_read_marks_the_snapshot_stale(): + """`auto_merge` decides whether an in_review card's hint says "merge #N" or "auto-merge + pending". A live reload changed it without a board write, and the snapshot kept the old + hint as current.""" + loop = BoardLoop({"auto_merge": False}) + loop._take_work_snapshot(_Cards()) + assert not work_snapshot.needs_refresh() + + assert loop.reload({"auto_merge": True}) == {"auto_merge": (False, True)} + + assert work_snapshot.needs_refresh() and _stale(work_snapshot.provider()) + + +def test_a_blocked_card_carries_its_reason_and_what_moves_it(): + """Blocked cards are ranked first, and they are the cards an agent most needs to act + on. Their hints were blank.""" + rows = [ + { + "id": "bd-t", + "title": "t", + "board_state": "blocked", + "blocked": True, + "blocked_class": "terminal", + "blocked_reason": "zombie drive", + }, + { + "id": "bd-s", + "title": "s", + "board_state": "blocked", + "blocked": True, + "blocked_class": "transient", + "blocked_reason": "coder timed out", + }, + ] + BoardLoop({})._take_work_snapshot(_Cards(rows)) + hints = {c["id"]: c["hint"] for c in work_snapshot.provider()} + assert hints == { + "bd-t": "needs a human (terminal): zombie drive", + "bd-s": "retries on its own (transient): coder timed out", + } + + +# ── the small items ──────────────────────────────────────────────────────────────────── + + +@requires_br +async def test_an_escalation_block_stamps_its_own_class_over_a_stale_one(tmp_path): + """(i) `block_from_review` added the flag and no class, so a stale `transient` left on + the card read as this block's class, and the sweep auto-healed a card whose automated + fixes were spent, when it should have paged a human.""" + board = BeadsBoard(repo=str(tmp_path), actor="test") + fid = _ready(board, tmp_path, "escalated", "e.py") + board.claim(fid) + board.open_review(fid, pr_url="https://github.com/o/r/pull/1") + board._run("update", fid, "--add-label", "blocked-class:transient") # left by a pre-fix unblock + + f = board.block_from_review(fid, "ci-fail: tests red at the top tier") + + assert f["blocked"] and f["blocked_class"] == "terminal" + assert [label for label in f["labels"] if label.startswith("blocked-class:")] == ["blocked-class:terminal"] + assert f["blocked_reason"] == "escalation exhausted: ci-fail: tests red at the top tier" + notified = [] + loop = BoardLoop({}) + loop._notify_operator = lambda fid_, text, **_kw: notified.append(fid_) + await loop._recover_blocked(board) + assert notified == [fid] and board.get_feature(fid)["blocked"] + + +def test_the_counter_survives_a_module_reload(): + """(ii) The revision lived in module globals. A plugin reload re-imports the module, and + the store (bumping) and the provider (reading) could end up on different counters.""" + work_snapshot.note_board_write() + before = work_snapshot.board_revision() + importlib.reload(work_snapshot) + assert work_snapshot.board_revision() == before + + +@requires_br +def test_boot_does_not_release_an_escalation_block_as_a_preflight_hold(tmp_path): + """(iv) The card was once held by a red preflight. Later its CI escalation was exhausted + and `block_from_review` blocked it. That wrote `escalation exhausted:`, not `blocked:`, + so the latest `blocked:` reason was still the old preflight hold, and a restart released + the card as an orphaned hold.""" + from project_board.loop import PREFLIGHT_BLOCK_PREFIX + + board = BeadsBoard(repo=str(tmp_path), actor="test") + fid = _ready(board, tmp_path, "once held", "h.py") + board.flag_blocked(fid, f"{PREFLIGHT_BLOCK_PREFIX} — the coder environment can't run the gate: tsc missing") + board.clear_blocked(fid) + board.claim(fid) + board.open_review(fid, pr_url="https://github.com/o/r/pull/2") + board.block_from_review(fid, "ci-fail: red at the top tier") + + BoardLoop({})._recover_preflight_holds(board) + + assert board.get_feature(fid)["blocked"], "an escalation block was released at boot as a preflight hold" diff --git a/tests/test_work_snapshot_staleness_401.py b/tests/test_work_snapshot_staleness_401.py new file mode 100644 index 0000000..0daa38d --- /dev/null +++ b/tests/test_work_snapshot_staleness_401.py @@ -0,0 +1,318 @@ +"""#401: the agent's working state showed board cards in states they had already left. + +Live on protoEngineer, 2026-09-07 07:15–07:17Z (audit log + agent.log): + +* 07:15:08 — the sweep published the snapshot, at the START of the sweep: bd-ezs7 + `in_progress`, bd-p8ft `blocked` (terminal). +* 07:16:40 — the PM itself called `board_unblock_feature(bd-p8ft)` → `ready`. +* 07:17:02 — the PM itself called `board_block_feature(bd-ezs7)` → `blocked`. +* 07:17:20 — the PM recorded "bd-p8ft is confirmed board_state=blocked / blocked_class=terminal" + and planned around it. Its injected working state still said so, and would until the next + sweep (300s later). It had made the transition itself 40 seconds earlier. + +The host reads the provider fresh on every turn (graph/work_providers.py). Nothing is cached +host-side. The staleness is the plugin's: the snapshot was refreshed only on the sweep, it +was taken before the sweep's own transitions, and nothing in it said how old it was. So two +cards that moved in OPPOSITE directions both read back in their old states. + +Now every board write this process makes (tool, route or loop, which all go through +`BeadsBoard._run`) bumps a revision. A snapshot read before that revision says it is STALE +instead of passing old states off as current. The loop's refresher republishes soon after, +and the sweep publishes after its own transitions. These drive the real `br` where the seam is +`br`. The review round's findings are in tests/test_work_snapshot_review_401.py. +""" + +from __future__ import annotations + +import asyncio +import logging +import shutil +import sys + +import pytest + +from project_board import store as store_mod +from project_board import work_snapshot, worktree +from project_board.loop import BoardLoop +from project_board.store import BeadsBoard + +requires_br = pytest.mark.skipif( + shutil.which(store_mod.BR) is None, + reason="real `br` (beads) CLI not on PATH — CI installs it and sets PB_REQUIRE_BR=1", +) + +_AC = "- WHEN x THE SYSTEM SHALL y" +_LOG = "protoagent.plugins.project_board" + + +@pytest.fixture(autouse=True) +def _fresh_snapshot(): + work_snapshot.reset() + yield + work_snapshot.reset() + + +def _cards(items) -> dict[str, str]: + return {i["id"]: i["state"] for i in items if isinstance(i, dict)} + + +def _status_line(items) -> str: + return items[0] if items and isinstance(items[0], str) else "" + + +def _ready(board: BeadsBoard, repo, title: str, path: str) -> str: + (repo / path).write_text("x = 1\n") + fid = board.create_feature(title, spec="s", acceptance_criteria=_AC, files_to_modify=[path])["id"] + board.mark_ready(fid) + return fid + + +@requires_br +async def test_the_agents_own_transitions_never_read_back_as_the_old_state(tmp_path, monkeypatch): + """The incident through real `br`: publish the board the way the sweep does, have the + PM's own tool edges move two cards in opposite directions, and read the working state + the PM's next turn would get.""" + board = BeadsBoard(repo=str(tmp_path), actor="test") + ezs7 = _ready(board, tmp_path, "make poll timeout a no-progress bound", "adapters.py") + p8ft = _ready(board, tmp_path, "record continuity entries", "conversations.py") + board.flag_blocked(p8ft, "open_review expects in_progress, got 'ready'", category="terminal") + + work_snapshot.publish(board.list_features()) # exactly what the sweep publishes + assert _cards(work_snapshot.provider()) == {ezs7: "ready", p8ft: "blocked"} + assert not _status_line(work_snapshot.provider()) # current → just the cards + + # What board_unblock_feature / board_block_feature call, 40 seconds apart. + board.clear_blocked(p8ft) + board.flag_blocked(ezs7, "zombie drive: no process, no branch, no PR", category="terminal") + + shown = work_snapshot.provider() + assert _status_line(shown).startswith("STALE"), ( + f"the working state presented pre-transition states as current: {shown}" + ) + assert "board_list" in _status_line(shown) # and says where the live record is + + # The loop's refresher brings it current: the cards as they ARE, no STALE line. + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: board) + loop = BoardLoop({}) + assert loop._work_snapshot_due() + await loop._publish_work_snapshot() + shown = work_snapshot.provider() + assert not _status_line(shown) + assert _cards(shown) == {ezs7: "blocked", p8ft: "ready"} + + +def test_a_write_that_lands_while_the_board_is_read_leaves_the_snapshot_stale(): + """The ordering the refresh relies on: the revision is read BEFORE the board. A write + that lands mid-read may or may not be in the rows, so the snapshot must not claim it.""" + revision = work_snapshot.board_revision() # the loop reads the revision first… + work_snapshot.note_board_write() # …a tool writes while the board is being listed… + work_snapshot.publish([{"id": "bd-1", "board_state": "ready", "title": "t"}], revision=revision) + + assert _status_line(work_snapshot.provider()).startswith("STALE") + assert work_snapshot.needs_refresh() + + +class _Board: + """A fake store whose writes move real (in-memory) state, so the snapshot can be + checked against what the board actually says after the sweep.""" + + def __init__(self, rows: dict[str, dict]): + self.rows = rows + + def list_features(self, state=None): + rows = [dict(r, id=fid) for fid, r in self.rows.items()] + return [r for r in rows if state is None or r["board_state"] == state] + + def live_cards(self): + return [r for r in self.list_features() if r["board_state"] not in ("done", "cancelled")] + + def get_feature(self, fid): + return dict(self.rows[fid], id=fid) if fid in self.rows else None + + def requeue(self, fid): + self.rows[fid]["board_state"] = "ready" + + def archive_stale(self, archive_after_days=7): + return [] + + +def _no_repo_side_effects(monkeypatch): + async def _no_pr(branch, *, cwd="."): + return "" + + monkeypatch.setattr(worktree, "pr_url_for_branch", _no_pr) + monkeypatch.setattr(worktree, "list_feature_worktrees", lambda repo, root: []) + + +async def test_the_sweep_publishes_after_its_own_transitions(monkeypatch): + """The sweep resets an orphaned in_progress card (no live drive, no PR) to ready. The + snapshot was taken BEFORE that, so the working state showed the card still being built + for a whole sweep interval.""" + board = _Board({"bd-orphan": {"title": "orphaned build", "board_state": "in_progress"}}) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: board) + _no_repo_side_effects(monkeypatch) + + await BoardLoop({})._sweep() + + assert board.rows["bd-orphan"]["board_state"] == "ready" # the sweep really moved it + assert _cards(work_snapshot.provider()) == {"bd-orphan": "ready"} + + +@pytest.fixture +def stalling_br(tmp_path, monkeypatch): + """The real `br`, until told to stall, the way #404's stalls looked. ``PB_STALL=`` + hangs that verb before it runs, and ``PB_STALL_ONCE=`` only its first call; + ``PB_STALL_AFTER=`` RUNS it (its write commits) and then hangs. The timeout drops to 1s and the TERM grace to 0.3s, so a stall costs a second + of #431's 45, and the store stops the child and raises ``BoardTimeout`` as it would live.""" + real = shutil.which(store_mod.BR) + once = tmp_path / "stalled-once" + script = tmp_path / "br-stalling" + script.write_text( + f"""#!/bin/sh +if [ -n "$PB_STALL" ] && [ "$1" = "$PB_STALL" ]; then exec sleep 60; fi +if [ -n "$PB_STALL_AFTER" ] && [ "$1" = "$PB_STALL_AFTER" ]; then "{real}" "$@" >/dev/null 2>&1; exec sleep 60; fi +if [ -n "$PB_STALL_ONCE" ] && [ "$1" = "$PB_STALL_ONCE" ] && [ ! -e "{once}" ]; then touch "{once}"; exec sleep 60; fi +exec "{real}" "$@" +""" + ) + script.chmod(0o755) + monkeypatch.setattr(store_mod, "BR", str(script)) + monkeypatch.setattr(store_mod, "_BR_TIMEOUT_S", 1.0) + monkeypatch.setattr(store_mod, "_BR_TERM_GRACE_S", 0.3) + monkeypatch.delenv("PB_STALL", raising=False) + monkeypatch.delenv("PB_STALL_AFTER", raising=False) + monkeypatch.delenv("PB_STALL_ONCE", raising=False) + + +@requires_br +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell wrapper") +async def test_a_write_that_stalls_after_landing_still_marks_the_snapshot_stale(tmp_path, monkeypatch, stalling_br): + """A `br` write can commit and then hang (#404). The store stops it and raises + ``BoardTimeout``, and the card HAS moved. The stale mark rides a `finally` in `_run`, so + the working state says it may be out of date instead of showing the card as it was. A + read that stalls moves nothing and marks nothing.""" + board = BeadsBoard(repo=str(tmp_path), actor="test") + fid = _ready(board, tmp_path, "card the PM blocks", "a.py") + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: board) + loop = BoardLoop({}) + assert await loop._publish_work_snapshot() + assert _cards(work_snapshot.provider()) == {fid: "ready"} and not _status_line(work_snapshot.provider()) + + before = work_snapshot.board_revision() + monkeypatch.setenv("PB_STALL", "list") + with pytest.raises(store_mod.BoardTimeout): + board.live_cards() + assert work_snapshot.board_revision() == before # a stalled read changes nothing + monkeypatch.delenv("PB_STALL") + + monkeypatch.setenv("PB_STALL_AFTER", "update") + with pytest.raises(store_mod.BoardTimeout): + board.flag_blocked(fid, "zombie drive: no process, no branch, no PR", category="terminal") + monkeypatch.delenv("PB_STALL_AFTER") + + assert board.get_feature(fid)["board_state"] == "blocked" # the write landed before the stall + assert _status_line(work_snapshot.provider()).startswith("STALE"), "a stalled write read back as current" + assert await loop._publish_work_snapshot() + assert _cards(work_snapshot.provider()) == {fid: "blocked"} and not _status_line(work_snapshot.provider()) + + +# ── with #431's stall handling (#404): the first stall ends the tick ──────────────────── + + +class _StalledSnapshot(_Board): + """A board every sweep pass reads fine, until the sweep's snapshot read stalls.""" + + def live_cards(self): + raise store_mod.BoardTimeout("`br list …` timed out after 45s and was stopped — the board store did not answer") + + +async def test_a_stall_in_the_sweeps_snapshot_read_ends_the_tick(monkeypatch, caplog): + """#431: a stalled store ends the tick at its first stall, because every later phase + would stall on it too, one timeout each. The sweep's own snapshot read swallowed its + stall as one more refresh failure, and the tick went on to the preflight and the claim + scan. It is the sweep's stall now: logged by the tick as one, and counted by the + refresher's backoff.""" + board = _StalledSnapshot({"bd-1": {"title": "ready card", "board_state": "ready"}}) + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: board) + _no_repo_side_effects(monkeypatch) + loop = BoardLoop({"merge_poll": False, "health_sweep_interval_s": 0.001}) + later: list[str] = [] + + async def _preflight(): + later.append("preflight") + + async def _claim_scan(): + later.append("claim scan") + return False + + monkeypatch.setattr(loop, "_maybe_preflight", _preflight) + monkeypatch.setattr(loop, "_spawn_ready", _claim_scan) + + with caplog.at_level(logging.WARNING, logger=_LOG): + assert await loop._tick() is False + + assert later == [], f"the tick went on past a stalled store: {later}" + [stall] = [r for r in caplog.records if "stalled on the board store" in r.message] + assert "health sweep" in stall.message and not any(r.exc_info for r in caplog.records) + assert loop._snapshot_failures == 1 + + +@requires_br +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX shell wrapper") +async def test_a_stalled_refresher_read_does_not_cost_the_claim_scan(tmp_path, monkeypatch, stalling_br, caplog): + """The refresher runs beside the ticks, as `start()` runs it, so its read can be the one + that stalls. That stall is the refresher's alone: one line, no traceback, a backoff, and + the refresher keeps running. The next tick still claims the card, and once the store + answers the snapshot comes back current. (#431's own end-to-end case, a stalled PR + reconcile read, is tests/test_br_timeout_404.py::test_after_a_stall_the_next_tick_claims.)""" + from project_board.loop import reconcile as reconcile_mod + + board = BeadsBoard(repo=str(tmp_path), actor="test") + fid = _ready(board, tmp_path, "the card to claim", "target.py") + monkeypatch.setattr("project_board.loop.get_store", lambda **_kw: board) + monkeypatch.setattr(reconcile_mod, "_SNAPSHOT_POLL_S", 0.01) + monkeypatch.setattr(work_snapshot, "MIN_INTERVAL_S", 0.0) + loop = BoardLoop( + { + "coder": "proto", + "repo": str(tmp_path), + "loop_enabled": True, + "merge_poll": True, + "merge_poll_interval_s": 0, + "health_sweep_interval_s": 0, + "preflight": False, + "max_pending_reviews": 0, + } + ) + dispatched: list[str] = [] + + async def _drive(feature): + dispatched.append(feature["id"]) + + monkeypatch.setattr(loop, "_drive", _drive) + monkeypatch.setenv("PB_STALL_ONCE", "list") + + with caplog.at_level(logging.WARNING, logger=_LOG): + refresher = asyncio.create_task(loop._keep_work_snapshot_current()) + try: + for _ in range(500): # the refresher reads first, and eats the one stall + if loop._snapshot_failures: + break + await asyncio.sleep(0.01) + assert loop._snapshot_failures == 1 and not refresher.done() + claimed = await loop._tick() + await asyncio.gather(*loop._drives) + for _ in range(500): + if work_snapshot.taken_at() is not None and not work_snapshot.needs_refresh(): + break + await asyncio.sleep(0.01) + shown = work_snapshot.provider() + finally: + loop._stop.set() + await refresher + + assert claimed is True and dispatched == [fid] + assert not _status_line(shown) and _cards(shown) == {fid: "in_progress"} + [stall] = [r for r in caplog.records if "timed out after 1s" in r.message] + assert "work snapshot refresh failed" in stall.message + assert not any(r.exc_info for r in caplog.records) diff --git a/work_snapshot.py b/work_snapshot.py index 91fafa2..50d978f 100644 --- a/work_snapshot.py +++ b/work_snapshot.py @@ -11,23 +11,45 @@ **The host calls a provider inline on EVERY turn, so it must be cheap and non-blocking.** Every board read is a ``br`` subprocess (``BoardStore._run`` is THE blocking seam), which is exactly what a provider may not do. So the provider never touches the store: the loop -publishes a snapshot on its periodic sweep and the provider returns that, already in memory. - -Consequences of that choice, both deliberate: - -- The snapshot is up to one sweep interval stale (``health_sweep_interval_s``, 300s by - default). For a "what am I on the hook for" hint that is fine — the board API and the - board tools remain the authoritative read. -- With the loop disabled nothing publishes, so the section is empty. That is honest: a - board nothing is driving is not a live commitment. - -Module-level rather than loop-instance state so a plugin reload swapping the loop object -does not strand the registered provider (it stays bound to this module's accessor). +publishes a snapshot and the provider returns that, already in memory. + +**How stale it can be, stated exactly (#401).** It used to be refreshed only by the health +sweep, every 300s, and at the START of the sweep, before the sweep's own transitions. +Nothing in it said how old it was. The PM unblocked one card and blocked another with its +own tools, and on its next turn its working state still showed both in their old states. +It believed that over its own tool results and planned against a board that no longer +existed. Now: + +- every store write that can change what a card projects as bumps a process-wide board + revision (``note_board_write``, from ``BeadsBoard._run``). Every write in this process + goes through there: the agent's tools, the operator's routes, the loop's own edges. So + does a live config change the hints read (``mark_stale``); +- a snapshot records the revision it was READ at, and while the board has moved past it + the provider leads with a STALE line naming when it was taken; +- the loop's refresher republishes after a change, at most every ``MIN_INTERVAL_S``, and in + any case at least every ``MAX_AGE_S``. It runs apart from the claim tick, so a loop paused + at its setup gate still refreshes, and it backs off while reads fail. + +So the snapshot is either marked STALE (a change this process made has not been picked up +yet) or at most ``MAX_AGE_S`` (+ one poll) old. The one thing the revision cannot see is a +writer outside this process, such as an operator's hand-run ``br``. Only that age bound +covers it. A snapshot older than ``STALE_AFTER_S`` (the refresher is failing, or has stopped) +is marked STALE too. With the loop disabled nothing publishes and the section stays empty. +That is honest: a board nothing is driving is not a live commitment. + +The state lives in a process-stable ``sys.modules`` slot (the ``store._br_lock`` / #178 +pattern). A plugin reload re-imports this module, and module globals would then give the +store (bumping) and the provider (reading) two different counters. """ from __future__ import annotations import logging +import sys +import threading +import time +import types +from datetime import datetime, timezone log = logging.getLogger("protoagent.plugins.project_board") @@ -41,19 +63,74 @@ # board with 90 live cards must not hand the host a 90-item list to trim every turn. MAX_ITEMS = 12 -_SNAPSHOT: list[dict] = [] - - -def publish(features) -> None: - """Called by the loop's sweep with the live board projection. Keeps only the live - states, orders them the way the board reasons about urgency (blocked first — a card - that cannot clear itself is the one the agent most needs to see, then in_review, - in_progress, ready), and trims to ``MAX_ITEMS``. Never raises: a bad snapshot must not - break the sweep.""" - global _SNAPSHOT +# The freshness bounds (#401). A change is picked up at most MIN_INTERVAL_S after the last +# refresh (changes are coalesced: a burst of writes costs one read). A quiet board is +# re-read every MAX_AGE_S, which is what bounds a writer this process can't see. A snapshot +# older than STALE_AFTER_S means the refresher is failing or stopped, and it says so. +MIN_INTERVAL_S = 5.0 +MAX_AGE_S = 60.0 +STALE_AFTER_S = 2 * MAX_AGE_S + +_SLOT_PREFIX = "project_board.work_snapshot::" + + +def _state(): + """The process-stable holder: ``snapshot`` (cards, taken_at, revision), ``revision`` + and its ``lock``. Installed atomically (``setdefault``, see ``store._br_lock``).""" + pkg = __name__.rsplit(".", 1)[0] if "." in __name__ else __name__ + name = _SLOT_PREFIX + pkg + holder = sys.modules.get(name) + if holder is None: + holder = types.ModuleType(name) + holder.__doc__ = "Process-stable holder for project_board's working-state snapshot (#401) — data, not code." + holder.snapshot = ([], None, 0) + holder.revision = 0 + holder.lock = threading.Lock() + holder = sys.modules.setdefault(name, holder) + return holder + + +def note_board_write() -> None: + """Record that the board changed. ``BeadsBoard._run`` calls it after every ``br`` write + that can move a card's state, title or labels. The call sits in a ``finally``, so it + also covers a write that raised or timed out and may still have landed.""" + state = _state() + with state.lock: + state.revision += 1 + + +def mark_stale() -> None: + """Record that what the snapshot RENDERS changed without a board write: a live config + reload of a knob the hints read (auto_merge, review_gate, …).""" + note_board_write() + + +def board_revision() -> int: + """The board revision right now. Read it BEFORE reading the board you publish.""" + return _state().revision + + +def taken_at() -> float | None: + """When the current snapshot was read (``time.time()``), or None before the first.""" + return _state().snapshot[1] + + +def publish(features, *, revision: int | None = None) -> None: + """Called by the loop with the live board projection. Keeps only the live states (and + a backlog card the board names a step for, #406), orders them the way the board reasons + about urgency (blocked first — a card that cannot clear itself is the one the agent most + needs to see, then in_review, in_progress, ready, and that backlog card last), and trims + to ``MAX_ITEMS``. Never raises: a bad snapshot must not break the refresher. + + ``revision`` is the ``board_revision()`` read BEFORE ``features`` was. A write that lands + while the board is being read may or may not be in those rows, so the snapshot counts as + stale from that write on, and never claims a state it might not include. Omitted means + the revision now (for a caller that reads with no concurrent writers).""" + if revision is None: + revision = board_revision() # PER-ITEM, not all-or-nothing. Building the whole list inside one try meant a single # malformed card (a None in the list, a non-dict row) aborted the entire update and left - # `_SNAPSHOT` holding its PREVIOUS value — so the agent kept being shown a stale board + # the snapshot holding its PREVIOUS value — so the agent kept being shown a stale board # indefinitely, with nothing in the working state to say so. One bad card must cost that # card, not the whole view. # A backlog card is normally not on the hook — except one the board says has a step @@ -92,16 +169,38 @@ def publish(features) -> None: skipped += 1 if skipped: log.warning("[project_board] work snapshot: skipped %d malformed feature row(s)", skipped) - _SNAPSHOT = built + _state().snapshot = (built, time.time(), revision) + + +def needs_refresh() -> bool: + """Whether the board changed since the snapshot was taken, or no snapshot was ever taken. + In memory, no I/O. The refresher adds the age bound (``MAX_AGE_S``) and its coalescing.""" + _cards, taken, revision = _state().snapshot + return taken is None or revision != board_revision() + +def provider() -> list: + """The registered work provider: an in-memory read, no I/O, no lock. -def provider() -> list[dict]: - """The registered work provider: an in-memory read, no I/O, no lock.""" - return list(_SNAPSHOT) + The cards, as dicts. A snapshot that is stale is led by ONE plain-string line saying so + and naming when it was taken: the board changed since it was read, or it is older than + ``STALE_AFTER_S``. It is the first item so a host that caps a provider's items can never + trim the warning off. Once the loop has republished, the cards are current and the line + is gone.""" + cards, taken, revision = _state().snapshot + if taken is None: + return list(cards) + if revision == board_revision() and time.time() - taken <= STALE_AFTER_S: + return list(cards) + stamp = datetime.fromtimestamp(taken, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + return [ + f"STALE — the board may have changed since this list was taken ({stamp}); " + "re-read with board_list / board_get_feature before acting on it", + *cards, + ] def reset() -> None: """Drop the snapshot — used when the loop stops, so a stopped board stops advertising work it is no longer driving.""" - global _SNAPSHOT - _SNAPSHOT = [] + _state().snapshot = ([], None, 0)