zyt/app/tests/test_video_contract.py

from __future__ import annotations

import logging
import sys
import threading
import time
from pathlib import Path
from types import SimpleNamespace

import pytest

PROJECT_ROOT = Path(__file__).resolve().parents[1]
SOURCE_ROOT = PROJECT_ROOT / "src"
if str(SOURCE_ROOT) not in sys.path:
    sys.path.insert(0, str(SOURCE_ROOT))

from doctor_workstation.video.launcher import (  # noqa: E402
    BackendMode,
    VideoCallLauncher,
    VideoCallRequest,
    VideoTicketError,
    normalize_backend_ticket,
)
from doctor_workstation.video.lifecycle import OrderedCallLifecycle  # noqa: E402
from doctor_workstation.video.security import (  # noqa: E402
    TrustedDocumentError,
    TrustedDocumentPolicy,
)


def test_companion_uses_legacy_safe_transcription_session_identity() -> None:
    """Generated session IDs stay below 32 chars so upgraded databases cannot truncate."""

    source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text(
        encoding="utf-8"
    )
    function_source = source.split("function newTranscriptionSessionId", 1)[1].split(
        "function requestTranscriptionStart", 1
    )[0]

    assert "replaceAll('-', '')" in function_source
    assert ".slice(0, 28)" in function_source
    assert "return `tr-${" in function_source


def test_companion_archives_cloud_video_local_mixed_audio_and_transcript() -> None:
    """A connected call starts three independent artifacts before hangup."""

    source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text(
        encoding="utf-8"
    )

    assert "context.createMediaStreamDestination()" in source
    assert "cloud.getAudioTrack({ processed: true })" in source
    assert "userId: activeConfig.targetUserId" in source
    assert "new MediaRecorder(destination.stream" in source
    assert "recorder.start(1000)" in source
    assert "bridge.startLocalAudioRecording(sessionId, mimeType)" in source
    assert "bridge.appendLocalAudioChunk(" in source
    assert "bridge.finishLocalAudioRecording(sessionId, totalBytes)" in source
    assert "operations.push(stopLocalRecording())" in source
    assert "operations.push(stopTranscription('completed'))" in source
    assert "Promise.allSettled(operations)" in source
    assert "腾讯云混流视频、本机语音录音和实时转写均已启动" in source


def test_companion_watches_room_id_for_the_entire_call_cycle() -> None:
    """A slowly-created TRTC room must still bind to the exact call record."""

    source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text(
        encoding="utf-8"
    )
    room_source = source.split("function readRoomId", 1)[1].split(
        "function handleStatusChanged", 1
    )[0]

    assert "TUIStore.watch(StoreName.CALL, roomIdWatchOptions)" in room_source
    assert "[NAME.ROOM_ID]: handleRoomIdChanged" in room_source
    assert "cycle === callCycleGeneration" in room_source
    assert "while (activeConfig && !endNotified" in room_source
    assert "attempt < 40" not in room_source
    assert "diagnosisId: activeConfig.diagnosisId" in room_source


def test_room_binding_is_acknowledged_and_transcriber_room_is_a_fallback() -> None:
    companion_source = (
        PROJECT_ROOT / "video_companion" / "src" / "main.ts"
    ).read_text(encoding="utf-8")
    window_source = (
        PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "window.py"
    ).read_text(encoding="utf-8")

    assert "observeRoomId(roomId)" in companion_source
    assert "onRealtimeTranscriberStarted: (roomId)" in companion_source
    assert "function roomBindingResult(" in companion_source
    assert "if (boundRoomId) return" in companion_source
    assert "roomBindingResult?.(" in window_source


def test_im_conversation_renders_deduplicated_video_call_status_timeline() -> None:
    """Video lifecycle feedback belongs in the IM timeline as local status events."""

    companion_root = PROJECT_ROOT / "video_companion" / "src"
    source = (companion_root / "main.ts").read_text(encoding="utf-8")
    app_source = (companion_root / "App.vue").read_text(encoding="utf-8")
    styles = (companion_root / "style.css").read_text(encoding="utf-8")

    timeline_source = source.split("function appendVideoCallStatus", 1)[1].split(
        "function onMessageReceived", 1
    )[0]
    assert "activeConfig.mode !== 'chat'" in timeline_source
    assert "local-video-call-${callCycleGeneration}-${callStatus}" in timeline_source
    assert "findIndex((item) => item.id === id)" in timeline_source
    assert "appendVideoCallStatus('starting', '正在创建安全视频通话')" in source
    assert "appendVideoCallStatus('dialing', '正在呼叫患者')" in source
    assert "appendVideoCallStatus('connected', '视频通话已接通')" in source
    assert "appendVideoCallStatus('ended', '视频通话已结束')" in source
    assert "appendVideoCallStatus('failed', `视频通话发起失败：${message}`)" in source
    assert "message.type === 'call-status'" in app_source
    assert 'class="call-status-event"' in app_source
    assert 'role="status"' in app_source
    assert "IM 已连接 · ${props.statusText.value}" in app_source
    assert ".message-row--call-status" in styles
    assert ".call-status-event--connected" in styles
    assert ".call-status-event--failed" in styles


def test_companion_local_recording_waits_for_real_audio_and_has_runtime_fallbacks() -> None:
    source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text(
        encoding="utf-8"
    )

    assert "querySelectorAll<HTMLMediaElement>('video, audio')" in source
    assert "stream.getAudioTracks()" in source
    assert "navigator.mediaDevices.getUserMedia" in source
    assert "await waitForCallAudioTracks(cloud, sessionId)" in source
    assert "localRecordingAttachedSourceCount <= 0" in source
    assert "localRecordingBytes < 1024" in source
    assert "已阻止上传空文件" in source


def test_qt_close_waits_for_local_audio_finish_before_destroying_webengine() -> None:
    """A title-bar/desktop hangup must keep accepting bridge chunks until COS ack."""

    source = (
        PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "window.py"
    ).read_text(encoding="utf-8")
    request_shutdown = source.split(
        "def _request_companion_shutdown", 1
    )[1].split("def _force_requested_shutdown", 1)[0]
    begin_shutdown = source.split("def _begin_shutdown", 1)[1].split(
        "def wait_for_lifecycles", 1
    )[0]
    close_event = source.split("def closeEvent", 1)[1].split(
        "else:", 1
    )[0]

    assert "window.doctorConsultation?.close?.()" in request_shutdown
    assert "self._shutdown_requested = True" in request_shutdown
    assert "self._closing = True" not in request_shutdown
    assert "event.ignore()" in close_event
    assert "self._request_companion_shutdown" in close_event
    assert "self._shutdown_timer.stop()" in begin_shutdown
    assert "window.doctorConsultation?.close?.()" not in begin_shutdown
    assert "if not self.open_im or self._shutdown_requested" in source


def test_local_audio_capture_keeps_and_persists_its_own_call_room_identity() -> None:
    """A later IM call cycle must not relabel an earlier recording."""

    source = (
        PROJECT_ROOT / "src" / "doctor_workstation" / "video" / "window.py"
    ).read_text(encoding="utf-8")

    assert "lifecycle: OrderedCallLifecycle" in source
    assert "lifecycle = capture.lifecycle" in source
    assert "call_record_id=call_record_id" in source
    assert 'room_id=lifecycle.current_room_id or ""' in source
    assert "store.bind_identity(" in source


def test_companion_shows_incremental_subtitles_but_only_persists_final_segments() -> None:
    main_source = (
        PROJECT_ROOT / "video_companion" / "src" / "main.ts"
    ).read_text(encoding="utf-8")
    component_source = (
        PROJECT_ROOT / "video_companion" / "src" / "App.vue"
    ).read_text(encoding="utf-8")
    handler = main_source.split("function handleTranscriberMessage", 1)[1].split(
        "function subscribeTranscriber", 1
    )[0]

    assert "showLiveCaption(message)" in handler
    assert "if (message.isCompleted !== true) return" in handler
    assert handler.index("showLiveCaption(message)") < handler.index(
        "if (message.isCompleted !== true) return"
    )
    assert 'aria-label="实时语音字幕"' in component_source
    assert "liveCaptions.value" in component_source
    assert "caption.speaker" in component_source
    assert "caption.text" in component_source


def test_companion_screenshot_requires_doctor_confirmation_before_upload() -> None:
    source = (PROJECT_ROOT / "video_companion" / "src" / "App.vue").read_text(
        encoding="utf-8"
    )
    capture = source.split("async function captureScreenshot", 1)[1].split(
        "function discardScreenshot", 1
    )[0]
    confirm = source.split("async function confirmScreenshot", 1)[1].split(
        "watch(", 1
    )[0]

    assert "screenshotPreview.value = canvas.toDataURL" in capture
    assert "onSaveScreenshot" not in capture
    assert "await props.onSaveScreenshot(screenshotPreview.value)" in confirm
    assert "确认画面后再保存到患者资料" in source
    assert "确认并上传" in source
    assert "取消" in source


def test_companion_loads_im_history_without_an_empty_first_page_cursor() -> None:
    source = (PROJECT_ROOT / "video_companion" / "src" / "main.ts").read_text(
        encoding="utf-8"
    )
    load_messages = source.split("async function loadMessages", 1)[1].split(
        "async function sendText", 1
    )[0]

    assert "nextReqMessageID: prepend ? nextReqMessageID : ''" not in load_messages
    assert "nextReqMessageID?: string" in load_messages
    assert (
        "if (prepend && nextReqMessageID) request.nextReqMessageID = nextReqMessageID"
        in load_messages
    )
    assert "chat.getMessageList(request)" in load_messages


def test_companion_preserves_im_scroll_position_for_live_and_older_messages() -> None:
    source = (PROJECT_ROOT / "video_companion" / "src" / "App.vue").read_text(
        encoding="utf-8"
    )
    watcher = source.split("() => props.messages.value.length", 1)[1].split(
        "watch(", 1
    )[0]
    load_earlier = source.split("async function loadEarlierMessages", 1)[1].split(
        "async function runAction", 1
    )[0]

    assert "if (!stickToMessageBottom.value) return" in watcher
    assert "stickToMessageBottom.value = false" in load_earlier
    assert "container.scrollHeight - previousHeight" in load_earlier
    assert '@scroll="handleMessageScroll"' in source
    assert '@click="loadEarlierMessages"' in source


def test_normalizes_admin_ticket_aliases_to_companion_contract() -> None:
    request = normalize_backend_ticket(
        {
            "sdkAppId": "1400123456",
            "userId": " doctor_42 ",
            "userSig": "short-lived-ticket",
            "patientUserId": " patient_8 ",
            "diagnosisId": 123,
            "patientId": 8,
        }
    )

    assert request == VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        patient_id=8,
    )
    assert request.to_web_config() == {
        "SDKAppID": 1400123456,
        "userID": "doctor_42",
        "userSig": "short-lived-ticket",
        "targetUserId": "patient_8",
        "diagnosisId": 123,
    }


def test_accepts_uppercase_aliases_and_nested_backend_envelope() -> None:
    request = VideoCallRequest.from_backend_ticket(
        {
            "data": {
                "SDKAppID": 1400123456,
                "userID": "doctor_42",
                "userSig": "ticket-value",
                "targetUserId": "patient_8",
            }
        },
        diagnosis_id="diagnosis-123",
        patient_id=8,
        backend_mode="embedded",
    )

    assert request.diagnosis_id == "diagnosis-123"
    assert request.backend_mode is BackendMode.EMBEDDED


def test_accepts_repository_call_ticket_object_without_importing_core_models() -> None:
    ticket = SimpleNamespace(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="ticket-value",
        patient_user_id="patient_8",
        diagnosis_id=123,
        raw={"sdkAppId": 1400123456},
    )

    request = normalize_backend_ticket(ticket, patient_id=8)

    assert request.patient_id == 8
    assert request.to_web_config()["targetUserId"] == "patient_8"


def test_secret_is_excluded_from_repr_and_safe_log_context() -> None:
    request = normalize_backend_ticket(
        {
            "sdkAppId": 1400123456,
            "userId": "doctor_42",
            "userSig": "never-write-this-value",
            "patientUserId": "patient_8",
        },
        diagnosis_id=123,
    )

    assert "never-write-this-value" not in repr(request)
    assert "never-write-this-value" not in str(request.safe_log_context())
    assert "user_sig" not in request.safe_log_context()


@pytest.mark.parametrize("forbidden_key", ["SDKSecretKey", "sdk_secret_key", "secretKey"])
def test_rejects_server_side_secret_material(forbidden_key: str) -> None:
    with pytest.raises(VideoTicketError, match="forbidden server-side secret"):
        normalize_backend_ticket(
            {
                "sdkAppId": 1400123456,
                "userId": "doctor_42",
                "userSig": "ticket-value",
                "patientUserId": "patient_8",
                "diagnosisId": 123,
                forbidden_key: "must-never-reach-a-client",
            }
        )


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("sdkAppId", 0),
        ("userId", ""),
        ("userSig", ""),
        ("patientUserId", " "),
        ("diagnosisId", None),
    ],
)
def test_rejects_incomplete_or_invalid_ticket(field: str, value: object) -> None:
    ticket: dict[str, object] = {
        "sdkAppId": 1400123456,
        "userId": "doctor_42",
        "userSig": "ticket-value",
        "patientUserId": "patient_8",
        "diagnosisId": 123,
    }
    ticket[field] = value

    with pytest.raises(VideoTicketError):
        normalize_backend_ticket(ticket)


def test_rejects_conflicting_aliases_and_modes() -> None:
    with pytest.raises(VideoTicketError, match="conflicting SDKAppID aliases"):
        normalize_backend_ticket(
            {
                "SDKAppID": 1400123456,
                "sdkAppId": 1400654321,
                "userID": "doctor_42",
                "userSig": "ticket-value",
                "targetUserId": "patient_8",
                "diagnosisId": 123,
            }
        )

    with pytest.raises(VideoTicketError, match="backend mode"):
        BackendMode.parse("native")


def test_launcher_rejects_browser_before_importing_window_or_writing_repository() -> None:
    class Repository:
        def start_call(self, **payload: object) -> None:
            raise AssertionError(f"unexpected repository write: {payload}")

    launcher = VideoCallLauncher(repository=Repository(), backend_mode="browser")
    with pytest.raises(VideoTicketError, match="one-time handoff"):
        launcher.prepare(
            {
                "sdkAppId": 1400123456,
                "userId": "doctor_42",
                "userSig": "ticket-value",
                "patientUserId": "patient_8",
                "diagnosisId": 123,
            }
        )


def test_call_lifecycle_is_fifo_daemon_and_never_blocks_caller() -> None:
    events: list[tuple[object, ...]] = []
    start_entered = threading.Event()
    release_start = threading.Event()

    class Repository:
        def start_call(
            self, diagnosis_id: int, patient_id: int, *, call_type: int
        ) -> dict[str, int]:
            start_entered.set()
            assert release_start.wait(2)
            events.append(("start", diagnosis_id, patient_id, call_type))
            return {"call_record_id": 900}

        def bind_call_room(
            self, diagnosis_id: int, room_id: str, *, call_record_id: int
        ) -> None:
            events.append(("bind", diagnosis_id, room_id, call_record_id))

        def end_call(self, diagnosis_id: int, *, call_record_id: int) -> None:
            events.append(("end", diagnosis_id, call_record_id))

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        patient_id=8,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    started_at = time.monotonic()
    start_future = lifecycle.start()
    bind_future = lifecycle.bind_room("456789")
    duplicate_bind = lifecycle.bind_room("456789")
    changed_bind = lifecycle.bind_room("another-room")
    end_future = lifecycle.end("test")
    elapsed = time.monotonic() - started_at

    assert start_entered.wait(1)
    assert elapsed < 0.2
    assert lifecycle.worker_is_daemon is True
    assert lifecycle.wait(0.01) is False
    assert duplicate_bind is bind_future
    assert changed_bind.result(timeout=0) is False
    assert lifecycle.current_room_id == "456789"

    release_start.set()
    assert start_future.result(timeout=2) is True
    assert bind_future.result(timeout=2) is True
    assert end_future.result(timeout=2) is True
    assert lifecycle.wait(1) is True
    assert lifecycle.current_room_id == "456789"

    assert events == [
        ("start", 123, 8, 2),
        ("bind", 123, "456789", 900),
        ("end", 123, 900),
    ]


def test_transcription_lifecycle_uses_start_record_id_and_remains_fifo() -> None:
    events: list[tuple[object, ...]] = []

    class Repository:
        def start_call(
            self, diagnosis_id: int, patient_id: int, *, call_type: int
        ) -> dict[str, object]:
            events.append(("start", diagnosis_id, patient_id, call_type))
            return {"data": {"callRecordId": 901}}

        def start_call_transcription(
            self,
            diagnosis_id: int,
            call_record_id: int,
            transcription_session_id: str,
            *,
            language: str,
        ) -> None:
            events.append(
                (
                    "transcription-start",
                    diagnosis_id,
                    call_record_id,
                    transcription_session_id,
                    language,
                )
            )

        def upsert_call_transcript_segments(
            self,
            diagnosis_id: int,
            call_record_id: int,
            transcription_session_id: str,
            segments: list[dict[str, object]],
        ) -> None:
            events.append(
                (
                    "segment",
                    diagnosis_id,
                    call_record_id,
                    transcription_session_id,
                    segments,
                )
            )

        def finish_call_transcription(
            self,
            diagnosis_id: int,
            call_record_id: int,
            transcription_session_id: str,
            expected_segment_count: int,
            *,
            status: str,
        ) -> None:
            events.append(
                (
                    "finish",
                    diagnosis_id,
                    call_record_id,
                    transcription_session_id,
                    expected_segment_count,
                    status,
                )
            )

        def end_call(self, diagnosis_id: int) -> None:
            events.append(("end", diagnosis_id))

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        patient_id=8,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    start = lifecycle.start()
    transcription_start = lifecycle.start_transcription("session-1")
    segment = lifecycle.save_transcript_segment(
        {
            "segment_id": "seg-1",
            "speaker_user_id": "patient_8",
            "speaker_role": "patient",
            "timestamp": 1200,
            "text": "patient words",
        }
    )
    duplicate = lifecycle.save_transcript_segment(
        {"segment_id": "seg-1", "text": "must not produce another write"}
    )
    finish = lifecycle.finish_transcription(status="completed")
    end = lifecycle.end("test")

    assert duplicate is segment
    assert start.result(timeout=2) is True
    assert transcription_start.result(timeout=2) is True
    assert segment.result(timeout=2) is True
    assert finish.result(timeout=2) is True
    assert end.result(timeout=2) is True
    assert lifecycle.wait(1) is True
    assert lifecycle.call_record_id == 901
    assert events == [
        ("start", 123, 8, 2),
        ("transcription-start", 123, 901, "session-1", "zh-CN"),
        (
            "segment",
            123,
            901,
            "session-1",
            [
                {
                    "segment_id": "seg-1",
                    "transcription_session_id": "session-1",
                    "speaker_user_id": "patient_8",
                    "speaker_role": "patient",
                    "timestamp": 1200,
                    "text": "patient words",
                }
            ],
        ),
        ("finish", 123, 901, "session-1", 1, "completed"),
        ("end", 123),
    ]


def test_end_auto_finishes_active_transcription_as_partial() -> None:
    events: list[tuple[object, ...]] = []

    class Repository:
        def start_call(
            self, diagnosis_id: int, patient_id: int, *, call_type: int
        ) -> dict[str, int]:
            events.append(("start", diagnosis_id, patient_id, call_type))
            return {"call_record_id": 902}

        def start_call_transcription(
            self,
            diagnosis_id: int,
            call_record_id: int,
            transcription_session_id: str,
            *,
            language: str,
        ) -> None:
            events.append(
                (
                    "transcription-start",
                    diagnosis_id,
                    call_record_id,
                    transcription_session_id,
                    language,
                )
            )

        def finish_call_transcription(
            self,
            diagnosis_id: int,
            call_record_id: int,
            transcription_session_id: str,
            expected_segment_count: int,
            *,
            status: str,
        ) -> None:
            events.append(
                (
                    "finish",
                    diagnosis_id,
                    call_record_id,
                    transcription_session_id,
                    expected_segment_count,
                    status,
                )
            )

        def end_call(self, diagnosis_id: int) -> None:
            events.append(("end", diagnosis_id))

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        patient_id=8,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    lifecycle.start()
    lifecycle.start_transcription("session-auto-partial")
    ended = lifecycle.end("window-closed")

    assert ended.result(timeout=2) is True
    assert lifecycle.wait(1) is True
    assert events == [
        ("start", 123, 8, 2),
        ("transcription-start", 123, 902, "session-auto-partial", "zh-CN"),
        ("finish", 123, 902, "session-auto-partial", 0, "partial"),
        ("end", 123),
    ]


def test_failed_start_prevents_bind_and_end_writes() -> None:
    events: list[str] = []

    class Repository:
        def start_call(self, diagnosis_id: int, *, call_type: int) -> None:
            del diagnosis_id, call_type
            events.append("start")
            raise RuntimeError("backend unavailable")

        def bind_call_room(self, diagnosis_id: int, room_id: str) -> None:
            del diagnosis_id, room_id
            events.append("bind")

        def end_call(self, diagnosis_id: int) -> None:
            del diagnosis_id
            events.append("end")

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))
    start_future = lifecycle.start()
    bind_future = lifecycle.bind_room("456789")
    end_future = lifecycle.end("test")

    with pytest.raises(RuntimeError, match="backend unavailable"):
        start_future.result(timeout=1)
    assert bind_future.result(timeout=1) is False
    assert end_future.result(timeout=1) is False
    assert lifecycle.wait(1) is True
    assert events == ["start"]


def test_explicit_cos_recording_failure_fails_room_binding_without_losing_call_identity() -> None:
    class Repository:
        def start_call(
            self, diagnosis_id: int, patient_id: int, *, call_type: int
        ) -> dict[str, int]:
            del diagnosis_id, patient_id, call_type
            return {"call_record_id": 904}

        def bind_call_room(
            self, diagnosis_id: int, room_id: str, *, call_record_id: int
        ) -> dict[str, object]:
            assert (diagnosis_id, room_id, call_record_id) == (123, "456789", 904)
            return {
                "call_record_id": 904,
                "cloud_recording": {
                    "started": False,
                    "message": "COS bucket is unavailable",
                },
            }

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        patient_id=8,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    with pytest.raises(RuntimeError, match="COS bucket is unavailable"):
        lifecycle.bind_room("456789").result(timeout=2)

    assert lifecycle.call_record_id == 904
    assert lifecycle.bound_room_id is None
    assert lifecycle.wait(1) is True


def test_failed_room_binding_releases_claim_and_can_retry_same_room() -> None:
    bind_attempts = 0

    class Repository:
        def start_call(
            self, diagnosis_id: int, patient_id: int, *, call_type: int
        ) -> dict[str, int]:
            del diagnosis_id, patient_id, call_type
            return {"call_record_id": 906}

        def bind_call_room(
            self, diagnosis_id: int, room_id: str, *, call_record_id: int
        ) -> dict[str, object]:
            nonlocal bind_attempts
            assert (diagnosis_id, room_id, call_record_id) == (123, "456789", 906)
            bind_attempts += 1
            if bind_attempts == 1:
                raise RuntimeError("temporary bind failure")
            return {
                "call_record_id": 906,
                "cloud_recording": {"started": True},
            }

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        patient_id=8,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    with pytest.raises(RuntimeError, match="temporary bind failure"):
        lifecycle.bind_room("456789").result(timeout=2)

    assert lifecycle.current_room_id is None
    assert lifecycle.bind_room("456789").result(timeout=2) is True
    assert lifecycle.bound_room_id == "456789"
    assert lifecycle.current_room_id == "456789"
    assert bind_attempts == 2
    assert lifecycle.wait(1) is True


def test_local_audio_upload_uses_exact_started_record_and_precedes_end(
    tmp_path: Path,
) -> None:
    events: list[tuple[object, ...]] = []

    class Repository:
        def start_call(
            self, diagnosis_id: int, patient_id: int, *, call_type: int
        ) -> dict[str, int]:
            events.append(("start", diagnosis_id, patient_id, call_type))
            return {"call_record_id": 905}

        def upload_call_recording(
            self,
            path: Path,
            diagnosis_id: int,
            *,
            call_record_id: int,
            mime_type: str,
        ) -> dict[str, object]:
            events.append(
                (
                    "local-audio",
                    path.read_bytes(),
                    diagnosis_id,
                    call_record_id,
                    mime_type,
                )
            )
            return {
                "completed": True,
                "call_record_id": call_record_id,
                "media_kind": "local_audio",
            }

        def end_call(self, diagnosis_id: int, *, call_record_id: int) -> None:
            events.append(("end", diagnosis_id, call_record_id))

    recording = tmp_path / "call-audio.webm"
    recording.write_bytes(b"opus-webm-audio")
    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        patient_id=8,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    lifecycle.start()
    uploaded = lifecycle.save_local_audio_recording(
        recording,
        mime_type="audio/webm;codecs=opus",
    )
    ended = lifecycle.end("doctor-hangup")

    assert uploaded.result(timeout=2) is True
    assert ended.result(timeout=2) is True
    assert lifecycle.wait(1) is True
    assert events == [
        ("start", 123, 8, 2),
        ("local-audio", b"opus-webm-audio", 123, 905, "audio/webm;codecs=opus"),
        ("end", 123, 905),
    ]


def test_video_screenshot_is_uploaded_and_appended_to_patient_tongue_images() -> None:
    events: list[tuple[object, ...]] = []

    class Repository:
        def start_call(self, diagnosis_id: int, *, call_type: int) -> dict[str, int]:
            events.append(("start", diagnosis_id, call_type))
            return {"call_record_id": 903}

        def upload_material_bytes(
            self,
            content: bytes,
            filename: str,
            material_type: str,
            cid: int = 0,
        ) -> str:
            events.append(("upload", content, filename, material_type, cid))
            return "/uploads/image/callshot-123.jpg"

        def add_doctor_note(
            self,
            diagnosis_id: int,
            content: str,
            tongue_images: list[str],
        ) -> None:
            events.append(("note", diagnosis_id, content, tongue_images))

        def end_call(self, diagnosis_id: int) -> None:
            events.append(("end", diagnosis_id))

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    lifecycle.start()
    screenshot = lifecycle.save_screenshot(b"jpeg-frame", "callshot-123.jpg")
    lifecycle.end("test")

    assert screenshot.result(timeout=2) == "/uploads/image/callshot-123.jpg"
    assert lifecycle.wait(1) is True
    assert events == [
        ("start", 123, 2),
        ("upload", b"jpeg-frame", "callshot-123.jpg", "image", 0),
        ("note", 123, "", ["/uploads/image/callshot-123.jpg"]),
        ("end", 123),
    ]


def test_start_rejects_missing_record_id_without_using_ticket_fallback() -> None:
    class Repository:
        def start_call(self, diagnosis_id: int, *, call_type: int) -> dict[str, object]:
            del diagnosis_id, call_type
            return {}

    request = VideoCallRequest(
        sdk_app_id=1400123456,
        user_id="doctor_42",
        user_sig="short-lived-ticket",
        target_user_id="patient_8",
        diagnosis_id=123,
        call_record_id=77,
    )
    lifecycle = OrderedCallLifecycle(request, Repository(), logging.getLogger(__name__))

    with pytest.raises(ValueError, match="startCall response did not include"):
        lifecycle.start().result(timeout=2)

    assert lifecycle.started is False


def test_https_document_policy_is_exact_and_origin_scoped() -> None:
    policy = TrustedDocumentPolicy.from_url(
        "https://RTC.Example.com/doctor-call/index.html?tenant=a#boot",
        is_local=False,
    )

    assert policy.allows_main_document(
        "https://rtc.example.com:443/doctor-call/index.html?tenant=a#ready"
    )
    assert not policy.allows_main_document(
        "https://rtc.example.com/doctor-call/index.html?tenant=b"
    )
    assert not policy.allows_main_document("https://rtc.example.com/other/index.html?tenant=a")
    assert policy.allows_origin("https://rtc.example.com")
    assert not policy.allows_origin("https://sub.rtc.example.com")
    assert not policy.allows_origin("http://rtc.example.com")

    with pytest.raises(TrustedDocumentError, match="HTTPS"):
        TrustedDocumentPolicy.from_url("http://rtc.example.com/doctor-call", is_local=False)


def test_local_document_policy_rejects_sibling_files(tmp_path: Path) -> None:
    index = tmp_path / "dist" / "index.html"
    index.parent.mkdir()
    index.touch()
    sibling = index.with_name("other.html")
    sibling.touch()
    policy = TrustedDocumentPolicy.from_url(index.as_uri(), is_local=True)

    assert policy.allows_main_document(index.as_uri())
    assert not policy.allows_main_document(sibling.as_uri())
    assert policy.allows_origin("file:///")