From fae84a45c3cac8dcca52b9f7394545f3f581fb1b Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 13 Sep 2026 13:25:37 -0400 Subject: [PATCH 1/2] correlator+summarizer: per-scene call-doc storage, fixes #96 and #114's real fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every scene of a multi-scene call correlates independently in upload.py's scene loop, but every scene's corr_debug was written flat onto the same shared call doc — scene 2's write silently clobbered scene 1's corr_path/corr_consensus/etc (#96), and summarizer.py read the whole call's raw transcript per linked call, mixing text from scenes the incident had nothing to do with, while ignoring transcript_corrected entirely (#114). Fix: thread a scene_index from both `for scene in scenes:` loops in upload.py down through _correlate_with_consensus -> incident_correlator.preview_correlation/correlate_call -> _build_context -> ctx["scene_index"]. incident_correlator._apply_and_log now writes, in the same Firestore call: - the existing flat corr_* fields, unchanged (last-scene-wins, the safe backward-compatible default for any reader that doesn't know about `scenes` yet) - a new nested `scenes.` entry with {transcript, incident_id, corr_debug}, via doc_set(..., merge=True). Firestore's DocumentReference.set(data, merge=True) recursively merges nested map fields by key (documented SDK behaviour, not assumed) — a write to scenes.1 merges alongside an existing scenes.0 instead of replacing the whole `scenes` map. scene_index defaults to 0 for every caller with no scene concept (the recorrelation sweep, the no-scenes-extracted orphan-check path), so a plain single-scene call still gets a one-entry `scenes` map equivalent to reading its flat fields today. admin.py's _call_summary exposes the new `scenes` list per call (each entry carrying the same corr_* field names as the flat fields, so the two shapes are interchangeable to the tally); the summary tally now iterates each call's scenes-if-present, else its own flat fields, so a 2-scene call with two different corr_path values counts as two data points instead of one blend. New `scene_decision_count` sits next to `linked_call_count` to make that distinction visible. summarizer.py's _scene_text_for_incident reads a linked call's `scenes` map to find the scene(s) whose corr_debug recorded a link into the specific incident being summarized, joining more than one if several scenes landed in the same incident. Falls back to transcript_corrected-or-transcript for a call doc with no `scenes` field (predates this change) — the one-liner half of #114, worth doing regardless since it stops raw-transcript summaries even for old-schema docs. Does not touch #80/#95/#102's existing ctx-threading fixes (embedding/severity/coords/LLM-prompt-transcript) — correct as-is, out of scope here. Tests: 14 new (test_per_scene_call_doc.py, test_summarizer_scene_transcript.py, additions to test_admin_debug_correlation.py) covering the merge shape, last-scene-wins flat-field backward compat, the admin tally's per-scene vs per-call counting (including old-schema fallback), and the summarizer's scene-specific text selection (including old-schema fallback). Full sandboxed suite: 364 -> 378 passed, all green. Fixes server-26#96, server-26#114 Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix --- .../app/internal/incident_correlator.py | 60 +++++++- drb-c2-core/app/internal/summarizer.py | 57 ++++++- drb-c2-core/app/routers/admin.py | 75 +++++++++- drb-c2-core/app/routers/upload.py | 19 ++- .../tests/test_admin_debug_correlation.py | 72 +++++++++ drb-c2-core/tests/test_per_scene_call_doc.py | 141 ++++++++++++++++++ .../tests/test_summarizer_scene_transcript.py | 124 +++++++++++++++ 7 files changed, 535 insertions(+), 13 deletions(-) create mode 100644 drb-c2-core/tests/test_per_scene_call_doc.py create mode 100644 drb-c2-core/tests/test_summarizer_scene_transcript.py diff --git a/drb-c2-core/app/internal/incident_correlator.py b/drb-c2-core/app/internal/incident_correlator.py index fc1142e..5de12f7 100644 --- a/drb-c2-core/app/internal/incident_correlator.py +++ b/drb-c2-core/app/internal/incident_correlator.py @@ -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,46 @@ 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.. Firestore's + `DocumentReference.set(data, merge=True)` recursively merges nested map + fields by key (confirmed against the documented set-with-merge semantics, + not assumed): 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 diff --git a/drb-c2-core/app/internal/summarizer.py b/drb-c2-core/app/internal/summarizer.py index 0cb8914..2d7f0cd 100644 --- a/drb-c2-core/app/internal/summarizer.py +++ b/drb-c2-core/app/internal/summarizer.py @@ -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..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 diff --git a/drb-c2-core/app/routers/admin.py b/drb-c2-core/app/routers/admin.py index bebfec8..eeb04c0 100644 --- a/drb-c2-core/app/routers/admin.py +++ b/drb-c2-core/app/routers/admin.py @@ -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` diff --git a/drb-c2-core/app/routers/upload.py b/drb-c2-core/app/routers/upload.py index a7b2d32..fc183b3 100644 --- a/drb-c2-core/app/routers/upload.py +++ b/drb-c2-core/app/routers/upload.py @@ -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) diff --git a/drb-c2-core/tests/test_admin_debug_correlation.py b/drb-c2-core/tests/test_admin_debug_correlation.py index 6b02fca..a274a84 100644 --- a/drb-c2-core/tests/test_admin_debug_correlation.py +++ b/drb-c2-core/tests/test_admin_debug_correlation.py @@ -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} diff --git a/drb-c2-core/tests/test_per_scene_call_doc.py b/drb-c2-core/tests/test_per_scene_call_doc.py new file mode 100644 index 0000000..5fbabdb --- /dev/null +++ b/drb-c2-core/tests/test_per_scene_call_doc.py @@ -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. 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 diff --git a/drb-c2-core/tests/test_summarizer_scene_transcript.py b/drb-c2-core/tests/test_summarizer_scene_transcript.py new file mode 100644 index 0000000..f421591 --- /dev/null +++ b/drb-c2-core/tests/test_summarizer_scene_transcript.py @@ -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"] From 0fe6d3b567dfc7df9aaea66f29733e7ee9867ec9 Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 13 Sep 2026 13:33:52 -0400 Subject: [PATCH 2/2] correlator: delete stale scenes on re-extraction instead of leaving them to rot (#96/#114) Review of #132 found a blocker: PATCH /calls/{id}/transcript wipes tags/severity/location/units/embedding before re-extraction, but not the new scenes map, and doc_set(merge=True) can only add/overwrite nested map keys, never remove one. A call corrected from 3 scenes to 1 kept scenes.1/scenes.2 with pre-correction transcripts and incident_ids forever -- corrupting the exact per-scene tally #96 exists to make trustworthy, and able to re-feed stale text into #114's summarizer fix if a stale scene's incident_id still names a real incident. Fix: fstore.doc_update(...,{"scenes": fstore.DELETE_FIELD}) -- a real delete, not a merge over an empty map. Added fstore.DELETE_FIELD (re-exports the real firebase_admin sentinel) and stubbed it in the sandboxed test conftest, which didn't have it. Also softened an overclaiming docstring: the Firestore nested-merge behavior is verified against the doc_set wrapper's pass-through, not against live Firestore. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix --- drb-c2-core/app/internal/firestore.py | 9 ++ .../app/internal/incident_correlator.py | 9 +- drb-c2-core/app/routers/calls.py | 9 ++ drb-c2-core/tests/conftest.py | 4 + .../test_reprocess_clears_stale_scenes.py | 89 +++++++++++++++++++ 5 files changed, 118 insertions(+), 2 deletions(-) create mode 100644 drb-c2-core/tests/test_reprocess_clears_stale_scenes.py diff --git a/drb-c2-core/app/internal/firestore.py b/drb-c2-core/app/internal/firestore.py index c3ab551..c9415c2 100644 --- a/drb-c2-core/app/internal/firestore.py +++ b/drb-c2-core/app/internal/firestore.py @@ -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) # --------------------------------------------------------------------------- diff --git a/drb-c2-core/app/internal/incident_correlator.py b/drb-c2-core/app/internal/incident_correlator.py index 5de12f7..3ea816a 100644 --- a/drb-c2-core/app/internal/incident_correlator.py +++ b/drb-c2-core/app/internal/incident_correlator.py @@ -1458,8 +1458,13 @@ async def _apply_and_log(decision: dict, ctx: dict) -> Optional[str]: same corr_debug — plus this scene's own transcript and the incident_id it resolved to — under scenes.. Firestore's `DocumentReference.set(data, merge=True)` recursively merges nested map - fields by key (confirmed against the documented set-with-merge semantics, - not assumed): a write of {"scenes": {"1": {...}}} merges into an existing + 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. diff --git a/drb-c2-core/app/routers/calls.py b/drb-c2-core/app/routers/calls.py index 9663981..b3efaa3 100644 --- a/drb-c2-core/app/routers/calls.py +++ b/drb-c2-core/app/routers/calls.py @@ -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. diff --git a/drb-c2-core/tests/conftest.py b/drb-c2-core/tests/conftest.py index 7364d68..41c939f 100644 --- a/drb-c2-core/tests/conftest.py +++ b/drb-c2-core/tests/conftest.py @@ -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() diff --git a/drb-c2-core/tests/test_reprocess_clears_stale_scenes.py b/drb-c2-core/tests/test_reprocess_clears_stale_scenes.py new file mode 100644 index 0000000..92477a8 --- /dev/null +++ b/drb-c2-core/tests/test_reprocess_clears_stale_scenes.py @@ -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