diff --git a/examples/coding-agent-opencode/README.md b/examples/coding-agent-opencode/README.md index c2ff3f1a0..a53d4d11d 100644 --- a/examples/coding-agent-opencode/README.md +++ b/examples/coding-agent-opencode/README.md @@ -230,12 +230,23 @@ Every case keeps what it observed in the JSON report: the assembled calls, the delta counts, the finish reason, the usage object, the error body, any retries, and the `x-sie-served-by` header when the endpoint sends one. -A server that is still loading the model answers `503 MODEL_LOADING`, so the -check first waits for a tiny request to be served, up to 15 minutes, and -records how long that took. A `429`, or a `503` other than `MODEL_LOADING`, -is retried after its `Retry-After`, up to 5 minutes per request, and each -retry is recorded. An error event in the middle of a stream is recorded as -that case's error. +A server that is still loading the model answers `503 MODEL_LOADING`, and a +gateway still starting a worker answers `503 provisioning`, so the check first +waits for a tiny request to be served, up to 15 minutes, and records how long +that took. A request refused as busy or rate-limited is retried: a `429` or +`503` whose code is `QUEUE_FULL`, `RESOURCE_EXHAUSTED` or +`rate_limit_exceeded`, or an error event with one of those codes that ends a +stream before any output. Each wait is the `Retry-After` header, in seconds or +until its future date, or else the error's `retry_after_s`, capped at 60 +seconds. Without either, the wait is 5 seconds, doubling with each retry. A request is retried at most 4 times and +waits at most 3 minutes in all. Each retry is recorded with its error and +wait in the case's `observed`, and the case's `retries` counts them. A case +whose request is still refused as busy after that, or whose stream ends in a +busy error event after some output, has the status `capacity` rather than +`fail`. The summary counts it apart, and it also makes the exit code 1. Other +errors, including `MODEL_LOADING` or `provisioning` after the warm-up, are not +retried, and an error event in the middle of a stream is recorded as that +case's error. ## What result to expect @@ -320,8 +331,9 @@ served by remote ``` - The one failure is the rate limit, not the feature. The check sends its - cases back to back and does not retry, and the `:relay` profile admits a - bounded number of requests a minute. The same `tool_choice: "required"` + cases back to back, and the `:relay` profile admits a bounded number of + requests a minute. The refusal came as an error event in the stream, which + the check at that commit did not retry. The same `tool_choice: "required"` request sent alone a few minutes later came back with `finish_reason: tool_calls` and the forced `get_weather` call. - A direct chat answer from the same endpoint carried `x-sie-served-by: diff --git a/examples/coding-agent-opencode/conformance.py b/examples/coding-agent-opencode/conformance.py index 054b99238..e996ff770 100755 --- a/examples/coding-agent-opencode/conformance.py +++ b/examples/coding-agent-opencode/conformance.py @@ -23,11 +23,14 @@ import argparse import json +import math import os import sys import time +from collections.abc import Callable from dataclasses import dataclass, field from datetime import UTC, datetime +from email.utils import parsedate_to_datetime from pathlib import Path from typing import Any @@ -39,7 +42,10 @@ REQUEST_TIMEOUT_S = 900.0 WARM_UP_TIMEOUT_S = 900.0 WARM_UP_POLL_S = 5.0 -RETRY_BUDGET_S = 300.0 +WARM_UP_CODES = frozenset({"MODEL_LOADING", "provisioning"}) +BUSY_CODES = frozenset({"QUEUE_FULL", "RESOURCE_EXHAUSTED", "rate_limit_exceeded"}) +BUSY_RETRIES = 4 +BUSY_WAIT_CAP_S = 180.0 RETRY_AFTER_DEFAULT_S = 5.0 RETRY_AFTER_MAX_S = 60.0 GENERATION_BUDGET = 4096 @@ -187,40 +193,83 @@ def _extra(obj: Any, key: str) -> Any: return extra.get(key) if isinstance(extra, dict) else None +def _seconds(value: Any) -> float | None: + try: + seconds = float(value) + except (TypeError, ValueError): + return None + return seconds if math.isfinite(seconds) and seconds >= 0 else None + + +def _retry_after(value: str | None) -> float | None: + seconds = _seconds(value) + if seconds is not None or value is None: + return seconds + try: + when = parsedate_to_datetime(value) + except (TypeError, ValueError): + return None + if when.tzinfo is None: + when = when.replace(tzinfo=UTC) + remaining = (when - datetime.now(UTC)).total_seconds() + return remaining if remaining > 0 else None + + def _error_payload(exc: Exception) -> dict[str, Any]: body = getattr(exc, "body", None) body = body if isinstance(body, dict) else {} inner = body.get("error") if isinstance(body.get("error"), dict) else body + status_error = isinstance(exc, openai.APIStatusError) + retry_after = _retry_after(exc.response.headers.get("retry-after")) if status_error else None return { - "status": exc.status_code if isinstance(exc, openai.APIStatusError) else None, - "code": inner.get("code") or (None if isinstance(exc, openai.APIStatusError) else type(exc).__name__), + "status": exc.status_code if status_error else None, + "code": inner.get("code") or (None if status_error else type(exc).__name__), "param": inner.get("param"), "message": inner.get("message") or str(exc), + "retry_after_s": retry_after if retry_after is not None else _seconds(inner.get("retry_after_s")), } -def _retry_wait(exc: openai.APIStatusError) -> float | None: - busy = exc.status_code == 429 or (exc.status_code == 503 and _error_payload(exc)["code"] != "MODEL_LOADING") - if not busy: +def _is_busy(error: dict[str, Any] | None) -> bool: + return error is not None and error["status"] in (None, 429, 503) and error["code"] in BUSY_CODES + + +def _delivered(result: StreamResult) -> bool: + return bool(result.content or result.reasoning or result.tool_calls) + + +def _busy_retry(result: StreamResult, retries: list[dict[str, Any]]) -> dict[str, Any] | None: + error = result.error + if error is None or not _is_busy(error) or _delivered(result) or len(retries) >= BUSY_RETRIES: return None - try: - wait = float(exc.response.headers.get("retry-after", RETRY_AFTER_DEFAULT_S)) - except ValueError: - wait = RETRY_AFTER_DEFAULT_S - return min(max(wait, 0.0), RETRY_AFTER_MAX_S) + hint = error["retry_after_s"] + wait = min(RETRY_AFTER_DEFAULT_S * 2 ** len(retries) if hint is None else hint, RETRY_AFTER_MAX_S) + if sum(retry["waited_s"] for retry in retries) + wait > BUSY_WAIT_CAP_S: + return None + return {**error, "waited_s": wait} -def _create(client: openai.OpenAI, result: StreamResult, **kwargs: Any) -> Any: +def _retrying(send: Callable[[], StreamResult]) -> StreamResult: started = time.monotonic() + retries: list[dict[str, Any]] = [] while True: - try: - return client.chat.completions.with_raw_response.create(**kwargs) - except openai.APIStatusError as exc: - wait = _retry_wait(exc) - if wait is None or time.monotonic() - started + wait > RETRY_BUDGET_S: - raise - result.retries.append({**_error_payload(exc), "waited_s": wait}) - time.sleep(wait) + result = send() + retry = _busy_retry(result, retries) + if retry is None: + result.retries = retries + result.seconds = time.monotonic() - started + return result + retries.append(retry) + time.sleep(retry["waited_s"]) + + +def _still_busy(result: StreamResult) -> str: + error = result.error or {} + refusal = f"{error.get('status')} {error.get('code')}: {error.get('message')}" + if _delivered(result): + return f"busy after some output, not retried: {refusal}" + waited = sum(retry["waited_s"] for retry in result.retries) + return f"still busy after {len(result.retries)} retries and {waited:g} s of waiting: {refusal}" def _keep_headers(result: StreamResult, raw: Any) -> None: @@ -228,12 +277,17 @@ def _keep_headers(result: StreamResult, raw: Any) -> None: def stream_chat(client: openai.OpenAI, model: str, messages: list[dict[str, Any]], **params: Any) -> StreamResult: + return _retrying(lambda: _stream_once(client, model, messages, **params)) + + +def complete_chat(client: openai.OpenAI, model: str, messages: list[dict[str, Any]], **params: Any) -> StreamResult: + return _retrying(lambda: _complete_once(client, model, messages, **params)) + + +def _stream_once(client: openai.OpenAI, model: str, messages: list[dict[str, Any]], **params: Any) -> StreamResult: result = StreamResult() - started = time.monotonic() try: - raw = _create( - client, - result, + raw = client.chat.completions.with_raw_response.create( model=model, messages=messages, stream=True, @@ -275,15 +329,13 @@ def stream_chat(client: openai.OpenAI, model: str, messages: list[dict[str, Any] call.arguments += piece.function.arguments except (openai.APIError, httpx.HTTPError) as exc: result.error = _error_payload(exc) - result.seconds = time.monotonic() - started return result -def complete_chat(client: openai.OpenAI, model: str, messages: list[dict[str, Any]], **params: Any) -> StreamResult: +def _complete_once(client: openai.OpenAI, model: str, messages: list[dict[str, Any]], **params: Any) -> StreamResult: result = StreamResult() - started = time.monotonic() try: - raw = _create(client, result, model=model, messages=messages, **params) + raw = client.chat.completions.with_raw_response.create(model=model, messages=messages, **params) _keep_headers(result, raw) completion = raw.parse() if completion.usage is not None: @@ -298,7 +350,6 @@ def complete_chat(client: openai.OpenAI, model: str, messages: list[dict[str, An result.reasoning = reasoning except (openai.APIError, httpx.HTTPError) as exc: result.error = _error_payload(exc) - result.seconds = time.monotonic() - started return result @@ -309,6 +360,8 @@ def _describe_calls(result: StreamResult) -> str: def _fail_on_error(name: str, result: StreamResult) -> Case | None: if result.error is None: return None + if _is_busy(result.error): + return Case(name, "capacity", _still_busy(result), result.observed()) error = result.error return Case( name, "fail", f"request failed: {error['status']} {error['code']}: {error['message']}", result.observed() @@ -394,6 +447,8 @@ def case_tool_result_round_trip( ) -> Case: name = "tool_result_round_trip" calls = first.calls() + if _is_busy(first.error): + return Case(name, "capacity", "not run, streaming_tool_call_deltas stayed busy", first.observed()) if first.error or not calls: return Case(name, "fail", "no tool call from streaming_tool_call_deltas to send a result for", first.observed()) assistant: dict[str, Any] = { @@ -516,6 +571,8 @@ def case_finish_reason_stop(client: openai.OpenAI, model: str) -> tuple[Case, St def case_finish_reason_tool_calls(first: Case) -> Case: name = "finish_reason_tool_calls" finish = first.observed.get("finish_reason") + if first.status == "capacity": + return Case(name, "capacity", "not judged, the tool-call turn stayed busy", {"finish_reason": finish}) if finish == "tool_calls": return Case(name, "pass", "finish_reason=tool_calls on the tool-call turn", {"finish_reason": finish}) return Case(name, "fail", f"tool-call turn ended with finish_reason={finish!r}", {"finish_reason": finish}) @@ -539,6 +596,8 @@ def case_finish_reason_length(client: openai.OpenAI, model: str) -> Case: def case_usage_on_final_chunk(stop: StreamResult) -> Case: name = "usage_on_final_chunk" observed = {"usage": stop.usage, "usage_on_last_chunk": stop.usage_on_last_chunk, "chunks": stop.chunks} + if _is_busy(stop.error): + return Case(name, "capacity", "not judged, the stop stream stayed busy", observed) if stop.error: return Case(name, "fail", "the stop stream failed, see finish_reason_stop", observed) usage = stop.usage or {} @@ -585,15 +644,18 @@ def case_include_reasoning(client: openai.OpenAI, model: str) -> tuple[Case, boo } accepted = with_flag.error is None for label, result in (("include_reasoning", with_flag), ("plain request", without_flag)): - if result.error: + if result.error and not _is_busy(result.error): error = result.error verdict = "refused" if label == "include_reasoning" else "failed" detail = f"{label} {verdict}: {error['status']} {error['code']}: {error['message']}" return Case(name, "fail", detail, observed), accepted - if whole.error: + if whole.error and not _is_busy(whole.error): error = whole.error detail = f"non-streamed include_reasoning failed: {error['status']} {error['code']}: {error['message']}" return Case(name, "fail", detail, observed), accepted + busy = next((result for result in (with_flag, without_flag, whole) if _is_busy(result.error)), None) + if busy is not None: + return Case(name, "capacity", _still_busy(busy), observed), accepted if observed["content_has_think_tags"]: return Case(name, "fail", "raw tags leaked into content", observed), accepted if without_flag.reasoning_deltas: @@ -645,6 +707,8 @@ def case_tool_choice_required(client: openai.OpenAI, model: str) -> Case: observed = result.observed() if result.error: error = result.error + if _is_busy(error): + return Case(name, "capacity", _still_busy(result), observed) if _is_refusal(error): observed["support"] = "refused" return Case(name, "recorded", f"refused: {error['status']} {error['code']}: {error['message']}", observed) @@ -689,7 +753,7 @@ def wait_until_loaded(client: openai.OpenAI, model: str) -> dict[str, Any]: attempts += 1 result = stream_chat(client, model, messages, max_tokens=4, temperature=0) elapsed = time.monotonic() - started - if (result.error or {}).get("code") != "MODEL_LOADING" or elapsed > WARM_UP_TIMEOUT_S: + if (result.error or {}).get("code") not in WARM_UP_CODES or elapsed > WARM_UP_TIMEOUT_S: return {"attempts": attempts, "seconds": round(elapsed, 1), "error": result.error} time.sleep(WARM_UP_POLL_S) @@ -734,7 +798,8 @@ def run(base_url: str, api_key: str, model: str) -> dict[str, Any]: cases.append(reasoning_case) cases.append(case_tool_choice_required(client, model)) - counts = {status: sum(1 for case in cases if case.status == status) for status in ("pass", "fail", "recorded")} + statuses = ("pass", "fail", "recorded", "capacity") + counts = {status: sum(1 for case in cases if case.status == status) for status in statuses} return { "base_url": base_url, "model": model, @@ -744,11 +809,16 @@ def run(base_url: str, api_key: str, model: str) -> dict[str, Any]: "served_by": tool_result.headers.get("x-sie-served-by"), "warm_up": warm_up, "tools_turn_params": tool_params, - "cases": [case.__dict__ for case in cases], + "cases": [{**case.__dict__, "retries": _retries_in(case.observed)} for case in cases], "summary": counts, } +def _retries_in(observed: dict[str, Any]) -> int: + nested = sum(_retries_in(value) for value in observed.values() if isinstance(value, dict)) + return len(observed.get("retries") or []) + nested + + def print_table(report: dict[str, Any]) -> None: width = max(len(case["name"]) for case in report["cases"]) print(f"model {report['model']} at {report['base_url']}") @@ -758,10 +828,12 @@ def print_table(report: dict[str, Any]) -> None: print(f"warm-up: {warm_up['attempts']} attempt(s), {warm_up['seconds']}s") print() for case in report["cases"]: - print(f" {case['name']:<{width}} {case['status']:<8} {case['detail']}") + retried = f" ({case['retries']} busy retries)" if case["retries"] else "" + print(f" {case['name']:<{width}} {case['status']:<8} {case['detail']}{retried}") summary = report["summary"] + capacity = f", {summary['capacity']} capacity" if summary["capacity"] else "" print() - print(f"{summary['pass']} pass, {summary['fail']} fail, {summary['recorded']} recorded") + print(f"{summary['pass']} pass, {summary['fail']} fail, {summary['recorded']} recorded{capacity}") def main() -> int: @@ -781,7 +853,7 @@ def main() -> int: args.report.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") print_table(report) print(f"report written to {args.report}") - return 1 if report["summary"]["fail"] else 0 + return 1 if report["summary"]["fail"] or report["summary"]["capacity"] else 0 if __name__ == "__main__": diff --git a/examples/coding-agent-opencode/tests/test_conformance.py b/examples/coding-agent-opencode/tests/test_conformance.py index b72c5a3e8..5e5839bf6 100644 --- a/examples/coding-agent-opencode/tests/test_conformance.py +++ b/examples/coding-agent-opencode/tests/test_conformance.py @@ -9,9 +9,13 @@ from __future__ import annotations import json +import sys import threading from collections.abc import Iterator +from datetime import UTC, datetime, timedelta +from email.utils import format_datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path from typing import Any import pytest @@ -25,6 +29,16 @@ LONG_ANSWER = ( "9.9 is the larger number, because its tenths digit, 9, is larger than the tenths digit of 9.11, which is 1." ) +RATE_LIMITED = {"status": 429, "code": "rate_limit_exceeded", "message": "rate limit exceeded, retry later"} +BUSY_EVENT = { + "error_event": { + "message": "the upstream is busy, please retry", + "type": "server_error", + "param": None, + "code": "RESOURCE_EXHAUSTED", + "retry_after_s": 12, + } +} def _words(text: str) -> int: @@ -40,10 +54,13 @@ def _reply(body: dict[str, Any]) -> dict[str, Any]: if body.get("tool_choice") == "auto" and KNOBS.get("auto") == "refuse": return {"status": 400, "code": "unsupported_field", "message": "field 'tool_choice' is not supported"} if body.get("tool_choice") == "required": + refusals = KNOBS.get("required_refusals", []) + refused = SEEN.setdefault("required_refusals", []) + if len(refused) < len(refusals): + refused.append(body) + return refusals[len(refused) - 1] if KNOBS.get("required") == "refuse": return {"status": 400, "code": "unsupported_field", "message": "tool_choice 'required' is not supported"} - if KNOBS.get("required") == "rate_limited": - return {"status": 429, "code": "rate_limit_exceeded", "message": "rate limit exceeded, retry later"} if KNOBS.get("required") == "ignore": return {"content": "Hello!", "finish": "stop"} return {"calls": [("call_req", "get_weather", {"city": "Paris"})], "finish": "tool_calls"} @@ -56,9 +73,9 @@ def _reply(body: dict[str, Any]) -> dict[str, Any]: return {"status": 500, "code": "INTERNAL_ERROR", "message": "worker failed"} if KNOBS.get("busy", 0) > len(SEEN.get("busy", [])): SEEN.setdefault("busy", []).append(body) - return {"status": 503, "code": "QUEUE_FULL", "message": "upstream busy"} + return {"status": 503, "code": "QUEUE_FULL", "message": "upstream busy", "headers": {"retry-after": "0"}} if KNOBS.get("midstream_error"): - return {"midstream_error": True} + return {"error_event": KNOBS["midstream_error"]} if tools and "Tokyo" in user_text: calls = [("call_1", "get_weather", {"city": "Paris"}), ("call_2", "get_weather", {"city": "Tokyo"})] if KNOBS.get("parallel") == "single": @@ -73,6 +90,8 @@ def _reply(body: dict[str, Any]) -> dict[str, Any]: return {"status": 400, "code": "INPUT_TOO_LONG", "message": "prompt exceeds the context window"} return {"content": "7741-ZEBRA", "finish": "stop"} if "9.11" in user_text: + if KNOBS.get("plain_busy") and not body.get("include_reasoning"): + return {"status": 503, "code": "QUEUE_FULL", "message": "upstream busy", "headers": {"retry-after": "0"}} mode = KNOBS.get("reasoning", "yes") reasoning = REASONING if mode in ("yes", "ignored", "stream_only") else "" answer = LONG_ANSWER if KNOBS.get("long_answer") else "9.9" @@ -90,7 +109,7 @@ def _reply(body: dict[str, Any]) -> dict[str, Any]: def _thinking_budget_spent(body: dict[str, Any], reply: dict[str, Any]) -> dict[str, Any]: think = KNOBS.get("think_tokens", 0) max_tokens = body.get("max_tokens") - if not think or max_tokens in (4, 8) or "status" in reply or reply.get("midstream_error"): + if not think or max_tokens in (4, 8) or "status" in reply or "error_event" in reply: return reply if max_tokens is None or max_tokens < think + 64: return {"content": "", "finish": "length", "hidden": "think " * (max_tokens or 0)} @@ -146,14 +165,17 @@ def do_POST(self) -> None: body = json.loads(self.rfile.read(int(self.headers["content-length"]))) KNOBS["_posts"] = KNOBS.get("_posts", 0) + 1 if KNOBS["_posts"] <= KNOBS.get("loading", 0): - error = {"message": "model is loading, please retry", "type": "server_error", "code": "MODEL_LOADING"} + code = KNOBS.get("loading_code", "MODEL_LOADING") + error = {"message": "model is loading, please retry", "type": "server_error", "code": code} self._json(503, {"error": error}) return + if KNOBS.get("busy_all"): + self._json(503, {"error": {"message": "upstream busy", "type": "server_error", "code": "QUEUE_FULL"}}) + return reply = _thinking_budget_spent(body, _reply(body)) if "status" in reply: error = {"message": reply["message"], "type": "invalid_request_error", "code": reply["code"]} - retry = {"retry-after": str(KNOBS.get("retry_after", 0))} if reply["status"] in (429, 503) else {} - self._json(reply["status"], {"error": error}, retry) + self._json(reply["status"], {"error": error}, reply.get("headers")) return if not body.get("stream"): message: dict[str, Any] = {"role": "assistant", "content": reply.get("content")} @@ -175,10 +197,12 @@ def do_POST(self) -> None: self.send_header("x-sie-upstream", "vendor-name-from-the-profile") self.end_headers() self._chunk([{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}]) - if reply.get("midstream_error"): + if "error_event" in reply: + if reply.get("partial"): + self._chunk([{"index": 0, "delta": {"content": reply["partial"]}, "finish_reason": None}]) self._chunk( [{"index": 0, "delta": {}, "finish_reason": None}], - error={"message": "upstream busy", "type": "server_error", "param": None, "code": "QUEUE_FULL"}, + error=reply["error_event"], request_id="req-stub", ) self.wfile.write(b"data: [DONE]\n\n") @@ -222,6 +246,13 @@ def reset_knobs() -> Iterator[None]: SEEN.clear() +@pytest.fixture +def waits(monkeypatch: pytest.MonkeyPatch) -> list[float]: + slept: list[float] = [] + monkeypatch.setattr(conformance.time, "sleep", slept.append) + return slept + + def _report(base_url: str, **knobs: Any) -> dict[str, Any]: KNOBS.update(knobs) return conformance.run(base_url, "", "stub-model") @@ -280,9 +311,10 @@ def test_tool_call_deltas_without_ids_fail(base_url: str) -> None: assert cases["parallel_tool_calls"]["status"] == "fail" -def test_model_loading_is_waited_out(base_url: str, monkeypatch: pytest.MonkeyPatch) -> None: +@pytest.mark.parametrize("code", ["MODEL_LOADING", "provisioning"]) +def test_model_loading_is_waited_out(base_url: str, monkeypatch: pytest.MonkeyPatch, code: str) -> None: monkeypatch.setattr(conformance, "WARM_UP_POLL_S", 0.01) - report = _report(base_url, loading=3) + report = _report(base_url, loading=3, loading_code=code) assert report["warm_up"]["attempts"] == 4 assert report["warm_up"]["error"] is None assert _failures({case["name"]: case for case in report["cases"]}) == set() @@ -292,8 +324,148 @@ def test_busy_refusals_are_retried_and_recorded(base_url: str) -> None: cases = _run(base_url, busy=2) assert _failures(cases) == set() assert cases["tool_choice_auto"]["status"] == "pass" - retries = cases["streaming_tool_call_deltas"]["observed"]["retries"] - assert [(retry["status"], retry["code"]) for retry in retries] == [(503, "QUEUE_FULL"), (503, "QUEUE_FULL")] + case = cases["streaming_tool_call_deltas"] + assert case["status"] == "pass" + assert case["retries"] == 2 + retries = case["observed"]["retries"] + assert [(retry["status"], retry["code"], retry["waited_s"]) for retry in retries] == [(503, "QUEUE_FULL", 0.0)] * 2 + assert cases["parallel_tool_calls"]["retries"] == 0 + + +def test_a_rate_limited_request_waits_out_its_retry_after(base_url: str, waits: list[float]) -> None: + limited = {**RATE_LIMITED, "headers": {"retry-after": "7"}} + case = _run(base_url, required_refusals=[limited, limited])["tool_choice_required"] + assert case["status"] == "recorded" + assert case["observed"]["support"] == "supported" + assert case["retries"] == 2 + assert waits == [7.0, 7.0] + + +def test_a_busy_error_event_before_any_output_is_retried(base_url: str, waits: list[float]) -> None: + case = _run(base_url, required_refusals=[BUSY_EVENT])["tool_choice_required"] + assert case["status"] == "recorded" + assert case["observed"]["support"] == "supported" + (retry,) = case["observed"]["retries"] + assert (retry["status"], retry["code"], retry["retry_after_s"]) == (None, "RESOURCE_EXHAUSTED", 12.0) + assert waits == [12.0] + + +@pytest.mark.parametrize( + ("refusal", "expected_waits"), + [ + (RATE_LIMITED, [5.0, 10.0, 20.0, 40.0]), + ({**RATE_LIMITED, "headers": {"retry-after": "Wed, 21 Oct 2015 07:28:00 GMT"}}, [5.0, 10.0, 20.0, 40.0]), + ({**RATE_LIMITED, "headers": {"retry-after": "nan"}}, [5.0, 10.0, 20.0, 40.0]), + ({**RATE_LIMITED, "headers": {"retry-after": "-3"}}, [5.0, 10.0, 20.0, 40.0]), + ({**RATE_LIMITED, "headers": {"retry-after": "60"}}, [60.0, 60.0, 60.0]), + ({**RATE_LIMITED, "headers": {"retry-after": "600"}}, [60.0, 60.0, 60.0]), + (BUSY_EVENT, [12.0, 12.0, 12.0, 12.0]), + ], +) +def test_a_request_still_busy_after_the_cap_is_a_capacity_failure( + base_url: str, waits: list[float], refusal: dict[str, Any], expected_waits: list[float] +) -> None: + report = _report(base_url, required_refusals=[refusal] * 10) + case = {case["name"]: case for case in report["cases"]}["tool_choice_required"] + assert case["status"] == "capacity" + assert "support" not in case["observed"] + assert case["retries"] == len(expected_waits) + assert waits == expected_waits + assert case["detail"].startswith(f"still busy after {len(expected_waits)} retries and {sum(expected_waits):g} s") + assert report["summary"] == {"pass": 10, "fail": 0, "recorded": 0, "capacity": 1} + + +def test_a_retry_after_date_is_waited_until(base_url: str, waits: list[float]) -> None: + when = format_datetime(datetime.now(UTC) + timedelta(seconds=30), usegmt=True) + case = _run(base_url, required_refusals=[{**RATE_LIMITED, "headers": {"retry-after": when}}])[ + "tool_choice_required" + ] + assert case["status"] == "recorded" + assert case["retries"] == 1 + assert len(waits) == 1 + assert 25.0 < waits[0] <= 30.0 + + +def test_a_busy_request_does_not_hide_a_refusal(base_url: str, waits: list[float]) -> None: + refused = _run(base_url, reasoning="refuse", plain_busy=True)["include_reasoning"] + assert refused["status"] == "fail" + assert refused["detail"].startswith("include_reasoning refused: 400 unsupported_field") + assert refused["retries"] == conformance.BUSY_RETRIES + KNOBS.clear() + SEEN.clear() + busy = _run(base_url, plain_busy=True)["include_reasoning"] + assert busy["status"] == "capacity" + assert busy["detail"].startswith("still busy after 4 retries") + + +@pytest.mark.parametrize( + "refusal", + [ + {"status": 429, "code": "insufficient_quota", "message": "you exceeded your current quota"}, + {"status": 503, "code": "transport_failure", "message": "upstream down", "headers": {"retry-after": "0"}}, + {"status": 500, "code": "RESOURCE_EXHAUSTED", "message": "out of memory", "headers": {"retry-after": "0"}}, + { + "error_event": { + "message": "generation failed", + "type": "server_error", + "param": None, + "code": "inference_error", + } + }, + ], +) +def test_other_errors_are_failures_and_not_retried(base_url: str, waits: list[float], refusal: dict[str, Any]) -> None: + case = _run(base_url, required_refusals=[refusal, refusal])["tool_choice_required"] + assert case["status"] == "fail" + assert case["retries"] == 0 + assert waits == [] + + +def test_a_busy_error_event_after_output_is_not_retried(base_url: str, waits: list[float]) -> None: + refusal = {"partial": "Hel", **BUSY_EVENT} + case = _run(base_url, required_refusals=[refusal, refusal])["tool_choice_required"] + assert case["status"] == "capacity" + assert ( + case["detail"] + == "busy after some output, not retried: None RESOURCE_EXHAUSTED: the upstream is busy, please retry" + ) + assert case["retries"] == 0 + assert case["observed"]["content"] == "Hel" + assert waits == [] + + +def test_an_endpoint_that_stays_busy_fails_on_capacity_not_behaviour( + base_url: str, + waits: list[float], + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + report = _report(base_url, busy_all=True) + cases = {case["name"]: case for case in report["cases"]} + assert {name: case["status"] for name, case in cases.items()} == { + "tool_choice_auto": "recorded", + "streaming_tool_call_deltas": "capacity", + "parallel_tool_calls": "capacity", + "tool_result_round_trip": "capacity", + "prompt_above_64k_tokens": "capacity", + "finish_reason_stop": "capacity", + "finish_reason_tool_calls": "capacity", + "finish_reason_length": "capacity", + "usage_on_final_chunk": "capacity", + "include_reasoning": "capacity", + "tool_choice_required": "capacity", + } + assert cases["streaming_tool_call_deltas"]["retries"] == conformance.BUSY_RETRIES + assert cases["include_reasoning"]["retries"] == 3 * conformance.BUSY_RETRIES + assert cases["finish_reason_tool_calls"]["retries"] == 0 + assert cases["usage_on_final_chunk"]["retries"] == 0 + assert report["summary"] == {"pass": 0, "fail": 0, "recorded": 1, "capacity": 10} + argv = ["conformance.py", "--base-url", base_url, "--model", "stub-model", "--report", str(tmp_path / "r.json")] + monkeypatch.setattr(sys, "argv", argv) + capsys.readouterr() + assert conformance.main() == 1 + assert "0 pass, 0 fail, 1 recorded, 10 capacity" in capsys.readouterr().out def test_a_failed_tools_turn_does_not_blame_tool_choice(base_url: str) -> None: @@ -304,11 +476,13 @@ def test_a_failed_tools_turn_does_not_blame_tool_choice(base_url: str) -> None: def test_an_error_event_mid_stream_is_recorded_not_raised(base_url: str) -> None: - cases = _run(base_url, midstream_error=True) + error = {"message": "generation failed", "type": "server_error", "param": None, "code": "inference_error"} + cases = _run(base_url, midstream_error=error) case = cases["streaming_tool_call_deltas"] assert case["status"] == "fail" - assert case["observed"]["error"]["code"] == "QUEUE_FULL" - assert case["observed"]["error"]["message"] == "upstream busy" + assert case["observed"]["error"]["code"] == "inference_error" + assert case["observed"]["error"]["message"] == "generation failed" + assert case["retries"] == 0 assert cases["finish_reason_stop"]["status"] == "pass" @@ -342,16 +516,6 @@ def test_required_refused_is_recorded_not_failed(base_url: str) -> None: assert cases["tool_choice_required"]["observed"]["error"]["status"] == 400 -def test_required_rate_limited_is_a_failure_not_a_refusal(base_url: str, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(conformance, "RETRY_BUDGET_S", 0.2) - cases = _run(base_url, required="rate_limited", retry_after=0.05) - case = cases["tool_choice_required"] - assert case["status"] == "fail" - assert "support" not in case["observed"] - assert case["observed"]["error"]["status"] == 429 - assert case["observed"]["retries"] - - def test_required_accepted_but_ignored_fails(base_url: str) -> None: cases = _run(base_url, required="ignore") assert cases["tool_choice_required"]["status"] == "fail"