logan/server-26
1
0
Fork 0
Code Issues 57 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

correlator+summarizer: per-scene call-doc storage, fixes #96 and #114's real fix

Browse Source 
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.<scene_index>` 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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Tbknwttzou4s46PAykmtix
This commit is contained in:
Logan Cusano
2026-09-13 13:25:37 -04:00
co-authored by Claude Sonnet 5
parent b7701b6d49
commit fae84a45c3
7 changed files with 535 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drb-c2-core/app/internal/incident_correlator.py
+58 -2
View File
@@ -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.<scene_index>. 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
drb-c2-core/app/internal/summarizer.py
+54 -3
View File
@@ -15,6 +15,39 @@ from app.internal import firestore as fstore
from app.config import settings
def _scene_sort_key(scene_index: str):
"""Numeric-first sort so a >=10-scene call's entries still read in order."""
return (0, int(scene_index)) if scene_index.isdigit() else (1, scene_index)
def _scene_text_for_incident(doc: dict, incident_id: str) -> Optional[str]:
"""
The text of `doc` (a call doc) that actually belongs to `incident_id`.
server-26#96 records, per scene, which incident_id that scene's
correlation decision resolved to (incident_correlator._apply_and_log's
`scenes.<index>.incident_id`). Use that to pick only the scene(s) of this
call that are genuinely part of this incident, joining more than one if
several scenes happened to link into the same incident.
Falls back to transcript_corrected-or-transcript when the call doc has no
`scenes` field (predates server-26#96) or — defensively — when it has one
but nothing in it names this incident_id (should not happen for a call_id
that's actually in this incident's call_ids, but silently dropping a
call's contribution to its own summary would be a worse failure mode than
falling back to the whole-call text).
"""
scenes = doc.get("scenes") or {}
matched = [
scene.get("transcript")
for _, scene in sorted(scenes.items(), key=lambda kv: _scene_sort_key(kv[0]))
if scene.get("incident_id") == incident_id and scene.get("transcript")
]
if matched:
return "\n".join(matched)
return doc.get("transcript_corrected") or doc.get("transcript")
async def summarizer_loop() -> None:
from app.internal.feature_flags import get_flags
interval = settings.summary_interval_minutes * 60
@@ -63,12 +96,30 @@ async def _summarize_incident(inc: dict) -> None:
if not call_ids:
return
# Fetch transcripts for all calls in this incident
# Fetch transcripts for all calls in this incident.
#
# server-26#114: a call links into an incident one SCENE at a time (see
# incident_correlator._apply_decision / server-26#96's `scenes` map on the
# call doc), and the same call_id can appear in more than one incident's
# call_ids — once per scene, each scene possibly landing in a different
# incident. Reading doc["transcript"] (the whole call, raw) meant an
# incident's summary was built partly on text from a DIFFERENT scene of
# that call that this incident has nothing to do with, and ignored
# transcript_corrected entirely.
#
# _scene_text_for_incident reads the specific scene(s) whose corr_debug
# recorded a link into THIS incident_id. For a call doc that predates
# this fix (no `scenes` field) it falls back to
# transcript_corrected-or-transcript — the one-liner half of #114, worth
# doing even for old-schema docs since it stops raw-transcript summaries.
transcripts: list[str] = []
for cid in call_ids:
doc = await fstore.doc_get("calls", cid)
if doc and doc.get("transcript"):
transcripts.append(doc["transcript"])
if not doc:
continue
text = _scene_text_for_incident(doc, incident_id)
if text:
transcripts.append(text)
if not transcripts:
# No transcripts yet — clear stale flag and wait for next pass
drb-c2-core/app/routers/admin.py
+69 -6
View File
@@ -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`
drb-c2-core/app/routers/upload.py
+17 -2
View File
@@ -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)
drb-c2-core/tests/test_admin_debug_correlation.py
+72
View File
@@ -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}
drb-c2-core/tests/test_per_scene_call_doc.py
+141
View File
@@ -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
drb-c2-core/tests/test_summarizer_scene_transcript.py
+124
View File
@@ -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"]