Merge pull request 'Explicit timeouts on every OpenAI and Gemini request (#153)' (#209) from fix/153-ai-request-timeouts into main

Reviewed-on: #209

drb-c2-core/app/config.py

@@ -30,6 +30,24 @@ class Settings(BaseSettings):
     openai_api_key: Optional[str] = None
     stt_model: str = "whisper-1"  # whisper-1 | gpt-4o-mini-transcribe | gpt-4o-transcribe
 
+    # AI request timeouts (server-26#153). Without these the OpenAI SDK waits
+    # up to 600s per attempt and retries twice, and the google-genai client
+    # waits forever, so a hung provider call had no upper bound at all — and
+    # recorrelation_sweep.py's hold-back could only guess at one. Every OpenAI
+    # client is built by app/internal/ai_clients.py and the one Gemini client
+    # by app/internal/gemini.py; both read these. A timeout raises like any
+    # other provider error, so every call site's existing fallback applies
+    # (rules-only correlation, raw transcript, no summary, ...).
+    # The timeout bounds each phase of one HTTP attempt (connect/write/read);
+    # these endpoints are non-streaming, so in practice it bounds the attempt.
+    stt_timeout_seconds: float = 120.0     # Whisper: uploads up to MAX_RECORDING_SECONDS of audio
+    openai_timeout_seconds: float = 60.0   # chat completions + embeddings
+    # Retries per OpenAI request (SDK default 2). The SDK retries timeouts too,
+    # so each one multiplies the worst case by another full timeout.
+    openai_max_retries: int = 1
+    # google-genai does not retry unless told to, so this is the whole bound.
+    gemini_timeout_seconds: float = 60.0
+
     # Google Maps (geocoding)
     google_maps_api_key: Optional[str] = None
     # 511NY developer key (server-26#183/#198). Issued with the NYSDOT Developer

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}],

drb-c2-core/app/routers/trips.py

@@ -531,8 +531,8 @@ async def trip_chat(
 
     events = await fstore.collection_list("trip_events", trip_id=trip_id)
 
-    from openai import AsyncOpenAI
-    oai = AsyncOpenAI(api_key=settings.openai_api_key)
+    from app.internal.ai_clients import async_openai_client
+    oai = async_openai_client()
 
     # Strip history to only user/assistant roles to prevent prompt injection
     safe_history = [

drb-c2-core/tests/test_ai_timeouts.py

@@ -0,0 +1,246 @@
+"""
+server-26#153 — no OpenAI or Gemini request ran with an explicit timeout.
+
+OpenAI's SDK waits 600s per attempt and retries twice (timeouts included);
+google-genai waits forever. A hung provider call therefore had no upper bound,
+and recorrelation_sweep's MIN_MINUTES_SINCE_PIPELINE_START was a guess against
+that tail. These tests pin:
+  - every client is built with the configured timeout / retry budget,
+  - a hung request really does give up (real SDK, local socket that never answers),
+  - a timeout takes the same fallback path as any other provider error,
+  - the sweep hold-back is derived from the timeouts, not hand-picked.
+"""
+import pathlib
+import re
+import socket
+import threading
+import time
+from unittest.mock import AsyncMock, patch
+
+import httpx
+import openai
+import pytest
+
+from app.config import settings
+from app.internal import ai_clients, ai_health, gemini, llm_correlator, recorrelation_sweep
+from app.internal import intelligence, summarizer, transcript_correction
+
+APP_DIR = pathlib.Path(__file__).resolve().parent.parent / "app"
+
+
+def _timeout_error() -> openai.APITimeoutError:
+    return openai.APITimeoutError(request=httpx.Request("POST", "https://api.openai.com/v1/x"))
+
+
+@pytest.fixture(autouse=True)
+def _reset_ai_health():
+    ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS}
+    yield
+    ai_health._state = {t: ai_health._default_state() for t in ai_health.TIERS}
+
+
+@pytest.fixture
+def openai_key():
+    with patch.object(settings, "openai_api_key", "sk-test"):
+        yield
+
+
+# ── Every client carries the configured limits ──────────────────────────────
+
+def test_openai_client_uses_configured_timeout_and_retries(openai_key):
+    with patch.object(settings, "openai_timeout_seconds", 42.0), \
+         patch.object(settings, "openai_max_retries", 1):
+        client = ai_clients.openai_client()
+    assert client.timeout == 42.0
+    assert client.max_retries == 1
+
+
+def test_openai_client_timeout_override_for_stt(openai_key):
+    client = ai_clients.openai_client(timeout=settings.stt_timeout_seconds)
+    assert client.timeout == settings.stt_timeout_seconds
+
+
+def test_async_openai_client_uses_configured_limits(openai_key):
+    with patch.object(settings, "openai_timeout_seconds", 7.0), \
+         patch.object(settings, "openai_max_retries", 0):
+        client = ai_clients.async_openai_client()
+    assert client.timeout == 7.0
+    assert client.max_retries == 0
+
+
+def test_defaults_are_not_the_sdk_defaults():
+    # SDK defaults: OpenAI 600s x 3 attempts, Gemini unbounded.
+    assert settings.stt_timeout_seconds < 600
+    assert settings.openai_timeout_seconds < 600
+    assert settings.openai_max_retries < openai.DEFAULT_MAX_RETRIES
+    assert settings.gemini_timeout_seconds > 0
+
+
+def test_gemini_client_sets_timeout_in_milliseconds():
+    saved = gemini._client
+    gemini._client = None
+    try:
+        with patch.object(settings, "gemini_api_key", "test-key"), \
+             patch.object(settings, "gemini_timeout_seconds", 45.0):
+            client = gemini._get_client()
+        assert client._api_client._http_options.timeout == 45_000
+    finally:
+        gemini._client = saved
+
+
+def test_no_openai_client_is_built_outside_ai_clients():
+    """Regression guard: a new `OpenAI(...)` elsewhere would silently run on
+    the 600s SDK default again."""
+    offenders = []
+    for path in APP_DIR.rglob("*.py"):
+        if path.name == "ai_clients.py":
+            continue
+        for n, line in enumerate(path.read_text().splitlines(), 1):
+            if re.search(r"\b(Async)?OpenAI\(", line) or re.search(r"genai\.Client\(", line):
+                if path.name == "gemini.py" and "genai.Client(" in line:
+                    continue
+                offenders.append(f"{path.relative_to(APP_DIR)}:{n}: {line.strip()}")
+    assert offenders == []
+
+
+# ── A hung request actually gives up ────────────────────────────────────────
+
+@pytest.fixture
+def hung_server():
+    """A TCP server that accepts connections and never sends a byte."""
+    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    srv.bind(("127.0.0.1", 0))
+    srv.listen(8)
+    conns = []
+    stop = threading.Event()
+
+    def accept():
+        srv.settimeout(0.1)
+        while not stop.is_set():
+            try:
+                conns.append(srv.accept()[0])
+            except OSError:
+                continue
+
+    t = threading.Thread(target=accept, daemon=True)
+    t.start()
+    yield f"http://127.0.0.1:{srv.getsockname()[1]}/v1"
+    stop.set()
+    t.join(timeout=2)
+    for c in conns:
+        c.close()
+    srv.close()
+
+
+def test_hung_openai_request_times_out_instead_of_waiting_600s(openai_key, hung_server, monkeypatch):
+    monkeypatch.setenv("OPENAI_BASE_URL", hung_server)
+    with patch.object(settings, "openai_timeout_seconds", 0.3), \
+         patch.object(settings, "openai_max_retries", 1):
+        client = ai_clients.openai_client()
+        t0 = time.monotonic()
+        with pytest.raises(openai.APITimeoutError):
+            client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": "x"}])
+        elapsed = time.monotonic() - t0
+    # 2 attempts x 0.3s + SDK backoff (<~1s); far below the 600s default.
+    assert elapsed < 5
+
+
+# ── Timeouts degrade exactly like other provider errors ────────────────────
+
+def test_timeout_messages_classify_as_transient():
+    assert ai_health.classify(str(_timeout_error())) == "transient"
+    assert ai_health.classify(str(httpx.ReadTimeout("The read operation timed out"))) == "transient"
+
+
+def _corr_ctx():
+    return {"call_id": "c1", "is_thin_call": False, "recent": [{"incident_id": "i1"}],
+            "all_active": [], "call_units": ["12"], "call_cleared": [], "tags": ["mva"]}
+
+
+@pytest.mark.asyncio
+async def test_gemini_timeout_in_decide_falls_back_to_rules():
+    def hang(model, prompt):
+        raise httpx.ReadTimeout("The read operation timed out")
+
+    with patch.object(settings, "gemini_api_key", "test-key"), \
+         patch.object(llm_correlator, "_build_decide_prompt", lambda ctx: "p"), \
+         patch.object(llm_correlator, "_sync_gemini", hang):
+        assert await llm_correlator.decide("c1", _corr_ctx()) is None  # None == rules_only
+    state = ai_health.snapshot()["correlation_cheap"]
+    assert state["consecutive_failures"] == 1
+    assert state["permanent"] is False
+
+
+@pytest.mark.asyncio
+async def test_gemini_timeout_in_tiebreak_keeps_the_rules_decision():
+    def hang(model, prompt):
+        raise httpx.ReadTimeout("The read operation timed out")
+
+    rules = {"action": "new", "matched_incident": None, "corr_debug": {}}
+    with patch.object(llm_correlator, "_build_tiebreak_prompt", lambda r, l, c: "p"), \
+         patch.object(llm_correlator, "_sync_gemini", hang):
+        out = await llm_correlator.tiebreak(rules, {"action": "orphan"}, _corr_ctx())
+    assert out is rules
+    assert ai_health.snapshot()["correlation_smart"]["permanent"] is False
+
+
+@pytest.mark.asyncio
+async def test_gemini_timeout_in_transcript_correction_keeps_raw_text():
+    def hang(model, prompt):
+        raise httpx.ReadTimeout("The read operation timed out")
+
+    text = "engine 5 respond to a structure fire at 12 main street smoke showing"
+    with patch.object(settings, "gemini_api_key", "test-key"), \
+         patch.object(transcript_correction, "resolve_context", AsyncMock(return_value={})), \
+         patch.object(transcript_correction, "_sync_gemini", hang):
+        assert await transcript_correction.correct("c1", text) == (None, None, False)
+
+
+class _TimingOutOpenAI:
+    """Stands in for the client ai_clients builds; every endpoint times out."""
+    def __init__(self, *a, **k):
+        self.chat = self.embeddings = self
+        self.completions = self
+
+    def create(self, **kwargs):
+        raise _timeout_error()
+
+
+def test_extraction_timeout_propagates_to_the_ai_health_path(openai_key):
+    # _sync_extract re-raises API failures so extract_scenes reports them to
+    # ai_health (tier "extraction"); a timeout must not be swallowed as "no scenes".
+    with patch.object(ai_clients, "openai_client", _TimingOutOpenAI):
+        with pytest.raises(openai.APITimeoutError):
+            intelligence._sync_extract("unit 12 on scene", "PD Dispatch", 1, "sys", None)
+
+
+def test_embedding_timeout_returns_none(openai_key):
+    with patch.object(ai_clients, "openai_client", _TimingOutOpenAI):
+        assert intelligence._sync_embed("unit 12 on scene") is None
+
+
+def test_summary_timeout_returns_none(openai_key):
+    with patch.object(ai_clients, "openai_client", _TimingOutOpenAI):
+        assert summarizer._sync_summarize({"type": "fire"}, ["engine 5 on scene"]) is None
+
+
+# ── The sweep hold-back is derived from the timeouts ────────────────────────
+
+def test_worst_case_bound_follows_the_settings():
+    with patch.object(settings, "stt_timeout_seconds", 100.0), \
+         patch.object(settings, "openai_timeout_seconds", 10.0), \
+         patch.object(settings, "openai_max_retries", 1), \
+         patch.object(settings, "gemini_timeout_seconds", 20.0), \
+         patch.object(settings, "stt_retry_on_degenerate", True):
+        bound = recorrelation_sweep.pipeline_worst_case_seconds(n_scenes=2)
+    # stt 2x2x100 + correction 20 + extraction 2x10 + embeds 2x2x10 + decide/tiebreak 2x20 + slack
+    assert bound == 400 + 20 + 20 + 40 + 40 + recorrelation_sweep._NON_AI_SLACK_SECONDS
+
+
+def test_hold_back_covers_the_worst_case_and_never_drops_below_old_floor():
+    bound_min = recorrelation_sweep.pipeline_worst_case_seconds() / 60
+    assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START >= bound_min
+    assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START >= 15
+    # ...and stays inside the scan window, or the sweep could never recover a
+    # call whose pipeline died after writing intelligence_started_at.
+    assert recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START < settings.recorrelation_scan_minutes

drb-c2-core/tests/test_recorrelation_sweep.py

@@ -66,11 +66,14 @@ async def test_sweep_pass_skips_a_call_whose_pipeline_just_started():
         "ended_at": _iso(now - timedelta(minutes=1)),
         "intelligence_started_at": _iso(now - timedelta(seconds=30)),
     }
+    # Relative to the hold-back, which is now derived from the AI timeouts
+    # (server-26#153) rather than a fixed 15 minutes.
+    past_hold = recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START + 4
     genuinely_orphaned_call = {
         "call_id": "call-genuine-orphan",
-        "started_at": _iso(now - timedelta(minutes=20)),
-        "ended_at": _iso(now - timedelta(minutes=19)),
-        "intelligence_started_at": _iso(now - timedelta(minutes=19)),
+        "started_at": _iso(now - timedelta(minutes=past_hold + 1)),
+        "ended_at": _iso(now - timedelta(minutes=past_hold)),
+        "intelligence_started_at": _iso(now - timedelta(minutes=past_hold)),
     }
 
     async def fake_collection_where(collection, clauses):