Logan CusanoandClaude Sonnet 5
241a15b8da
correct()'s prompt says "Do NOT expand ten-codes" and "NEVER add information", but nothing checked the model's output against its own rules -- raw["corrected"] was accepted verbatim past a non-empty/changed check, and the "changed" field it returns was logged for audit and never validated. Caught live: the same call's raw vs corrected transcript showed "10-7" rewritten to "10-13" in one place and "10-4" in another, plus a bare "7" expanded into "ShotSpotter" -- alongside genuinely good fixes (Holmes Street and 4th and Rowe -> Home Street and Forest Ave, from this system's own vocabulary). A wrong 10-13 standing in for a real 10-7 reads exactly as trustworthy as a correct transcript, which is worse than leaving the raw mishearing in place. _code_tokens() extracts every ten-code/signal-shaped token from the original and corrected text/segments; any change to that set discards the correction and falls back to raw. Checked independently for the joined text and for segments, consistent with the existing all-or-nothing segment-alignment rule. Does not catch a wrong word swapped for another equally plausible non-code word -- that class still depends entirely on the model following its own prompt. Filed server-26#161 for the broader STT/audio quality initiative this belongs alongside. Verified: 426 pass, 0 fail (4 new tests). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
282 lines
11 KiB
Python
282 lines
11 KiB
Python
"""
|
|
Unit tests for the transcript correction pass (server-26#36).
|
|
|
|
Two properties carry real risk and are pinned hardest here:
|
|
|
|
* SCOPE RESOLUTION — talkgroup reference data must rank ABOVE system data.
|
|
A system spanning several counties can have a talkgroup covering one
|
|
municipality, and burying that municipality's streets under a county-wide
|
|
list is the failure this whole feature exists to avoid.
|
|
|
|
* SEGMENT ALIGNMENT — scene extraction maps scenes to transmissions by index
|
|
(segment_indices), so a corrected array of the wrong length would silently
|
|
attribute the wrong audio to a scene. Anything but an exact 1:1 match must
|
|
be discarded whole.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from app.internal import transcript_correction as tc
|
|
|
|
SYSTEM = {
|
|
"vocabulary": ["Croton-Harmon", "Metro-North"],
|
|
"ten_codes": {"10-4": "acknowledged", "10-13": "officer needs assistance"},
|
|
"area_context": {
|
|
"county": "Westchester",
|
|
"state": "New York",
|
|
"local_knowledge": [
|
|
{"term": "Route 9", "meaning": "north-south state highway"},
|
|
{"term": "Saw Mill Parkway"},
|
|
],
|
|
},
|
|
"config": {
|
|
"talkgroups": [
|
|
{
|
|
"id": 9048,
|
|
"name": "Ossining - Police Dispatch",
|
|
"vocabulary": ["Snowden Avenue", "Croton-Harmon"],
|
|
"area_context": {
|
|
"municipality": "Ossining",
|
|
"local_knowledge": [{"term": "Sing Sing", "meaning": "state prison"}],
|
|
},
|
|
},
|
|
{"id": 9600, "name": "Harrison - Police/EMS Dispatch"},
|
|
{"id": 9563, "ten_codes": {"10-4": "on scene"}},
|
|
]
|
|
},
|
|
}
|
|
|
|
|
|
def _system(doc=SYSTEM):
|
|
return patch.object(tc.fstore, "doc_get_cached", AsyncMock(return_value=doc))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _api_key():
|
|
"""
|
|
The dev venv has no GEMINI_API_KEY, and correct() returns early without one
|
|
— which would make every assertion below pass for the wrong reason.
|
|
"""
|
|
with patch.object(tc.settings, "gemini_api_key", "test-key"):
|
|
yield
|
|
|
|
|
|
# ── Scope resolution ────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_vocabulary_ranks_above_system():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
assert ctx["vocabulary"][0] == "Snowden Avenue", "talkgroup terms must come first"
|
|
assert "Metro-North" in ctx["vocabulary"], "system terms are still inherited"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_terms_are_not_repeated():
|
|
"""Croton-Harmon is on both scopes; it should appear once, at talkgroup rank."""
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
assert [t.lower() for t in ctx["vocabulary"]].count("croton-harmon") == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_area_precedes_system_area():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
joined = "\n".join(ctx["area_lines"])
|
|
assert joined.index("Ossining") < joined.index("Westchester")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_without_own_data_inherits_system():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9600)
|
|
assert ctx["vocabulary"] == ["Croton-Harmon", "Metro-North"]
|
|
assert any("Westchester" in line for line in ctx["area_lines"])
|
|
assert ctx["area"].get("municipality") is None, "inherits, invents nothing"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_talkgroup_ten_code_overrides_system_meaning():
|
|
with _system():
|
|
ctx = await tc.resolve_context("sys-1", 9563)
|
|
assert ctx["ten_codes"]["10-4"] == "on scene"
|
|
assert ctx["ten_codes"]["10-13"] == "officer needs assistance"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("system_id, tgid", [(None, 9048), ("sys-1", None)])
|
|
async def test_missing_scope_is_not_an_error(system_id, tgid):
|
|
with _system():
|
|
ctx = await tc.resolve_context(system_id, tgid)
|
|
assert isinstance(ctx["vocabulary"], list)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unconfigured_system_yields_empty_context():
|
|
with _system(doc=None):
|
|
ctx = await tc.resolve_context("sys-1", 9048)
|
|
assert ctx == {
|
|
"vocabulary": [], "ten_codes": {}, "area_lines": [],
|
|
"area": {}, "system_area": {}, "tg_area": {},
|
|
}
|
|
|
|
|
|
# ── Correction behaviour ────────────────────────────────────────────────────
|
|
|
|
def _gemini(payload):
|
|
return patch.object(tc, "_sync_gemini", lambda model, prompt: payload)
|
|
|
|
|
|
SEGS = [{"start": 0.0, "end": 1.0, "text": "Headquarters, 11-9."},
|
|
{"start": 1.0, "end": 2.0, "text": "Shout out to Optum."},
|
|
{"start": 2.0, "end": 3.0, "text": "360 north, back to Rose."}]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_short_transcript_is_never_sent():
|
|
"""9 of 29 calls in the sample window were <=3 words. Nothing to correct."""
|
|
with patch.object(tc, "_sync_gemini") as m:
|
|
out = await tc.correct("c1", "10-4.", None, system_id="sys-1")
|
|
assert out == (None, None, False)
|
|
m.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_by_setting():
|
|
with patch.object(tc.settings, "transcript_correction_enabled", False), \
|
|
patch.object(tc, "_sync_gemini") as m:
|
|
assert await tc.correct("c1", "a b c d e", None) == (None, None, False)
|
|
m.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_segments_corrected_when_lengths_match():
|
|
payload = {"corrected": "Headquarters, 11-9. Show it out to Ossining. 360 north, back to Route 9.",
|
|
"segments": ["Headquarters, 11-9.", "Show it out to Ossining.", "360 north, back to Route 9."]}
|
|
with _system(), _gemini(payload):
|
|
text, segs, not_speech = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1", talkgroup_id=9048)
|
|
assert not_speech is False
|
|
assert segs is not None and len(segs) == 3
|
|
assert segs[1]["text"] == "Show it out to Ossining."
|
|
assert segs[1]["start"] == 1.0, "timing must survive correction untouched"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wrong_segment_count_is_discarded_whole():
|
|
"""A short array would silently misattribute audio to the wrong scene."""
|
|
payload = {"corrected": "fine", "segments": ["only", "two"]}
|
|
with _system(), _gemini(payload):
|
|
text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1")
|
|
assert segs is None
|
|
assert text == "fine", "the joined correction still stands"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_string_segment_entries_are_discarded():
|
|
payload = {"corrected": None, "segments": ["ok", 42, "ok"]}
|
|
with _system(), _gemini(payload):
|
|
_, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1")
|
|
assert segs is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unchanged_segments_report_no_correction():
|
|
payload = {"corrected": None, "segments": [s["text"] for s in SEGS]}
|
|
with _system(), _gemini(payload):
|
|
text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1")
|
|
assert (text, segs) == (None, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_echoed_transcript_counts_as_no_change():
|
|
with _system(), _gemini({"corrected": " x y z w "}):
|
|
text, _, _ = await tc.correct("c1", "x y z w", None, system_id="sys-1")
|
|
assert text is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_not_speech_is_surfaced():
|
|
with _system(), _gemini({"corrected": None, "not_speech": True}):
|
|
_, _, not_speech = await tc.correct("c1", "10-11. 10-12. 10-13. 10-14.", None, system_id="sys-1")
|
|
assert not_speech is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_failure_leaves_the_transcript_alone():
|
|
"""Correction is an improvement, never a dependency."""
|
|
def boom(model, prompt):
|
|
raise RuntimeError("gemini exploded")
|
|
with _system(), patch.object(tc, "_sync_gemini", boom):
|
|
assert await tc.correct("c1", "x y z w", SEGS, system_id="sys-1") == (None, None, False)
|
|
|
|
|
|
# ── Code-token guard (server-26#162) ────────────────────────────────────────
|
|
# Caught live: the same call came back with "10-7" rewritten to "10-13" in one
|
|
# place and "10-4" in another. A real code swapped for a different real code
|
|
# reads exactly as trustworthy as a correct one — worse than leaving the raw
|
|
# mishearing in place, since nothing downstream can tell it happened.
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_changed_ten_code_is_discarded():
|
|
payload = {"corrected": "10-13, we're back in town."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct("c1", "10-7, we're back in town.", None, system_id="sys-1")
|
|
assert text is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invented_code_token_is_discarded():
|
|
"""Nothing code-shaped in the original — the model added one from nothing."""
|
|
payload = {"corrected": "ShotSpotter, 10-4, group of 3 shooting outside."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct("c1", "Seven, group of 3 shooting outside.", None, system_id="sys-1")
|
|
assert text is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legitimate_place_correction_with_unchanged_codes_still_applies():
|
|
"""The guard must not collateral-damage a correction that never touches
|
|
a code token — Home/Forest for Holmes/4th-and-Rowe is exactly the kind of
|
|
fix this pass exists to make."""
|
|
payload = {"corrected": "10-13 coming over on Home Street and Forest Ave, 4-2."}
|
|
with _system(), _gemini(payload):
|
|
text, _, _ = await tc.correct(
|
|
"c1", "10-13 coming over on Holmes Street and 4th and Rowe, 4-2.",
|
|
None, system_id="sys-1",
|
|
)
|
|
assert text == "10-13 coming over on Home Street and Forest Ave, 4-2."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_segment_code_change_discards_segments_only():
|
|
"""A code change in one segment discards the whole segments array (same
|
|
all-or-nothing rule as a length mismatch), but the independently-checked
|
|
joined correction still stands if it kept its own codes intact. The
|
|
joined `text`/`corrected` pair here is deliberately code-free — this test
|
|
isolates the segment-level guard, not the joined-text one."""
|
|
payload = {
|
|
"corrected": "Show it out to Ossining, back to Route 9.",
|
|
"segments": ["Headquarters, 10-13.", "Show it out to Ossining.", "360 north, back to Route 9."],
|
|
}
|
|
with _system(), _gemini(payload):
|
|
text, segs, _ = await tc.correct("c1", "x y z w", SEGS, system_id="sys-1", talkgroup_id=9048)
|
|
assert segs is None
|
|
assert text == "Show it out to Ossining, back to Route 9."
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reference_data_reaches_the_prompt():
|
|
seen = {}
|
|
def capture(model, prompt):
|
|
seen["prompt"] = prompt
|
|
return {"corrected": None}
|
|
with _system(), patch.object(tc, "_sync_gemini", capture):
|
|
await tc.correct("c1", "x y z w", None, system_id="sys-1",
|
|
talkgroup_id=9048, talkgroup_name="Ossining - Police Dispatch")
|
|
p = seen["prompt"]
|
|
assert "Snowden Avenue" in p and "Ossining - Police Dispatch" in p
|
|
assert "Sing Sing — state prison" in p, "a term without its meaning is half the information"
|
|
assert "Ossining, Westchester, New York" in p, "state must reach the prompt (server-26#36)"
|
|
assert "10-13=officer needs assistance" in p
|