"""
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