@@ -7,6 +7,15 @@ from google.cloud.firestore_v1.base_query import FieldFilter
|
|||||||
from app.config import settings
|
from app.config import settings
|
||||||
from app.internal.logger import logger
|
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)
|
# In-memory TTL cache for rarely-changing documents (systems, nodes config)
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|||||||
@@ -709,6 +709,7 @@ async def correlate_call(
|
|||||||
embedding: Optional[list] = None,
|
embedding: Optional[list] = None,
|
||||||
severity: Optional[str] = None,
|
severity: Optional[str] = None,
|
||||||
transcript: Optional[str] = None,
|
transcript: Optional[str] = None,
|
||||||
|
scene_index: int = 0,
|
||||||
) -> 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.
|
||||||
@@ -718,6 +719,11 @@ async def correlate_call(
|
|||||||
Callers that re-correlate a whole call rather than a scene — the
|
Callers that re-correlate a whole call rather than a scene — the
|
||||||
recorrelation sweep — pass the call doc's stored values explicitly; they are
|
recorrelation sweep — pass the call doc's stored values explicitly; they are
|
||||||
no longer read from the doc inside _build_context.
|
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(
|
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,
|
||||||
@@ -726,6 +732,7 @@ async def correlate_call(
|
|||||||
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, transcript=transcript,
|
embedding=embedding, severity=severity, transcript=transcript,
|
||||||
|
scene_index=scene_index,
|
||||||
)
|
)
|
||||||
decision = _run_decision(ctx)
|
decision = _run_decision(ctx)
|
||||||
return await _apply_and_log(decision, ctx)
|
return await _apply_and_log(decision, ctx)
|
||||||
@@ -750,6 +757,7 @@ async def preview_correlation(
|
|||||||
embedding: Optional[list] = None,
|
embedding: Optional[list] = None,
|
||||||
severity: Optional[str] = None,
|
severity: Optional[str] = None,
|
||||||
transcript: Optional[str] = None,
|
transcript: Optional[str] = None,
|
||||||
|
scene_index: int = 0,
|
||||||
) -> 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.
|
||||||
@@ -763,6 +771,15 @@ async def preview_correlation(
|
|||||||
matched_incident the candidate incident doc (action == "link")
|
matched_incident the candidate incident doc (action == "link")
|
||||||
incident_type resolved type after tag inference (action == "new")
|
incident_type resolved type after tag inference (action == "new")
|
||||||
corr_debug fields to persist on the call doc
|
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(
|
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,
|
||||||
@@ -771,6 +788,7 @@ async def preview_correlation(
|
|||||||
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, transcript=transcript,
|
embedding=embedding, severity=severity, transcript=transcript,
|
||||||
|
scene_index=scene_index,
|
||||||
)
|
)
|
||||||
decision = _run_decision(ctx)
|
decision = _run_decision(ctx)
|
||||||
return {"decision": decision, "ctx": ctx}
|
return {"decision": decision, "ctx": ctx}
|
||||||
@@ -806,6 +824,7 @@ async def _build_context(
|
|||||||
embedding: Optional[list] = None,
|
embedding: Optional[list] = None,
|
||||||
severity: Optional[str] = None,
|
severity: Optional[str] = None,
|
||||||
transcript: Optional[str] = None,
|
transcript: Optional[str] = None,
|
||||||
|
scene_index: int = 0,
|
||||||
) -> 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)
|
||||||
@@ -883,6 +902,9 @@ async def _build_context(
|
|||||||
"incident_type": incident_type, "location": location,
|
"incident_type": incident_type, "location": location,
|
||||||
"location_coords": location_coords, "reassignment": reassignment,
|
"location_coords": location_coords, "reassignment": reassignment,
|
||||||
"create_if_new": create_if_new,
|
"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]:
|
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)
|
incident_id = await _apply_decision(decision, ctx)
|
||||||
corr_debug = decision.get("corr_debug") or {}
|
corr_debug = decision.get("corr_debug") or {}
|
||||||
if corr_debug:
|
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:
|
try:
|
||||||
await fstore.doc_set("calls", ctx["call_id"], corr_debug)
|
await fstore.doc_set("calls", ctx["call_id"], updates)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warning(f"Could not write corr_debug for call {ctx['call_id']}: {e}")
|
logger.warning(f"Could not write corr_debug for call {ctx['call_id']}: {e}")
|
||||||
return incident_id
|
return incident_id
|
||||||
|
|||||||
@@ -15,6 +15,39 @@ from app.internal import firestore as fstore
|
|||||||
from app.config import settings
|
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:
|
async def summarizer_loop() -> None:
|
||||||
from app.internal.feature_flags import get_flags
|
from app.internal.feature_flags import get_flags
|
||||||
interval = settings.summary_interval_minutes * 60
|
interval = settings.summary_interval_minutes * 60
|
||||||
@@ -63,12 +96,30 @@ async def _summarize_incident(inc: dict) -> None:
|
|||||||
if not call_ids:
|
if not call_ids:
|
||||||
return
|
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] = []
|
transcripts: list[str] = []
|
||||||
for cid in call_ids:
|
for cid in call_ids:
|
||||||
doc = await fstore.doc_get("calls", cid)
|
doc = await fstore.doc_get("calls", cid)
|
||||||
if doc and doc.get("transcript"):
|
if not doc:
|
||||||
transcripts.append(doc["transcript"])
|
continue
|
||||||
|
text = _scene_text_for_incident(doc, incident_id)
|
||||||
|
if text:
|
||||||
|
transcripts.append(text)
|
||||||
|
|
||||||
if not transcripts:
|
if not transcripts:
|
||||||
# No transcripts yet — clear stale flag and wait for next pass
|
# No transcripts yet — clear stale flag and wait for next pass
|
||||||
|
|||||||
@@ -97,8 +97,49 @@ async def debug_correlation(
|
|||||||
def _strip(doc: dict) -> dict:
|
def _strip(doc: dict) -> dict:
|
||||||
return {k: v for k, v in doc.items() if k != "embedding"}
|
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 {
|
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"),
|
"call_id": call.get("call_id"),
|
||||||
"started_at": call.get("started_at"),
|
"started_at": call.get("started_at"),
|
||||||
"ended_at": call.get("ended_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 [])]
|
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]
|
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:
|
def _span_minutes(inc: dict) -> float:
|
||||||
stamps = sorted(
|
stamps = sorted(
|
||||||
s for s in ((c.get("started_at") or "") for c in (inc.get("calls_detail") or [])) if s
|
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_systems_only": ai_systems_only,
|
||||||
"ai_enabled_system_ids": sorted(ai_systems),
|
"ai_enabled_system_ids": sorted(ai_systems),
|
||||||
"linked_call_count": len(linked),
|
"linked_call_count": len(linked),
|
||||||
"corr_path": _tally(c.get("corr_path") for c in linked),
|
# server-26#96 — tallied over scene_entries (one entry per scene of a
|
||||||
"corr_fit_signal": _tally(c.get("corr_fit_signal") for c in linked),
|
# multi-scene call, from its `scenes` map; one entry per call when it
|
||||||
"corr_consensus": _tally(c.get("corr_consensus") for c in linked),
|
# has none) rather than over `linked` directly, so a 2-scene call
|
||||||
"corr_llm_action": _tally(c.get("corr_llm_action") for c in linked),
|
# 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
|
# server-26#115 — this IS the number the escape-hatch fix exists to
|
||||||
# produce: why each llm=orphan/rules=new call escaped the gate.
|
# 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
|
# server-26#127 — shadow-mode chatter classifier. The target
|
||||||
# population is non-events, which land as orphans or single-call
|
# population is non-events, which land as orphans or single-call
|
||||||
# incidents, NOT as a slice of every linked call -- tally `orphans`
|
# incidents, NOT as a slice of every linked call -- tally `orphans`
|
||||||
|
|||||||
@@ -260,6 +260,15 @@ async def patch_transcript(
|
|||||||
"vehicles": [],
|
"vehicles": [],
|
||||||
"embedding": None,
|
"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.
|
# Unlink from ALL current incidents so re-correlation starts clean.
|
||||||
# Handles both old single incident_id and new incident_ids list.
|
# Handles both old single incident_id and new incident_ids list.
|
||||||
|
|||||||
@@ -240,6 +240,7 @@ async def _correlate_with_consensus(
|
|||||||
embedding: Optional[list] = None,
|
embedding: Optional[list] = None,
|
||||||
severity: Optional[str] = None,
|
severity: Optional[str] = None,
|
||||||
transcript: Optional[str] = None,
|
transcript: Optional[str] = None,
|
||||||
|
scene_index: int = 0,
|
||||||
) -> 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.
|
||||||
@@ -248,6 +249,11 @@ async def _correlate_with_consensus(
|
|||||||
|
|
||||||
Falls back to rules-only when GEMINI_API_KEY is absent, the call is
|
Falls back to rules-only when GEMINI_API_KEY is absent, the call is
|
||||||
content-free (thin), or any LLM call fails.
|
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
|
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,
|
location_coords=location_coords, units=units, vehicles=vehicles,
|
||||||
cleared_units=cleared_units, reassignment=reassignment,
|
cleared_units=cleared_units, reassignment=reassignment,
|
||||||
embedding=embedding, severity=severity, transcript=transcript,
|
embedding=embedding, severity=severity, transcript=transcript,
|
||||||
|
scene_index=scene_index,
|
||||||
)
|
)
|
||||||
ctx = preview["ctx"]
|
ctx = preview["ctx"]
|
||||||
rules_decision = preview["decision"]
|
rules_decision = preview["decision"]
|
||||||
@@ -365,7 +372,10 @@ async def _run_extraction_pipeline(
|
|||||||
)
|
)
|
||||||
|
|
||||||
# Step 3: Correlate each scene to an incident independently.
|
# 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"])
|
all_tags.extend(scene["tags"])
|
||||||
# When dispatch is pulling a unit to a NEW call (reassignment), suppress unit
|
# 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.
|
# 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"),
|
embedding=scene.get("embedding"),
|
||||||
severity=scene.get("severity"),
|
severity=scene.get("severity"),
|
||||||
transcript=scene.get("transcript"),
|
transcript=scene.get("transcript"),
|
||||||
|
scene_index=scene_index,
|
||||||
)
|
)
|
||||||
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)
|
||||||
@@ -485,7 +496,10 @@ async def _run_intelligence_pipeline(
|
|||||||
incident_ids: list[str] = []
|
incident_ids: list[str] = []
|
||||||
all_tags: list[str] = []
|
all_tags: list[str] = []
|
||||||
if _flag("correlation_enabled"):
|
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"])
|
all_tags.extend(scene["tags"])
|
||||||
is_reassignment = bool(scene.get("reassignment"))
|
is_reassignment = bool(scene.get("reassignment"))
|
||||||
corr_units = [] if is_reassignment else scene.get("units")
|
corr_units = [] if is_reassignment else scene.get("units")
|
||||||
@@ -506,6 +520,7 @@ async def _run_intelligence_pipeline(
|
|||||||
embedding=scene.get("embedding"),
|
embedding=scene.get("embedding"),
|
||||||
severity=scene.get("severity"),
|
severity=scene.get("severity"),
|
||||||
transcript=scene.get("transcript"),
|
transcript=scene.get("transcript"),
|
||||||
|
scene_index=scene_index,
|
||||||
)
|
)
|
||||||
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)
|
||||||
|
|||||||
@@ -34,6 +34,10 @@ except ModuleNotFoundError:
|
|||||||
# into dicts that tests compare against, and a MagicMock compares unequal
|
# into dicts that tests compare against, and a MagicMock compares unequal
|
||||||
# to itself across attribute accesses.
|
# to itself across attribute accesses.
|
||||||
_fs.SERVER_TIMESTAMP = "__SERVER_TIMESTAMP__"
|
_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 = ModuleType("firebase_admin.auth")
|
||||||
_auth.verify_id_token = MagicMock()
|
_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_consensus"] == "rules_only"
|
||||||
assert detail["corr_llm_reasoning"] is None
|
assert detail["corr_llm_reasoning"] is None
|
||||||
assert detail["corr_llm_action"] 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