test(llm): live per-request timeout coverage for all providers

Two live checks per provider: a generous timeout asserted at the SDK
call boundary, and a tight timeout that must abort well under the 600s
client default. Gemini's async transport can be aiohttp, so its tight
timeout surfaces as asyncio.TimeoutError rather than httpx.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/live_llm/test_live_timeouts.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import time
+from typing import Any
+
+import anthropic
+import httpx
+import openai
+import pytest
+
+from src.llm.request_builder import execute_completion
+
+from .conftest import make_backend, require_provider_key, wrap_async_method
+from .model_matrix import LiveModelSpec, ProviderName, get_live_model_specs
+
+pytestmark = [pytest.mark.live_llm]
+
+GENEROUS_TIMEOUT_SECONDS = 120
+TIGHT_TIMEOUT_SECONDS = 0.01
+# Well under the 600s client default; generous enough to absorb SDK retries.
+TIGHT_TIMEOUT_WALL_CLOCK_LIMIT_SECONDS = 30
+
+TIMEOUT_EXCEPTIONS: dict[ProviderName, tuple[type[BaseException], ...]] = {
+    "anthropic": (anthropic.APITimeoutError,),
+    "openai": (openai.APITimeoutError,),
+    # google-genai raises httpx or aiohttp timeouts depending on its transport;
+    # aiohttp surfaces as asyncio.TimeoutError (== builtins.TimeoutError).
+    "gemini": (httpx.TimeoutException, TimeoutError),
+}
+
+PROVIDER_MARKS = {
+    "anthropic": pytest.mark.requires_anthropic,
+    "openai": pytest.mark.requires_openai,
+    "gemini": pytest.mark.requires_gemini,
+}
+
+
+def representative_specs() -> list[Any]:
+    """One spec per provider — timeout plumbing is transport-level, not model-level."""
+    params: list[Any] = []
+    for provider in ("anthropic", "openai", "gemini"):
+        specs = get_live_model_specs(provider=provider)
+        if not specs:
+            continue
+        params.append(
+            pytest.param(specs[0], marks=PROVIDER_MARKS[provider], id=specs[0].id)
+        )
+    return params
+
+
+def assert_timeout_reached_sdk(
+    model_spec: LiveModelSpec, call_kwargs: dict[str, Any], timeout_seconds: float
+) -> None:
+    if model_spec.provider == "gemini":
+        http_options = call_kwargs["config"]["http_options"]
+        assert http_options.timeout == int(timeout_seconds * 1000)
+    else:
+        assert call_kwargs["timeout"] == timeout_seconds
+
+
+def sdk_call_target(backend: Any, model_spec: LiveModelSpec) -> tuple[Any, str]:
+    if model_spec.provider == "gemini":
+        return backend._client.aio.models, "generate_content"
+    if model_spec.provider == "anthropic":
+        return backend._client.messages, "create"
+    return backend._client.chat.completions, "create"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("model_spec", representative_specs())
+async def test_live_provider_timeout_reaches_the_wire(
+    model_spec: LiveModelSpec,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    require_provider_key(model_spec)
+    backend, config = make_backend(
+        model_spec, provider_params={"timeout": GENEROUS_TIMEOUT_SECONDS}
+    )
+    target, attribute = sdk_call_target(backend, model_spec)
+    calls = wrap_async_method(monkeypatch, target, attribute)
+
+    result = await execute_completion(
+        backend,
+        config,
+        messages=[{"role": "user", "content": "Reply with the single word: ok"}],
+        max_tokens=256,
+    )
+
+    assert isinstance(result.content, str)
+    assert result.content.strip()
+    assert len(calls) == 1
+    assert_timeout_reached_sdk(model_spec, calls[0]["kwargs"], GENEROUS_TIMEOUT_SECONDS)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("model_spec", representative_specs())
+async def test_live_tight_provider_timeout_aborts_request(
+    model_spec: LiveModelSpec,
+) -> None:
+    require_provider_key(model_spec)
+    backend, config = make_backend(
+        model_spec, provider_params={"timeout": TIGHT_TIMEOUT_SECONDS}
+    )
+
+    started = time.monotonic()
+    with pytest.raises(TIMEOUT_EXCEPTIONS[model_spec.provider]):
+        await execute_completion(
+            backend,
+            config,
+            messages=[{"role": "user", "content": "Reply with the single word: ok"}],
+            max_tokens=256,
+        )
+    elapsed = time.monotonic() - started
+
+    assert (
+        elapsed < TIGHT_TIMEOUT_WALL_CLOCK_LIMIT_SECONDS
+    ), f"tight timeout took {elapsed:.1f}s — per-request timeout likely not applied"