diff --git a/drb-c2-core/app/config.py b/drb-c2-core/app/config.py index 76bc34c..1cbba33 100644 --- a/drb-c2-core/app/config.py +++ b/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 diff --git a/drb-c2-core/app/internal/ai_clients.py b/drb-c2-core/app/internal/ai_clients.py new file mode 100644 index 0000000..544ce20 --- /dev/null +++ b/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, + ) diff --git a/drb-c2-core/app/internal/gemini.py b/drb-c2-core/app/internal/gemini.py index a1c577c..5966d27 100644 --- a/drb-c2-core/app/internal/gemini.py +++ b/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 diff --git a/drb-c2-core/app/internal/intelligence.py b/drb-c2-core/app/internal/intelligence.py index 30daf5e..cef7cbc 100644 --- a/drb-c2-core/app/internal/intelligence.py +++ b/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: diff --git a/drb-c2-core/app/internal/recorrelation_sweep.py b/drb-c2-core/app/internal/recorrelation_sweep.py index cdbc5a1..a4d9fed 100644 --- a/drb-c2-core/app/internal/recorrelation_sweep.py +++ b/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: diff --git a/drb-c2-core/app/internal/summarizer.py b/drb-c2-core/app/internal/summarizer.py index acb6ebe..4361391 100644 --- a/drb-c2-core/app/internal/summarizer.py +++ b/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}], diff --git a/drb-c2-core/app/internal/transcription.py b/drb-c2-core/app/internal/transcription.py index fc7a604..7218bb6 100644 --- a/drb-c2-core/app/internal/transcription.py +++ b/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", diff --git a/drb-c2-core/app/internal/vocabulary_learner.py b/drb-c2-core/app/internal/vocabulary_learner.py index fa90dc1..790f99a 100644 --- a/drb-c2-core/app/internal/vocabulary_learner.py +++ b/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}], diff --git a/drb-c2-core/app/routers/trips.py b/drb-c2-core/app/routers/trips.py index 4f72e05..d7eef96 100644 --- a/drb-c2-core/app/routers/trips.py +++ b/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 = [ diff --git a/drb-c2-core/tests/test_ai_timeouts.py b/drb-c2-core/tests/test_ai_timeouts.py new file mode 100644 index 0000000..ffb4b56 --- /dev/null +++ b/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 diff --git a/drb-c2-core/tests/test_recorrelation_sweep.py b/drb-c2-core/tests/test_recorrelation_sweep.py index eeddfce..c187946 100644 --- a/drb-c2-core/tests/test_recorrelation_sweep.py +++ b/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):