From fae84a45c3cac8dcca52b9f7394545f3f581fb1b Mon Sep 17 00:00:00 2001 From: Logan Cusano Date: Sun, 13 Sep 2026 13:25:37 -0400 Subject: [PATCH] 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"]