correlator: judge each scene on its own embedding + severity (#80, #95) #105

Merged
logan merged 1 commits from fix/scene-context-leak-80-95 into main 2026-09-06 17:49:24 -04:00
5 changed files with 98 additions and 3 deletions
Showing only changes of commit a9d1d2475a - Show all commits
@@ -668,10 +668,17 @@ async def correlate_call(
vehicles: Optional[list[str]] = None, vehicles: Optional[list[str]] = None,
cleared_units: Optional[list[str]] = None, cleared_units: Optional[list[str]] = None,
reassignment: bool = False, reassignment: bool = False,
embedding: Optional[list] = None,
severity: Optional[str] = None,
) -> Optional[str]: ) -> Optional[str]:
""" """
Link call_id to an existing incident or create a new one. Link call_id to an existing incident or create a new one.
Thin wrapper: builds context → runs rules decision → commits. Thin wrapper: builds context → runs rules decision → commits.
``embedding`` and ``severity`` are the SCENE's own values (server-26#80/#95).
Callers that re-correlate a whole call rather than a scene — the
recorrelation sweep — pass the call doc's stored values explicitly; they are
no longer read from the doc inside _build_context.
""" """
ctx = await _build_context( ctx = await _build_context(
call_id=call_id, units=units, vehicles=vehicles, cleared_units=cleared_units, call_id=call_id, units=units, vehicles=vehicles, cleared_units=cleared_units,
@@ -679,6 +686,7 @@ async def correlate_call(
system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
tags=tags, incident_type=incident_type, location=location, tags=tags, incident_type=incident_type, location=location,
reassignment=reassignment, create_if_new=create_if_new, reassignment=reassignment, create_if_new=create_if_new,
embedding=embedding, severity=severity,
) )
decision = _run_decision(ctx) decision = _run_decision(ctx)
return await _apply_and_log(decision, ctx) return await _apply_and_log(decision, ctx)
@@ -700,6 +708,8 @@ async def preview_correlation(
vehicles: Optional[list[str]] = None, vehicles: Optional[list[str]] = None,
cleared_units: Optional[list[str]] = None, cleared_units: Optional[list[str]] = None,
reassignment: bool = False, reassignment: bool = False,
embedding: Optional[list] = None,
severity: Optional[str] = None,
) -> dict: ) -> dict:
""" """
Run the rules engine and return the decision WITHOUT committing to Firestore. Run the rules engine and return the decision WITHOUT committing to Firestore.
@@ -720,6 +730,7 @@ async def preview_correlation(
system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name, system_id=system_id, talkgroup_id=talkgroup_id, talkgroup_name=talkgroup_name,
tags=tags, incident_type=incident_type, location=location, tags=tags, incident_type=incident_type, location=location,
reassignment=reassignment, create_if_new=create_if_new, reassignment=reassignment, create_if_new=create_if_new,
embedding=embedding, severity=severity,
) )
decision = _run_decision(ctx) decision = _run_decision(ctx)
return {"decision": decision, "ctx": ctx} return {"decision": decision, "ctx": ctx}
@@ -752,6 +763,8 @@ async def _build_context(
location: Optional[str], location: Optional[str],
reassignment: bool, reassignment: bool,
create_if_new: bool, create_if_new: bool,
embedding: Optional[list] = None,
severity: Optional[str] = None,
) -> dict: ) -> dict:
now = reference_time or datetime.now(timezone.utc) now = reference_time or datetime.now(timezone.utc)
window = timedelta(hours=settings.correlation_window_hours) window = timedelta(hours=settings.correlation_window_hours)
@@ -777,11 +790,20 @@ async def _build_context(
all_active = _drop_capped(all_active, now) all_active = _drop_capped(all_active, now)
recent = [inc for inc in all_active if _within_window_of(inc, now, window)] recent = [inc for inc in all_active if _within_window_of(inc, now, window)]
call_embedding = call_doc.get("embedding") # embedding and severity come from the SCENE being correlated, not the call
# doc — server-26#80 / #95. intelligence.py writes only the primary scene's
# embedding and severity to calls/{id}, so reading them back here handed
# every non-primary scene the primary scene's semantic vector and severity
# rung: a scene about a different event scored against the wrong incident on
# the embedding path (:1166/:1205/:1533) and could inherit a minor/moderate/
# major severity it never had, clearing the creation gate on borrowed
# weight. Same failure and same fix as the #87 coords leak directly below —
# a scene that passes none has none, and is judged thin on its own signal.
call_embedding = embedding
call_units = units if units is not None else (call_doc.get("units") or []) call_units = units if units is not None else (call_doc.get("units") or [])
call_vehicles = vehicles if vehicles is not None else (call_doc.get("vehicles") or []) call_vehicles = vehicles if vehicles is not None else (call_doc.get("vehicles") or [])
call_cleared = cleared_units if cleared_units is not None else (call_doc.get("cleared_units") or []) call_cleared = cleared_units if cleared_units is not None else (call_doc.get("cleared_units") or [])
call_severity = call_doc.get("severity") or "routine" call_severity = severity or "routine"
# A string that is not a place is not a location anywhere downstream — not # A string that is not a place is not a location anywhere downstream — not
# in the fit tests, not in the thin-call test, not in the LLM prompt, and # in the fit tests, not in the thin-call test, not in the LLM prompt, and
# not on the incident. Its coordinates go with it: coords are geocoded # not on the incident. Its coordinates go with it: coords are geocoded
+10 -1
View File
@@ -24,7 +24,16 @@ from app.internal.incident_correlator import clean_location, location_is_unit
_PROMPT_TEMPLATE = """You are analyzing a P25 public safety radio recording. The audio was transcribed by Whisper through a digital radio vocoder, which introduces errors. Each numbered transmission is a separate PTT press from a different radio. _PROMPT_TEMPLATE = """You are analyzing a P25 public safety radio recording. The audio was transcribed by Whisper through a digital radio vocoder, which introduces errors. Each numbered transmission is a separate PTT press from a different radio.
SCENE DETECTION: SCENE DETECTION:
A busy dispatch channel sometimes captures back-to-back conversations about multiple concurrent incidents in a single recording. Detect whether this recording contains ONE scene (all transmissions relate to a single event) or MULTIPLE scenes (clearly distinct dispatch conversations with different units being assigned, different locations, different event types). Assign short status transmissions (10-4, en route, acknowledgements) with no clear scene context to the most recent scene before them in the list. A busy dispatch channel sometimes captures back-to-back conversations about multiple concurrent incidents in a single recording. Your default is ONE scene. Return MULTIPLE scenes ONLY when the recording clearly contains two or more SEPARATE EVENTS — different incidents at different places, with no shared units, no shared subject, and no conversational thread connecting them.
These do NOT make a new scene — keep them in the same scene:
- a different unit or speaker joining the same event
- a follow-up transmission about the same job (records check, case number, tow/mileage, a unit clearing, an ETA, a location correction)
- the same subject or location being discussed again minutes later
- an administrative or status exchange that follows an event on the same channel
If you are unsure whether two exchanges are one event or two, treat them as ONE.
Assign short status transmissions (10-4, en route, acknowledgements) with no clear scene context to the most recent scene before them in the list.
Always respond with the scenes array, even for a single scene. Always respond with the scenes array, even for a single scene.
@@ -90,6 +90,11 @@ async def _recorrelate_orphan(call: dict) -> bool:
return False return False
# All data needed for correlation was stored by the first-pass extraction. # All data needed for correlation was stored by the first-pass extraction.
# embedding/severity are no longer read from the call doc inside
# _build_context (server-26#80/#95) — the sweep re-links a whole call, not a
# scene, so it passes the call doc's stored (primary-scene) values here. It
# is link-only (create_if_new=False), so a borrowed severity cannot open a
# new incident off this path.
incident_id = await incident_correlator.correlate_call( incident_id = await incident_correlator.correlate_call(
call_id = call_id, call_id = call_id,
node_id = call.get("node_id", ""), node_id = call.get("node_id", ""),
@@ -101,6 +106,8 @@ async def _recorrelate_orphan(call: dict) -> bool:
location = call.get("location"), location = call.get("location"),
location_coords= call.get("location_coords"), location_coords= call.get("location_coords"),
cleared_units = call.get("cleared_units") or [], cleared_units = call.get("cleared_units") or [],
embedding = call.get("embedding"),
severity = call.get("severity"),
reference_time = started_at, # anchor window to when the call happened reference_time = started_at, # anchor window to when the call happened
create_if_new = False, # never create — link-only create_if_new = False, # never create — link-only
) )
+7
View File
@@ -114,6 +114,8 @@ async def _correlate_with_consensus(
vehicles: Optional[list] = None, vehicles: Optional[list] = None,
cleared_units: Optional[list] = None, cleared_units: Optional[list] = None,
reassignment: bool = False, reassignment: bool = False,
embedding: Optional[list] = None,
severity: Optional[str] = None,
) -> Optional[str]: ) -> Optional[str]:
""" """
Consensus correlator: runs the rules engine and the cheap LLM in sequence. Consensus correlator: runs the rules engine and the cheap LLM in sequence.
@@ -131,6 +133,7 @@ async def _correlate_with_consensus(
tags=tags, incident_type=incident_type, location=location, tags=tags, incident_type=incident_type, location=location,
location_coords=location_coords, units=units, vehicles=vehicles, location_coords=location_coords, units=units, vehicles=vehicles,
cleared_units=cleared_units, reassignment=reassignment, cleared_units=cleared_units, reassignment=reassignment,
embedding=embedding, severity=severity,
) )
ctx = preview["ctx"] ctx = preview["ctx"]
rules_decision = preview["decision"] rules_decision = preview["decision"]
@@ -221,6 +224,8 @@ async def _run_extraction_pipeline(
vehicles=scene.get("vehicles"), vehicles=scene.get("vehicles"),
cleared_units=scene.get("cleared_units"), cleared_units=scene.get("cleared_units"),
reassignment=is_reassignment, reassignment=is_reassignment,
embedding=scene.get("embedding"),
severity=scene.get("severity"),
) )
if incident_id and incident_id not in incident_ids: if incident_id and incident_id not in incident_ids:
incident_ids.append(incident_id) incident_ids.append(incident_id)
@@ -336,6 +341,8 @@ async def _run_intelligence_pipeline(
vehicles=scene.get("vehicles"), vehicles=scene.get("vehicles"),
cleared_units=scene.get("cleared_units"), cleared_units=scene.get("cleared_units"),
reassignment=is_reassignment, reassignment=is_reassignment,
embedding=scene.get("embedding"),
severity=scene.get("severity"),
) )
if incident_id and incident_id not in incident_ids: if incident_id and incident_id not in incident_ids:
incident_ids.append(incident_id) incident_ids.append(incident_id)
@@ -254,6 +254,56 @@ async def test_a_scene_with_no_location_does_not_inherit_the_call_docs_pin():
assert ctx["is_thin_call"] is True assert ctx["is_thin_call"] is True
@pytest.mark.asyncio
async def test_a_scene_does_not_inherit_the_call_docs_embedding_or_severity():
"""
server-26#80 / #95. Same shape as the #87 coords leak above:
intelligence.py writes only the PRIMARY scene's embedding and severity to
calls/{id}. A non-primary scene being correlated must be judged on its own
embedding (or none) and its own severity — not the call doc's — or a scene
about a different event scores against the wrong incident on the embedding
path and can inherit a minor/moderate/major rung it never had, clearing the
creation gate on borrowed weight.
"""
with patch("app.internal.incident_correlator.fstore") as mock_fstore:
mock_fstore.doc_get = AsyncMock(
return_value={"embedding": [0.1] * 1536, "severity": "major"}
)
mock_fstore.collection_list = AsyncMock(return_value=[])
ctx = await _build_context(
call_id="call-scene-2", units=None, vehicles=None, cleared_units=None,
location_coords=None, reference_time=NOW,
system_id="sys-1", talkgroup_id=383, talkgroup_name=DISPATCH_TG,
tags=[], incident_type="police", location=None,
reassignment=False, create_if_new=True,
embedding=None, severity=None,
)
assert ctx["call_embedding"] is None
assert ctx["call_severity"] == "routine"
assert ctx["is_thin_call"] is True
@pytest.mark.asyncio
async def test_a_scene_is_judged_on_its_own_embedding_and_severity():
"""The other half of #80/#95: the scene's own values are what land in ctx."""
scene_vec = [0.9] * 1536
with patch("app.internal.incident_correlator.fstore") as mock_fstore:
mock_fstore.doc_get = AsyncMock(
return_value={"embedding": [0.1] * 1536, "severity": "routine"}
)
mock_fstore.collection_list = AsyncMock(return_value=[])
ctx = await _build_context(
call_id="call-scene-2", units=None, vehicles=None, cleared_units=None,
location_coords=None, reference_time=NOW,
system_id="sys-1", talkgroup_id=383, talkgroup_name=DISPATCH_TG,
tags=[], incident_type="police", location=None,
reassignment=False, create_if_new=True,
embedding=scene_vec, severity="major",
)
assert ctx["call_embedding"] == scene_vec
assert ctx["call_severity"] == "major"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_a_bare_number_never_becomes_an_incident_location_or_title(): async def test_a_bare_number_never_becomes_an_incident_location_or_title():
inc = await _create(tags=["flames"], location="49", coords=None, inc = await _create(tags=["flames"], location="49", coords=None,