from __future__ import annotations

import pytest

pytest.importorskip("aiohttp")

import json
import zipfile

from easycat.debug._issues import build_issues as _build_issues
from easycat.debug._turn_timeline import (
    build_timeline as _build_timeline,
)
from easycat.debug._turn_timeline import (
    summarise_turns as _summarise_turns,
)
from easycat.debug.bundle import RunBundle
from easycat.debug.export import export_debug_bundle
from easycat.debugger._records import (
    _SEARCH_MAX_QUERY_LEN,
    _SEARCH_SCAN_LIMIT,
    _build_transcript,
    _compile_search_regex,
    _filter_records,
    _search_records,
)
from easycat.debugger._sources import _session_source
from easycat.runtime import InMemoryRingBuffer
from easycat.runtime.artifacts import InMemoryArtifactStore
from easycat.runtime.records import (
    ErrorInfo,
    FrameworkTransitionRecord,
    JournalRecord,
    JournalRecordKind,
    TimingInfo,
)

from ._server_helpers import _build_voice_bundle


def test_filter_records_by_stage():
    records = [
        {"sequence": 1, "name": "stage_start", "data": {"stage": "tts"}},
        {"sequence": 2, "name": "stage_complete", "data": {"stage": "stt"}},
        {"sequence": 3, "name": "tts_frame", "data": {"stage": "tts"}},
    ]
    out = _filter_records(
        records, stage="tts", turn_id=None, name=None, from_seq=None, to_seq=None
    )
    assert [r["sequence"] for r in out] == [1, 3]


def test_filter_records_by_sequence_range():
    records = [{"sequence": i, "data": {}} for i in range(10)]
    out = _filter_records(records, stage=None, turn_id=None, name=None, from_seq=3, to_seq=6)
    assert [r["sequence"] for r in out] == [3, 4, 5, 6]


def test_filter_records_by_sequence_range_ignores_malformed_sequence_values():
    records = [
        {"sequence": "2", "data": {}},
        {"sequence": True, "data": {}},
        {"sequence": 3, "data": {}},
    ]

    out = _filter_records(records, stage=None, turn_id=None, name=None, from_seq=1, to_seq=4)

    assert [r["sequence"] for r in out] == [3]


def test_filter_records_by_multiple_names():
    records = [
        {"sequence": 1, "name": "vad_start_speaking", "data": {}},
        {"sequence": 2, "name": "tts_audio", "data": {}},
        {"sequence": 3, "name": "stt_partial", "data": {}},
        {"sequence": 4, "name": "bot_started_speaking", "data": {}},
    ]
    out = _filter_records(
        records,
        stage=None,
        turn_id=None,
        name=["vad_start_speaking", "stt_partial"],
        from_seq=None,
        to_seq=None,
    )
    assert [r["sequence"] for r in out] == [1, 3]


def test_build_issues_surfaces_errors_empty_stt_and_latency():
    records = [
        {
            "sequence": 1,
            "name": "stage_start",
            "turn_id": "t1",
            "data": {"stage": "vad"},
            "timing": {"wall_ns": 0},
        },
        {
            "sequence": 2,
            "name": "vad_stop_speaking",
            "turn_id": "t1",
            "data": {},
            "timing": {"wall_ns": 1_000_000_000},
        },
        {
            "sequence": 3,
            "name": "stt_final",
            "turn_id": "t1",
            "data": {"text": ""},
            "timing": {"wall_ns": 3_000_000_000},
        },
        {
            "sequence": 4,
            "name": "stage_start",
            "turn_id": "t1",
            "data": {"stage": "agent"},
            "timing": {"wall_ns": 3_100_000_000},
        },
        {
            "sequence": 5,
            "name": "tool_call_failed",
            "turn_id": "t1",
            "data": {"stage": "agent"},
            "timing": {"wall_ns": 3_200_000_000},
            "error": {"type": "RuntimeError", "message": "tool exploded"},
        },
        {
            "sequence": 6,
            "name": "stage_complete",
            "turn_id": "t1",
            "data": {"stage": "agent"},
            "timing": {"wall_ns": 12_500_000_000},
        },
    ]

    out = _build_issues(records)

    codes = {issue["code"] for issue in out["issues"]}
    # The shared issue engine uses unified codes (any errored record ->
    # record_error; per-stage latency
    # -> slow_milestone with the stage in the ``metric`` field).  Its full code
    # surface (barge-in, audio-health, latency) is covered by tests/debug/.
    assert {"record_error", "empty_stt_final", "slow_turn"} <= codes
    assert out["summary"]["error"] >= 1
    assert out["summary"]["warning"] >= 2
    assert out["total"] == len(out["issues"])


def test_transcript_ignores_malformed_turn_ids():
    records = [
        {
            "sequence": 1,
            "name": "stt_final",
            "turn_id": ["bad"],
            "data": {"text": "ignored"},
        },
        {
            "sequence": 2,
            "name": "agent_final",
            "turn_id": {"id": "bad"},
            "data": {"text": "also ignored"},
        },
        {
            "sequence": 3,
            "name": "stt_final",
            "turn_id": "turn-1",
            "data": {"text": "hello"},
        },
    ]

    out = _build_transcript(records)

    assert len(out) == 1
    assert out[0]["turn_id"] == "turn-1"
    assert out[0]["user"] == "hello"


def test_build_transcript_delegates_to_shared_helper():
    """The server transcript projection is the shared dependency-free helper.

    ``_build_transcript`` must be byte-identical to
    ``debug/_turn_timeline.extract_turn_transcripts`` so the SPA transcript
    panel and the two-source ``easycat diff`` agree on what each turn said.
    """
    import json as _json

    from easycat.debug._turn_timeline import extract_turn_transcripts

    records = [
        {"sequence": 1, "name": "stt_final", "turn_id": "t1", "data": {"text": "hi there"}},
        {
            "sequence": 2,
            "name": "agent_delta",
            "turn_id": "t1",
            "data": {"text": "hello", "type": "TEXT_DELTA"},
        },
        {
            "sequence": 3,
            "name": "agent_delta",
            "turn_id": "t1",
            "data": {"text": " back", "type": "TEXT_DELTA"},
        },
        {
            "sequence": 4,
            "name": "stage_complete",
            "turn_id": "t2",
            "data": {"stage": "agent", "response": "second answer"},
        },
        {"sequence": 5, "name": "stt_final", "turn_id": ["bad"], "data": {"text": "dropped"}},
    ]

    server_out = _build_transcript(records)
    shared_out = extract_turn_transcripts(records)
    # Byte-identical JSON serialization confirms the lift preserved the shape.
    assert _json.dumps(server_out) == _json.dumps(shared_out)
    assert server_out[0]["agent"] == "hello back"


def test_summarise_turns_tracks_audio_bytes():
    records = [
        {
            "sequence": 1,
            "turn_id": "t1",
            "name": "stage_start",
            "data": {"stage": "stt"},
            "timing": {"wall_ns": 1_000_000},
        },
        {
            "sequence": 2,
            "turn_id": "t1",
            "name": "tts_frame",
            "data": {"stage": "tts", "audio_bytes": 320},
            "timing": {"wall_ns": 5_000_000},
        },
        {
            "sequence": 3,
            "turn_id": "t1",
            "name": "tts_frame",
            "data": {"stage": "tts", "audio_bytes": 640},
            "timing": {"wall_ns": 9_000_000},
        },
    ]
    turns = _summarise_turns(records)
    assert len(turns) == 1
    assert turns[0]["turn_id"] == "t1"
    assert turns[0]["tts_audio_bytes"] == 960
    assert turns[0]["wall_ms"] == 8.0
    assert turns[0]["stage_counts"] == {"stt": 1, "tts": 2}


def test_summarise_turns_ignores_malformed_sequence_values():
    records = [
        {"sequence": "2", "turn_id": "t1", "name": "stage_start"},
        {"sequence": True, "turn_id": "t1", "name": "agent_delta"},
        {"sequence": 1, "turn_id": "t1", "name": "stage_complete"},
    ]

    (turn,) = _summarise_turns(records)

    assert turn["first_sequence"] == 1
    assert turn["last_sequence"] == 1


def test_serve_bundle_raises_helpful_error_when_aiohttp_missing(monkeypatch, tmp_path):
    """If aiohttp is unavailable, the entry points must fail with a clear
    message rather than a bare ImportError."""
    bundle_path = tmp_path / "empty.zip"
    # Build a minimal bundle so _bundle_source succeeds before _serve fails.
    with zipfile.ZipFile(bundle_path, "w") as zf:
        zf.writestr("manifest.json", json.dumps({"format_version": 1}))
        zf.writestr("journal.ndjson", b"")

    import easycat.debugger.server as srv

    def _no_aiohttp(*_a, **_kw):
        raise ImportError("simulated missing aiohttp")

    monkeypatch.setattr(srv, "_make_app", lambda *_a, **_kw: _no_aiohttp())
    with pytest.raises(ImportError):
        srv.serve_bundle(bundle_path, open_browser=False)


def test_debugger_source_session_adapts_live_journal():
    """Live-session source should snapshot the journal each call and pull
    artifact bytes from the session's artifact store."""
    artifact_store = InMemoryArtifactStore()
    journal = InMemoryRingBuffer(capacity=8, artifact_store=artifact_store)
    ref = artifact_store.put(b"hello-world", artifact_class="replay_critical")

    class _StubSession:
        session_id = "stub-1"
        is_running = True
        turn_state = "IDLE"
        _artifact_store = artifact_store

        @property
        def journal(self):
            return journal

    source = _session_source(_StubSession())
    assert source.manifest()["session_id"] == "stub-1"
    assert source.artifact(ref) == b"hello-world"
    # Empty journal returns no records.
    assert source.records() == []
    # Adding a record makes it visible on the next records() call —
    # this is the polling contract live sources rely on.
    journal.append(
        kind=JournalRecordKind.EVENT,
        name="test",
        session_id="stub-1",
        error=ErrorInfo(
            type="ExceptionGroup",
            message="pipeline failed",
            children=(ErrorInfo(type="ValueError", message="bad input"),),
        ),
    )
    records = source.records()
    record = next(r for r in records if r["name"] == "test")
    assert record["error"]["children"][0]["type"] == "ValueError"
    assert record["error"]["children"][0]["message"] == "bad input"


class _ListJournal:
    """Journal stub returning pre-built records for both live and export reads."""

    def __init__(self, records):
        self._records = list(records)

    def read(self, start=0, limit=None):
        out = self._records[start:]
        if limit is not None:
            out = out[:limit]
        return out

    @property
    def latest_sequence(self):
        return self._records[-1].sequence if self._records else 0


class _DualSession:
    """Session stub the live debugger source and the export bundle both read."""

    session_id = "sess-agree"
    is_running = True
    turn_state = "IDLE"
    _artifact_store = None
    _debug = "light"
    _easycat_config = None
    _config = None

    def __init__(self, records):
        self._records = records

    @property
    def journal(self):
        return _ListJournal(self._records)

    @property
    def _journal(self):
        return _ListJournal(self._records)


def test_live_and_bundle_serialization_agree_with_tags_and_subclass_fields(tmp_path):
    """The live debugger view and the exported bundle must serialize the *same*
    record identically — including ``tags`` and record-subclass fields.

    Regression for #28: the server used to walk a hardcoded attribute tuple that
    dropped ``tags`` and ``FrameworkTransitionRecord`` fields
    (``framework`` / ``direction`` / ``bridge_latency_ms``), so a record looked
    narrower live than in a bundle. Both paths now share
    :func:`easycat.debug._serialize.record_to_dict`, so they must agree.
    """
    records = [
        JournalRecord(
            sequence=1,
            session_id="sess-agree",
            kind=JournalRecordKind.EVENT,
            name="stt_final",
            timing=TimingInfo(wall_ns=1, mono_ns=2, cpu_ns=3),
            turn_id="turn-1",
            data={"text": "hello"},
            tags=frozenset({"debug", "runtime"}),
        ),
        FrameworkTransitionRecord(
            sequence=2,
            session_id="sess-agree",
            name="framework_enter",
            timing=TimingInfo(wall_ns=4, mono_ns=5, cpu_ns=6),
            turn_id="turn-1",
            tags=frozenset({"agent"}),
            framework="openai_agents",
            direction="enter",
            bridge_latency_ms=12.5,
        ),
    ]
    session = _DualSession(records)

    # Live debugger serialization (what the /api/records + WS ticks emit).
    live = _session_source(session).records()

    # Export bundle serialization (journal.ndjson lines).
    path = tmp_path / "export.zip"
    export_debug_bundle(session, path)
    with zipfile.ZipFile(path, "r") as zf:
        lines = zf.read("journal.ndjson").decode("utf-8").splitlines()
    bundle = [json.loads(line) for line in lines]

    # The whole point of #28: identical shape live vs. bundle.
    assert live == bundle

    # The server now GAINS the fields it used to drop.
    base, transition = live
    assert base["tags"] == ["debug", "runtime"]
    assert transition["tags"] == ["agent"]
    assert transition["framework"] == "openai_agents"
    assert transition["direction"] == "enter"
    assert transition["bridge_latency_ms"] == 12.5


def test_journal_view_exposes_latest_sequence():
    """``JournalView`` re-exposes the backend's O(1) ``latest_sequence`` so
    live-tailing callers can detect growth without re-reading the journal."""
    from easycat.runtime import JournalView

    journal = InMemoryRingBuffer(capacity=8)
    view = JournalView(journal)
    assert view.latest_sequence == 0
    seq = journal.append(kind=JournalRecordKind.EVENT, name="a", session_id="s")
    assert view.latest_sequence == seq
    journal.append(kind=JournalRecordKind.EVENT, name="b", session_id="s")
    assert view.latest_sequence == seq + 1


def test_session_source_progress_is_cheap_and_tracks_growth():
    """The live source must report growth via the O(1) ``progress()`` probe
    without re-reading and re-serializing the whole journal each tick.

    Regression for the WebSocket busy-poll: it previously called
    ``records()`` (full ``read()`` + ``_record_to_dict`` per record) every
    500ms just to compare a count.
    """
    from easycat.runtime import JournalView

    journal = InMemoryRingBuffer(capacity=8)

    class _StubSession:
        session_id = "stub-progress"
        is_running = True
        turn_state = "IDLE"
        _artifact_store = None

        @property
        def journal(self):
            return JournalView(journal)

    source = _session_source(_StubSession())

    # progress() must not fall back to serializing records: spy on read().
    calls: list[int] = []
    real_read = journal.read

    def _counting_read(*a, **k):
        calls.append(1)
        return real_read(*a, **k)

    journal.read = _counting_read  # type: ignore[method-assign]

    latest_seq, count = source.progress()
    assert (latest_seq, count) == (0, 0)
    journal.append(kind=JournalRecordKind.EVENT, name="a", session_id="s")
    latest_seq, count = source.progress()
    assert latest_seq == 1
    assert count == 1
    # The cheap probe must never have read/serialized the journal.
    assert calls == []


async def test_timeline_helpers_are_shared_with_cli(tmp_path):
    """The waterfall math lives once in ``debug/_turn_timeline``.

    The debugger endpoints and ``easycat bundles show`` / ``inspect``
    must compute identical per-turn spans, and ``turn_waterfall`` decorates
    the shared timeline with milestone deltas for the CLI.
    """
    from easycat.debug import _turn_timeline

    bundle_path = await _build_voice_bundle(tmp_path)
    records = list(RunBundle.load(bundle_path).records())
    timeline = _turn_timeline.build_timeline(records)
    waterfall = _turn_timeline.turn_waterfall(records)

    assert waterfall, "expected at least one turn"
    by_turn = {turn["turn_id"]: turn for turn in timeline}
    for turn in waterfall:
        assert turn["spans"] == by_turn[turn["turn_id"]]["spans"]
        assert set(turn["milestones"]) == {
            "vad_endpoint_to_stt_final_ms",
            "stt_final_to_agent_request_ms",
            "agent_request_to_first_token_ms",
            "agent_first_token_to_tts_first_byte_ms",
            "vad_endpoint_to_tts_first_byte_ms",
            "user_speech_start_to_bot_stopped_ms",
        }
    # A real voice turn reaches TTS, so at least one turn resolves the
    # full VAD endpoint → TTS first byte delta.
    assert any(
        turn["milestones"]["vad_endpoint_to_tts_first_byte_ms"] is not None for turn in waterfall
    )


async def test_api_timeline_carries_per_turn_milestones(tmp_path):
    """``/api/timeline`` now serves ``turn_waterfall``: each turn keeps its
    stage spans (so the SPA waterfall is unaffected) AND a ``milestones``
    block the critical-path panel renders."""
    from aiohttp.test_utils import TestClient, TestServer

    from easycat.debugger.server import _bundle_source, _make_app

    bundle_path = await _build_voice_bundle(tmp_path)
    app = _make_app(_bundle_source(bundle_path))

    async with TestClient(TestServer(app)) as client:
        body = await (await client.get("/api/timeline")).json()

    assert "timeline" in body
    assert body["timeline"], "expected at least one turn"
    for turn in body["timeline"]:
        assert "spans" in turn
        assert set(turn["milestones"]) == {
            "vad_endpoint_to_stt_final_ms",
            "stt_final_to_agent_request_ms",
            "agent_request_to_first_token_ms",
            "agent_first_token_to_tts_first_byte_ms",
            "vad_endpoint_to_tts_first_byte_ms",
            "user_speech_start_to_bot_stopped_ms",
        }
    assert any(
        turn["milestones"]["vad_endpoint_to_tts_first_byte_ms"] is not None
        for turn in body["timeline"]
    )


def test_build_issues_returns_stable_rollup_shape():
    """``/api/issues`` serves the ``{issues, summary, total}`` contract."""
    records = [
        {"sequence": 1, "name": "error", "turn_id": "t1", "error": {"type": "BoomError"}},
        {"sequence": 2, "name": "stt_final", "turn_id": "t1", "data": {"text": ""}},
    ]
    report = _build_issues(records)
    assert set(report) == {"issues", "summary", "total"}
    assert report["total"] == 2
    assert report["summary"] == {"error": 1, "warning": 1, "info": 0}
    # Errors sort before warnings.
    assert [issue["severity"] for issue in report["issues"]] == ["error", "warning"]


def test_filter_records_negative_offset_raises():
    with pytest.raises(ValueError, match="offset"):
        _filter_records(
            [],
            stage=None,
            turn_id=None,
            name=None,
            from_seq=None,
            to_seq=None,
            offset=-1,
        )


def test_filter_records_zero_limit_raises():
    with pytest.raises(ValueError, match="limit"):
        _filter_records(
            [],
            stage=None,
            turn_id=None,
            name=None,
            from_seq=None,
            to_seq=None,
            limit=0,
        )


def _search_sample_records() -> list[dict]:
    return [
        {"sequence": 1, "name": "stt_final", "turn_id": "t1", "data": {"text": "hello world"}},
        {
            "sequence": 2,
            "name": "agent_error",
            "turn_id": "t1",
            "data": {},
            "error": {"type": "TimeoutError", "message": "request timed out"},
        },
        {"sequence": 3, "name": "tts_frame", "turn_id": "t2", "data": {"codec": "pcm"}},
    ]


def test_search_records_matches_data_substring():
    matched, truncated = _search_records(_search_sample_records(), query="hello")
    assert [r["sequence"] for r in matched] == [1]
    assert matched[0]["_match_fields"] == ["data"]
    assert truncated is False


def test_search_records_matches_error_fields():
    matched, _ = _search_records(_search_sample_records(), query="timed out")
    assert [r["sequence"] for r in matched] == [2]
    assert matched[0]["_match_fields"] == ["error"]


def test_search_records_matches_name_and_turn():
    matched, _ = _search_records(_search_sample_records(), query="t1")
    assert [r["sequence"] for r in matched] == [1, 2]
    assert all("turn_id" in r["_match_fields"] for r in matched)


def test_search_records_regex_matches():
    matched, _ = _search_records(_search_sample_records(), query="timeout|pcm", use_regex=True)
    assert [r["sequence"] for r in matched] == [2, 3]


def test_search_records_invalid_regex_raises():
    with pytest.raises(ValueError, match="invalid regex"):
        _search_records(_search_sample_records(), query="[", use_regex=True)


def test_search_records_rejects_catastrophic_regex():
    records = [{"sequence": 1, "name": "a" * 32 + "!", "data": {}}]
    # Build the nested-repeat pattern from runtime values so the test source
    # holds no static catastrophic-backtracking regex literal for analyzers to
    # flag. ``reps`` is opaque to static analysis but equals 1, yielding
    # ``(a+)+$``.
    reps = len(records)
    nested_repeat = "(a" + "+" * reps + ")" + "+" * reps + "$"
    with pytest.raises(ValueError, match="unsafe regex"):
        _search_records(records, query=nested_repeat, use_regex=True)


def test_search_records_rejects_quantified_alternation_regex():
    records = [{"sequence": 1, "name": "a" * 32 + "!", "data": {}}]
    # Constructed at runtime (see above) to avoid a static ReDoS literal;
    # equivalent to ``(a|aa)+$``.
    reps = len(records)
    alternation = "(a|" + "a" * (reps + 1) + ")" + "+" * reps + "$"
    with pytest.raises(ValueError, match="unsafe regex"):
        _search_records(records, query=alternation, use_regex=True)


def test_search_records_rejects_adjacent_optional_repeats():
    # ``a?`` repeated then an equal run of required ``a`` -- the classic
    # exponential-backtracking shape -- where the repeats are adjacent siblings
    # rather than nested. Built from runtime multiplication so no static ReDoS
    # literal exists in the source; equivalent to ``"a?" * 24 + "a" * 24``.
    n = 24 * len("x")
    pattern = "a?" * n + "a" * n
    with pytest.raises(ValueError, match="unsafe regex"):
        _compile_search_regex(pattern)


def test_search_records_rejects_overlong_query():
    too_long = "a" * (_SEARCH_MAX_QUERY_LEN + 1)
    with pytest.raises(ValueError, match="query too long"):
        _search_records(_search_sample_records(), query=too_long, use_regex=True)
    with pytest.raises(ValueError, match="query too long"):
        _search_records(_search_sample_records(), query=too_long)


def test_search_records_empty_query_matches_nothing():
    matched, truncated = _search_records(_search_sample_records(), query="")
    assert matched == []
    assert truncated is False


def test_search_records_errors_only():
    # ``t1`` appears in two records but only the error one survives ``errors_only``.
    matched, _ = _search_records(_search_sample_records(), query="t1", errors_only=True)
    assert [r["sequence"] for r in matched] == [2]


def test_search_records_does_not_mutate_source():
    records = _search_sample_records()
    _search_records(records, query="hello")
    assert all("_match_fields" not in r for r in records)


def test_search_records_sets_scan_truncated_past_limit():
    big = [
        {"sequence": i, "name": "x", "data": {"k": "match"}} for i in range(_SEARCH_SCAN_LIMIT + 5)
    ]
    matched, truncated = _search_records(big, query="match")
    assert truncated is True
    assert len(matched) == _SEARCH_SCAN_LIMIT


def test_summarise_turns_dedupes_t3_8_interrupt_fanout():
    """T3.8 fans an InterruptSignal across 8 stages, so a single
    barge-in would appear as 9 interruptions (8 control_signal + 1
    legacy interruption event).  ``_summarise_turns`` must dedupe by
    ``signal_id`` and report 1.
    """
    sig_id = "barge-1"
    records = [
        {
            "sequence": 1,
            "turn_id": "t1",
            "name": "interruption",
            "data": {},
            "timing": {"wall_ns": 1_000_000},
        },
    ]
    # Fan-out from T3.8: one control_signal per stage, all sharing
    # the same signal_id.
    for i, stage in enumerate(
        ("transport", "tts", "agent", "turn", "stt", "vad", "audio", "telephony")
    ):
        records.append(
            {
                "sequence": 2 + i,
                "turn_id": "t1",
                "kind": "control",
                "name": "control_signal",
                "data": {
                    "stage": stage,
                    "observed_stage": stage,
                    "signal_kind": "interrupt",
                    "signal_id": sig_id,
                },
                "timing": {"wall_ns": (2 + i) * 1_000_000},
            }
        )
    out = _summarise_turns(records)
    assert len(out) == 1
    assert out[0]["interruption_count"] == 1


def test_summarise_turns_counts_legacy_interruption_when_no_signals():
    """Older bundles have only the legacy ``interruption`` event with
    no ``control_signal`` records.  The counter should still find them."""
    records = [
        {
            "sequence": 1,
            "turn_id": "t1",
            "name": "interruption",
            "data": {},
            "timing": {"wall_ns": 1_000_000},
        },
    ]
    out = _summarise_turns(records)
    assert len(out) == 1
    assert out[0]["interruption_count"] == 1


def test_build_timeline_skips_control_signal_records_for_spans():
    """``control_signal`` from a barge-in shouldn't generate synthetic
    instant-spans for stages that had no real pipeline activity in
    that turn (e.g. telephony in a pure-WS session)."""
    records = [
        {
            "sequence": 1,
            "turn_id": "t1",
            "name": "stage_start",
            "data": {"stage": "tts"},
            "timing": {"wall_ns": 1_000_000},
        },
        {
            "sequence": 2,
            "turn_id": "t1",
            "name": "stage_complete",
            "data": {"stage": "tts"},
            "timing": {"wall_ns": 5_000_000},
        },
        # Barge-in fan-out reaches telephony but it had no real activity.
        {
            "sequence": 3,
            "turn_id": "t1",
            "kind": "control",
            "name": "control_signal",
            "data": {
                "stage": "telephony",
                "observed_stage": "telephony",
                "signal_kind": "interrupt",
                "signal_id": "barge-1",
            },
            "timing": {"wall_ns": 6_000_000},
        },
    ]
    timeline = _build_timeline(records)
    assert len(timeline) == 1
    stage_names = [s["stage"] for s in timeline[0]["spans"]]
    assert "tts" in stage_names
    assert "telephony" not in stage_names
