diff --git a/changelog.d/429.md b/changelog.d/429.md new file mode 100644 index 0000000..386c961 --- /dev/null +++ b/changelog.d/429.md @@ -0,0 +1,12 @@ +Max-mode now tells the loop why its candidates failed, instead of reporting "no diff". It used to +swallow each candidate's error, so when every candidate failed the loop saw a capability failure +and climbed a tier. That happened on a spent quota or a model the provider refuses, while a working +sibling at the same rung sat idle. It also hid timeouts from the decompose ask, and seam failures +from the pre-model block. When every candidate raises, the loop now takes the edge of the most +specific failure among them, as it would for a single dispatch. In order, that is a refused model +(even beside a quota), a quota, a timeout, a dispatch failure the loop does not retry, a retryable +one, then any other board error. A raw non-board error keeps the old verdict, so shutdown and +cancel still win. Only a fan-out where at least one candidate ran and came back empty is certain +to climb as a capability failure. A climb after a timeout now fans out again at the new rung. Its +"previous attempt timed out" note describes the candidate that actually timed out, and no longer +follows the card into later drives. diff --git a/coder_seam.py b/coder_seam.py index 6414460..6dfe7a2 100644 --- a/coder_seam.py +++ b/coder_seam.py @@ -495,6 +495,16 @@ def progress_stop_reason(fid: str | None, gen: int, reason) -> None: b.stop_reason = str(reason)[:200] +# The stop reason the dispatch stamps on a gen its watchdog killed. The loop's timeout +# note (#146) is mined from THAT gen — in a max-mode fan-out the last gen may be a +# sibling that failed fast on something else (#425 review). +TIMED_OUT_REASON = "timed out" + + +def _stamp_timeout(fid: str | None, gen: int, timeout) -> None: + progress_stop_reason(fid, gen, f"{TIMED_OUT_REASON} after {timeout}s") + + # ── Persist finished gens to the bead (#226) ──────────────────────────────────── # The WRITE side of the coder-monitor history (#226): when a gen finishes, its live # snapshot is serialized to a `coder-monitor: {…}` JSON bead comment so the drawer @@ -751,6 +761,7 @@ async def _tool_cb(event): # client + SIGKILLed the tree on its way out — nothing for the board to clean up. raise except asyncio.TimeoutError: + _stamp_timeout(fid, gen, timeout) raise worktree.CoderTimeout(f"coder timed out after {timeout}s") except Exception as exc: # noqa: BLE001 — normalise every below-seam failure to the # adapter path's contract: nothing propagates raw; a dispatch failure surfaces as @@ -843,6 +854,9 @@ async def _dispatch_coder_tapped_legacy( return await worktree.dispatch_coder( coder, worktree_path, prompt, timeout=timeout, env_passthrough=env_passthrough ) + except worktree.CoderTimeout: + _stamp_timeout(fid, gen, timeout) + raise finally: progress_end(fid, gen) # the gen must close on EVERY exit path (panel: orphaned gens) @@ -881,6 +895,7 @@ async def _tool_cb(event): # client requires a number (it logs int(timeout)), so "unbounded" rides a 24h # sentinel instead of the 600s floor the first tap draft imposed (panel round 2). prompt_timeout = timeout or getattr(scoped, "timeout_s", None) or 86400.0 + timed_out = False try: coro = client.prompt( prompt, @@ -902,6 +917,7 @@ async def _tool_cb(event): pass raise except asyncio.TimeoutError: + timed_out = True raise worktree.CoderTimeout(f"coder timed out after {timeout}s") except (AcpError, DelegateError) as exc: raise worktree.WorktreeError(f"coder dispatch failed: {exc}") @@ -913,9 +929,12 @@ async def _tool_cb(event): # Stash whatever stop-reason / dead-end signal the ACP client reports (#198) # — sampled on EVERY exit so an empty reply still records WHY the coder # stopped. Best-effort getattr: a host without the attribute yields None. - progress_stop_reason( - fid, gen, getattr(client, "last_stop_reason", None) or getattr(client, "last_dead_end", None) - ) + if timed_out: + _stamp_timeout(fid, gen, timeout) # the pooled client's last reason is not THIS turn's + else: + progress_stop_reason( + fid, gen, getattr(client, "last_stop_reason", None) or getattr(client, "last_dead_end", None) + ) progress_end(fid, gen) try: await adapter.teardown(scoped) # #1 lifecycle rule: reap the worktree-scoped subprocess diff --git a/docs/configuration.md b/docs/configuration.md index 8093471..0d07faa 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -238,8 +238,22 @@ the quota backoff is spent — a marked provider still gets its one real attempt card blocks (the operator may have repointed it), and a dispatch it serves clears the mark. If every provider on the rung refused its model on this card, the card blocks under `dispatch-infra` naming them, and does not climb: that is a config problem, and a stronger rung would only hide it. If some were only rate-limited, it -blocks as `rate_limit`, which the sweep heals on its own. Max-mode swallows each candidate's -error, so neither rotation applies there. +blocks as `rate_limit`, which the sweep heals on its own. + +Max-mode (`max_mode_n > 1`) follows the same rules. If every candidate raised, the loop takes +the edge of the most specific failure among them, exactly as it would for a single dispatch. +The precedence is: + +1. a refused model, even beside a quota; +2. a quota; +3. a timeout; +4. a dispatch failure the loop does not retry; +5. a retryable one; +6. any other board error, as raised. + +A raw error that is not a board error keeps the old "no diff" verdict. Only a fan-out where +at least one candidate ran and came back with nothing is guaranteed to be a capability failure +that climbs. A climb after a timeout is a fresh build, so the stronger rung fans out again. A card's STARTING rung comes from its difficulty, so a `medium` card never touches rung 1. diff --git a/loop/_common.py b/loop/_common.py index 7d8a3fa..b73b18d 100644 --- a/loop/_common.py +++ b/loop/_common.py @@ -835,6 +835,56 @@ def _next_rung_cursor() -> int: _ROTATABLE_CATEGORIES = frozenset({"rate_limit", "provider_unavailable"}) +def provider_failure_category(exc: BaseException) -> str | None: + """The provider class (``rate_limit`` / ``provider_unavailable``) of a coder DISPATCH + failure, or ``None`` when ``exc`` is anything else — the one definition of "the + provider failed, not the model". The drive rotates on it; max-mode asks it of every + candidate before it decides what the drive is told (#425).""" + text = str(exc) + if not text.startswith("coder dispatch failed"): + return None + category = classify(text).category + return category if category in _ROTATABLE_CATEGORIES else None + + +def representative_failure(errors: list[Exception]) -> Exception | None: + """The ONE error that speaks for a max-mode fan-out in which EVERY candidate raised + (#425). No candidate returned anything to judge, so the drive is handed this instead + of a "no diff" and its own handling applies, as for a single dispatch. The most + specific edge wins: + + 0. a provider that refused its model — beside a quota too: the provider is at least + partly broken, and a capability climb is wrong either way; + 1. a spent quota — rotate within the rung, or back off; + 2. a timeout — so #378's timeout counter, and its decompose ask, see it; + 3. a dispatch failure the drive does not retry — #339 blocks it as pre-model unless a + candidate reached the model; + 4. any other dispatch failure (a retryable one: back off and re-run); + 5. any other ``WorktreeError``, as it was raised. + + Ties go to the earliest candidate. Only a ``WorktreeError`` can speak: anything else + (a raw ``BrokenPipeError`` from an untapped host, say) returns None, and the caller + keeps its old "no diff" verdict — which the drive's shutdown and cancel checks see + first, where a raw error would skip them and block the card as `unexpected`. Pure, + like ``rotation_target``, so the order is testable without driving a card.""" + + def rank(exc: Exception) -> int: + category = provider_failure_category(exc) + if category == "provider_unavailable": + return 0 + if category == "rate_limit": + return 1 + if isinstance(exc, worktree.CoderTimeout): + return 2 + text = str(exc) + if text.startswith("coder dispatch failed"): + return 4 if classify(text).retryable else 3 + return 5 + + speakers = [e for e in errors if isinstance(e, worktree.WorktreeError)] + return min(speakers, key=rank) if speakers else None + + # ── #420: remember a provider that can't serve its model ──────────────────────────── # Rotation alone rediscovers a dead provider card by card: the rung cursor spreads cards # across a rung's siblings, so every card that happens to open on it pays one failed @@ -1263,6 +1313,8 @@ def _inbox_db_path(): "rotation_target", "prefer_live_sibling", "_ROTATABLE_CATEGORIES", + "provider_failure_category", + "representative_failure", "_next_rung_cursor", "_PROVIDER_DOWN_TTL_S", "_PROVIDER_DOWN", diff --git a/loop/drive.py b/loop/drive.py index 47d8ff3..f921dfc 100644 --- a/loop/drive.py +++ b/loop/drive.py @@ -1116,6 +1116,12 @@ async def _drive(self, feature: dict): wt = branch = None pr_url = None # set once open_pr returns — the cancel paths below close it (#211) keep_wt = False # reuse the worktree on a goal-fix retry (keep the impl; add tests) + # "A prior attempt timed out" (#146) for the NEXT dispatch after a timeout climb. + # Drive-local on purpose (#425 review): it rode `_ci_feedback`, which persists + # across drives and marks a carried-forward FIX — so one all-timeout fan-out + # switched max-mode off for that card for good, and every later drive still + # opened with "a PREVIOUS attempt TIMED OUT". A timeout climb is a fresh build. + timeout_note = "" try: while True: # Rebuild the prompt each attempt so a re-dispatch (CI bounce, @@ -1123,7 +1129,7 @@ async def _drive(self, feature: dict): # _ci_feedback + _ci_prior_diff. Fetch this area's distilled lessons # from the KG (best-effort, async) and inject them — the flywheel READ. lessons = await self._fetch_kg_lessons(feature) - prompt = self._build_prompt(feature, lessons=lessons) + prompt = self._build_prompt(feature, lessons=lessons, timeout_note=timeout_note) # Which PROVIDER at this rung. `siblings` are interchangeable delegates for # the same capability tier (#362); `sib` advances only on a provider failure # (below) or round-robin across dispatches, never on a capability failure — @@ -1278,6 +1284,7 @@ async def _drive(self, feature: dict): # dispatch scheduled on worker threads land before the drive # proceeds toward open_pr (the pre-offload ordering). await self._await_bg_records(fid) + timeout_note = "" # the dispatch that carried it has run — consumed # The provider served this dispatch, so a down mark on it (#420) — set # by another card, or one that was stale when every sibling was marked # and this one got the rung's last attempt — no longer holds. @@ -1567,7 +1574,7 @@ async def _drive(self, feature: dict): dispatch_error = str(exc).startswith("coder dispatch failed") if policy.category == "provider_unavailable" and not dispatch_error: policy = classify(str(exc), provider_rules=False) - provider_failure = dispatch_error and policy.category in _ROTATABLE_CATEGORIES + provider_failure = provider_failure_category(exc) is not None # A dispatch that failed on a fix round's KEPT worktree left it untouched: # its files hold the implementation and the feedback says so. Whatever # re-dispatches next — a sibling, or the same provider after a backoff — @@ -1641,6 +1648,7 @@ async def _drive(self, feature: dict): ) tier = nxt retries = 0 + timeout_note = "" # a new rung: an older timeout note is stale spent_here.clear() # a NEW rung has its own providers (#362) refused_here.clear() # Fresh per-tier budgets on the climb — mirrors the @@ -1835,15 +1843,18 @@ async def _drive(self, feature: dict): # about a worktree that no longer exists, so clear it (+ the prior diff) # to keep prompt and worktree consistent. keep_wt_class = str(exc).startswith(("goal verification failed", "requirements unresolved")) + timeout_note = "" # a new rung: an older timeout note is stale if keep_wt_class and wt is not None: keep_wt = True # reuse the verified worktree; _ci_feedback is truthful elif isinstance(exc, worktree.CoderTimeout): # A timeout carries NO diff and NO CI output, so the stronger tier # would otherwise get a BYTE-IDENTICAL prompt — blind to the fact a # prior attempt ran out of time and what it was doing when killed - # (#146). Seed the CI/review-bounce feedback lever with the ring - # buffer's timeout context so the escalated dispatch leads with it. - self._ci_feedback[fid] = self._timeout_escalation_context(fid) + # (#146). Lead the escalated dispatch with the ring buffer's timeout + # context — as a drive-local note, NOT `_ci_feedback`: this is a fresh + # build, so a max-mode/solve board fans out again at the new rung. + timeout_note = self._timeout_escalation_context(fid) + self._ci_feedback.pop(fid, None) # fresh worktree ahead: no fix to carry self._ci_prior_diff.pop(fid, None) # a timeout produced no diff to echo back else: # Fresh worktree ahead — drop any gate-fix feedback describing the @@ -2241,7 +2252,10 @@ async def _dispatch_max_mode( pass; ADR 0064), else the best-of-N LLM judge; the winner is PROMOTED into the canonical ``feat-`` worktree / ``feat/`` branch (so the rest of the lifecycle is unchanged) and the losers are reaped. All-empty → ``NoChangesError``, which - ``_drive`` escalates/blocks exactly like a single coder that produced nothing. + ``_drive`` escalates/blocks exactly like a single coder that produced nothing — + unless EVERY candidate raised, when the one error that speaks for them + (``representative_failure``) is re-raised instead, so the drive handles it as it + would a single dispatch's (#425). Returns (canonical_wt, canonical_branch, winner_reply). The fan-out is bounded by ``max_concurrent`` × ``max_mode_n`` coders; size those to the host.""" @@ -2271,6 +2285,26 @@ async def _dispatch_max_mode( if idx is None: for cid in cand_ids: await worktree.reap_feature_worktree(repo, self.root, cid) + # #425: "no diff" is a CAPABILITY verdict, and the drive climbs a rung on it. That is + # only true when a candidate RETURNED — ran, and came back with nothing. When EVERY + # candidate raised there is nothing to judge, and swallowing the errors hid what + # actually happened: a quota (#362) or a refused model (#420) climbed instead of + # rotating, a timeout never reached #378's counter, a pre-model seam failure never + # reached #339's block. So hand the drive ONE of those errors — the most specific + # edge, `representative_failure` — and its own handling applies exactly as for a + # single dispatch. A CancelledError child neither returned nor raised, and a raw + # (non-WorktreeError) error can't speak for the fan-out: both keep the old verdict. + raised = [r for r in results if isinstance(r, Exception)] + rep = representative_failure(raised) if len(raised) == len(results) else None + if rep is not None: + log.info( + "[project_board] %s max-mode: all %d candidates raised — handing the drive the " + "failure that speaks for them, not a no-diff: %s", + fid, + n, + str(rep)[:160], + ) + raise rep raise worktree.NoChangesError(f"max-mode: all {n} candidates produced no diff") log.info("[project_board] %s max-mode: candidate %d/%d wins → promoting", fid, idx, n) win_wt, win_branch = cands[idx] @@ -2361,16 +2395,22 @@ def _timeout_escalation_context(self, fid: str) -> str: attempt ran out of time, how long it ran, or what it was doing when killed. Mine the progress ring buffer (``coder_seam.progress_snapshot``) for the timed-out gen's elapsed time, the last tool in flight, and the thought tail, - and lead the re-dispatch with them. Returned for injection into - ``_ci_feedback`` so it rides the exact same prompt path a CI/review bounce - uses — no new plumbing. Best-effort: a missing/empty snapshot still yields a - usable "prior attempt timed out, produced no diff" note — a monitor read must - never break escalation.""" + and lead the re-dispatch with them. Returned as the drive's timeout note, which + rides the same rejected-attempt block a CI/review bounce uses. Best-effort: a + missing/empty snapshot still yields a usable "prior attempt timed out, produced + no diff" note — a monitor read must never break escalation. + + Mined from the gen that TIMED OUT — the seam stamps its stop reason — not simply + the last one: in a max-mode fan-out that is just the last candidate, which may + have failed fast on something else, and the note then described the wrong + attempt ("ran ~0.0s", no tool; #425 review). The last gen is only the fallback + for a buffer that carries no stamp (an older gen, an untapped host).""" try: gens = coder_seam.progress_snapshot(fid).get("gens") or [] except Exception: # noqa: BLE001 — a monitor read must never break escalation gens = [] - gen = gens[-1] if gens else {} + timed_out = [g for g in gens if str(g.get("stop_reason") or "").startswith(coder_seam.TIMED_OUT_REASON)] + gen = (timed_out or gens or [{}])[-1] elapsed = gen.get("elapsed_s") ran_for = f"ran ~{elapsed}s and " if elapsed is not None else "" lines = [ diff --git a/loop/prompt.py b/loop/prompt.py index 70b1f60..e69e1d8 100644 --- a/loop/prompt.py +++ b/loop/prompt.py @@ -37,14 +37,20 @@ def _build_task_prompt(self, feature: dict) -> str: f"{criteria_block}" ) - def _build_prompt(self, feature: dict, lessons: str = "") -> str: + def _build_prompt(self, feature: dict, lessons: str = "", timeout_note: str = "") -> str: """An imperative, fully-specified instruction (ProtoMaker discipline). A passive 'implement this feature' + a vague spec makes a coder produce nothing; naming the files + a direct 'make the edits now' makes it act. ``lessons`` (distilled gotchas from the knowledge graph, fetched async in ``_drive``) is injected so a coder gets this area's known failure modes on - attempt 1 — the read half of the flywheel (retro grounds → coder heeds).""" + attempt 1 — the read half of the flywheel (retro grounds → coder heeds). + + ``timeout_note`` is the drive's note that a PRIOR attempt timed out (#146). It + rides the same rejected-attempt block as CI/fix feedback, but it is NOT that + feedback: ``_ci_feedback`` persists across drives and marks a carried-forward + FIX (which disables max-mode/solve fan-out), while a timeout climb is a fresh + build — so the note lives only in the drive that climbed (#425 review).""" files = feature.get("files_to_modify") or [] files_block = ( "\n".join(f"- {f}" for f in files) if files else "(none listed — create the files the task requires)" @@ -112,7 +118,7 @@ def _build_prompt(self, feature: dict, lessons: str = "") -> str: pending = _PENDING_FEEDBACK.pop(fid, None) if pending: self._ci_feedback[fid] = pending - ci = self._ci_feedback.get(fid) + ci = self._ci_feedback.get(fid) or timeout_note prior = self._ci_prior_diff.get(fid) prior_block = ( f"\n### The diff that failed (your previous attempt — fix it, don't restart from scratch)\n" diff --git a/tests/test_loop.py b/tests/test_loop.py index 78fae08..840d3e8 100644 --- a/tests/test_loop.py +++ b/tests/test_loop.py @@ -2611,9 +2611,10 @@ async def _dispatch(c, wt, prompt, *, timeout=None, env_passthrough=()): assert "produced NO diff" in escalated assert "Read" in escalated and "loop.py" in escalated assert "still mapping the dispatch flow" in escalated - # r3: it arrived via `_ci_feedback`, so it rides the standard rejected-attempt block. + # r3: it rides the standard rejected-attempt block — but as the drive's own note, NOT + # `_ci_feedback`, which persists and would switch fan-out off for the card (#425 review). assert "previous attempt was REJECTED" in escalated - assert "still mapping the dispatch flow" in loop._ci_feedback.get("bd-1", "") + assert "bd-1" not in loop._ci_feedback async def test_drive_tier_climb_grants_a_fresh_window_despite_stale_budget_labels(monkeypatch): diff --git a/tests/test_max_mode_provider_425.py b/tests/test_max_mode_provider_425.py new file mode 100644 index 0000000..54841eb --- /dev/null +++ b/tests/test_max_mode_provider_425.py @@ -0,0 +1,558 @@ +"""#425: max-mode must hand the drive what killed its candidates, not a "no diff". + +Max-mode builds a card N ways at once and swallows each candidate's error. With every +candidate dead the drive saw only `NoChangesError("max-mode: all N candidates produced +no diff")` — a CAPABILITY verdict — and climbed a rung. When they all died because the +PROVIDER did (a spent quota, #362; a model it can't serve, #420) that says nothing about +the model, and the single-dispatch path already rotates within the rung on both. The +review of #421 reproduced it: rung `[codex, …]`, `max_mode_n: 2`, codex refusing its +model — the drive dispatched `codex, codex`, escalated, then `opus, opus`, escalated +again. No mark, no rotation. The same swallow hid a timeout from #378's counter and a +pre-model seam failure from #339's block. + +When EVERY candidate raised, max-mode now re-raises the one error that speaks for them +(`representative_failure`, most specific edge first). Only a candidate that RETURNED — +ran, and came back with nothing — keeps the capability verdict. +""" + +from __future__ import annotations + +import asyncio +import dataclasses + +import pytest + +from project_board import coder_seam, worktree +import project_board.loop as loop_mod + +from test_loop import _DEAD_MODEL, _SESSION_LIMIT, _rung_env + +_MAX_LADDER = {"coders": {"smart": ["codex", "sonnet"], "reasoning": ["opus"]}, "max_mode_n": 2} +_TIMEOUT = "coder timed out after 1800s" +_SEAM = "coder dispatch failed: adapter rejected the session (unknown error)" # terminal +_RESET = "coder dispatch failed: connection reset by peer" # transient → retryable + + +def _blocks(store): + return [c for c in store.calls if c[0] == "flag_blocked"] + + +def _no_diffs(monkeypatch, loop): + """Every candidate worktree comes back empty — the only way max-mode reaches its + all-failed edge. (The judge is the selector's empty-check when no gate is set.)""" + + async def _judge(feature, base, worktrees): + return None + + monkeypatch.setattr(loop, "_judge_candidates", _judge) + + +def _record_sleeps(monkeypatch) -> list[float]: + slept: list[float] = [] + + async def _sleep(delay): + slept.append(delay) + + monkeypatch.setattr("project_board.loop.asyncio.sleep", _sleep) + return slept + + +# ── precedence 0: a refused model ────────────────────────────────────────────────────── + + +async def test_max_mode_rotates_past_a_refusing_provider_instead_of_climbing(monkeypatch): + """The reproduction from the #421 review. Both candidates on codex are refused, so the + card moves to the sibling at the SAME rung — ladder untouched — and codex is marked, so + the next card starts on sonnet instead of paying to rediscover it.""" + seen: list[tuple[str, int]] = [] # (coder, climbs so far) per candidate dispatch + + async def _dispatch(coder, wt, prompt, **kw): + seen.append((coder, len(store.escalated))) + if coder == "codex": + raise worktree.WorktreeError(_DEAD_MODEL) + return "" # sonnet runs, and comes back empty — ends the drive + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-mm", "title": "t", "spec": "s"}) + + assert seen[:4] == [("codex", 0), ("codex", 0), ("sonnet", 0), ("sonnet", 0)], seen + assert loop_mod.provider_is_down("codex") + + +async def test_a_refusal_beside_a_quota_takes_the_refusal_edge(monkeypatch): + """One candidate refused, the other rate-limited: the provider is at least partly + broken, and a capability climb is wrong either way. The refusal speaks — rotate and + mark — rather than the quota's backoff on a provider that refuses its model.""" + seen: list[tuple[str, int]] = [] + + async def _dispatch(coder, wt, prompt, **kw): + seen.append((coder, len(store.escalated))) + if coder == "codex": + raise worktree.WorktreeError(_SESSION_LIMIT if kw.get("gen") == 1 else _DEAD_MODEL) + return "" + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + slept = _record_sleeps(monkeypatch) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-md", "title": "t", "spec": "s"}) + + assert seen[:4] == [("codex", 0), ("codex", 0), ("sonnet", 0), ("sonnet", 0)], seen + assert loop_mod.provider_is_down("codex") and slept == [] + + +async def test_max_mode_on_a_one_provider_board_blocks_a_dead_provider_as_infra(monkeypatch): + """No sibling to rotate to: the card blocks the way a single dispatch does — under + `dispatch-infra`, naming the fix — instead of as an unexplained `terminal` no-diff.""" + seen: list[str] = [] + + async def _dispatch(coder, wt, prompt, **kw): + seen.append(coder) + raise worktree.WorktreeError(_DEAD_MODEL) + + loop, store = _rung_env(monkeypatch, _dispatch, cfg={"coder": "codex", "max_mode_n": 2}) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-m1", "title": "t", "spec": "s"}) + + assert seen == ["codex", "codex"] # one fan-out, then the block + blocked = _blocks(store) + assert len(blocked) == 1 and blocked[0][3] == "dispatch-infra", blocked + assert blocked[0][2].startswith("provider unavailable") + assert "does not exist or you do not have access" in blocked[0][2] # the evidence rides along + assert store.escalated == [] + + +# ── precedence 1: a spent quota ──────────────────────────────────────────────────────── + + +async def test_max_mode_quota_on_every_candidate_switches_provider_without_a_backoff(monkeypatch): + """A spent quota: every candidate rate-limited means the provider is out, not that the + model failed. Switch to the sibling at once — no 60s backoff on the exhausted provider, + and no climb.""" + seen: list[tuple[str, int]] = [] + + async def _dispatch(coder, wt, prompt, **kw): + seen.append((coder, len(store.escalated))) + if coder == "codex": + raise worktree.WorktreeError(_SESSION_LIMIT) + return "" + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + slept = _record_sleeps(monkeypatch) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-mq", "title": "t", "spec": "s"}) + + assert seen[:4] == [("codex", 0), ("codex", 0), ("sonnet", 0), ("sonnet", 0)], seen + assert slept == [], "a quota with a sibling left must rotate, not back off" + assert not loop_mod.provider_is_down("codex") # a quota is not a refusal — never marked + + +async def test_a_quota_beside_a_timeout_takes_the_quota_edge(monkeypatch): + """A rate limit outranks a timeout: the provider being out is the more specific news, + and the timeout candidate may only have been starved by it. Rotate — and the timeout + is not counted against the card's size.""" + seen: list[tuple[str, int]] = [] + + async def _dispatch(coder, wt, prompt, **kw): + seen.append((coder, len(store.escalated))) + if coder == "codex": + if kw.get("gen") == 1: + raise worktree.CoderTimeout(_TIMEOUT) + raise worktree.WorktreeError(_SESSION_LIMIT) + return "" + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-mt", "title": "t", "spec": "s"}) + + assert seen[:4] == [("codex", 0), ("codex", 0), ("sonnet", 0), ("sonnet", 0)], seen + assert "bd-mt" not in loop._timeout_attempts + + +# ── precedence 2: a timeout ──────────────────────────────────────────────────────────── + + +async def test_a_timeout_beside_a_seam_failure_reaches_the_timeout_handling(monkeypatch): + """A timeout outranks a seam failure, so #378 sees it: the card's timeout count moves, + and — the model having worked before the clock ran out — the climb carries #146's + timeout context instead of a byte-identical prompt.""" + prompts: list[tuple[str, str]] = [] + + async def _dispatch(coder, wt, prompt, *, fid=None, gen=1, **kw): + prompts.append((coder, prompt)) + if coder == "codex" and gen == 1: + coder_seam.progress_begin(fid, gen) + coder_seam.progress_tool(fid, gen, {"phase": "start", "id": "t1", "name": "Edit"}) + raise worktree.CoderTimeout(_TIMEOUT) + if coder == "codex": + raise worktree.WorktreeError(_SEAM) + return "" + + async def _no_commits(*_a, **_kw): + raise worktree.NoChangesError("coder produced no commits vs base — nothing to PR") + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + monkeypatch.setattr(worktree, "open_pr", _no_commits) # end the drive after the climb + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-mto", "title": "t", "spec": "s"}) + + assert loop._timeout_attempts.get("bd-mto") == 1, "the timeout never reached #378's counter" + assert [e[0] for e in store.escalated][:1] == ["bd-mto"] + climbed = [p for c, p in prompts if c == "opus"] + assert climbed and "TIMED OUT" in climbed[0] + + +# ── precedence 3 and 4: dispatch failures ────────────────────────────────────────────── + + +async def test_a_seam_failure_beside_a_retryable_one_is_blocked_as_pre_model(monkeypatch): + """A dispatch failure the drive does not retry outranks one it would: with no model + activity anywhere in the fan-out, #339 blocks it for triage — one fan-out, no backoff + re-runs, no climb.""" + seen: list[str] = [] + + async def _dispatch(coder, wt, prompt, **kw): + seen.append(coder) + raise worktree.WorktreeError(_RESET if kw.get("gen") == 1 else _SEAM) + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + slept = _record_sleeps(monkeypatch) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-ms", "title": "t", "spec": "s"}) + + assert seen == ["codex", "codex"] and slept == [] and store.escalated == [] + blocked = _blocks(store) + assert len(blocked) == 1 and blocked[0][3] == "dispatch-infra", blocked + assert "adapter rejected the session" in blocked[0][2] + + +async def test_a_retryable_failure_on_every_candidate_is_retried_not_climbed(monkeypatch): + """Every candidate hit a network blip: the transient edge applies, as for a single + dispatch — back off and re-run the fan-out, then block as `transient`, which the sweep + heals. A blip is not a capability ceiling.""" + seen: list[str] = [] + + async def _dispatch(coder, wt, prompt, **kw): + seen.append(coder) + raise worktree.WorktreeError(_RESET) + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + slept = _record_sleeps(monkeypatch) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-mr", "title": "t", "spec": "s"}) + + attempts = loop_mod.classify(_RESET).max_attempts + assert set(seen) == {"codex"} and len(seen) == 2 * attempts, seen + assert len(slept) == attempts - 1 and store.escalated == [] + blocked = _blocks(store) + assert len(blocked) == 1 and blocked[0][3] == "transient" + + +# ── the fallback: a candidate RETURNED ───────────────────────────────────────────────── + + +async def test_a_candidate_that_returned_keeps_it_a_capability_failure(monkeypatch): + """One candidate was refused, the other RAN and produced nothing. The model had its + shot and missed, so this is exactly the capability failure it always was: it climbs, + and the provider — which served a candidate — is not marked.""" + seen: list[str] = [] + + async def _dispatch(coder, wt, prompt, **kw): + seen.append(coder) + if coder == "codex" and kw.get("gen") == 1: + raise worktree.WorktreeError(_DEAD_MODEL) + return "" + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-mx", "title": "t", "spec": "s"}) + + assert seen[:2] == ["codex", "codex"] and "opus" in seen, f"a capability failure must climb, got {seen}" + assert "sonnet" not in seen + assert not loop_mod.provider_is_down("codex") + + +# ── the policy on its own ────────────────────────────────────────────────────────────── + + +def test_the_representative_is_the_most_specific_failure(): + refused = worktree.WorktreeError(_DEAD_MODEL) + quota = worktree.WorktreeError(_SESSION_LIMIT) + timeout = worktree.CoderTimeout(_TIMEOUT) + seam = worktree.WorktreeError(_SEAM) + blip = worktree.WorktreeError(_RESET) + other = worktree.WorktreeError("worktree add failed: fatal: invalid reference") + ladder = [refused, quota, timeout, seam, blip, other] + for i, expected in enumerate(ladder): + # each one beats everything after it, in any order + rest = ladder[i:] + assert loop_mod.representative_failure(list(reversed(rest))) is expected, expected + # ties go to the earliest candidate + first, second = worktree.WorktreeError(_RESET), worktree.WorktreeError(_RESET) + assert loop_mod.representative_failure([first, second]) is first + # only a WorktreeError can speak for the fan-out: a raw error never does + raw = BrokenPipeError("[Errno 32] Broken pipe") + assert loop_mod.representative_failure([raw, raw]) is None + assert loop_mod.representative_failure([raw, blip]) is blip + + +def test_provider_failure_category_is_the_drives_own_definition(): + """The predicate max-mode ranks with is the one the drive rotates on: only a coder + DISPATCH failure of a provider class counts — the same refusal words in a reviewer's + gap, and every other dispatch failure, do not.""" + assert loop_mod.provider_failure_category(worktree.WorktreeError(_DEAD_MODEL)) == "provider_unavailable" + assert loop_mod.provider_failure_category(worktree.WorktreeError(_SESSION_LIMIT)) == "rate_limit" + gap = worktree.WorktreeError("goal verification failed: no test covers the model_not_found branch") + assert loop_mod.provider_failure_category(gap) is None + assert loop_mod.provider_failure_category(worktree.WorktreeError(_SEAM)) is None + assert loop_mod.provider_failure_category(worktree.CoderTimeout(_TIMEOUT)) is None + + +# ── #425 review: the climb after a timeout, and what the fan-out leaves behind ─────── + + +def _tool(fid, gen, name="Edit"): + coder_seam.progress_begin(fid, gen) + coder_seam.progress_tool(fid, gen, {"phase": "start", "id": f"t{gen}", "name": name}) + + +async def test_a_climb_after_an_all_timeout_fan_out_still_fans_out(monkeypatch): + """The timeout climb used to put its note in `_ci_feedback` — the slot that marks a + carried-forward FIX, which switches fan-out off. So the climbed rung ran ONE opus + dispatch instead of N. A timeout climb is a fresh build: the stronger rung fans out.""" + seen: list[tuple[str, int]] = [] + + async def _dispatch(coder, wt, prompt, *, fid=None, gen=1, **kw): + seen.append((coder, gen)) + if coder == "codex": + _tool(fid, gen) + raise worktree.CoderTimeout(_TIMEOUT) + return "" + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-fo", "title": "t", "spec": "s"}) + + assert seen == [("codex", 1), ("codex", 2), ("opus", 1), ("opus", 2)], seen + + +async def test_a_later_drive_of_the_card_fans_out_again_without_a_stale_note(monkeypatch): + """`_ci_feedback` outlives the drive, so the note also switched fan-out off for every + LATER drive of the card (a sweep requeue, an operator unblock) and opened each prompt + with "a PREVIOUS attempt TIMED OUT". The note is the climbing drive's alone.""" + seen: list[tuple[str, int, bool]] = [] + + async def _dispatch(coder, wt, prompt, *, fid=None, gen=1, **kw): + seen.append((coder, gen, "TIMED OUT" in prompt)) + _tool(fid, gen) + if coder == "opus": + raise worktree.WorktreeError(_SEAM) # the climbed rung dies another way + raise worktree.CoderTimeout(_TIMEOUT) + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-rd", "title": "t", "spec": "s"}) + assert [s[2] for s in seen if s[0] == "opus"] == [True, True] # the climb carried the note + first = len(seen) + await loop._drive({"id": "bd-rd", "title": "t", "spec": "s"}) # the requeue + + assert seen[first : first + 2] == [("codex", 1, False), ("codex", 2, False)], seen[first:] + + +async def test_the_timeout_note_is_spent_once_the_climbed_dispatch_has_run(monkeypatch): + """The note is for the dispatch right after the climb. Once that dispatch has run, a + retry of the same rung (here: its reply was empty) must not open with it again.""" + prompts: list[tuple[str, str]] = [] + + async def _host_dispatch(coder, wt, prompt, *, timeout=None, env_passthrough=()): + prompts.append((coder, prompt)) + if coder == "a": + coder_seam.progress_tool("bd-sp", 1, {"phase": "start", "id": "t1", "name": "Edit"}) + raise worktree.CoderTimeout(_TIMEOUT) + return "" # b runs and replies with nothing → the empty-reply same-tier retry + + async def _no_commits(*_a, **_kw): + raise worktree.NoChangesError("coder produced no commits vs base — nothing to PR") + + async def _unused(*_a, **_kw): + raise AssertionError("the real tap is under test") + + real_tapped = coder_seam.dispatch_coder_tapped + loop, store = _rung_env(monkeypatch, _unused, cfg={"coders": {"smart": "a", "reasoning": "b"}}, tiers=["reasoning"]) + monkeypatch.setattr(coder_seam, "dispatch_coder_tapped", real_tapped) + monkeypatch.setattr(worktree, "dispatch_coder", _host_dispatch) + monkeypatch.setattr(worktree, "open_pr", _no_commits) + await loop._drive({"id": "bd-sp", "title": "t", "spec": "s"}) + + b = [p for c, p in prompts if c == "b"] + assert len(b) >= 2, prompts + assert "TIMED OUT" in b[0] and "TIMED OUT" not in b[1] + + +@dataclasses.dataclass +class _Coder: + name: str + workdir: str = "" + + +async def test_the_climb_note_describes_the_candidate_that_timed_out(monkeypatch): + """Through the REAL tap. The note was mined from the LAST gen — here the candidate that + failed fast on a seam error — so it told the stronger model "ran ~0.0s", no tool. The + seam now stamps the gen its watchdog killed, and the note comes from that one.""" + real_tapped = coder_seam.dispatch_coder_tapped + prompts: list[tuple[str, str]] = [] + + async def _seam(coder, prompt, *, timeout=None, on_tool=None, on_thought=None, on_text=None): + prompts.append((coder.name, prompt)) + if coder.name == "codex": + if coder.workdir.endswith(".c0"): + await on_tool({"phase": "start", "id": "t1", "name": "Edit", "input": {"path": "big_module.py"}}) + await on_thought("refactoring big_module.py section by section") + await asyncio.Event().wait() # hangs until the watchdog fires + raise RuntimeError("adapter rejected the session (unknown error)") + return "done" + + async def _unused(*_a, **_kw): + raise AssertionError("the real tap is under test") + + loop, store = _rung_env(monkeypatch, _unused, cfg=dict(_MAX_LADDER, coder_timeout_s=0.2), tiers=["reasoning"]) + monkeypatch.setattr(coder_seam, "dispatch_coder_tapped", real_tapped) + monkeypatch.setattr(coder_seam, "_import_dispatch_tapped", lambda: _seam) + monkeypatch.setattr(loop, "_resolve_delegate", lambda name, expect: _Coder(name)) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-tn", "title": "t", "spec": "s"}) + + climbed = [p for c, p in prompts if c == "opus"] + assert climbed and "TIMED OUT" in climbed[0] + assert "Edit" in climbed[0] and "big_module.py" in climbed[0] and "refactoring big_module.py" in climbed[0] + + +async def test_every_candidate_is_reaped_before_the_drive_sees_the_failure(monkeypatch): + """Re-raising must not leak a worktree: all N candidates are reaped, nothing promoted, + BEFORE the drive handles the representative (here: blocks the one-provider board).""" + + async def _dispatch(coder, wt, prompt, **kw): + raise worktree.WorktreeError(_DEAD_MODEL) + + loop, store = _rung_env(monkeypatch, _dispatch, cfg={"coder": "codex", "max_mode_n": 3}) + _no_diffs(monkeypatch, loop) + created: list[str] = [] + + async def _create(repo, base, fid, root, title="", **_kw): + created.append(fid) + return ("/wt/feat-" + fid, "feat/" + fid) + + async def _reap(repo, root, fid): + store.calls.append(("reap", fid)) + + async def _promote(*_a, **_kw): + raise AssertionError("an all-raised fan-out promotes nothing") + + monkeypatch.setattr(worktree, "create_worktree", _create) + monkeypatch.setattr(worktree, "reap_feature_worktree", _reap) + monkeypatch.setattr(worktree, "promote_worktree", _promote) + await loop._drive({"id": "bd-rp", "title": "t", "spec": "s"}) + + names = store.names() + reaps = [c[1] for c in store.calls if c[0] == "reap"] + assert sorted(reaps) == sorted(created) == ["bd-rp.c0", "bd-rp.c1", "bd-rp.c2"] + assert max(i for i, n in enumerate(names) if n == "reap") < names.index("flag_blocked") + assert loop._inflight == {} + + +async def test_a_cancelled_child_is_neither_returned_nor_raised(monkeypatch): + """A CancelledError child did not fail: the fan-out keeps its old verdict rather than + letting the other candidate's refusal speak for both.""" + + async def _dispatch(coder, wt, prompt, **kw): + if coder == "codex" and kw.get("gen") == 1: + raise asyncio.CancelledError() + if coder == "codex": + raise worktree.WorktreeError(_DEAD_MODEL) + return "" + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-cx", "title": "t", "spec": "s"}) + + assert store.escalated and not loop_mod.provider_is_down("codex") + + +async def test_a_raw_error_on_every_candidate_keeps_the_shutdown_edge(monkeypatch): + """An untapped host can surface a raw error (not a WorktreeError). Re-raised, it skipped + the drive's shutdown and cancel checks and blocked the card as `unexpected` in the + middle of a shutdown. It keeps the old no-diff verdict, which those checks see first.""" + + async def _dispatch(coder, wt, prompt, **kw): + loop._shutting_down = True + raise BrokenPipeError("[Errno 32] Broken pipe") + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-sd", "title": "t", "spec": "s"}) + + assert _blocks(store) == [], "a shutdown mid-fan-out flagged the card blocked" + + +async def test_a_candidate_that_returned_through_the_real_tap_keeps_it_a_capability_failure(monkeypatch): + """The fallback, through the REAL `dispatch_coder_tapped` (the untapped host path) + rather than a stand-in for it: one candidate refused, the other RAN and returned + nothing. The card climbs as a capability failure — never a rotation to the sibling.""" + seen: list[str] = [] + + async def _host_dispatch(coder, wt, prompt, *, timeout=None, env_passthrough=()): + seen.append(coder) + if coder == "codex" and wt.endswith(".c0"): + raise worktree.WorktreeError(_DEAD_MODEL) + return "" + + async def _unused(*_a, **_kw): + raise AssertionError("the real tap is under test") + + real_tapped = coder_seam.dispatch_coder_tapped + loop, store = _rung_env(monkeypatch, _unused, cfg=_MAX_LADDER, tiers=["reasoning"]) + monkeypatch.setattr(coder_seam, "dispatch_coder_tapped", real_tapped) + monkeypatch.setattr(worktree, "dispatch_coder", _host_dispatch) + _no_diffs(monkeypatch, loop) + await loop._drive({"id": "bd-rt", "title": "t", "spec": "s"}) + + assert "sonnet" not in seen, f"a fan-out where a candidate returned rotated instead of climbing: {seen}" + assert store.escalated and "opus" in seen + assert not loop_mod.provider_is_down("codex") + + +# ── #429 × #435: every candidate times out, twice ───────────────────────────────────── + + +@pytest.mark.skipif( + not hasattr(loop_mod, "TOO_WIDE_CLASS"), reason="needs #435's too-wide park; runs once both are on the branch" +) +async def test_a_fan_out_that_keeps_timing_out_is_parked_for_its_split(monkeypatch): + """Merged behaviour. Alone, #425's fix hands the drive a CoderTimeout, and a card whose + fan-outs keep timing out blocks `transient` — the sweep then rebuilds it twice more, N + candidates each time. With #435, the second fresh timeout parks it for a split: the + first fan-out climbs (fanning out again at the new rung), the second parks.""" + seen: list[tuple[str, int]] = [] + + async def _dispatch(coder, wt, prompt, *, fid=None, gen=1, **kw): + seen.append((coder, gen)) + _tool(fid, gen) + raise worktree.CoderTimeout(_TIMEOUT) + + loop, store = _rung_env(monkeypatch, _dispatch, cfg=_MAX_LADDER, tiers=["reasoning"]) + _no_diffs(monkeypatch, loop) + asked: list[tuple[str, int]] = [] + budgets: dict[str, int] = {} + + def _ask(fid, *, timeouts): + asked.append((fid, timeouts)) + return {"id": "bd-split", "board_state": "backlog"} + + store.request_decomposition = _ask + store.record_budget = lambda fid, kind, n: budgets.__setitem__(f"{fid}:{kind}", n) + store.get_feature = lambda fid: {"id": fid, "labels": [], "board_state": "in_progress"} + await loop._drive({"id": "bd-tw", "title": "t", "spec": "s"}) + + assert seen == [("codex", 1), ("codex", 2), ("opus", 1), ("opus", 2)], seen + assert asked == [("bd-tw", 2)] and len(store.escalated) == 1 + assert _blocks(store)[-1][3] == loop_mod.TOO_WIDE_CLASS