From 38638b3f8bf4d50c60524ea53865f6ea82a03fe8 Mon Sep 17 00:00:00 2001 From: "Michael J. Jabbour" Date: Wed, 19 Aug 2026 03:28:45 -0400 Subject: [PATCH] fix: report retry count on llm:response so a retried call stops looking clean MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 30-minute failure was logged as a success. In session `eec9ae98` the resume-2 turn spent 14:19:26 -> 14:49:44 on three sequential 600 s timeouts and emitted one `llm:response` with `duration_ms: 1818942` (30.3 min) and `status: ok` -- a single "successful" call swallowing three timeouts, hiding the failure from every latency metric that keys on status. The same forensic section reports 4 of 10 provider failures never reached `events.jsonl` at all. Root cause, from `amplifier_core.utils.retry.retry_with_backoff`: `on_retry` is invoked only before a retry *sleep*. Two paths raise without ever calling it -- a non-retryable error, and the terminal failure once retries are exhausted. So the most important failure in any sequence is precisely the one the `provider:retry` stream never sees. Meanwhile `elapsed_ms` already spans the whole retry loop, so a long duration was indistinguishable from a slow model, and `llm:response` -- the one event every consumer already reads -- said nothing about attempts at all. Track the retry count inside the call and surface it as `retries` on `llm:response`: on the success path and on both error paths, so the two paths `on_retry` never reaches are covered. A 30-minute success now describes itself: `status: ok, duration_ms: 1818942, retries: 3`. Additive, no contract change. `retries` is always present -- an absent key is indistinguishable from an old producer, so consumers never have to guess whether zero means "none" or "not reported". Five tests cover the shapes that mattered in the incident: a success that swallowed retries reports them; a clean success reports `retries: 0`; a terminal failure reports what it cost (the only record, since `on_retry` never fires there); a non-retryable failure reports zero; and duration and retries are reported together, because neither is actionable alone. Also collapses one pre-existing `ruff format` drift in the touched function, which is why the repo's format drift drops from 6 files to 5. Verified: `ruff check` clean, 298 passed (baseline 293). 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- amplifier_module_provider_vllm/__init__.py | 19 ++- tests/test_retry_observability.py | 185 +++++++++++++++++++++ 2 files changed, 201 insertions(+), 3 deletions(-) create mode 100644 tests/test_retry_observability.py diff --git a/amplifier_module_provider_vllm/__init__.py b/amplifier_module_provider_vllm/__init__.py index ce02664..8a8f65d 100644 --- a/amplifier_module_provider_vllm/__init__.py +++ b/amplifier_module_provider_vllm/__init__.py @@ -929,9 +929,7 @@ def _build_continuation_input( # If we extracted any assistant content, add as a spec-compliant message item if assistant_content: - continuation_input.append( - _build_assistant_message_item(assistant_content) - ) + continuation_input.append(_build_assistant_message_item(assistant_content)) return continuation_input @@ -1618,8 +1616,17 @@ async def _do_complete(): retryable=True, ) from e + # Retries that happened INSIDE this one logical call. Without it a call + # that swallowed three 600 s timeouts reports as a single clean success: + # session `eec9ae98` recorded `duration_ms: 1818942` (30.3 min) with + # `status: ok`, which hides the failure from every latency metric that + # keys on status. + retry_count = 0 + async def _on_retry(attempt: int, delay: float, error: kernel_errors.LLMError): """Callback invoked before each retry sleep.""" + nonlocal retry_count + retry_count = attempt if self.coordinator and hasattr(self.coordinator, "hooks"): await self.coordinator.hooks.emit( PROVIDER_RETRY, @@ -1926,6 +1933,10 @@ async def _on_retry(attempt: int, delay: float, error: kernel_errors.LLMError): "usage": event_usage, "status": "ok", "duration_ms": elapsed_ms, + # How many attempts this "success" actually cost. `elapsed_ms` + # already spans the whole retry loop, so without this a long + # duration is indistinguishable from a slow model. + "retries": retry_count, "continuation_count": continuation_count if continuation_count > 0 else None, @@ -1947,6 +1958,7 @@ async def _on_retry(attempt: int, delay: float, error: kernel_errors.LLMError): { "status": "error", "duration_ms": elapsed_ms, + "retries": retry_count, "error": str(e), "provider": self.name, "model": params["model"], @@ -1967,6 +1979,7 @@ async def _on_retry(attempt: int, delay: float, error: kernel_errors.LLMError): { "status": "error", "duration_ms": elapsed_ms, + "retries": retry_count, "error": error_msg, "provider": self.name, "model": params["model"], diff --git a/tests/test_retry_observability.py b/tests/test_retry_observability.py new file mode 100644 index 0000000..0c93a5a --- /dev/null +++ b/tests/test_retry_observability.py @@ -0,0 +1,185 @@ +"""A call that retried internally must say so on the way out. + +From the forensic report on session `eec9ae98`: + +> **A 30-minute failure was logged as success.** The resume-2 turn spent +> 14:19:26 -> 14:49:44 on three sequential 600 s timeouts. The eventual response +> recorded `duration_ms: 1818942` (30.3 min) with `status: ok` -- one +> "successful" call swallowing three timeouts, hiding the failure from any +> latency metric. + +`elapsed_ms` already spans the whole retry loop, so a long duration was +indistinguishable from a slow model. `provider:retry` events exist, but they are +a separate stream that has to be joined by hand -- and `retry_with_backoff` only +invokes `on_retry` *before a retry sleep*, so a terminal failure (retries +exhausted) and a non-retryable error emit nothing at all. The one event every +consumer already reads, `llm:response`, said nothing. + +It now carries `retries`, on success and on both error paths. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, patch + +import httpx +import openai +import pytest +from amplifier_core import llm_errors as kernel_errors +from amplifier_core.message_models import ChatRequest, Message + +from amplifier_module_provider_vllm import VLLMProvider + + +class FakeHooks: + def __init__(self) -> None: + self.events: list[tuple[str, dict[str, Any]]] = [] + + async def emit(self, name: str, payload: dict[str, Any]) -> None: + self.events.append((name, payload)) + + +class FakeCoordinator: + def __init__(self) -> None: + self.hooks = FakeHooks() + + +class DummyResponse: + def __init__(self) -> None: + self.output: list[Any] = [] + self.usage = SimpleNamespace(input_tokens=10, output_tokens=5) + self.status = "completed" + self.id = "resp_test" + + +def _provider() -> tuple[VLLMProvider, FakeCoordinator]: + provider = VLLMProvider( + base_url="http://localhost:8000/v1", + config={ + "max_retries": 3, + "min_retry_delay": 0.01, + "max_retry_delay": 1.0, + "use_streaming": False, + }, + ) + coordinator = FakeCoordinator() + provider.coordinator = coordinator # type: ignore[assignment] + return provider, coordinator + + +def _request() -> ChatRequest: + return ChatRequest(messages=[Message(role="user", content="Hello")]) + + +def _timeout_error() -> openai.APITimeoutError: + return openai.APITimeoutError( + request=httpx.Request("POST", "http://localhost:8000/v1/responses") + ) + + +def _responses(coordinator: FakeCoordinator) -> list[dict[str, Any]]: + return [ + payload for name, payload in coordinator.hooks.events if name == "llm:response" + ] + + +def test_a_success_that_swallowed_retries_reports_them() -> None: + """The incident's shape: one `status: ok` hiding three timeouts.""" + provider, coordinator = _provider() + provider.client.responses.create = AsyncMock( # type: ignore[method-assign] + side_effect=[_timeout_error(), _timeout_error(), DummyResponse()] + ) + + with patch("asyncio.sleep", new_callable=AsyncMock): + asyncio.run(provider.complete(_request())) + + responses = _responses(coordinator) + assert len(responses) == 1 + assert responses[0]["status"] == "ok" + assert responses[0]["retries"] == 2, ( + "a success that cost three attempts reported as a clean single call; " + "duration alone cannot distinguish it from a slow model" + ) + + +def test_a_clean_success_reports_no_retries() -> None: + """The field must be present and zero, not absent. + + An absent key is indistinguishable from an old producer; consumers should + never have to guess whether zero means "none" or "not reported". + """ + provider, coordinator = _provider() + provider.client.responses.create = AsyncMock(return_value=DummyResponse()) # type: ignore[method-assign] + + asyncio.run(provider.complete(_request())) + + responses = _responses(coordinator) + assert responses[0]["status"] == "ok" + assert responses[0]["retries"] == 0 + + +def test_a_terminal_failure_reports_what_it_cost() -> None: + """`on_retry` never fires for the final attempt, so this is the only record. + + `retry_with_backoff` raises without notifying once retries are exhausted -- + the most important failure in the sequence is the one the retry stream never + sees. + """ + provider, coordinator = _provider() + provider.client.responses.create = AsyncMock(side_effect=_timeout_error()) # type: ignore[method-assign] + + with ( + patch("asyncio.sleep", new_callable=AsyncMock), + pytest.raises(kernel_errors.LLMError), + ): + asyncio.run(provider.complete(_request())) + + responses = _responses(coordinator) + assert responses[-1]["status"] == "error" + assert responses[-1]["retries"] == 3, ( + "a failure after four attempts must record the three retries it burned" + ) + + +def test_a_non_retryable_failure_reports_zero() -> None: + """Raised immediately, never retried -- and `on_retry` never fires here either.""" + provider, coordinator = _provider() + provider.client.responses.create = AsyncMock( # type: ignore[method-assign] + side_effect=openai.AuthenticationError( + "Invalid key", + response=httpx.Response( + status_code=401, + request=httpx.Request("POST", "http://localhost:8000/v1/responses"), + ), + body=None, + ) + ) + + with pytest.raises(kernel_errors.AuthenticationError): + asyncio.run(provider.complete(_request())) + + responses = _responses(coordinator) + assert responses[-1]["status"] == "error" + assert responses[-1]["retries"] == 0 + + +def test_duration_and_retries_are_reported_together() -> None: + """Neither number is actionable alone. + + 30 minutes with `retries: 0` is a slow model; 30 minutes with `retries: 3` + is three dead connections. The incident could not tell those apart. + """ + provider, coordinator = _provider() + provider.client.responses.create = AsyncMock( # type: ignore[method-assign] + side_effect=[_timeout_error(), DummyResponse()] + ) + + with patch("asyncio.sleep", new_callable=AsyncMock): + asyncio.run(provider.complete(_request())) + + payload = _responses(coordinator)[0] + assert {"status", "duration_ms", "retries"} <= payload.keys() + assert payload["retries"] == 1