@@ -7,6 +7,15 @@ from google.cloud.firestore_v1.base_query import FieldFilter
|
||||
from app.config import settings
|
||||
from app.internal.logger import logger
|
||||
|
||||
# Re-exported so callers never need their own `firebase_admin.firestore` import
|
||||
# just to delete a field. server-26#96/#114 review: `doc_set(..., merge=True)`
|
||||
# merges nested maps by key but can never REMOVE one — writing `{"scenes": {}}`
|
||||
# to clear a map is a no-op, not a delete. Use `doc_update(coll, id, {"field":
|
||||
# fstore.DELETE_FIELD})` (or doc_set + merge, DELETE_FIELD works under both)
|
||||
# whenever a re-extraction/reprocess path needs a stale nested field gone
|
||||
# rather than merged over.
|
||||
DELETE_FIELD = fs.DELETE_FIELD
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# In-memory TTL cache for rarely-changing documents (systems, nodes config)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -709,6 +709,7 @@ async def correlate_call(
|
||||
embedding: Optional[list] = None,
|
||||
severity: Optional[str] = None,
|
||||
transcript: Optional[str] = None,
|
||||
scene_index: int = 0,
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
Link call_id to an existing incident or create a new one.
|
||||
@@ -718,6 +719,11 @@ async def correlate_call(
|
||||
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.
|
||||
|
||||
``scene_index`` (server-26#96) identifies which scene of the call this
|
||||
decision belongs to for the per-scene ``scenes`` map written by
|
||||
_apply_and_log. Defaults to 0 — correct for every caller here, since this
|
||||
entry point always re-correlates a call as a single unit, not a scene loop.
|
||||
"""
|
||||
ctx = await _build_context(
|
||||
call_id=call_id, units=units, vehicles=vehicles, cleared_units=cleared_units,
|
||||
@@ -726,6 +732,7 @@ async def correlate_call(
|
||||
tags=tags, incident_type=incident_type, location=location,
|
||||
reassignment=reassignment, create_if_new=create_if_new,
|
||||
embedding=embedding, severity=severity, transcript=transcript,
|
||||
scene_index=scene_index,
|
||||
)
|
||||
decision = _run_decision(ctx)
|
||||
return await _apply_and_log(decision, ctx)
|
||||
@@ -750,6 +757,7 @@ async def preview_correlation(
|
||||
embedding: Optional[list] = None,
|
||||
severity: Optional[str] = None,
|
||||
transcript: Optional[str] = None,
|
||||
scene_index: int = 0,
|
||||
) -> dict:
|
||||
"""
|
||||
Run the rules engine and return the decision WITHOUT committing to Firestore.
|
||||
@@ -763,6 +771,15 @@ async def preview_correlation(
|
||||
matched_incident the candidate incident doc (action == "link")
|
||||
incident_type resolved type after tag inference (action == "new")
|
||||
corr_debug fields to persist on the call doc
|
||||
|
||||
``scene_index`` (server-26#96) — which scene of the call (upload.py's
|
||||
``for scene_index, scene in enumerate(scenes):`` loop) this call is. It
|
||||
rides through ctx to _apply_and_log, which uses it as the key under the
|
||||
call doc's ``scenes`` map so each scene's corr_debug/transcript lands in
|
||||
its own map entry instead of colliding on the shared flat fields. Defaults
|
||||
to 0 for callers with no scene concept (a single-scene call, or the
|
||||
no-scenes-extracted correlation attempt) — equivalent to today's
|
||||
behaviour for those calls.
|
||||
"""
|
||||
ctx = await _build_context(
|
||||
call_id=call_id, units=units, vehicles=vehicles, cleared_units=cleared_units,
|
||||
@@ -771,6 +788,7 @@ async def preview_correlation(
|
||||
tags=tags, incident_type=incident_type, location=location,
|
||||
reassignment=reassignment, create_if_new=create_if_new,
|
||||
embedding=embedding, severity=severity, transcript=transcript,
|
||||
scene_index=scene_index,
|
||||
)
|
||||
decision = _run_decision(ctx)
|
||||
return {"decision": decision, "ctx": ctx}
|
||||
@@ -806,6 +824,7 @@ async def _build_context(
|
||||
embedding: Optional[list] = None,
|
||||
severity: Optional[str] = None,
|
||||
transcript: Optional[str] = None,
|
||||
scene_index: int = 0,
|
||||
) -> dict:
|
||||
now = reference_time or datetime.now(timezone.utc)
|
||||
window = timedelta(hours=settings.correlation_window_hours)
|
||||
@@ -883,6 +902,9 @@ async def _build_context(
|
||||
"incident_type": incident_type, "location": location,
|
||||
"location_coords": location_coords, "reassignment": reassignment,
|
||||
"create_if_new": create_if_new,
|
||||
# server-26#96 — which scene of the call this decision is for. Carried
|
||||
# through so _apply_and_log can key the per-scene write correctly.
|
||||
"scene_index": scene_index,
|
||||
}
|
||||
|
||||
|
||||
@@ -1419,12 +1441,51 @@ def _run_decision(ctx: dict) -> dict:
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]:
|
||||
"""Commit a rules decision and persist the corr_debug fields to the call doc."""
|
||||
"""
|
||||
Commit a rules decision and persist the corr_debug fields to the call doc.
|
||||
|
||||
server-26#96: every scene of a multi-scene call reaches this function
|
||||
independently (upload.py's ``for scene_index, scene in enumerate(scenes):``
|
||||
loop → _correlate_with_consensus → preview_correlation/apply_correlation),
|
||||
and every scene writes to the SAME call doc. Writing corr_debug only at
|
||||
the flat top level meant scene 2's write silently clobbered scene 1's
|
||||
corr_path/corr_consensus/etc — the call doc ended up describing a splice
|
||||
of decisions, not any one of them.
|
||||
|
||||
Fix: write the flat fields exactly as before (kept for any reader that
|
||||
doesn't know about `scenes` yet — last-scene-wins, same as pre-#96
|
||||
behaviour, a safe backward-compatible default) AND additionally nest the
|
||||
same corr_debug — plus this scene's own transcript and the incident_id it
|
||||
resolved to — under scenes.<scene_index>. Firestore's
|
||||
`DocumentReference.set(data, merge=True)` recursively merges nested map
|
||||
fields by key. Verified against `internal/firestore.py`'s `doc_set`
|
||||
wrapper (a straight `ref.set(data, merge=merge)` pass-through — no
|
||||
`update()`, no read-modify-write, nothing that would change this) and
|
||||
against the documented set-with-merge semantics; NOT exercised against a
|
||||
live Firestore instance (no SDK available in the sandboxes this landed
|
||||
from — review flagged this distinction explicitly). A write of
|
||||
{"scenes": {"1": {...}}} merges into an existing
|
||||
{"scenes": {"0": {...}}} to produce {"scenes": {"0": {...}, "1": {...}}}
|
||||
rather than replacing the whole `scenes` map, so scene 0's and scene 1's
|
||||
entries land side by side instead of colliding like the flat fields do.
|
||||
`scene_index` defaults to 0 (see preview_correlation/correlate_call), so a
|
||||
plain single-scene call still gets a `scenes` map — just with one entry,
|
||||
equivalent to reading the flat fields today.
|
||||
"""
|
||||
incident_id = await _apply_decision(decision, ctx)
|
||||
corr_debug = decision.get("corr_debug") or {}
|
||||
if corr_debug:
|
||||
scene_index = ctx.get("scene_index", 0)
|
||||
updates = dict(corr_debug)
|
||||
updates["scenes"] = {
|
||||
str(scene_index): {
|
||||
"transcript": ctx.get("scene_transcript"),
|
||||
"incident_id": incident_id,
|
||||
"corr_debug": corr_debug,
|
||||
}
|
||||
}
|
||||
try:
|
||||
await fstore.doc_set("calls", ctx["call_id"], corr_debug)
|
||||
await fstore.doc_set("calls", ctx["call_id"], updates)
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not write corr_debug for call {ctx['call_id']}: {e}")
|
||||
return incident_id
|
||||
|
||||
@@ -15,6 +15,39 @@ from app.internal import firestore as fstore
|
||||
from app.config import settings
|
||||
|
||||
|
||||
def _scene_sort_key(scene_index: str):
|
||||
"""Numeric-first sort so a >=10-scene call's entries still read in order."""
|
||||
return (0, int(scene_index)) if scene_index.isdigit() else (1, scene_index)
|
||||
|
||||
|
||||
def _scene_text_for_incident(doc: dict, incident_id: str) -> Optional[str]:
|
||||
"""
|
||||
The text of `doc` (a call doc) that actually belongs to `incident_id`.
|
||||
|
||||
server-26#96 records, per scene, which incident_id that scene's
|
||||
correlation decision resolved to (incident_correlator._apply_and_log's
|
||||
`scenes.<index>.incident_id`). Use that to pick only the scene(s) of this
|
||||
call that are genuinely part of this incident, joining more than one if
|
||||
several scenes happened to link into the same incident.
|
||||
|
||||
Falls back to transcript_corrected-or-transcript when the call doc has no
|
||||
`scenes` field (predates server-26#96) or — defensively — when it has one
|
||||
but nothing in it names this incident_id (should not happen for a call_id
|
||||
that's actually in this incident's call_ids, but silently dropping a
|
||||
call's contribution to its own summary would be a worse failure mode than
|
||||
falling back to the whole-call text).
|
||||
"""
|
||||
scenes = doc.get("scenes") or {}
|
||||
matched = [
|
||||
scene.get("transcript")
|
||||
for _, scene in sorted(scenes.items(), key=lambda kv: _scene_sort_key(kv[0]))
|
||||
if scene.get("incident_id") == incident_id and scene.get("transcript")
|
||||
]
|
||||
if matched:
|
||||
return "\n".join(matched)
|
||||
return doc.get("transcript_corrected") or doc.get("transcript")
|
||||
|
||||
|
||||
async def summarizer_loop() -> None:
|
||||
from app.internal.feature_flags import get_flags
|
||||
interval = settings.summary_interval_minutes * 60
|
||||
@@ -63,12 +96,30 @@ async def _summarize_incident(inc: dict) -> None:
|
||||
if not call_ids:
|
||||
return
|
||||
|
||||
# Fetch transcripts for all calls in this incident
|
||||
# Fetch transcripts for all calls in this incident.
|
||||
#
|
||||
# server-26#114: a call links into an incident one SCENE at a time (see
|
||||
# incident_correlator._apply_decision / server-26#96's `scenes` map on the
|
||||
# call doc), and the same call_id can appear in more than one incident's
|
||||
# call_ids — once per scene, each scene possibly landing in a different
|
||||
# incident. Reading doc["transcript"] (the whole call, raw) meant an
|
||||
# incident's summary was built partly on text from a DIFFERENT scene of
|
||||
# that call that this incident has nothing to do with, and ignored
|
||||
# transcript_corrected entirely.
|
||||
#
|
||||
# _scene_text_for_incident reads the specific scene(s) whose corr_debug
|
||||
# recorded a link into THIS incident_id. For a call doc that predates
|
||||
# this fix (no `scenes` field) it falls back to
|
||||
# transcript_corrected-or-transcript — the one-liner half of #114, worth
|
||||
# doing even for old-schema docs since it stops raw-transcript summaries.
|
||||
transcripts: list[str] = []
|
||||
for cid in call_ids:
|
||||
doc = await fstore.doc_get("calls", cid)
|
||||
if doc and doc.get("transcript"):
|
||||
transcripts.append(doc["transcript"])
|
||||
if not doc:
|
||||
continue
|
||||
text = _scene_text_for_incident(doc, incident_id)
|
||||
if text:
|
||||
transcripts.append(text)
|
||||
|
||||
if not transcripts:
|
||||
# No transcripts yet — clear stale flag and wait for next pass
|
||||
|
||||
@@ -97,8 +97,49 @@ async def debug_correlation(
|
||||
def _strip(doc: dict) -> dict:
|
||||
return {k: v for k, v in doc.items() if k != "embedding"}
|
||||
|
||||
def _call_summary(call: dict) -> dict:
|
||||
def _scene_summary(scene_index: str, scene: dict) -> dict:
|
||||
"""
|
||||
One scene's own correlation record, from the call doc's `scenes` map
|
||||
(server-26#96). Same corr_* field names as _call_summary's flat
|
||||
fields below, deliberately — a scene entry and a scene-less call
|
||||
summary are interchangeable data points to the tally functions.
|
||||
"""
|
||||
corr_debug = scene.get("corr_debug") or {}
|
||||
return {
|
||||
"scene_index": scene_index,
|
||||
"transcript": scene.get("transcript"),
|
||||
"incident_id": scene.get("incident_id"),
|
||||
"corr_path": corr_debug.get("corr_path"),
|
||||
"corr_incident_idle_min": corr_debug.get("corr_incident_idle_min"),
|
||||
"corr_distance_km": corr_debug.get("corr_distance_km"),
|
||||
"corr_score": corr_debug.get("corr_score"),
|
||||
"corr_candidates": corr_debug.get("corr_candidates"),
|
||||
"corr_shared_units": corr_debug.get("corr_shared_units"),
|
||||
"corr_fit_signal": corr_debug.get("corr_fit_signal"),
|
||||
"corr_matched_units": corr_debug.get("corr_matched_units"),
|
||||
"corr_consensus": corr_debug.get("corr_consensus"),
|
||||
"corr_llm_reasoning": corr_debug.get("corr_llm_reasoning"),
|
||||
"corr_llm_action": corr_debug.get("corr_llm_action"),
|
||||
"corr_rules_action": corr_debug.get("corr_rules_action"),
|
||||
"corr_gate_veto": corr_debug.get("corr_gate_veto"),
|
||||
}
|
||||
|
||||
def _call_summary(call: dict) -> dict:
|
||||
# server-26#96 — per-scene records, keyed by scene index as written by
|
||||
# incident_correlator._apply_and_log. Present only on calls that went
|
||||
# through correlation after this fix landed; absent (None) on older
|
||||
# call docs, which the tally below falls back for. Sorted numerically
|
||||
# so a >=10-scene call still reads in scene order.
|
||||
scenes_map = call.get("scenes") or {}
|
||||
scenes = [
|
||||
_scene_summary(idx, s)
|
||||
for idx, s in sorted(
|
||||
scenes_map.items(),
|
||||
key=lambda kv: (0, int(kv[0])) if kv[0].isdigit() else (1, kv[0]),
|
||||
)
|
||||
] or None
|
||||
return {
|
||||
"scenes": scenes,
|
||||
"call_id": call.get("call_id"),
|
||||
"started_at": call.get("started_at"),
|
||||
"ended_at": call.get("ended_at"),
|
||||
@@ -271,6 +312,21 @@ async def debug_correlation(
|
||||
linked = [c for inc in incident_records for c in (inc.get("calls_detail") or [])]
|
||||
call_counts = [len(inc.get("call_ids") or []) for inc in incident_records]
|
||||
|
||||
def _tally_entries(call_summary: dict) -> list:
|
||||
"""
|
||||
server-26#96 — the unit correlation actually decided over is the
|
||||
scene, not the call. A call summary carrying a `scenes` list (every
|
||||
call correlated after this fix) contributes one entry per scene, each
|
||||
with its own corr_path/corr_consensus/etc, instead of the single flat
|
||||
record that used to blend every scene's last write together. A call
|
||||
summary with no `scenes` (a call doc from before this fix) falls back
|
||||
to contributing itself as one entry — identical to pre-#96 behaviour.
|
||||
"""
|
||||
scenes = call_summary.get("scenes")
|
||||
return scenes if scenes else [call_summary]
|
||||
|
||||
scene_entries = [entry for c in linked for entry in _tally_entries(c)]
|
||||
|
||||
def _span_minutes(inc: dict) -> float:
|
||||
stamps = sorted(
|
||||
s for s in ((c.get("started_at") or "") for c in (inc.get("calls_detail") or [])) if s
|
||||
@@ -302,13 +358,20 @@ async def debug_correlation(
|
||||
"ai_systems_only": ai_systems_only,
|
||||
"ai_enabled_system_ids": sorted(ai_systems),
|
||||
"linked_call_count": len(linked),
|
||||
"corr_path": _tally(c.get("corr_path") for c in linked),
|
||||
"corr_fit_signal": _tally(c.get("corr_fit_signal") for c in linked),
|
||||
"corr_consensus": _tally(c.get("corr_consensus") for c in linked),
|
||||
"corr_llm_action": _tally(c.get("corr_llm_action") for c in linked),
|
||||
# server-26#96 — tallied over scene_entries (one entry per scene of a
|
||||
# multi-scene call, from its `scenes` map; one entry per call when it
|
||||
# has none) rather than over `linked` directly, so a 2-scene call
|
||||
# with two different corr_path values counts as two data points
|
||||
# instead of one blended flat record. scene_decision_count makes that
|
||||
# distinction visible next to linked_call_count.
|
||||
"scene_decision_count": len(scene_entries),
|
||||
"corr_path": _tally(e.get("corr_path") for e in scene_entries),
|
||||
"corr_fit_signal": _tally(e.get("corr_fit_signal") for e in scene_entries),
|
||||
"corr_consensus": _tally(e.get("corr_consensus") for e in scene_entries),
|
||||
"corr_llm_action": _tally(e.get("corr_llm_action") for e in scene_entries),
|
||||
# server-26#115 — this IS the number the escape-hatch fix exists to
|
||||
# produce: why each llm=orphan/rules=new call escaped the gate.
|
||||
"corr_gate_veto": _tally(c.get("corr_gate_veto") for c in linked),
|
||||
"corr_gate_veto": _tally(e.get("corr_gate_veto") for e in scene_entries),
|
||||
# server-26#127 — shadow-mode chatter classifier. The target
|
||||
# population is non-events, which land as orphans or single-call
|
||||
# incidents, NOT as a slice of every linked call -- tally `orphans`
|
||||
|
||||
@@ -260,6 +260,15 @@ async def patch_transcript(
|
||||
"vehicles": [],
|
||||
"embedding": None,
|
||||
})
|
||||
# server-26#96/#114 review: doc_set(merge=True) can only ADD/overwrite keys
|
||||
# in a nested map, never remove one, so the fields above get cleared but a
|
||||
# prior `scenes` map would survive re-extraction forever. A call corrected
|
||||
# from 3 scenes down to 1 would keep scenes.1/scenes.2 with pre-correction
|
||||
# transcripts and incident_ids -- corrupting the exact per-scene tally #96
|
||||
# exists to make trustworthy, and re-feeding stale text into #114's
|
||||
# summarizer fix if a stale scene's incident_id still names a real
|
||||
# incident. Must be a real delete, not a merge over an empty map.
|
||||
await fstore.doc_update("calls", call_id, {"scenes": fstore.DELETE_FIELD})
|
||||
|
||||
# Unlink from ALL current incidents so re-correlation starts clean.
|
||||
# Handles both old single incident_id and new incident_ids list.
|
||||
|
||||
@@ -240,6 +240,7 @@ async def _correlate_with_consensus(
|
||||
embedding: Optional[list] = None,
|
||||
severity: Optional[str] = None,
|
||||
transcript: Optional[str] = None,
|
||||
scene_index: int = 0,
|
||||
) -> Optional[str]:
|
||||
"""
|
||||
Consensus correlator: runs the rules engine and the cheap LLM in sequence.
|
||||
@@ -248,6 +249,11 @@ async def _correlate_with_consensus(
|
||||
|
||||
Falls back to rules-only when GEMINI_API_KEY is absent, the call is
|
||||
content-free (thin), or any LLM call fails.
|
||||
|
||||
``scene_index`` (server-26#96) — which scene of the call this is, from the
|
||||
caller's ``enumerate(scenes)`` loop. Threaded through so the call doc's
|
||||
per-scene ``scenes`` map records this scene's own corr_debug/transcript
|
||||
instead of colliding with every other scene's write on the flat fields.
|
||||
"""
|
||||
from app.internal import incident_correlator, llm_correlator
|
||||
|
||||
@@ -258,6 +264,7 @@ async def _correlate_with_consensus(
|
||||
location_coords=location_coords, units=units, vehicles=vehicles,
|
||||
cleared_units=cleared_units, reassignment=reassignment,
|
||||
embedding=embedding, severity=severity, transcript=transcript,
|
||||
scene_index=scene_index,
|
||||
)
|
||||
ctx = preview["ctx"]
|
||||
rules_decision = preview["decision"]
|
||||
@@ -365,7 +372,10 @@ async def _run_extraction_pipeline(
|
||||
)
|
||||
|
||||
# Step 3: Correlate each scene to an incident independently.
|
||||
for scene in scenes:
|
||||
# server-26#96: scene_index is threaded through so each scene's
|
||||
# corr_debug/transcript lands in its own entry of the call doc's
|
||||
# `scenes` map instead of clobbering every other scene's write.
|
||||
for scene_index, scene in enumerate(scenes):
|
||||
all_tags.extend(scene["tags"])
|
||||
# When dispatch is pulling a unit to a NEW call (reassignment), suppress unit
|
||||
# overlap so the new scene doesn't chain into the unit's previous incident.
|
||||
@@ -388,6 +398,7 @@ async def _run_extraction_pipeline(
|
||||
embedding=scene.get("embedding"),
|
||||
severity=scene.get("severity"),
|
||||
transcript=scene.get("transcript"),
|
||||
scene_index=scene_index,
|
||||
)
|
||||
if incident_id and incident_id not in incident_ids:
|
||||
incident_ids.append(incident_id)
|
||||
@@ -485,7 +496,10 @@ async def _run_intelligence_pipeline(
|
||||
incident_ids: list[str] = []
|
||||
all_tags: list[str] = []
|
||||
if _flag("correlation_enabled"):
|
||||
for scene in scenes:
|
||||
# server-26#96: scene_index is threaded through so each scene's
|
||||
# corr_debug/transcript lands in its own entry of the call doc's
|
||||
# `scenes` map instead of clobbering every other scene's write.
|
||||
for scene_index, scene in enumerate(scenes):
|
||||
all_tags.extend(scene["tags"])
|
||||
is_reassignment = bool(scene.get("reassignment"))
|
||||
corr_units = [] if is_reassignment else scene.get("units")
|
||||
@@ -506,6 +520,7 @@ async def _run_intelligence_pipeline(
|
||||
embedding=scene.get("embedding"),
|
||||
severity=scene.get("severity"),
|
||||
transcript=scene.get("transcript"),
|
||||
scene_index=scene_index,
|
||||
)
|
||||
if incident_id and incident_id not in incident_ids:
|
||||
incident_ids.append(incident_id)
|
||||
|
||||
@@ -34,6 +34,10 @@ except ModuleNotFoundError:
|
||||
# into dicts that tests compare against, and a MagicMock compares unequal
|
||||
# to itself across attribute accesses.
|
||||
_fs.SERVER_TIMESTAMP = "__SERVER_TIMESTAMP__"
|
||||
# Same reasoning as SERVER_TIMESTAMP above: a distinct sentinel, not a
|
||||
# MagicMock, so `fstore.DELETE_FIELD is fs.DELETE_FIELD` and dict/`is`
|
||||
# comparisons against it in tests (server-26#96/#114, PR #132) behave.
|
||||
_fs.DELETE_FIELD = "__DELETE_FIELD__"
|
||||
|
||||
_auth = ModuleType("firebase_admin.auth")
|
||||
_auth.verify_id_token = MagicMock()
|
||||
|
||||
@@ -89,3 +89,75 @@ async def test_debug_correlation_llm_fields_absent_when_rules_only():
|
||||
assert detail["corr_consensus"] == "rules_only"
|
||||
assert detail["corr_llm_reasoning"] is None
|
||||
assert detail["corr_llm_action"] is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# server-26#96 — the summary tally must count per-scene decisions, not the
|
||||
# one blended flat record a multi-scene call used to leave behind.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_debug_correlation_exposes_scenes_and_tallies_each_as_its_own_datapoint():
|
||||
"""A 2-scene call: the flat fields still show last-scene-wins (unchanged
|
||||
behaviour for old readers), but the summary tally must see two distinct
|
||||
corr_path/corr_consensus data points, not one blend."""
|
||||
call = {
|
||||
"call_id": "call-1",
|
||||
# Flat fields — last scene wins, kept as-is for backward compat.
|
||||
"corr_path": "slow",
|
||||
"corr_consensus": "tiebreak",
|
||||
"scenes": {
|
||||
"0": {
|
||||
"transcript": "scene zero",
|
||||
"incident_id": "inc-1",
|
||||
"corr_debug": {"corr_path": "new", "corr_consensus": "agreed"},
|
||||
},
|
||||
"1": {
|
||||
"transcript": "scene one",
|
||||
"incident_id": "inc-1",
|
||||
"corr_debug": {"corr_path": "slow", "corr_consensus": "tiebreak"},
|
||||
},
|
||||
},
|
||||
}
|
||||
result = await _run([_incident(["call-1"])], {"call-1": call})
|
||||
|
||||
detail = result["incidents"][0]["calls_detail"][0]
|
||||
assert detail["corr_path"] == "slow" # flat field: last scene wins
|
||||
assert len(detail["scenes"]) == 2
|
||||
assert detail["scenes"][0]["corr_path"] == "new"
|
||||
assert detail["scenes"][1]["corr_path"] == "slow"
|
||||
|
||||
summary = result["summary"]
|
||||
assert summary["linked_call_count"] == 1 # still one CALL
|
||||
assert summary["scene_decision_count"] == 2 # but two DECISIONS
|
||||
assert summary["corr_path"] == {"new": 1, "slow": 1}
|
||||
assert summary["corr_consensus"] == {"agreed": 1, "tiebreak": 1}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_debug_correlation_tally_falls_back_for_single_scene_call():
|
||||
"""A plain single-scene call has no `scenes` field at all — the tally
|
||||
must fall back to its flat fields as one data point, same as pre-#96."""
|
||||
call = {"call_id": "call-2", "corr_path": "fast/single", "corr_consensus": "rules_only"}
|
||||
result = await _run([_incident(["call-2"])], {"call-2": call})
|
||||
|
||||
detail = result["incidents"][0]["calls_detail"][0]
|
||||
assert detail["scenes"] is None
|
||||
|
||||
summary = result["summary"]
|
||||
assert summary["linked_call_count"] == 1
|
||||
assert summary["scene_decision_count"] == 1
|
||||
assert summary["corr_path"] == {"fast/single": 1}
|
||||
assert summary["corr_consensus"] == {"rules_only": 1}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_debug_correlation_tally_handles_old_schema_call_with_no_scenes_field():
|
||||
"""A call doc written before server-26#96 has never heard of `scenes` —
|
||||
must behave identically to the single-scene case, not error."""
|
||||
old_call = {"call_id": "call-3", "corr_path": "cross-tg", "corr_consensus": "agreed"}
|
||||
result = await _run([_incident(["call-3"])], {"call-3": old_call})
|
||||
|
||||
summary = result["summary"]
|
||||
assert summary["scene_decision_count"] == 1
|
||||
assert summary["corr_path"] == {"cross-tg": 1}
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
"""
|
||||
server-26#96 — every scene of a multi-scene call writes corr_debug onto the
|
||||
SAME call doc via incident_correlator._apply_and_log, last-scene-wins. The
|
||||
fix additionally nests each scene's corr_debug/transcript/incident_id under
|
||||
scenes.<scene_index> on the call doc, keyed so Firestore's
|
||||
`set(merge=True)` (a recursive merge of nested map fields — this is the
|
||||
behaviour these tests assume and pin) lands each scene in its own map entry
|
||||
instead of colliding.
|
||||
|
||||
Firestore itself isn't available in this sandbox (see tests/conftest.py), so
|
||||
`_fake_doc_set` below implements that documented recursive-merge semantics by
|
||||
hand and is used as the fstore stand-in — these tests both exercise
|
||||
_apply_and_log's write shape AND pin the merge behaviour it depends on.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import patch
|
||||
|
||||
from app.internal import incident_correlator
|
||||
|
||||
|
||||
def _merge(dst: dict, src: dict) -> None:
|
||||
"""Firestore DocumentReference.set(data, merge=True) semantics: nested
|
||||
map fields are merged recursively by key, not replaced wholesale."""
|
||||
for k, v in src.items():
|
||||
if isinstance(v, dict) and isinstance(dst.get(k), dict):
|
||||
_merge(dst[k], v)
|
||||
else:
|
||||
dst[k] = v
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiscene_call_lands_each_scene_distinctly_and_flat_fields_last_write_wins():
|
||||
docs: dict[tuple, dict] = {}
|
||||
|
||||
async def fake_doc_set(collection, doc_id, data, merge=True):
|
||||
docs.setdefault((collection, doc_id), {})
|
||||
_merge(docs[(collection, doc_id)], data)
|
||||
|
||||
decision0 = {
|
||||
"action": "orphan", "matched_incident": None, "incident_type": None,
|
||||
"corr_debug": {"corr_path": "new", "corr_consensus": "agreed"},
|
||||
}
|
||||
ctx0 = {"call_id": "call-1", "scene_index": 0, "scene_transcript": "scene zero text"}
|
||||
|
||||
decision1 = {
|
||||
"action": "orphan", "matched_incident": None, "incident_type": None,
|
||||
"corr_debug": {"corr_path": "slow", "corr_consensus": "tiebreak"},
|
||||
}
|
||||
ctx1 = {"call_id": "call-1", "scene_index": 1, "scene_transcript": "scene one text"}
|
||||
|
||||
with patch.object(incident_correlator, "fstore") as mock_fstore:
|
||||
mock_fstore.doc_set = fake_doc_set
|
||||
await incident_correlator._apply_and_log(decision0, ctx0)
|
||||
await incident_correlator._apply_and_log(decision1, ctx1)
|
||||
|
||||
doc = docs[("calls", "call-1")]
|
||||
|
||||
# Flat top-level fields: unchanged behaviour, last scene's write wins —
|
||||
# the safe backward-compatible default for any reader that doesn't yet
|
||||
# know about `scenes`.
|
||||
assert doc["corr_path"] == "slow"
|
||||
assert doc["corr_consensus"] == "tiebreak"
|
||||
|
||||
# New `scenes` map: both scenes present, distinct, uncorrupted by the
|
||||
# second write.
|
||||
assert set(doc["scenes"].keys()) == {"0", "1"}
|
||||
assert doc["scenes"]["0"]["corr_debug"]["corr_path"] == "new"
|
||||
assert doc["scenes"]["0"]["corr_debug"]["corr_consensus"] == "agreed"
|
||||
assert doc["scenes"]["0"]["transcript"] == "scene zero text"
|
||||
assert doc["scenes"]["1"]["corr_debug"]["corr_path"] == "slow"
|
||||
assert doc["scenes"]["1"]["corr_debug"]["corr_consensus"] == "tiebreak"
|
||||
assert doc["scenes"]["1"]["transcript"] == "scene one text"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_scene_entry_records_which_incident_it_resolved_to():
|
||||
"""summarizer.py (#114) needs this to pick the right scene per incident."""
|
||||
docs: dict[tuple, dict] = {}
|
||||
|
||||
async def fake_doc_set(collection, doc_id, data, merge=True):
|
||||
docs.setdefault((collection, doc_id), {})
|
||||
_merge(docs[(collection, doc_id)], data)
|
||||
|
||||
with patch.object(incident_correlator, "fstore") as mock_fstore, \
|
||||
patch.object(incident_correlator, "_apply_decision", return_value="inc-42"):
|
||||
mock_fstore.doc_set = fake_doc_set
|
||||
decision = {
|
||||
"action": "new", "matched_incident": None, "incident_type": "fire",
|
||||
"corr_debug": {"corr_path": "new"},
|
||||
}
|
||||
ctx = {"call_id": "call-2", "scene_index": 0, "scene_transcript": "structure fire"}
|
||||
incident_id = await incident_correlator._apply_and_log(decision, ctx)
|
||||
|
||||
assert incident_id == "inc-42"
|
||||
assert docs[("calls", "call-2")]["scenes"]["0"]["incident_id"] == "inc-42"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_single_scene_call_still_gets_a_scenes_map_equivalent_to_flat_fields():
|
||||
"""scene_index defaults to 0 for every caller with no scene concept, so a
|
||||
plain single-scene call is one entry in `scenes` — equivalent to reading
|
||||
the flat fields, not a behaviour change for that population."""
|
||||
docs: dict[tuple, dict] = {}
|
||||
|
||||
async def fake_doc_set(collection, doc_id, data, merge=True):
|
||||
docs.setdefault((collection, doc_id), {})
|
||||
_merge(docs[(collection, doc_id)], data)
|
||||
|
||||
decision = {
|
||||
"action": "orphan", "matched_incident": None, "incident_type": None,
|
||||
"corr_debug": {"corr_path": "fast/thin", "corr_consensus": "rules_only"},
|
||||
}
|
||||
ctx = {"call_id": "call-3", "scene_transcript": "10-4"} # no scene_index key at all
|
||||
|
||||
with patch.object(incident_correlator, "fstore") as mock_fstore:
|
||||
mock_fstore.doc_set = fake_doc_set
|
||||
await incident_correlator._apply_and_log(decision, ctx)
|
||||
|
||||
doc = docs[("calls", "call-3")]
|
||||
assert doc["corr_path"] == "fast/thin"
|
||||
assert doc["scenes"] == {
|
||||
"0": {
|
||||
"transcript": "10-4",
|
||||
"incident_id": None,
|
||||
"corr_debug": {"corr_path": "fast/thin", "corr_consensus": "rules_only"},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_empty_corr_debug_writes_nothing_same_as_before():
|
||||
"""Preserve the pre-#96 short-circuit: no corr_debug means no write at
|
||||
all, flat or nested."""
|
||||
with patch.object(incident_correlator, "fstore") as mock_fstore, \
|
||||
patch.object(incident_correlator, "_apply_decision", return_value=None):
|
||||
mock_fstore.doc_set = None # would raise TypeError if ever called
|
||||
decision = {"action": "orphan", "matched_incident": None, "incident_type": None, "corr_debug": {}}
|
||||
ctx = {"call_id": "call-4", "scene_index": 0, "scene_transcript": "x"}
|
||||
result = await incident_correlator._apply_and_log(decision, ctx)
|
||||
|
||||
assert result is None
|
||||
@@ -0,0 +1,89 @@
|
||||
"""
|
||||
server-26#96/#114 review (PR #132): `PATCH /calls/{id}/transcript` clears
|
||||
stale intelligence fields before re-extraction runs, but `doc_set(...,
|
||||
merge=True)` can only add/overwrite keys in a nested map, never remove one.
|
||||
A call corrected from 3 scenes down to 1 would keep `scenes.1`/`scenes.2`
|
||||
with pre-correction transcripts and incident_ids forever -- corrupting the
|
||||
per-scene tally #96 exists to make trustworthy, and re-feeding stale text
|
||||
into #114's summarizer fix if a stale scene's incident_id still names a real
|
||||
incident. The fix deletes the field with `fstore.DELETE_FIELD` instead of
|
||||
merging over it with an empty map (which is a no-op).
|
||||
"""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi import BackgroundTasks
|
||||
|
||||
from app.internal import firestore as fstore
|
||||
from app.routers.calls import TranscriptUpdate, patch_transcript
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transcript_correction_deletes_the_scenes_field_not_merges_over_it():
|
||||
call = {
|
||||
"call_id": "call-1",
|
||||
"system_id": "sys-1",
|
||||
"node_id": "node-1",
|
||||
"transcript": "old raw text",
|
||||
# Simulates a prior 3-scene call, per #96's schema.
|
||||
"scenes": {
|
||||
"0": {"transcript": "scene zero", "incident_id": "inc-a", "corr_debug": {}},
|
||||
"1": {"transcript": "scene one", "incident_id": "inc-b", "corr_debug": {}},
|
||||
},
|
||||
}
|
||||
|
||||
doc_set_calls: list[tuple] = []
|
||||
doc_update_calls: list[tuple] = []
|
||||
|
||||
async def fake_doc_get(collection, doc_id):
|
||||
if collection == "calls" and doc_id == "call-1":
|
||||
return call
|
||||
return None
|
||||
|
||||
async def fake_doc_set(collection, doc_id, data, merge=True):
|
||||
doc_set_calls.append((collection, doc_id, data))
|
||||
|
||||
async def fake_doc_update(collection, doc_id, data):
|
||||
doc_update_calls.append((collection, doc_id, data))
|
||||
|
||||
fake_flags = (None, lambda name: name == "correlation_enabled")
|
||||
|
||||
with patch("app.routers.calls.fstore.doc_get", new=fake_doc_get), \
|
||||
patch("app.routers.calls.fstore.doc_set", new=fake_doc_set), \
|
||||
patch("app.routers.calls.fstore.doc_update", new=fake_doc_update), \
|
||||
patch("app.internal.feature_flags.resolve_flags", new=AsyncMock(return_value=fake_flags)):
|
||||
result = await patch_transcript(
|
||||
call_id="call-1",
|
||||
body=TranscriptUpdate(transcript="corrected text"),
|
||||
background_tasks=BackgroundTasks(),
|
||||
_={},
|
||||
)
|
||||
|
||||
assert result == {"ok": True, "call_id": "call-1"}
|
||||
|
||||
# The stale scenes map must be DELETED, not merged over with {} (a no-op
|
||||
# under Firestore's set(merge=True) semantics) and not left untouched by
|
||||
# a doc_set call that never mentions it.
|
||||
scenes_deletions = [
|
||||
(coll, doc_id, data) for (coll, doc_id, data) in doc_update_calls
|
||||
if coll == "calls" and doc_id == "call-1" and "scenes" in data
|
||||
]
|
||||
assert len(scenes_deletions) == 1, (
|
||||
f"expected exactly one doc_update clearing 'scenes', got {doc_update_calls}"
|
||||
)
|
||||
assert scenes_deletions[0][2]["scenes"] is fstore.DELETE_FIELD
|
||||
|
||||
# And no doc_set call should paper over the same field with an empty map
|
||||
# instead -- that would silently do nothing and leave stale scenes intact.
|
||||
for (coll, doc_id, data) in doc_set_calls:
|
||||
if coll == "calls" and doc_id == "call-1":
|
||||
assert "scenes" not in data, (
|
||||
"a doc_set (merge=True) write must never carry 'scenes' -- "
|
||||
"merging {} over an existing map is a no-op, not a delete"
|
||||
)
|
||||
|
||||
|
||||
def test_delete_field_is_the_real_firestore_sentinel():
|
||||
"""Catches an import-path typo turning this into a silent no-op sentinel."""
|
||||
from firebase_admin import firestore as fs
|
||||
assert fstore.DELETE_FIELD is fs.DELETE_FIELD
|
||||
@@ -0,0 +1,124 @@
|
||||
"""
|
||||
server-26#114 — the incident summarizer used to read doc["transcript"] (the
|
||||
WHOLE call, raw) for every linked call, so a multi-scene call contributed
|
||||
text from scenes it wasn't part of into an incident's summary, and
|
||||
transcript_corrected was never consulted at all.
|
||||
|
||||
Fix: _scene_text_for_incident reads the server-26#96 `scenes` map to find the
|
||||
scene(s) that actually resolved into a given incident_id, and falls back to
|
||||
transcript_corrected-or-transcript for a call doc with no `scenes` field
|
||||
(predates #96).
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from app.internal import summarizer
|
||||
from app.internal.summarizer import _scene_text_for_incident
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _scene_text_for_incident — pure function, no Firestore
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def test_picks_the_scene_that_linked_to_this_incident():
|
||||
doc = {
|
||||
"transcript": "whole raw transcript blend",
|
||||
"transcript_corrected": "whole corrected transcript blend",
|
||||
"scenes": {
|
||||
"0": {"transcript": "scene zero text", "incident_id": "inc-A", "corr_debug": {}},
|
||||
"1": {"transcript": "scene one text", "incident_id": "inc-B", "corr_debug": {}},
|
||||
},
|
||||
}
|
||||
assert _scene_text_for_incident(doc, "inc-A") == "scene zero text"
|
||||
assert _scene_text_for_incident(doc, "inc-B") == "scene one text"
|
||||
|
||||
|
||||
def test_joins_multiple_scenes_linked_to_the_same_incident_in_scene_order():
|
||||
doc = {
|
||||
"scenes": {
|
||||
"1": {"transcript": "second", "incident_id": "inc-A"},
|
||||
"0": {"transcript": "first", "incident_id": "inc-A"},
|
||||
},
|
||||
}
|
||||
assert _scene_text_for_incident(doc, "inc-A") == "first\nsecond"
|
||||
|
||||
|
||||
def test_old_schema_doc_falls_back_to_transcript_corrected_over_transcript():
|
||||
doc = {"transcript": "raw", "transcript_corrected": "corrected"}
|
||||
assert _scene_text_for_incident(doc, "inc-A") == "corrected"
|
||||
|
||||
|
||||
def test_old_schema_doc_with_only_raw_transcript_still_returns_it():
|
||||
doc = {"transcript": "raw only"}
|
||||
assert _scene_text_for_incident(doc, "inc-A") == "raw only"
|
||||
|
||||
|
||||
def test_scenes_present_but_none_match_falls_back_defensively():
|
||||
"""Should not happen for a call_id genuinely in this incident's call_ids,
|
||||
but silently dropping the call's contribution would be worse than a
|
||||
whole-call fallback."""
|
||||
doc = {
|
||||
"transcript": "raw",
|
||||
"transcript_corrected": "corrected",
|
||||
"scenes": {"0": {"transcript": "x", "incident_id": "inc-OTHER"}},
|
||||
}
|
||||
assert _scene_text_for_incident(doc, "inc-A") == "corrected"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _summarize_incident — end to end with fstore/Gemini mocked
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_summarize_incident_uses_scene_specific_text_for_a_multiscene_call():
|
||||
"""
|
||||
call-1 is a 2-scene call: scene 0 linked into inc-OTHER, scene 1 linked
|
||||
into inc-1 (the incident being summarized). Only scene 1's text may reach
|
||||
the model.
|
||||
"""
|
||||
call_1 = {
|
||||
"call_id": "call-1",
|
||||
"transcript": "scene zero text scene one text", # the old, wrong, whole-call blend
|
||||
"scenes": {
|
||||
"0": {"transcript": "scene zero text", "incident_id": "inc-OTHER"},
|
||||
"1": {"transcript": "scene one text", "incident_id": "inc-1"},
|
||||
},
|
||||
}
|
||||
|
||||
async def fake_doc_get(collection, doc_id):
|
||||
assert collection == "calls"
|
||||
return call_1 if doc_id == "call-1" else None
|
||||
|
||||
with patch("app.internal.feature_flags.get_flags",
|
||||
AsyncMock(return_value={"summaries_enabled": True})), \
|
||||
patch.object(summarizer, "fstore") as fs, \
|
||||
patch.object(summarizer, "_sync_summarize", return_value="a summary") as sync:
|
||||
fs.doc_get = AsyncMock(side_effect=fake_doc_get)
|
||||
fs.doc_set = AsyncMock()
|
||||
await summarizer._summarize_incident({"incident_id": "inc-1", "call_ids": ["call-1"]})
|
||||
|
||||
sync.assert_called_once()
|
||||
_inc_arg, transcripts_arg = sync.call_args.args
|
||||
assert transcripts_arg == ["scene one text"]
|
||||
assert "scene zero text scene one text" not in transcripts_arg
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_summarize_incident_falls_back_for_old_schema_call_doc():
|
||||
"""A call doc with no `scenes` field at all — summarizer must still work,
|
||||
using transcript_corrected over raw transcript."""
|
||||
call_1 = {"call_id": "call-1", "transcript": "raw", "transcript_corrected": "corrected"}
|
||||
|
||||
async def fake_doc_get(collection, doc_id):
|
||||
return call_1 if doc_id == "call-1" else None
|
||||
|
||||
with patch("app.internal.feature_flags.get_flags",
|
||||
AsyncMock(return_value={"summaries_enabled": True})), \
|
||||
patch.object(summarizer, "fstore") as fs, \
|
||||
patch.object(summarizer, "_sync_summarize", return_value="a summary") as sync:
|
||||
fs.doc_get = AsyncMock(side_effect=fake_doc_get)
|
||||
fs.doc_set = AsyncMock()
|
||||
await summarizer._summarize_incident({"incident_id": "inc-1", "call_ids": ["call-1"]})
|
||||
|
||||
_inc_arg, transcripts_arg = sync.call_args.args
|
||||
assert transcripts_arg == ["corrected"]
|
||||
Reference in New Issue
Block a user