Explicit timeouts on every OpenAI and Gemini request (#153)

No AI client set a timeout: OpenAI's SDK waits 600s per attempt and retries
twice (timeouts included); google-genai waits forever. A hung provider call
had no upper bound, so recorrelation_sweep's 15-minute pipeline hold-back was
a guess against an unbounded tail.

- config: stt_timeout_seconds (120), openai_timeout_seconds (60),
  openai_max_retries (1), gemini_timeout_seconds (60).
- app/internal/ai_clients.py builds every OpenAI/AsyncOpenAI client with them
  (transcription, extraction, embeddings, summarizer, vocabulary learner,
  trip chat). gemini.py's shared client gets HttpOptions(timeout=ms).
- A timeout raises like any other provider error, so existing fallbacks are
  unchanged: rules-only correlation, rules decision on tiebreak, raw
  transcript on correction, ai_health "transient".
- MIN_MINUTES_SINCE_PIPELINE_START is now derived from those timeouts
  (pipeline_worst_case_seconds, 3-scene budget + 2 min non-AI slack), floor
  15: 21 min with the defaults. Loop warns if it reaches the scan window.
- tests/test_ai_timeouts.py: client limits, a real hung-socket request that
  gives up in <5s, timeout fallback per tier, the derived bound, and a guard
  against any new OpenAI( outside ai_clients.py.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

drb-c2-core/app/internal/ai_clients.py

@@ -0,0 +1,44 @@
+"""
+One place every OpenAI client is built, so none of them runs on SDK-default
+timeouts (server-26#153).
+
+The OpenAI SDK's defaults are a 600s read timeout and 2 retries, which it also
+applies to timeouts: one hung Whisper request could hold a call's pipeline for
+half an hour, twice over with stt_retry_on_degenerate. That unbounded tail is
+what recorrelation_sweep.py's MIN_MINUTES_SINCE_PIPELINE_START had to guess
+against. With these settings it is derived instead (see
+recorrelation_sweep.pipeline_worst_case_seconds).
+
+A timeout raises openai.APITimeoutError, an ordinary Exception, so each call
+site's existing failure path handles it unchanged: ai_health.classify() reads
+"Request timed out." as transient.
+
+The Gemini client is built in app/internal/gemini.py with the same intent.
+"""
+from typing import Optional
+
+from app.config import settings
+
+
+def _timeout(timeout: Optional[float]) -> float:
+    return settings.openai_timeout_seconds if timeout is None else timeout
+
+
+def openai_client(timeout: Optional[float] = None):
+    """Synchronous OpenAI client. `timeout` overrides openai_timeout_seconds (STT passes its own)."""
+    from openai import OpenAI  # lazy, like every call site before this module
+    return OpenAI(
+        api_key=settings.openai_api_key,
+        timeout=_timeout(timeout),
+        max_retries=settings.openai_max_retries,
+    )
+
+
+def async_openai_client(timeout: Optional[float] = None):
+    """Async OpenAI client, same limits as openai_client()."""
+    from openai import AsyncOpenAI
+    return AsyncOpenAI(
+        api_key=settings.openai_api_key,
+        timeout=_timeout(timeout),
+        max_retries=settings.openai_max_retries,
+    )

drb-c2-core/app/internal/gemini.py

@@ -44,7 +44,18 @@ def _get_client():
     with _client_lock:
         if _client is None:
             from google import genai  # lazy — only when a Gemini call is made
-            _client = genai.Client(api_key=settings.gemini_api_key)
+            from google.genai import types
+            # server-26#153: google-genai's default is no timeout at all, so a
+            # hung call held its pipeline (and the sweep hold-back sized
+            # against it) open indefinitely. HttpOptions.timeout is in ms.
+            # The client does not retry unless retry_options is set, so this
+            # is the whole bound per request. A timeout raises from
+            # generate_json like any API error, so every caller's fallback
+            # (rules-only correlation, uncorrected transcript) is unchanged.
+            _client = genai.Client(
+                api_key=settings.gemini_api_key,
+                http_options=types.HttpOptions(timeout=int(settings.gemini_timeout_seconds * 1000)),
+            )
         return _client
 
 

drb-c2-core/app/internal/intelligence.py

@@ -938,7 +938,7 @@ def _sync_extract(
 ) -> list[dict]:
     """Call GPT-4o-mini and return a list of scene dicts."""
     from app.config import settings
-    from openai import OpenAI
+    from app.internal.ai_clients import openai_client
 
     if not settings.openai_api_key:
         logger.warning("OPENAI_API_KEY not set — intelligence extraction disabled.")
@@ -956,7 +956,7 @@ def _sync_extract(
     )
 
     try:
-        client = OpenAI(api_key=settings.openai_api_key)
+        client = openai_client()
         response = client.chat.completions.create(
             model="gpt-4o-mini",
             messages=[{"role": "user", "content": prompt}],
@@ -985,13 +985,13 @@ def _sync_extract(
 def _sync_embed(text: str) -> Optional[list[float]]:
     """Generate a text-embedding-3-small vector for semantic similarity."""
     from app.config import settings
-    from openai import OpenAI
+    from app.internal.ai_clients import openai_client
 
     if not settings.openai_api_key:
         return None
 
     try:
-        client = OpenAI(api_key=settings.openai_api_key)
+        client = openai_client()
         result = client.embeddings.create(model="text-embedding-3-small", input=text)
         return result.data[0].embedding
     except Exception as e:

drb-c2-core/app/internal/recorrelation_sweep.py

@@ -14,6 +14,7 @@ Never creates new incidents — link-only.  Zero LLM tokens (uses pre-computed
 talkgroup strings, haversine math, and stored embeddings).
 """
 import asyncio
+import math
 from datetime import datetime, timezone, timedelta
 from typing import Optional
 from app.internal.logger import logger
@@ -33,16 +34,54 @@ from app.config import settings
 # itself failed) is NOT held back by this — absence isn't evidence of an
 # in-flight pipeline, and #131's own bug predates this field existing.
 #
-# 15, not 5: neither OpenAI's Whisper client nor Gemini's call in
-# llm_correlator.py sets a request timeout (server-26#153), so a hung call can
-# run well past a few minutes on SDK-default retries, and this constant is a
-# guess against that unbounded tail, not a measured bound. Raising it costs
-# nothing on the recovery side: a call that finished processing (linked OR
-# genuinely orphaned) always has corr_path set (_apply_and_log writes it even
-# on the orphan action), so it's already excluded by the
-# `not c.get("corr_path")` filter below and never reaches this check at all —
-# this constant only ever delays calls that are still actually running.
-MIN_MINUTES_SINCE_PIPELINE_START = 15
+# Derived, not guessed (server-26#153). Every AI request the real-time
+# pipeline makes now has an explicit timeout (config.py: stt_timeout_seconds,
+# openai_timeout_seconds, openai_max_retries, gemini_timeout_seconds), so the
+# longest a call can stay "started but no corr_path yet" has an upper bound:
+# pipeline_worst_case_seconds() below. The hold-back is that bound rounded up,
+# never below the old hand-picked 15.
+#
+# Raising it costs nothing on the recovery side: a call that finished
+# processing (linked OR genuinely orphaned) always has corr_path set
+# (_apply_and_log writes it even on the orphan action), so it's already
+# excluded by the `not c.get("corr_path")` filter below and never reaches this
+# check at all — this constant only ever delays calls that are still actually
+# running, or whose pipeline died after writing the marker.
+
+# Scenes whose embeddings are computed before the first scene's correlation
+# commits (intelligence.extract_scenes embeds every scene, then upload.py
+# correlates them in order; scene 0's commit writes corr_path). Extraction
+# rarely returns more than 2; 3 is the budget, not a hard cap.
+_BOUND_SCENES = 3
+# Non-AI work on the same path: GCS download, Firestore reads/writes, geocoding
+# and place verification (httpx, 5s timeouts each).
+_NON_AI_SLACK_SECONDS = 120.0
+_FLOOR_MINUTES = 15
+
+
+def pipeline_worst_case_seconds(n_scenes: int = _BOUND_SCENES) -> float:
+    """Upper bound on intelligence_started_at -> first corr_path write, from
+    the configured AI timeouts. Order follows routers/upload.py
+    _run_intelligence_pipeline."""
+    openai_attempts = 1 + max(0, settings.openai_max_retries)
+    stt_rounds = 2 if settings.stt_retry_on_degenerate else 1
+    stt = stt_rounds * openai_attempts * settings.stt_timeout_seconds
+    # Transcript correction (inside transcribe_call). gemini.generate_json only
+    # re-issues a request when the model rejects a thinking level — a fast
+    # 400, not a timeout — and google-genai itself does not retry.
+    correction = settings.gemini_timeout_seconds
+    extraction = openai_attempts * settings.openai_timeout_seconds
+    embeddings = n_scenes * openai_attempts * settings.openai_timeout_seconds
+    correlation = 2 * settings.gemini_timeout_seconds  # decide + tiebreak, scene 0
+    return stt + correction + extraction + embeddings + correlation + _NON_AI_SLACK_SECONDS
+
+
+def _derive_min_minutes() -> int:
+    return max(_FLOOR_MINUTES, math.ceil(pipeline_worst_case_seconds() / 60))
+
+
+# 21 with the defaults (stt 120s, openai 60s x2 attempts, gemini 60s, 3 scenes).
+MIN_MINUTES_SINCE_PIPELINE_START = _derive_min_minutes()
 
 # Standard link-only retry budget before a call is tombstoned corr_path="unlinked".
 MAX_SWEEP_ATTEMPTS = 3
@@ -66,8 +105,17 @@ async def recorrelation_loop() -> None:
     logger.info(
         f"Re-correlation sweep started — "
         f"interval: {settings.summary_interval_minutes}m, "
-        f"scan window: {settings.recorrelation_scan_minutes}m"
+        f"scan window: {settings.recorrelation_scan_minutes}m, "
+        f"pipeline hold-back: {MIN_MINUTES_SINCE_PIPELINE_START}m"
     )
+    if MIN_MINUTES_SINCE_PIPELINE_START >= settings.recorrelation_scan_minutes:
+        # A call is held back longer than the window that finds it, so the
+        # sweep can never recover a call whose pipeline died mid-flight.
+        logger.warning(
+            f"Re-correlation sweep: pipeline hold-back ({MIN_MINUTES_SINCE_PIPELINE_START}m) "
+            f">= scan window ({settings.recorrelation_scan_minutes}m) — lower the AI "
+            f"timeouts or raise RECORRELATION_SCAN_MINUTES"
+        )
     while True:
         await asyncio.sleep(interval)
         try:

drb-c2-core/app/internal/summarizer.py

@@ -213,7 +213,7 @@ async def _resolve_stale_incidents() -> None:
 
 def _sync_summarize(inc: dict, transcripts: list[str]) -> Optional[str]:
     from app.config import settings
-    from openai import OpenAI
+    from app.internal.ai_clients import openai_client
 
     if not settings.openai_api_key:
         return None
@@ -241,7 +241,7 @@ Write a concise factual summary of this incident in 2-4 sentences. Include:
 Be factual. Do not speculate beyond what the transcripts say. Do not use bullet points."""
 
     try:
-        client = OpenAI(api_key=settings.openai_api_key)
+        client = openai_client()
         response = client.chat.completions.create(
             model="gpt-4o-mini",
             messages=[{"role": "user", "content": prompt}],

drb-c2-core/app/internal/transcription.py

@@ -271,8 +271,8 @@ def _sync_transcribe(
     """
     from google.cloud import storage as gcs
     from google.oauth2 import service_account
-    from openai import OpenAI
     from app.config import settings
+    from app.internal.ai_clients import openai_client
 
     if not settings.openai_api_key:
         logger.warning("OPENAI_API_KEY not set — transcription disabled.")
@@ -315,9 +315,11 @@ def _sync_transcribe(
         # gpt-4o-transcribe and gpt-4o-mini-transcribe only support json/text.
         use_verbose = settings.stt_model == "whisper-1"
 
-        openai_client = OpenAI(api_key=settings.openai_api_key)
+        # Own timeout, not openai_timeout_seconds: this request uploads the
+        # whole recording (server-26#153).
+        stt_client = openai_client(timeout=settings.stt_timeout_seconds)
         with open(tmp_path, "rb") as f:
-            response = openai_client.audio.transcriptions.create(
+            response = stt_client.audio.transcriptions.create(
                 model=settings.stt_model,
                 file=f,
                 language="en",

drb-c2-core/app/internal/vocabulary_learner.py

@@ -470,7 +470,7 @@ def _diff_new_terms(original: str, corrected: str) -> list[str]:
 
 def _sync_bootstrap(system_name: str, system_type: str, area_hint: str) -> list[str]:
     from app.config import settings as cfg
-    from openai import OpenAI
+    from app.internal.ai_clients import openai_client
 
     if not cfg.openai_api_key:
         return []
@@ -481,7 +481,7 @@ def _sync_bootstrap(system_name: str, system_type: str, area_hint: str) -> list[
         area_hint=area_hint,
     )
     try:
-        client = OpenAI(api_key=cfg.openai_api_key)
+        client = openai_client()
         response = client.chat.completions.create(
             model="gpt-4o",
             messages=[{"role": "user", "content": prompt}],
@@ -504,7 +504,7 @@ def _sync_induct(
 ) -> list[dict]:
     """Returns [{term, meaning}] — a bare string is still accepted from the model."""
     from app.config import settings as cfg
-    from openai import OpenAI
+    from app.internal.ai_clients import openai_client
 
     if not cfg.openai_api_key:
         return []
@@ -518,7 +518,7 @@ def _sync_induct(
         transcript_block=transcript_block[:8000],
     )
     try:
-        client = OpenAI(api_key=cfg.openai_api_key)
+        client = openai_client()
         response = client.chat.completions.create(
             model="gpt-4o-mini",
             messages=[{"role": "user", "content": prompt}],