diff --git a/CHANGELOG.md b/CHANGELOG.md index 3bcbf19..62e5816 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,16 @@ All notable changes to this project are documented here. The format follows `/help` stays visible. One more row, at 80x25, shows them all. - `packaging` is a declared dependency. It was already installed through other packages; `update` now imports it directly. +- **The judge goes back to Groq once its minute limit resets.** A 429 whose + `Retry-After` is at most 120 seconds (60 when it names none) now switches to the + local model only until that wait has passed: `Groq limit reached — judging with + local ollama qwen3.5:9b until it resets (~30s)`. The first request after it asks + Groq again, from one thread only; if Groq answers, the judge stays on it and says + so once (`Groq answering again — local ollama qwen3.5:9b stood in for 4 requests`), + on stderr, in the TUI and in the report's footer. Another short 429 starts a new + wait, quietly. Nobody waits: the local model answers while Groq cools down. A daily + limit (a longer `Retry-After`), a 413, a rejected key, a 5xx or no answer still + switches for the rest of the run, with the same notice as before. ## [0.4.10] - 2026-10-01 diff --git a/README.md b/README.md index 2b522e1..205bcf5 100644 --- a/README.md +++ b/README.md @@ -466,7 +466,10 @@ in the environment or a `.env` file, never from config, and is never printed. **When the judge stops answering.** If the configured judge hits its rate limit or otherwise stops answering, the run switches at once to `qwen3.5:9b` in your local Ollama, if it is installed (`ollama pull qwen3.5:9b`; proofpath never pulls it). It says so on stderr, in the TUI -above the prompt, and in the report. The next run tries the configured API again first. +above the prompt, and in the report. A per-minute limit (a 429 that asks for a wait of +two minutes or less) only switches until it resets: the run then tries the configured +API again, goes back to it if it answers, and says that too. The next run tries the +configured API again first. `proofpath config set judge.fallback off` turns this off. Without a local model to switch to, the run says so and tells you how to add one, on the same channels. diff --git a/docs/superpowers/OPEN-ITEMS.md b/docs/superpowers/OPEN-ITEMS.md index 299764d..65452cb 100644 --- a/docs/superpowers/OPEN-ITEMS.md +++ b/docs/superpowers/OPEN-ITEMS.md @@ -626,3 +626,21 @@ over "uv tool and pipx only". Design: `docs/superpowers/specs/2026-10-05-update- to the tool environment with pip, and the browser extra is installed that way on consent. - **Behaviour:** running the command updates straight away; `--check` only reports. + +## 22. Judge: back to the provider after a minute limit — 2026-10-05 + +The user asked whether the judge could return to Groq once its per-minute limit resets +instead of staying on the local `qwen3.5:9b` for the rest of the run. They chose +**option A, a cooldown with return**, over waiting on the provider (B) and over A plus +proactive pacing from the `x-ratelimit-*` headers (C). A 429 whose `Retry-After` is at +most 120 s switches only until that time has passed. A daily limit, a 413, a key error +and a server error keep today's sticky switch. Nobody waits, and the provider is asked +again only once the time it named has passed. Design: +`docs/superpowers/specs/2026-10-05-judge-cooldown-design.md`. It amends evidence-search +spec §11. + +Follow-up (review, 2026-10-05, left open on purpose): `_come_back` announces the back +notice outside the lock. If a concurrent re-switch announces its own notice inside that +microsecond window, the TUI line can show the back notice while `switched` (the +footer) holds the cooling one. Fix: announce under a separate `_announce_lock`, and only +when `switched` still equals the notice. diff --git a/docs/superpowers/specs/2026-09-28-evidence-search-design.md b/docs/superpowers/specs/2026-09-28-evidence-search-design.md index 8931299..28c601c 100644 --- a/docs/superpowers/specs/2026-09-28-evidence-search-design.md +++ b/docs/superpowers/specs/2026-09-28-evidence-search-design.md @@ -189,7 +189,10 @@ judge-written queries and translations). - **Trigger.** Any `JudgeUnavailable` from the primary provider: a 429, a quota error, or the provider not answering. The switch lasts for the rest of the run, so - a rate-limited Groq is not hit again on every batch. + a rate-limited Groq is not hit again on every batch. A 429 with a `Retry-After` of + at most 120 s (or none) is the exception: it switches only until that wait has + passed, then the provider is asked again (amended by + `2026-10-05-judge-cooldown-design.md`). The switch is immediate: with a fallback available, the primary gets no retry and no `Retry-After` wait. The failed request is sent again to the local model, so no batch is lost, and the notice appears before the model starts loading. diff --git a/docs/superpowers/specs/2026-10-05-judge-cooldown-design.md b/docs/superpowers/specs/2026-10-05-judge-cooldown-design.md new file mode 100644 index 0000000..b8bf7fe --- /dev/null +++ b/docs/superpowers/specs/2026-10-05-judge-cooldown-design.md @@ -0,0 +1,81 @@ +# Judge: back to the provider once its minute limit resets — design (2026-10-05) + +This amends evidence-search spec §11 ("Local fallback judge"), which made the switch +to the local model last for the rest of the run. OPEN-ITEMS §22 points here. + +## 1. Decision (user, 2026-10-05) + +The user asked whether the judge could go back to Groq once its per-minute limit +resets, instead of staying on `qwen3.5:9b` for the rest of the run. They chose option +A, **a cooldown with return**, over two others: waiting on the provider (B), and A plus +proactive pacing from the `x-ratelimit-*` headers (C). + +§11 had two goals, and both are kept: + +- **Nobody waits.** With a fallback behind it, the primary still gets no retry and no + `Retry-After` sleep. While the provider cools down, the local model answers. +- **The provider is not hammered.** It is asked again only once the time it named + itself has passed, and never on every batch. + +## 2. Behaviour + +1. **Which failures cool down.** Only a **429** whose `Retry-After` is at most + `COOLDOWN_MAX_S = 120` seconds. Groq's per-minute token and request limits fit this. + A 429 without the header counts as `COOLDOWN_DEFAULT_S = 60`. The switch stays + **sticky** for the rest of the run, exactly as before, for: + - a 429 with a longer `Retry-After` (a daily limit: the RPD or TPD tier); + - **413** (the request alone breaks the TPM, so asking again cannot help); + - 401 and 403 (the key); + - 5xx, a timeout, no answer, or an empty answer. +2. **Carrying the wait.** `JudgeUnavailable` gains `retry_after: float | None`. + `JudgeClient` fills it from the same `_retry_after_seconds` parse it already uses, + including when it is `fail_fast`. Never a body, a URL or a key. +3. **Routing.** `FallbackClient` keeps `_cooldown_until`, a clock reading or `None`. + - On a cooling failure it switches as today and sets + `_cooldown_until = now + retry_after`. That covers the notice, the warm-up and the + local answer for the failed request. + - While `now < _cooldown_until`, requests go to the local model. + - The first request at or after `_cooldown_until` goes to the primary again. + - If it answers, routing returns to the primary. A **back notice** is announced + once: `Groq answering again — local ollama stood in for N requests`. + - If it fails again with a cooling 429, a new cooldown starts from the new + `Retry-After`, and that request goes to the local model. No second switch + notice, and no second warm-up. + - If it fails in a sticky way, the switch becomes sticky from then on, as today. + - The clock is injected (`clock: Callable[[], float] = time.monotonic`) so tests + need no sleep. The existing lock guards every read and write of + `_cooldown_until`. Only one thread probes the primary when the cooldown ends; + the others keep using the local model until the probe has answered. +4. **Notices.** + - The switch notice for a cooling 429 says the switch is temporary: + `Groq limit reached — judging with local ollama until it resets (~Ns)`. + - `switched` keeps the latest notice: the switch, or the back notice once the + provider answers again. The report footer therefore tells the truth at the end + of the run. + - Every opinion's `model` field still names the model that gave it, and the + report's cost line still splits `api_calls` from `local_calls`. + - A sticky switch keeps today's wording, unchanged. +5. **`reset()`** (a new run) clears `_cooldown_until`, as it already clears the rest. +6. **Local failures.** These are unchanged (`_dead`, `JUDGE_FALLBACK_FAILED`). If the + local model is dead while the provider cools down, the request fails as today. The + next request after the cooldown still probes the provider. + +## 3. Tests (no network, no sleep) + +Use `respx`, or the existing fake clients, plus a fake clock. They cover: + +- a short 429 switches and then comes back after `Retry-After`, with one switch + notice and one back notice; +- requests inside the cooldown go to the local model and send nothing to the primary; +- a second short 429 at the probe restarts the cooldown with no new notice; +- a long 429, a 413, a 401 and a 5xx each stay sticky with today's notice; +- a 429 without `Retry-After` uses 60 s; +- `reset()` clears the cooldown; +- with concurrent requests at the end of the cooldown, only one probes; +- `JudgeUnavailable.retry_after` is filled on the `fail_fast` path. + +## 4. Out of scope + +- Proactive pacing from `x-ratelimit-remaining-tokens` (option C). +- Waiting on the provider (option B). +- Any change to the batch caps (`TOKEN_CAP`, `_REVIEW_TOKENS`). diff --git a/src/proofpath/cli.py b/src/proofpath/cli.py index 1724526..789dd48 100644 --- a/src/proofpath/cli.py +++ b/src/proofpath/cli.py @@ -577,8 +577,9 @@ def on_event(event: events.Event) -> None: if isinstance(event, events.StageEnd): ui.stage_row(human, event.name, event.by, event.summary, event.elapsed) elif isinstance(event, events.Note) and event.notice: - # The judge switched to its local fallback: said at once, on stderr, and - # never dropped, not even under ``-q`` (spec section 11). + # The judge switched to its local fallback, or back to its provider after + # a cooldown: said at once, on stderr, and never dropped, not even under + # ``-q`` (spec section 11, judge-cooldown spec). ui.kv(out, "judge", event.text, err=True) elif isinstance(event, events.Note): ui.note(human, event.text) diff --git a/src/proofpath/judge.py b/src/proofpath/judge.py index 3ed1d79..0bb8cd7 100644 --- a/src/proofpath/judge.py +++ b/src/proofpath/judge.py @@ -17,6 +17,7 @@ import importlib.resources import json +import math import re import string import threading @@ -59,12 +60,24 @@ class JudgeUnavailable(JudgeError): # noqa: N818 - reads as a state, not as a f a notice can print (``HTTP 500``, ``timeout``, ``empty answer``): never a body, never a URL, never a key. The fallback reads both to say why it switched without parsing its own message back. + + ``retry_after`` is the wait a 429 named in its ``Retry-After`` header, in seconds, + or ``None`` when the last answer named none. The fallback reads it to know when + the provider may be asked again (judge-cooldown spec 2.2). A number, never a body. """ - def __init__(self, detail: str, *, status: int | None = None, cause: str = "") -> None: + def __init__( + self, + detail: str, + *, + status: int | None = None, + cause: str = "", + retry_after: float | None = None, + ) -> None: super().__init__(detail) self.status = status self.cause = cause or (f"HTTP {status}" if status is not None else "no answer") + self.retry_after = retry_after #: ``think: false`` in Ollama's OpenAI-compatible terms, sent to every Ollama model @@ -175,6 +188,10 @@ class Completion: prompt_tokens: int completion_tokens: int model: str + #: Who answered, as an opinion names it (``"ollama qwen3.5:9b"``): stamped by the + #: client that answered, so a fallback that changed hands while the request was + #: out cannot name the wrong judge afterwards. Empty from a caller's own fake. + by: str = "" @dataclass @@ -411,7 +428,11 @@ def complete( last = "no request was made" status: int | None = None cause = "" + retry_after: float | None = None while attempt < _MAX_ATTEMPTS: + # Only the last failure's wait is carried: a 429 followed by a 5xx gave up + # on the 5xx, which named none. + retry_after = None payload: dict[str, Any] = { "model": self._config.model, "messages": messages, @@ -467,9 +488,9 @@ def complete( continue if status == 429: header = response.headers.get("Retry-After", "") - seconds = _retry_after_seconds(header) if header else None + retry_after = _retry_after_seconds(header) if header else None # A bare 429 still means "slow down", so wait longer than a 5xx. - delay = 2.0 * 2.0**attempt if seconds is None else seconds + delay = 2.0 * 2.0**attempt if retry_after is None else retry_after elif status >= 500: delay = _backoff(attempt) else: @@ -481,7 +502,9 @@ def complete( attempt += 1 if attempt < _MAX_ATTEMPTS: self._wait(delay) - raise JudgeUnavailable(last, status=status, cause=cause) + # Raised the same way after ``fail_fast``'s break: that is the path the + # fallback's primary takes, and the one whose wait it needs. + raise JudgeUnavailable(last, status=status, cause=cause, retry_after=retry_after) def _wait(self, delay: float) -> None: delay = min(delay, self._max_wait) @@ -537,6 +560,7 @@ def _record(self, response: httpx.Response, messages: list[dict[str, str]]) -> C prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, model=model, + by=f"{self.provider} {self.model}", ) @@ -556,6 +580,25 @@ def _record(self, response: httpx.Response, messages: list[dict[str, str]]) -> C #: What ``switched`` says once the local model has failed too: the primary's reason, #: then the local one. "Judging with local ollama" would then be false (final review). JUDGE_FALLBACK_FAILED = "{head}, and local ollama {model} failed too ({cause})" +#: The judge-cooldown amendment (2026-10-05): a 429 whose ``Retry-After`` is at most +#: ``COOLDOWN_MAX_S`` is a per-minute limit (Groq's RPM and TPM), so the switch it +#: causes is temporary and its notice says so; the primary is asked again once that +#: wait has passed. A longer wait is a daily limit (RPD, TPD) that asking again within +#: a run cannot beat, and a 429 that names no wait is taken to mean a minute. +COOLDOWN_MAX_S = 120.0 +COOLDOWN_DEFAULT_S = 60.0 +_UNTIL = " until it resets (~{seconds}s)" +JUDGE_FALLBACK_COOLING = _LIMIT_HEAD + _JUDGING + _UNTIL +#: Said once, when the primary answers the first request after its cooldown. It +#: replaces the switch notice in ``switched``, so the footer of a run that came back +#: does not say the local model was still judging. +JUDGE_FALLBACK_BACK = ( + "{provider} answering again — local ollama {model} stood in for {count} {noun}" +) +#: The same return when the local model was latched dead while the primary cooled +#: down. "Stood in for 0 requests" would drop the one fact the footer must keep: the +#: local model failed (rule 6). +JUDGE_FALLBACK_BACK_DEAD = "{provider} answering again — local ollama {model} failed ({cause})" #: The install/pull hint: offered once, beside the original error, when there is no #: fallback for a reason the user can fix by adding the local model -- never when #: they chose ``judge.fallback = off`` themselves, and never a notice of a switch @@ -689,6 +732,19 @@ class _Plan: hint: str | None = None +@dataclass(frozen=True) +class _Handover: + """What the thread that switched still owes the others, outside the lock.""" + + notice: str + #: The gate this switch made, to open once announced and warmed: exactly this + #: one. ``self._ready`` may already be a later switch's gate by then, and opening + #: that one instead would leave this switch's waiters to time out. + gate: threading.Event + #: ``False`` when the local model is latched dead: there is nothing to load. + warm: bool + + class FallbackClient: """A primary ``JudgeClient`` that switches, once and at once, to a local model. @@ -702,8 +758,13 @@ class FallbackClient: and its error says why no fallback was used. A local model that fails too is latched dead for the run: every later call raises at once with the same detail. - The switch is sticky for the life of the client, which is one run: whatever - builds a judge for a run builds a new client or calls :meth:`reset`. + The switch is sticky for the life of the client, which is one run, with one + exception (judge-cooldown spec, 2026-10-05): a 429 that names a wait of at most + ``COOLDOWN_MAX_S`` -- a per-minute limit -- only switches until that wait has + passed. The first call after it asks the primary again, one thread at a time; + if it answers, the primary judges again and the back notice goes out through + the same listener. Whatever builds a judge for a run builds a new client or + calls :meth:`reset`. """ def __init__( @@ -715,6 +776,7 @@ def __init__( ollama_url: str | None = None, local_factory: Callable[[str, str], JudgeClient] = local_client, warm: Callable[[str, str], None] | None = warm_local, + clock: Callable[[], float] = time.monotonic, ) -> None: self._primary = primary self._primary_timeout = primary.timeout @@ -748,6 +810,27 @@ def __init__( self._notice: str | None = None #: The combined detail once the local model has failed too; ``None`` before. self._dead: str | None = None + # The judge-cooldown amendment. A switch is either sticky (any failure but a + # short 429) or cooling: the clock reading from which the primary may be asked + # again. Neither, with a local client, means the primary answered again. All + # three, and ``_probing``, are read and written under the lock only. Injected + # so a test crosses a cooldown without sleeping. + self._clock = clock + self._sticky = False + self._cooldown_until: float | None = None + # True while one thread asks the primary again; every other thread stays on + # the local model until that probe has answered, so the end of a cooldown + # costs the provider one request, not one per waiting thread. + self._probing = False + # Requests handed to the local model in this period (a switch to its return), + # for the back notice. Counted when routed, so one still out at the return is + # in it; ``_period`` keeps a late one from counting toward the next period. + self._stood_in = 0 + self._period = 0 + # The client whose answer landed last: ``cost.model`` names it. + self._answered: JudgeClient | None = None + # The local failure that latched ``_dead``, for a later switch's notice. + self._dead_cause = "" # Local answers from before a ``reset``: the client is gone, the spend is not. self._spent = JudgeCost() @@ -759,13 +842,25 @@ def __repr__(self) -> str: __str__ = __repr__ + def _standing_in(self) -> JudgeClient | None: + """The local client while it judges in the primary's place, else ``None``. + + Not merely "a local client exists": once the primary answers again after a + cooldown, the client is kept (its spend, its warm model) but no longer judges, + and an opinion must name the model that gave it. + """ + if self._sticky or self._cooldown_until is not None: + return self._local + return None + @property def provider(self) -> str: - return "ollama" if self._local is not None else self._primary.provider + return "ollama" if self._standing_in() is not None else self._primary.provider @property def model(self) -> str: - return self._local.model if self._local is not None else self._primary.model + local = self._standing_in() + return local.model if local is not None else self._primary.model @property def cost(self) -> JudgeCost: @@ -775,18 +870,21 @@ def cost(self) -> JudgeCost: if self._local is not None: parts.append(self._local.cost) local = self._spent.calls + (self._local.cost.calls if self._local is not None else 0) + standing = self._standing_in() + answered = self._answered + source = answered if answered is not None else standing return JudgeCost( calls=sum(part.calls for part in parts), prompt_tokens=sum(part.prompt_tokens for part in parts), completion_tokens=sum(part.completion_tokens for part in parts), waited_s=sum(part.waited_s for part in parts), - model=self._local.cost.model if self._local is not None else primary.model, + model=source.cost.model if source is not None else primary.model, local_calls=local, ) def reset(self) -> None: - """Back to the primary, for a new run: the notice, the plan and a dead local - model are all forgotten.""" + """Back to the primary, for a new run: the notice, the plan, a cooldown and a + dead local model are all forgotten.""" with self._lock: if self._local is not None: spent = self._local.cost @@ -798,6 +896,16 @@ def reset(self) -> None: self._local = None self._plan = None self._dead = None + self._sticky = False + self._cooldown_until = None + # A probe still out belongs to the run before: its answer must not announce + # a return in this one (``_come_back`` checks the flag). + self._probing = False + self._stood_in = 0 + # A new period: an answer still out from the last run counts toward neither. + self._period += 1 + self._answered = None + self._dead_cause = "" self.switched = None self.hint = None self._primary_detail = "" @@ -843,26 +951,40 @@ def ask(client: JudgeClient) -> Completion: reasoning_effort=reasoning_effort, ) - local = self._route() + local, probe, period = self._route() if local is None: try: - return ask(self._primary) + answer = ask(self._primary) except JudgeUnavailable as exc: - local, notice = self._switch(exc) - if notice is not None: + local, handover, period = self._switch(exc, probe=probe) + except BaseException: + if probe: + # Not an answer and not a failure the cooldown knows: let the next + # call probe instead, rather than stay local for the run. + self._end_probe() + raise + else: + with self._lock: + self._answered = self._primary + if probe: + self._come_back() + return answer + if handover is not None: try: - self._announce(notice) - if self._warm is not None: + self._announce(handover.notice) + if handover.warm and self._warm is not None: # After the notice, before the request: the user reads the # line while the model loads (Amendment B 4). Outside the lock. self._warm(self._ollama_url, local.model) finally: - self._ready.set() + handover.gate.set() # A thread that did not switch waits for the notice and the warm-up, bounded # by the warm-up's own timeout (fix round 2). self._ready.wait(timeout=WARM_WAIT_S) - if self._dead is not None: - raise JudgeUnavailable(self._dead) + dead = self._dead + if dead is not None: + self._uncount(period, local) + raise JudgeUnavailable(dead) try: answer = ask(local) except JudgeUnavailable as exc: @@ -871,24 +993,60 @@ def ask(client: JudgeClient) -> Completion: detail = f"{self._primary_detail}; local ollama {local.model}: {exc}" # "Judging with local ollama" is no longer true, so the notice the report # carries says both failed instead (final review, minor). + self._uncount(period, local) self._both_failed(local.model, exc) if exc.status is None or exc.status >= 500 or exc.status in _DEAD_STATUSES: # Latched: a dead or stuck Ollama costs this run one wait, not one per # batch. Anything else failed this request only (fix round 2). - self._dead = detail + with self._lock: + self._dead = detail + self._dead_cause = exc.cause raise JudgeUnavailable(detail, status=exc.status, cause=exc.cause) from None - if self._notice is not None: - # Answering again after a failure that was about one request only. - self.switched = self._notice + with self._lock: + self._answered = local + # Only within the period that routed it: an answer that lands after the + # probe brought the primary back, or after a later switch, must not put + # this period's notice back over the newer one. + if self._counted(period, local) and self._notice is not None: + # Answering again after a failure that was about one request only. + self.switched = self._notice return answer + def _hand_over(self) -> int | None: + """Count one request handed to the local model; called under the lock. + + The period it counts in, or ``None`` when the model is latched dead and will + be refused the request before it is asked. + """ + if self._dead is not None: + return None + self._stood_in += 1 + return self._period + + def _counted(self, period: int | None, local: JudgeClient) -> bool: + """Whether a request counted in ``period`` still belongs to the open period; + called under the lock.""" + return period is not None and period == self._period and self._standing_in() is local + + def _uncount(self, period: int | None, local: JudgeClient) -> None: + """A counted request that got no answer did not stand in after all. Once the + back notice is out its count is said, so only an open period is corrected.""" + with self._lock: + if self._counted(period, local): + self._stood_in -= 1 + def _both_failed(self, model: str, failure: JudgeUnavailable) -> None: """Say in ``switched`` that the local model failed too, after the primary.""" primary = self._primary_failure if primary is None: # pragma: no cover - a local client exists only after a switch return head = _head(self._primary.provider, primary) - self.switched = JUDGE_FALLBACK_FAILED.format(head=head, model=model, cause=failure.cause) + with self._lock: + if self._standing_in() is None: + return # the primary answers again: the back notice is the true one + self.switched = JUDGE_FALLBACK_FAILED.format( + head=head, model=model, cause=failure.cause + ) def _announce(self, notice: str) -> None: """Hand the notice to the listener, once, and never at the answer's expense.""" @@ -897,11 +1055,19 @@ def _announce(self, notice: str) -> None: with suppress(Exception): self.on_switch(notice) - def _route(self) -> JudgeClient | None: - """The local client once switched; otherwise ``None``, with the plan made.""" + def _route(self) -> tuple[JudgeClient | None, bool, int | None]: + """Who takes this call: the local client while it stands in, else ``None`` for + the primary, with the plan made. The flag is ``True`` for the one call that asks + the primary again once a cooldown has run out (judge-cooldown spec 2.3); the + number is the period a local request was counted in (``_hand_over``).""" with self._lock: - if self._local is not None: - return self._local + if self._sticky: + return self._local, False, self._hand_over() + if self._cooldown_until is not None: + if self._probing or self._clock() < self._cooldown_until: + return self._local, False, self._hand_over() + self._probing = True + return None, True, None if self._plan is None: self._plan = self._choose() if self._plan.model is not None: @@ -913,7 +1079,84 @@ def _route(self) -> JudgeClient | None: self._primary.timeout = httpx.Timeout( PRIMARY_READ_TIMEOUT, connect=PRIMARY_CONNECT_TIMEOUT ) - return None + return None, False, None + + def _come_back(self) -> None: + """The probe was answered: the primary judges again, and the user is told once. + + The local client is kept, not closed: a thread that was routed to it before + the probe answered may still be using it, and its spend belongs to the run. + """ + with self._lock: + local = self._local + if not self._probing or local is None: + return # a reset came first; that run's return is not this run's news + self._probing = False + self._cooldown_until = None + self._notice = None + count = self._stood_in + if self._dead is not None: + back = JUDGE_FALLBACK_BACK_DEAD.format( + provider=_display(self._primary.provider), + model=local.model, + cause=self._dead_cause, + ) + else: + back = JUDGE_FALLBACK_BACK.format( + provider=_display(self._primary.provider), + model=local.model, + count=count, + noun="request" if count == 1 else "requests", + ) + self.switched = back + self._announce(back) + + def _end_probe(self) -> None: + with self._lock: + self._probing = False + + def _probe_failed(self, failure: JudgeUnavailable) -> tuple[JudgeClient, int | None] | None: + """The primary failed the probe: the local client answers this request too. + + ``None`` when a reset came first, so the failure is the new run's first one. + A short 429 starts a new cooldown from its own ``Retry-After``, with no second + notice and no second warm-up: the user has been told and the model is loaded. + Any other failure makes the switch sticky, as it would have been the first + time, and says so: "until it resets" is no longer true. + """ + notice: str | None = None + with self._lock: + local = self._local + if not self._probing or local is None: + return None + self._probing = False + self._primary_detail = str(failure) + self._primary_failure = failure + cooldown = _cooldown_s(failure) + if cooldown is not None: + self._cooldown_until = self._clock() + cooldown + else: + self._cooldown_until = None + self._sticky = True + notice = self._notice_for(failure, local.model, None) + self._notice = notice + self.switched = notice + period = self._hand_over() + if notice is not None: + self._announce(notice) + return local, period + + def _notice_for(self, failure: JudgeUnavailable, model: str, cooldown: float | None) -> str: + """The notice for a switch made now; called under the lock. + + A local model latched dead stays dead for the run (review, important 1), so a + later switch says both failed -- what ``switched`` already says -- instead of + announcing a judge that will refuse every request. + """ + if self._dead is None: + return _notice(self._primary.provider, model, failure, cooldown) + head = _head(self._primary.provider, failure) + return JUDGE_FALLBACK_FAILED.format(head=head, model=model, cause=self._dead_cause) def _choose(self) -> _Plan: """The configured model, once ``/api/tags`` confirms it is installed. @@ -935,8 +1178,15 @@ def _choose(self) -> _Plan: return _Plan(None, f"{setting} is not installed", hint=JUDGE_FALLBACK_HINT_MISSING) return _Plan(setting) # as written: the user's own spelling of the name - def _switch(self, failure: JudgeUnavailable) -> tuple[JudgeClient, str | None]: - """Switch once: the local client, and the notice when *this* call switched. + def _switch( + self, failure: JudgeUnavailable, *, probe: bool = False + ) -> tuple[JudgeClient, _Handover | None, int | None]: + """Switch once: the local client, the handover when *this* call switched, and + the period the request was counted in. + + A short 429 (``_cooldown_s``) switches only until the wait it named has + passed; anything else switches for the run. A failed ``probe`` is not a new + switch: :meth:`_probe_failed` decides what it changes. Raises the primary's error, with the reason, when there is no fallback. When that reason is one the user can fix -- Ollama unreachable, the model not @@ -950,10 +1200,16 @@ def _switch(self, failure: JudgeUnavailable) -> tuple[JudgeClient, str | None]: switch: the listener runs after the lock is released, so a slow or blocking listener can never hold another thread's switch (fix round 1, C). """ + if probe: + stood = self._probe_failed(failure) + if stood is not None: + return stood[0], None, stood[1] hint: str | None = None with self._lock: - if self._local is not None: - return self._local, None # another thread switched while we waited + standing = self._standing_in() + if standing is not None: + # Another thread switched while we waited. + return standing, None, self._hand_over() plan = self._plan if plan is None or plan.model is None: why = _WHY_OFF if plan is None else plan.why @@ -966,15 +1222,32 @@ def _switch(self, failure: JudgeUnavailable) -> tuple[JudgeClient, str | None]: f"{failure}; no local fallback: {why}", status=failure.status, cause=failure.cause, + retry_after=failure.retry_after, ) else: - notice = _notice(self._primary.provider, plan.model, failure) + cooldown = _cooldown_s(failure) + notice = self._notice_for(failure, plan.model, cooldown) self._primary_detail = str(failure) self._primary_failure = failure self._notice = notice - self._local = self._local_factory(plan.model, self._ollama_url) + if self._local is None: + self._local = self._local_factory(plan.model, self._ollama_url) + gate = self._ready + else: + # A switch after the primary came back: the same client (a thread + # may still hold it), warmed again, behind a fresh gate so the + # threads that find this switch wait for its notice too. The old + # gate is left to the switch that made it (``_Handover.gate``). + gate = self._ready = threading.Event() + if cooldown is None: + self._sticky = True + else: + self._cooldown_until = self._clock() + cooldown + self._period += 1 + self._stood_in = 0 self.switched = notice - return self._local, notice + handover = _Handover(notice, gate, warm=self._dead is None) + return self._local, handover, self._hand_over() if hint is not None: self._announce(hint) raise error from None @@ -990,9 +1263,31 @@ def _head(provider: str, failure: JudgeUnavailable) -> str: return _DOWN_HEAD.format(provider=name, cause=failure.cause) -def _notice(provider: str, model: str, failure: JudgeUnavailable) -> str: - """The switch notice: why the primary was left, and who judges now.""" - return _head(provider, failure) + _JUDGING.format(model=model) +def _notice( + provider: str, model: str, failure: JudgeUnavailable, cooldown: float | None = None +) -> str: + """The switch notice: why the primary was left, who judges now, and, for a + temporary switch, roughly for how long. Rounded up and never below one second, + so neither a sub-second wait nor ``Retry-After: 0`` is announced as ``~0s``.""" + notice = _head(provider, failure) + _JUDGING.format(model=model) + if cooldown is None: + return notice + return notice + _UNTIL.format(seconds=max(1, math.ceil(cooldown))) + + +def _cooldown_s(failure: JudgeUnavailable) -> float | None: + """How long the local model stands in before the primary is asked again, or + ``None`` for a switch that lasts the run (judge-cooldown spec 2.1). + + Only a 429 cools down. A 413 is in ``LIMIT_STATUSES`` too, but it means the + request alone busts the TPM: the same request asked again a minute later fails + the same way. The key (401, 403), a 5xx, a timeout or an empty answer is not + something a wait is known to fix either. + """ + if failure.status != 429: + return None + wait = COOLDOWN_DEFAULT_S if failure.retry_after is None else failure.retry_after + return wait if wait <= COOLDOWN_MAX_S else None def build_client( @@ -1197,6 +1492,9 @@ def __init__( self.unavailable = False self.detail = "" self.status: int | None = None + #: Who gave the last answer this judge received (``Completion.by``), for what + #: it wrote outside an opinion: the summary. + self.answered_by = "" @property def name(self) -> str: @@ -1322,6 +1620,7 @@ def summarize(self, report_markdown: str, *, max_tokens: int = _SUMMARY_TOKENS) # provider's own exception can carry the key. self._give_up(f"{type(exc).__name__} from the judge client") return "" + self.answered_by = completion.by or self.name text = completion.text.strip() if not text: # A 200 whose content is whitespace is not a summary with nothing to say. @@ -1440,7 +1739,10 @@ def _collect( # Rule 1 applies to the judge too: no quoted reason, no opinion. self.skipped.append(f"{ident} came back with no rationale") continue - into[ident] = JudgeOpinion(label=label, rationale=rationale, model=self.name) + # Named by the answer, not by ``self.name`` read now: the client may have + # changed hands since this answer left it (judge-cooldown review). + by = completion.by or self.name + into[ident] = JudgeOpinion(label=label, rationale=rationale, model=by) def _render(item: JudgeItem) -> str: diff --git a/src/proofpath/report.py b/src/proofpath/report.py index ee70856..532cf20 100644 --- a/src/proofpath/report.py +++ b/src/proofpath/report.py @@ -364,7 +364,8 @@ class Report: search: SearchSummary | None = None # The notice of a judge that switched to its local fallback (spec section 11), or # ``None``. A run that changed who gives its second opinion says so in the report, - # not only in a line that scrolled past while it ran. + # not only in a line that scrolled past while it ran. After a cooldown it is the + # back notice instead: the latest one, so the footer is true at the end of the run. judge_notice: str | None = None def exit_code(self) -> int: diff --git a/src/proofpath/tui/app.py b/src/proofpath/tui/app.py index 8ef225d..c5c798d 100644 --- a/src/proofpath/tui/app.py +++ b/src/proofpath/tui/app.py @@ -415,6 +415,9 @@ def __init__(self, text: str) -> None: def _judge_switched(self, text: str) -> None: """The summary's judge switched to its local model; say so above the bar. + The back notice of a cooldown that ended (judge-cooldown spec) arrives here + too, and replaces the line the same way. + Called on the summary's worker thread, before the model is loaded. It only posts a message -- ``post_message`` is thread-safe and never waits -- so the worker is not held, and the widget is touched on the loop alone. @@ -1053,7 +1056,7 @@ async def _run_summary( self._summarising.discard(run_id) if text: label = Text( - f"(model-written, {judge.name}) ", + f"(model-written, {judge.answered_by or judge.name}) ", style=self._theme.tone("muted") if self._out.color else "", ) line = ui.kv_text(self._out, "summary", Text.assemble(label, text)) diff --git a/src/proofpath/tui/widgets/judge_notice.py b/src/proofpath/tui/widgets/judge_notice.py index 1158a56..ae47628 100644 --- a/src/proofpath/tui/widgets/judge_notice.py +++ b/src/proofpath/tui/widgets/judge_notice.py @@ -2,7 +2,9 @@ Spec section 11: the switch is never silent. The run's own block carries the line as a note too, but that scrolls away with the log; this row sits directly above the bar, -right-aligned, from the moment the run switches until the next run starts. +right-aligned, from the moment the run switches until the next run starts. A switch +for a short 429 is temporary (judge-cooldown spec): the back notice then replaces +the line, through the same path. """ from __future__ import annotations @@ -34,7 +36,9 @@ def rows(self) -> int: def show(self, text: str) -> None: self.text = text - # ``caution``: the run is still judged, by a different model than configured. + # ``caution``: some of the run's second opinions came from a model other than + # the one configured. True of the back notice too: the provider judges again, + # but the report still mixes two judges. style = self._theme.tone("caution") if self._coloured else "" self.update(Text(text, style=style, no_wrap=True, overflow="ellipsis")) self.display = True diff --git a/src/proofpath/verify.py b/src/proofpath/verify.py index 4e49963..a512bdb 100644 --- a/src/proofpath/verify.py +++ b/src/proofpath/verify.py @@ -1559,7 +1559,8 @@ def _summarise( summary=text, # Named only when there is prose to attribute: a paragraph nobody wrote has # no author, and a model named beside an empty summary would read as one. - summary_model=judge.name if text else None, + # Named by the answer, not by the client's state now (judge-cooldown review). + summary_model=(judge.answered_by or judge.name) if text else None, stages=( *report.stages, Stage(name=SUMMARISING, by=judge.name, summary=summary, elapsed=elapsed), @@ -1577,10 +1578,13 @@ def _summarise( def _switch_notice(emit: Listener) -> Callable[[str], None]: - """The judge's switch to its local fallback, as the run's one notice (spec 11). + """The judge's switch to its local fallback, as the run's notice (spec 11). Called on the thread that switched, before the local model is loaded, so a - front end can show the line while the wait it announces is still ahead. + front end can show the line while the wait it announces is still ahead. After a + temporary switch (a short 429, judge-cooldown spec) the same listener carries the + back notice once the provider answers again, so the line on screen is replaced by + the one that is true now, as ``switched`` is for the report's footer. """ def notice(text: str) -> None: diff --git a/tests/test_judge.py b/tests/test_judge.py index e234945..cffddfe 100644 --- a/tests/test_judge.py +++ b/tests/test_judge.py @@ -161,7 +161,11 @@ def test_complete_sends_the_json_schema_and_fills_usage_from_the_response() -> N # Ollama takes no key, so no Authorization header may be sent. assert "authorization" not in route.calls.last.request.headers assert done == judge.Completion( - text='{"opinions": []}', prompt_tokens=120, completion_tokens=7, model="qwen3.5:9b" + text='{"opinions": []}', + prompt_tokens=120, + completion_tokens=7, + model="qwen3.5:9b", + by="ollama qwen3.5:9b", # who answered, as an opinion names it ) assert client.cost.calls == 1 assert client.cost.prompt_tokens == 120 @@ -438,13 +442,58 @@ def test_three_429s_give_up_with_judge_unavailable() -> None: ) slept: list[float] = [] client = judge.JudgeClient(OLLAMA, None, sleep=slept.append) - with pytest.raises(judge.JudgeUnavailable, match="429"): + with pytest.raises(judge.JudgeUnavailable, match="429") as caught: client.complete(MESSAGES) assert route.call_count == 3 assert slept == [1.0, 1.0] # no wait after the last attempt assert client.cost.calls == 0 # The waits happened, so the report must still show them after giving up. assert client.cost.waited_s == 2.0 + # The wait the provider named travels with the error (cooldown spec 2.2). + assert caught.value.retry_after == 1.0 + + +def caught_retry_after(client: judge.JudgeClient) -> float | None: + """``retry_after`` of the ``JudgeUnavailable`` one more call to ``client`` raises.""" + with pytest.raises(judge.JudgeUnavailable) as caught: + client.complete(MESSAGES) + return caught.value.retry_after + + +@pytest.mark.parametrize( + ("headers", "seconds"), + [({"Retry-After": "17"}, 17.0), ({}, None), ({"Retry-After": "soon"}, None)], +) +def test_a_fail_fast_429_carries_its_retry_after( + headers: dict[str, str], seconds: float | None +) -> None: + """Cooldown spec 2.2: the fallback's primary is ``fail_fast``, and that is the path + whose error must say when the provider will answer again: one call, no sleep.""" + with respx.mock() as router: + route = router.post(OLLAMA_URL).mock( + return_value=httpx.Response(429, headers=headers, json={}) + ) + slept: list[float] = [] + client = judge.JudgeClient(OLLAMA, None, sleep=slept.append) + client.fail_fast = True + assert caught_retry_after(client) == seconds + assert (route.call_count, slept) == (1, []) + + +def test_retry_after_belongs_to_the_last_failure_only() -> None: + """A 429 followed by a 5xx gave up on the 5xx, which named no wait.""" + with respx.mock() as router: + router.post(OLLAMA_URL).mock( + side_effect=[ + httpx.Response(429, headers={"Retry-After": "5"}, json={}), + httpx.Response(500), + httpx.Response(500), + ] + ) + client = judge.JudgeClient(OLLAMA, None, sleep=lambda _: None) + with pytest.raises(judge.JudgeUnavailable) as caught: + client.complete(MESSAGES) + assert (caught.value.status, caught.value.retry_after) == (500, None) @respx.mock diff --git a/tests/test_judge_fallback.py b/tests/test_judge_fallback.py index f7050be..d908f64 100644 --- a/tests/test_judge_fallback.py +++ b/tests/test_judge_fallback.py @@ -56,6 +56,11 @@ SCHEMA = {"type": "object", "properties": {"opinions": {"type": "array"}}} OPINION = json.dumps({"opinions": [{"id": "c1", "label": "NEI", "rationale": "silent"}]}) LIMIT = f"Groq limit reached — judging with local ollama {LOCAL}" +# A 429 without ``Retry-After`` cools down for the default 60 s (cooldown spec 2.1), +# so the switch it causes says it is temporary. +COOLING = f"{LIMIT} until it resets (~60s)" +# A sticky switch, for the surfaces that must keep showing today's wording. +KEY_401 = f"Groq rejected the key (HTTP 401) — judging with local ollama {LOCAL}" @pytest.fixture @@ -95,16 +100,17 @@ def fallback( setting: str = LOCAL, sleep: Sleeps | None = None, on_switch: Callable[[str], None] | None = None, + clock: Callable[[], float] = time.monotonic, ) -> judge.FallbackClient: """A Groq primary behind the fallback, exactly as ``build_client`` wires it.""" primary = judge.JudgeClient(GROQ, KEY, sleep=sleep or Sleeps()) - return judge.FallbackClient(primary, fallback=setting, on_switch=on_switch) + return judge.FallbackClient(primary, fallback=setting, on_switch=on_switch, clock=clock) def routes( mock: respx.MockRouter, *, - groq: httpx.Response | Exception | list[httpx.Response] | None = None, + groq: httpx.Response | Exception | list[httpx.Response | Exception] | None = None, tags: httpx.Response | Exception | None = None, local: httpx.Response | None = None, ) -> tuple[respx.Route, respx.Route, respx.Route, respx.Route]: @@ -138,7 +144,7 @@ def test_the_default_config_falls_back_to_qwen3_5(mock: respx.MockRouter) -> Non client.complete(MESSAGES) assert tags.call_count == 1 # only to confirm it is installed assert json.loads(local.calls.last.request.content)["model"] == "qwen3.5:9b" - assert client.switched == "Groq limit reached — judging with local ollama qwen3.5:9b" + assert client.switched == COOLING # a bare 429: the default 60 s cooldown client.close() @@ -201,8 +207,8 @@ def test_a_429_switches_at_once_and_the_opinion_names_the_local_model( opinions = judge.Judge(client).review([judge.JudgeItem("c1", "x", "y", Label.NEI, "low")]) assert opinions["c1"].model == f"ollama {LOCAL}" - assert client.switched == LIMIT - assert seen == [LIMIT] + assert client.switched == COOLING + assert seen == [COOLING] assert groq.call_count == 1 # no retry assert sleep.calls == [] # and no Retry-After wait assert warm.call_count == 1 @@ -323,7 +329,7 @@ def broken(text: str) -> None: _, _, _, local = routes(mock) client = fallback(on_switch=broken) assert client.complete(MESSAGES).text == OPINION - assert client.switched == LIMIT + assert client.switched == COOLING assert local.call_count == 1 @@ -483,7 +489,7 @@ def work() -> None: assert errors == [] assert done == [OPINION, OPINION] - assert seen == [LIMIT] + assert seen == [COOLING] assert groq.call_count == 2 assert warm.call_count == 1 assert local.call_count == 2 @@ -665,7 +671,9 @@ def test_a_users_own_override_is_honoured_as_written(mock: respx.MockRouter) -> assert tags.call_count == 1 assert json.loads(local.calls.last.request.content)["model"] == "my-judge:latest" - assert client.switched == "Groq limit reached — judging with local ollama my-judge:latest" + assert client.switched == ( + "Groq limit reached — judging with local ollama my-judge:latest until it resets (~60s)" + ) def test_an_untagged_override_matches_its_latest_tag(mock: respx.MockRouter) -> None: @@ -744,7 +752,7 @@ def test_reset_goes_back_to_the_primary_and_forgets_the_notice(mock: respx.MockR seen: list[str] = [] client = fallback(on_switch=seen.append) client.complete(MESSAGES) - assert client.switched == LIMIT + assert client.switched == COOLING client.reset() assert client.switched is None @@ -753,7 +761,7 @@ def test_reset_goes_back_to_the_primary_and_forgets_the_notice(mock: respx.MockR assert groq.call_count == 2 assert local.call_count == 1 - assert seen == [LIMIT] + assert seen == [COOLING] # What the local model already answered is not forgotten by the reset. assert (client.cost.calls, client.cost.local_calls) == (2, 1) @@ -765,25 +773,28 @@ def test_reset_goes_back_to_the_primary_and_forgets_the_notice(mock: respx.MockR ) +@pytest.mark.parametrize( + ("failure", "notice"), [(httpx.Response(429), COOLING), (httpx.Response(401), KEY_401)] +) def test_a_run_that_switches_emits_one_notice_and_the_footer_carries_it( - mock: respx.MockRouter, + mock: respx.MockRouter, failure: httpx.Response, notice: str ) -> None: - routes(mock, local=_ok(QUERIES)) + routes(mock, groq=failure, local=_ok(QUERIES)) built = searching(StubSearcher()) built.judge = judge.Judge(judge.build_client(GROQ, KEY)) events: list[Event] = [] report = verify_mod.verify(CLAIM, built, on_event=events.append) notices = [event for event in events if isinstance(event, Note) and event.notice] - assert [note.text for note in notices] == [LIMIT] - assert report.judge_notice == LIMIT + assert [note.text for note in notices] == [notice] + assert report.judge_notice == notice assert report.api_calls == 0 # Groq answered nothing; the one answer was local assert report.judge_cost is not None and report.judge_cost.local_calls == 1 footer = render_footer(report) - assert footer.judge_notice == LIMIT + assert footer.judge_notice == notice assert footer.local_calls == 1 markdown = render_markdown(report) - assert f"- judge fallback: {LIMIT}" in markdown + assert f"- judge fallback: {notice}" in markdown assert "- local calls: 1" in markdown @@ -798,7 +809,7 @@ def test_a_second_run_on_the_same_engine_starts_on_the_primary_again( events: list[Event] = [] second = verify_mod.verify(CLAIM, built, on_event=events.append) - assert first.judge_notice == LIMIT + assert first.judge_notice == COOLING assert second.judge_notice is None assert not [event for event in events if isinstance(event, Note) and event.notice] assert groq.call_count == 2 @@ -807,10 +818,17 @@ def test_a_second_run_on_the_same_engine_starts_on_the_primary_again( runner = CliRunner() +@pytest.mark.parametrize( + ("failure", "notice"), [(httpx.Response(429), COOLING), (httpx.Response(401), KEY_401)] +) def test_the_cli_prints_the_notice_on_stderr( - mock: respx.MockRouter, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + mock: respx.MockRouter, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + failure: httpx.Response, + notice: str, ) -> None: - routes(mock, local=_ok(QUERIES)) + routes(mock, groq=failure, local=_ok(QUERIES)) monkeypatch.setenv("GROQ_API_KEY", KEY.value) monkeypatch.setenv("PROOFPATH_CONFIG_DIR", str(tmp_path / "conf")) built = searching(StubSearcher()) @@ -825,9 +843,9 @@ def fake_default(config: Config, **kwargs: Any) -> verify_mod.Engine: target.write_text(CLAIM, encoding="utf-8") result = runner.invoke(app, ["check", str(target), "--judge", "--format", "json"]) - assert LIMIT in result.stderr + assert notice in result.stderr # stdout is the one JSON document and nothing else; the notice rides inside it. - assert json.loads(result.stdout)["judge_notice"] == LIMIT + assert json.loads(result.stdout)["judge_notice"] == notice assert KEY.value not in result.output @@ -997,7 +1015,7 @@ def test_a_local_model_that_answers_again_is_reported_as_judging_again( client.complete(MESSAGES) assert client.switched is not None and "failed too" in client.switched client.complete(MESSAGES) - assert client.switched == LIMIT + assert client.switched == COOLING def test_a_run_whose_local_model_fails_too_says_both_failed_in_the_report( @@ -1029,3 +1047,526 @@ def close(self) -> None: built = judge.Judge(Down()) # type: ignore[arg-type] assert built.queries(["a claim"]) == {} assert built.unavailable and built.status == 413 + + +# --- the cooldown: back to the provider once its minute limit resets ---------------- +# docs/superpowers/specs/2026-10-05-judge-cooldown-design.md. The clock is a fake moved +# by hand, so a cooldown passes without a single sleep; every number below is a whole +# second added to a whole start, so no float rounding decides a boundary. + + +class Clock: + """The fallback's clock, moved by hand.""" + + def __init__(self) -> None: + self.now = 1000.0 + + def __call__(self) -> float: + return self.now + + +def _limited(seconds: str) -> httpx.Response: + return httpx.Response(429, headers={"Retry-After": seconds}) + + +def _cooling(seconds: int) -> str: + return f"{LIMIT} until it resets (~{seconds}s)" + + +def _back(count: int) -> str: + noun = "request" if count == 1 else "requests" + return f"Groq answering again — local ollama {LOCAL} stood in for {count} {noun}" + + +def _groq_ok() -> httpx.Response: + return _ok(OPINION, model=GROQ.model) + + +def test_a_short_429_switches_then_comes_back_once_retry_after_has_passed( + mock: respx.MockRouter, +) -> None: + """Cooldown spec 2.3: one switch notice, one back notice, and the provider judges + again from the first request at the time it named itself.""" + clock = Clock() + seen: list[str] = [] + groq, _, warm, local = routes(mock, groq=[_limited("30"), _groq_ok(), _groq_ok()]) + client = fallback(on_switch=seen.append, clock=clock) + client.complete(MESSAGES) + assert client.switched == _cooling(30) + assert (client.provider, client.model) == ("ollama", LOCAL) + + clock.now += 30 + client.complete(MESSAGES) + client.complete(MESSAGES) + + assert seen == [_cooling(30), _back(1)] + assert client.switched == _back(1) # the footer tells the truth at the end of the run + assert (client.provider, client.model) == ("groq", GROQ.model) + assert (groq.call_count, local.call_count, warm.call_count) == (3, 1, 1) + assert (client.cost.calls, client.cost.local_calls) == (3, 1) + + +def test_each_opinion_names_the_model_that_gave_it_across_the_cooldown( + mock: respx.MockRouter, +) -> None: + clock = Clock() + routes(mock, groq=[_limited("30"), _groq_ok()]) + reviewer = judge.Judge(fallback(clock=clock)) + items = [judge.JudgeItem("c1", "x", "y", Label.NEI, "low")] + assert reviewer.review(items)["c1"].model == f"ollama {LOCAL}" + clock.now += 30 + assert reviewer.review(items)["c1"].model == f"groq {GROQ.model}" + + +def test_requests_inside_the_cooldown_go_to_the_local_model_only(mock: respx.MockRouter) -> None: + clock = Clock() + groq, _, _, local = routes(mock, groq=_limited("30")) + client = fallback(clock=clock) + client.complete(MESSAGES) + clock.now += 29.5 + client.complete(MESSAGES) + client.complete(MESSAGES) + + assert groq.call_count == 1 # nothing reaches the provider before its own time + assert local.call_count == 3 + + +def test_a_second_short_429_at_the_probe_restarts_the_cooldown_quietly( + mock: respx.MockRouter, +) -> None: + """Cooldown spec 2.3: the new wait comes from the new ``Retry-After``; no second + notice, no second warm-up, and the request that probed is answered locally.""" + clock = Clock() + sleep = Sleeps() + seen: list[str] = [] + groq, _, warm, local = routes(mock, groq=[_limited("30"), _limited("10"), _groq_ok()]) + client = fallback(on_switch=seen.append, clock=clock, sleep=sleep) + client.complete(MESSAGES) + clock.now += 30 + assert client.complete(MESSAGES).text == OPINION + assert (groq.call_count, local.call_count, warm.call_count) == (2, 2, 1) + assert seen == [_cooling(30)] + assert sleep.calls == [] # the probe is fail-fast too: no Retry-After wait + + clock.now += 9 + client.complete(MESSAGES) + assert groq.call_count == 2 # the new wait, not the old one, is honoured + clock.now += 1 + client.complete(MESSAGES) + + assert groq.call_count == 3 + assert seen == [_cooling(30), _back(3)] + assert client.switched == _back(3) + + +@pytest.mark.parametrize( + ("failure", "notice"), + [ + # A daily limit (RPD, TPD): asking again within minutes cannot help. + (_limited("3600"), LIMIT), + (_limited("121"), LIMIT), # one second past COOLDOWN_MAX_S + (httpx.Response(413), LIMIT), # the request alone busts the TPM + ( + httpx.Response(401), + f"Groq rejected the key (HTTP 401) — judging with local ollama {LOCAL}", + ), + ( + httpx.Response(500), + f"Groq did not answer (HTTP 500) — judging with local ollama {LOCAL}", + ), + ], +) +def test_every_other_failure_stays_sticky_with_todays_notice( + mock: respx.MockRouter, failure: httpx.Response, notice: str +) -> None: + clock = Clock() + seen: list[str] = [] + groq, _, _, local = routes(mock, groq=failure) + client = fallback(on_switch=seen.append, clock=clock) + client.complete(MESSAGES) + clock.now += 86_400 + client.complete(MESSAGES) + + assert groq.call_count == 1 + assert local.call_count == 2 + assert seen == [notice] + assert client.switched == notice + + +@pytest.mark.parametrize( + ("failure", "seconds"), + [(httpx.Response(429), 60), (_limited("120"), 120)], +) +def test_a_429_cools_for_its_retry_after_or_60_s_without_one( + mock: respx.MockRouter, failure: httpx.Response, seconds: int +) -> None: + clock = Clock() + start = clock.now + groq, _, _, _ = routes(mock, groq=[failure, _groq_ok()]) + client = fallback(clock=clock) + client.complete(MESSAGES) + assert client.switched == _cooling(seconds) + + clock.now = start + seconds - 1 + client.complete(MESSAGES) + assert groq.call_count == 1 + clock.now = start + seconds + client.complete(MESSAGES) + assert groq.call_count == 2 + + +def test_a_probe_that_fails_for_good_makes_the_switch_sticky(mock: respx.MockRouter) -> None: + """Cooldown spec 2.3: a sticky failure at the probe ends the cooldown for the run. + ``switched`` takes today's wording, since "until it resets" is no longer true.""" + clock = Clock() + seen: list[str] = [] + down = f"Groq did not answer (HTTP 500) — judging with local ollama {LOCAL}" + groq, _, warm, local = routes(mock, groq=[_limited("30"), httpx.Response(500)]) + client = fallback(on_switch=seen.append, clock=clock) + client.complete(MESSAGES) + clock.now += 30 + assert client.complete(MESSAGES).text == OPINION + clock.now += 86_400 + client.complete(MESSAGES) + + assert (groq.call_count, local.call_count, warm.call_count) == (2, 3, 1) + assert seen == [_cooling(30), down] + assert client.switched == down + + +def test_a_429_after_the_return_switches_again_with_its_own_notice( + mock: respx.MockRouter, +) -> None: + """Back on the provider means a later limit is a new switch, told as one: otherwise + ``switched`` would still say the provider was answering.""" + clock = Clock() + seen: list[str] = [] + groq, _, warm, local = routes(mock, groq=[_limited("30"), _groq_ok(), _limited("45")]) + client = fallback(on_switch=seen.append, clock=clock) + client.complete(MESSAGES) + clock.now += 30 + client.complete(MESSAGES) + client.complete(MESSAGES) + + assert seen == [_cooling(30), _back(1), _cooling(45)] + assert client.switched == _cooling(45) + assert (groq.call_count, local.call_count, warm.call_count) == (3, 2, 2) + + +def test_reset_clears_the_cooldown(mock: respx.MockRouter) -> None: + """Cooldown spec 2.5: a new run starts on the provider at once, and a cooldown + from the run before neither delays it nor earns it a back notice.""" + clock = Clock() + seen: list[str] = [] + groq, _, _, _ = routes(mock, groq=[_limited("30"), _groq_ok(), _limited("30")]) + client = fallback(on_switch=seen.append, clock=clock) + client.complete(MESSAGES) + client.reset() + client.complete(MESSAGES) # the same instant: no wait is left over + assert groq.call_count == 2 + assert client.switched is None + assert seen == [_cooling(30)] + + client.complete(MESSAGES) # and the next limit is this run's own switch + assert seen == [_cooling(30), _cooling(30)] + + +def test_only_one_thread_probes_when_the_cooldown_ends(mock: respx.MockRouter) -> None: + """Cooldown spec 2.3: while one thread asks the provider again, the others keep + using the local model; they neither queue behind the probe nor send a second one.""" + clock = Clock() + probing, release = threading.Event(), threading.Event() + asked: list[int] = [] + + def provider(request: httpx.Request) -> httpx.Response: + asked.append(1) + if len(asked) == 1: + return _limited("30") + if len(asked) == 2: + probing.set() + assert release.wait(timeout=10) # the probe stays out until released + return _groq_ok() + + groq, _, _, local = routes(mock) + groq.mock(side_effect=provider) + seen: list[str] = [] + client = fallback(on_switch=seen.append, clock=clock) + client.complete(MESSAGES) + clock.now += 30 + finished = threading.Semaphore(0) + done: list[str] = [] + errors: list[BaseException] = [] + + def work() -> None: + try: + done.append(client.complete(MESSAGES).text) + except BaseException as exc: # pragma: no cover - reported below + errors.append(exc) + finally: + finished.release() + + threads = [threading.Thread(target=work) for _ in range(3)] + for thread in threads: + thread.start() + assert probing.wait(timeout=5) + # Two threads finish on the local model while the probe is still out. + assert finished.acquire(timeout=5) + assert finished.acquire(timeout=5) + assert len(asked) == 2 # the switch, and the one probe + assert local.call_count == 3 + release.set() + for thread in threads: + thread.join(timeout=10) + + assert errors == [] + assert done == [OPINION] * 3 + assert seen == [_cooling(30), _back(3)] + + +# --- the cooldown, review round 1 ---------------------------------------------------- + + +class Ticking(Clock): + """Moves on 31 s at every read: a run's next judge call finds the 30 s cooldown + of its last one over, with nothing in the run there to move a clock by hand.""" + + def __call__(self) -> float: + now = self.now + self.now += 31 + return now + + +class Watched(Clock): + """Says when it is read. The routing reads it under the lock, the moment before + a call that stays on the local model reads the warm-up gate.""" + + def __init__(self) -> None: + super().__init__() + self.read = threading.Event() + + def __call__(self) -> float: + self.read.set() + return super().__call__() + + +def _held_second_local( + entered: threading.Event, gate: threading.Event +) -> Callable[[httpx.Request], httpx.Response]: + """A local route whose second request stays out until ``gate`` opens.""" + asked: list[int] = [] + + def answer(request: httpx.Request) -> httpx.Response: + asked.append(1) + if len(asked) == 2: + entered.set() + assert gate.wait(timeout=10) + return _ok(OPINION) + + return answer + + +def test_a_dead_local_model_is_never_announced_as_judging_again( + mock: respx.MockRouter, +) -> None: + """Review, important 1: latched dead stays dead for the run. A later switch says + both failed, in the notice and in ``switched``, and loads nothing. The return in + between works without the local model.""" + clock = Clock() + seen: list[str] = [] + both = f"Groq limit reached, and local ollama {LOCAL} failed too (HTTP 404)" + # Not "stood in for 0 requests": the footer would lose that the local model failed. + back = f"Groq answering again — local ollama {LOCAL} failed (HTTP 404)" + groq, _, warm, local = routes( + mock, + groq=[_limited("30"), _groq_ok(), _limited("30")], + local=httpx.Response(404), + ) + client = fallback(on_switch=seen.append, clock=clock) + with pytest.raises(judge.JudgeUnavailable): + client.complete(MESSAGES) + assert client.switched == both + clock.now += 30 + assert client.complete(MESSAGES).text == OPINION # Groq is back; nothing stood in + assert client.switched == back + with pytest.raises(judge.JudgeUnavailable): + client.complete(MESSAGES) + + assert client.switched == both + assert seen == [_cooling(30), back, both] + assert (groq.call_count, warm.call_count, local.call_count) == (3, 1, 1) + + +def test_an_answer_in_flight_at_the_return_is_named_for_the_model_that_gave_it( + mock: respx.MockRouter, +) -> None: + """Review, important 2: an opinion names the client that answered it, not the one + judging by the time it is read.""" + clock = Clock() + entered, gate = threading.Event(), threading.Event() + _, _, _, local = routes(mock, groq=[_limited("30"), _groq_ok()]) + local.mock(side_effect=_held_second_local(entered, gate)) + client = fallback(clock=clock) + items = [judge.JudgeItem("c1", "x", "y", Label.NEI, "low")] + client.complete(MESSAGES) # the switch + late: dict[str, judge.JudgeOpinion] = {} + thread = threading.Thread(target=lambda: late.update(judge.Judge(client).review(items))) + thread.start() + assert entered.wait(timeout=5) # a local answer, still out + clock.now += 30 + probed = judge.Judge(client).review(items) # the probe: Groq is back + gate.set() + thread.join(timeout=10) + + assert probed["c1"].model == f"groq {GROQ.model}" + assert late["c1"].model == f"ollama {LOCAL}" + assert client.cost.model == LOCAL # the last answer to land was the local one + + +def test_a_switch_after_a_return_never_strands_a_thread_on_the_first_warm_up( + mock: respx.MockRouter, +) -> None: + """Review, important 3: a thread waiting on the first switch's warm-up is woken by + that switch, even when a second switch was made while the model still loaded.""" + clock = Watched() + warming, release = threading.Event(), threading.Event() + warms: list[int] = [] + + def slow_first_warm(request: httpx.Request) -> httpx.Response: + warms.append(1) + if len(warms) == 1: + warming.set() + assert release.wait(timeout=10) + return httpx.Response(200, json={}) + + _, _, warm, _ = routes(mock, groq=[_limited("30"), _groq_ok(), _limited("30")]) + warm.mock(side_effect=slow_first_warm) + client = fallback(clock=clock) + done: list[str] = [] + switcher = threading.Thread(target=lambda: done.append(client.complete(MESSAGES).text)) + switcher.start() + assert warming.wait(timeout=5) + clock.read.clear() + waiter = threading.Thread( + target=lambda: done.append(client.complete(MESSAGES).text), daemon=True + ) + waiter.start() + assert clock.read.wait(timeout=5) # routed to the local model, now at the gate + clock.now += 30 + client.complete(MESSAGES) # the probe: Groq answers + client.complete(MESSAGES) # a new limit: the second switch, warmed at once + release.set() + switcher.join(timeout=5) + waiter.join(timeout=5) + + assert not waiter.is_alive() # not left to time out after WARM_WAIT_S + assert done == [OPINION, OPINION] + assert warm.call_count == 2 + + +@pytest.mark.parametrize("seconds", ["0", "0.4"]) +def test_a_wait_under_a_second_is_announced_as_about_one( + mock: respx.MockRouter, seconds: str +) -> None: + routes(mock, groq=_limited(seconds)) + client = fallback(clock=Clock()) + client.complete(MESSAGES) + assert client.switched == _cooling(1) + + +def test_the_back_notice_counts_each_cooldown_period_on_its_own( + mock: respx.MockRouter, +) -> None: + """Review, nit 5: a request the local model took during the period counts even when + its answer lands after the return, and never toward the next period.""" + clock = Clock() + seen: list[str] = [] + entered, gate = threading.Event(), threading.Event() + _, _, _, local = routes(mock, groq=[_limited("30"), _groq_ok(), _limited("30"), _groq_ok()]) + local.mock(side_effect=_held_second_local(entered, gate)) + client = fallback(on_switch=seen.append, clock=clock) + client.complete(MESSAGES) # period 1: the switch + thread = threading.Thread(target=client.complete, args=(MESSAGES,)) + thread.start() + assert entered.wait(timeout=5) # period 1: a second request, still out + clock.now += 30 + client.complete(MESSAGES) # the probe: back, with both of period 1's requests + client.complete(MESSAGES) # period 2: a new switch + gate.set() + thread.join(timeout=10) # period 1's answer lands during period 2 + clock.now += 30 + client.complete(MESSAGES) # the probe again + + assert seen == [_cooling(30), _back(2), _cooling(30), _back(1)] + + +def test_a_probe_that_raises_something_else_leaves_the_next_call_to_probe( + mock: respx.MockRouter, +) -> None: + """Not an answer and not a ``JudgeUnavailable``: the error reaches the caller, and + the cooldown is not left waiting for a probe that will never answer.""" + clock = Clock() + groq, _, _, local = routes(mock, groq=[_limited("30"), RuntimeError("bug"), _groq_ok()]) + client = fallback(clock=clock) + client.complete(MESSAGES) + clock.now += 30 + with pytest.raises(RuntimeError): + client.complete(MESSAGES) + client.complete(MESSAGES) + + assert groq.call_count == 3 + assert (client.provider, local.call_count) == ("groq", 1) + + +def test_a_run_whose_judge_comes_back_says_so_in_the_report(mock: respx.MockRouter) -> None: + """The back notice takes the switch notice's road: a note during the run, the + report's ``judge_notice``, its footer and its markdown. The summary, written after + the return, is named for Groq.""" + routes( + mock, + groq=[_limited("30"), _ok("A short summary.", model=GROQ.model)], + local=_ok(QUERIES), + ) + built = searching(StubSearcher()) + built.judge = judge.Judge(fallback(clock=Ticking())) + events: list[Event] = [] + report = verify_mod.verify(CLAIM, built, summarize=True, on_event=events.append) + + notices = [event.text for event in events if isinstance(event, Note) and event.notice] + assert notices == [_cooling(30), _back(1)] + assert report.judge_notice == _back(1) + assert render_footer(report).judge_notice == _back(1) + assert f"- judge fallback: {_back(1)}" in render_markdown(report) + assert report.summary_model == f"groq {GROQ.model}" + assert report.judge_cost is not None + assert (report.api_calls, report.judge_cost.local_calls) == (1, 1) + + +def test_the_cli_prints_the_back_notice_on_stderr( + mock: respx.MockRouter, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + routes( + mock, + groq=[_limited("30"), _ok("A short summary.", model=GROQ.model)], + local=_ok(QUERIES), + ) + monkeypatch.setenv("GROQ_API_KEY", KEY.value) + monkeypatch.setenv("PROOFPATH_CONFIG_DIR", str(tmp_path / "conf")) + built = searching(StubSearcher()) + + def fake_default(config: Config, **kwargs: Any) -> verify_mod.Engine: + # A clock that crosses the cooldown between the run's two judge calls (the + # queries, then the summary); everything else is the CLI's own. + built.judge = judge.Judge(fallback(clock=Ticking())) + built.escalate = kwargs.get("escalate", True) + return built + + monkeypatch.setattr(verify_mod.Engine, "default", staticmethod(fake_default)) + target = tmp_path / "claim.txt" + target.write_text(CLAIM, encoding="utf-8") + result = runner.invoke( + app, ["check", str(target), "--judge", "--summarize", "--format", "json"] + ) + + assert _cooling(30) in result.stderr + assert _back(1) in result.stderr + assert json.loads(result.stdout)["judge_notice"] == _back(1) diff --git a/tests/test_tui_app.py b/tests/test_tui_app.py index b335a7d..2983715 100644 --- a/tests/test_tui_app.py +++ b/tests/test_tui_app.py @@ -623,6 +623,8 @@ class FakeJudge: PARAGRAPH = "Three references do not say this." name = "fake judge-1" + #: Who gave the last answer, as ``Judge.answered_by``: empty until one arrives. + answered_by = "" def __init__(self, *, text: str = PARAGRAPH, detail: str = "") -> None: self._text = text @@ -634,6 +636,8 @@ def __init__(self, *, text: str = PARAGRAPH, detail: str = "") -> None: def summarize(self, report_markdown: str, *, max_tokens: int = 1500) -> str: self.seen = report_markdown + if self._text: + self.answered_by = self.name return self._text def close(self) -> None: @@ -4173,6 +4177,9 @@ async def test_ctrl_l_clears_a_collapsed_panel_and_keeps_an_open_one() -> None: # --- the local fallback judge's notice (spec section 11, Task 11) -------------------- SWITCHED = "Groq limit reached — judging with local ollama qwen3.5:9b" +# What a real client says for a bare 429: a per-minute limit, so the switch is +# temporary and the notice says so (judge-cooldown spec, the default 60 s). +COOLING = f"{SWITCHED} until it resets (~60s)" def _bottom_order(app: ProofpathApp) -> list[str]: @@ -4265,7 +4272,36 @@ async def test_the_rich_footer_keeps_its_height_under_the_notice() -> None: assert notice.region.bottom <= app.query_one(PromptFrame).region.y -async def test_each_run_tries_the_api_first_and_the_old_notice_is_cleared() -> None: +async def test_the_back_notice_replaces_the_switch_notice_above_the_prompt() -> None: + """Judge-cooldown spec: the provider answering again comes down the same road as + the switch, and the row above the bar then says what is true now.""" + back = "Groq answering again — local ollama qwen3.5:9b stood in for 3 requests" + app, schedulers = build_app() + async with app.run_test(size=SIZE) as pilot: + notice = app.query_one(JudgeNotice) + scheduler = schedulers[0] + await submit(pilot, "/check draft.md") + run = scheduler.runs[0] + scheduler.push(run, Note(COOLING, notice=True)) + await pilot.pause() + assert notice.text == COOLING + scheduler.push(run, Note(back, notice=True)) + await pilot.pause() + + assert notice.display and notice.text == back + assert back in _notes(app) + + +@pytest.mark.parametrize( + ("status", "switched"), + [ + (429, COOLING), # a minute limit: temporary + (401, "Groq rejected the key (HTTP 401) — judging with local ollama qwen3.5:9b"), + ], +) +async def test_each_run_tries_the_api_first_and_the_old_notice_is_cleared( + status: int, switched: str +) -> None: """Amendment B 9, end to end on the real scheduler and the real pipeline halves. Run 1's Groq answers 429 and the run switches to the local model; run 2's Groq @@ -4299,7 +4335,7 @@ def engine_factory(run: Run) -> Engine: with respx.mock(assert_all_called=False) as mock: groq = mock.post("https://api.groq.com/openai/v1/chat/completions").mock( - side_effect=[httpx.Response(429), ok("openai/gpt-oss-120b")] + side_effect=[httpx.Response(status), ok("openai/gpt-oss-120b")] ) mock.get("http://localhost:11434/api/tags").mock( return_value=httpx.Response(200, json={"models": [{"name": "qwen3.5:9b"}]}) @@ -4319,8 +4355,8 @@ def engine_factory(run: Run) -> Engine: first = lambda: app._scheduler.runs[0] # noqa: E731 - read after it exists await until(pilot, lambda: first().state in ("done", "failed"), "run 1") assert first().state == "done", first().error - assert first().report is not None and first().report.judge_notice == SWITCHED - assert notice.display and notice.text == SWITCHED + assert first().report is not None and first().report.judge_notice == switched + assert notice.display and notice.text == switched await submit(pilot, CLAIM) second = lambda: app._scheduler.runs[1] # noqa: E731 @@ -4380,7 +4416,7 @@ def factory() -> Judge: await until(pilot, lambda: paragraph in _kv_text(app), "the local summary") notice = app.query_one(JudgeNotice) await until(pilot, lambda: notice.display, "the notice") - assert notice.text == SWITCHED + assert notice.text == COOLING text = _kv_text(app) assert f"(model-written, ollama qwen3.5:9b) {paragraph}" in text