Logan CusanoandClaude Opus 5.5
278edf48dc
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>
101 lines
4.5 KiB
Python
101 lines
4.5 KiB
Python
"""
|
|
server-26#131 — the re-correlation sweep's orphan filter checked incident_id/
|
|
incident_ids/corr_path but had no way to tell "never processed" apart from
|
|
"real-time pipeline (routers/upload.py _run_intelligence_pipeline) is still
|
|
mid-flight". Racing the sweep against an in-flight real-time correlation could
|
|
land the same call on two different incidents — the exact duplicate-link bug
|
|
#131 found in 3 live dumps (~2% of linked calls). This pins the fix: a call
|
|
whose intelligence_started_at marker is recent is held back from the sweep
|
|
regardless of how orphaned it otherwise looks.
|
|
"""
|
|
from datetime import datetime, timezone, timedelta
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app.internal import recorrelation_sweep
|
|
|
|
|
|
def _iso(dt: datetime) -> str:
|
|
return dt.isoformat()
|
|
|
|
|
|
class TestPipelineLikelyStillRunning:
|
|
def test_recent_marker_is_still_running(self):
|
|
now = datetime.now(timezone.utc)
|
|
call = {"intelligence_started_at": _iso(now - timedelta(minutes=1))}
|
|
assert recorrelation_sweep._pipeline_likely_still_running(call, now) is True
|
|
|
|
def test_old_marker_is_not_still_running(self):
|
|
now = datetime.now(timezone.utc)
|
|
call = {"intelligence_started_at": _iso(now - timedelta(minutes=30))}
|
|
assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False
|
|
|
|
def test_marker_exactly_at_the_threshold_is_not_held_back(self):
|
|
# age_minutes < MIN_MINUTES_SINCE_PIPELINE_START (strict), so exactly
|
|
# at the threshold is old enough to release — pins the boundary so it
|
|
# can't drift to <= by accident and silently double the hold time.
|
|
now = datetime.now(timezone.utc)
|
|
threshold = recorrelation_sweep.MIN_MINUTES_SINCE_PIPELINE_START
|
|
call = {"intelligence_started_at": _iso(now - timedelta(minutes=threshold))}
|
|
assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False
|
|
|
|
def test_no_marker_at_all_is_not_held_back(self):
|
|
"""A pre-#131 call doc, or the marker write itself failed — absence
|
|
isn't evidence of an in-flight pipeline, so the sweep must still be
|
|
able to pick these up (that's its whole job)."""
|
|
now = datetime.now(timezone.utc)
|
|
assert recorrelation_sweep._pipeline_likely_still_running({}, now) is False
|
|
|
|
def test_unparseable_marker_is_not_held_back(self):
|
|
now = datetime.now(timezone.utc)
|
|
call = {"intelligence_started_at": "not-a-timestamp"}
|
|
assert recorrelation_sweep._pipeline_likely_still_running(call, now) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sweep_pass_skips_a_call_whose_pipeline_just_started():
|
|
"""Integration-shaped: a call that looks orphaned by every OTHER filter
|
|
(no incident_ids, no corr_path, no skip_reason, under the attempt budget)
|
|
but has a fresh intelligence_started_at must not reach correlate_call —
|
|
that's the race #131 found."""
|
|
now = datetime.now(timezone.utc)
|
|
racing_call = {
|
|
"call_id": "call-racing",
|
|
"started_at": _iso(now - timedelta(minutes=2)),
|
|
"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=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):
|
|
assert collection == "calls"
|
|
return [racing_call, genuinely_orphaned_call]
|
|
|
|
correlate_calls: list[str] = []
|
|
|
|
async def fake_correlate_call(**kwargs):
|
|
correlate_calls.append(kwargs["call_id"])
|
|
return None # no match — exercises the "not linked" branch too
|
|
|
|
doc_sets: list[tuple] = []
|
|
|
|
async def fake_doc_set(collection, doc_id, data, merge=True):
|
|
doc_sets.append((collection, doc_id, data))
|
|
|
|
with patch.object(recorrelation_sweep, "fstore") as mock_fstore, \
|
|
patch("app.internal.incident_correlator.correlate_call", fake_correlate_call):
|
|
mock_fstore.collection_where = fake_collection_where
|
|
mock_fstore.doc_set = fake_doc_set
|
|
await recorrelation_sweep._run_sweep_pass()
|
|
|
|
assert correlate_calls == ["call-genuine-orphan"]
|