"""Phase 4: listen-window clock + deferred quantitative assessment."""

import asyncio
from types import SimpleNamespace

from core.app_runtime.assessment import (
    build_metrics,
    clamp_listen_ratio,
    listen_window_ms,
    resolve_expected_text,
    text_similarity_v0,
)
from core.app_runtime.session import TaskSession, begin_session


CLIPS = [
    "/english_learning/shadowing/lessons/unit01/01.mp3",
    "/english_learning/shadowing/lessons/unit01/02.mp3",
]


def test_resolve_expected_text_from_transcripts_not_path():
    task = {
        "transcripts": {"01.mp3": "Hello, how are you today?", "01": "Hello, how are you today?"},
        "sentences": ["/english_learning/shadowing/lessons/unit01/01.mp3"],
    }
    expected = resolve_expected_text(task, CLIPS, 0)
    assert expected == "Hello, how are you today?"
    assert not expected.endswith(".mp3")
    assert resolve_expected_text(task, CLIPS, 0) != CLIPS[0]


def test_resolve_expected_text_rejects_path_as_original():
    task = {
        "transcripts": {"01.mp3": "/english_learning/shadowing/lessons/unit01/01.mp3"},
        "sentences": [CLIPS[0]],
    }
    assert resolve_expected_text(task, CLIPS, 0) is None


def test_resolve_expected_text_from_cache_txt(tmp_path):
    (tmp_path / "01.txt").write_text("I like to read English books.", encoding="utf-8")
    expected = resolve_expected_text({}, CLIPS, 0, cache_dir=str(tmp_path))
    assert expected == "I like to read English books."


def test_resolve_expected_text_sentences_fallback():
    task = {"sentences": ["Hello, how are you today?", "I like to read English books."]}
    assert resolve_expected_text(task, CLIPS, 1) == "I like to read English books."


def test_text_similarity_metrics_have_no_passed_or_grade():
    metrics = build_metrics("Hello, how are you today?", "Hello, how are you today?")
    assert len(metrics) == 1
    assert metrics[0]["method"] == "text_similarity_v0"
    assert 0.0 <= metrics[0]["value"] <= 1.0
    assert "passed" not in metrics[0]
    assert "grade" not in metrics[0]
    ratio = text_similarity_v0("hello how are you today", "Hello, how are you today?")
    assert ratio > 0.9


def test_build_metrics_empty_when_missing_asr_or_expected():
    assert build_metrics("", "Hello") == []
    assert build_metrics("Hello", "") == []
    assert build_metrics("Hello", CLIPS[0]) == []


def test_listen_ratio_is_clamped():
    assert clamp_listen_ratio(0.5) == 1.2
    assert clamp_listen_ratio(3.0) == 1.5
    assert clamp_listen_ratio(1.3) == 1.3
    assert listen_window_ms(1000, 0.5) == max(800, int(1000 * 1.2))
    assert listen_window_ms(1000, 3.0) == int(1000 * 1.5)
    assert listen_window_ms(100, 1.2) == 800
    # 听窗随上一句实际播放时长变化，不是固定 2.5s；倍速已含在 wall-clock duration 里。
    assert listen_window_ms(2050, 1.2) == int(2050 * 1.2)
    assert listen_window_ms(4000, 1.2) == int(4000 * 1.2)
    assert listen_window_ms(2050, 1.2, speed=0.9, wall_clock=True) == int(2050 * 1.2)
    assert listen_window_ms(1800, 1.2, speed=0.9, wall_clock=False) == int(round(1800 / 0.9) * 1.2)


def _shadowing_conn(tmp_path, monkeypatch):
    monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
    monkeypatch.setattr("core.app_runtime.session._schedule_upsert", lambda _payload: None)
    monkeypatch.setattr("core.audio_archive.get_project_dir", lambda: str(tmp_path))
    conn = SimpleNamespace(
        session_id="ws-1",
        device_id="mac",
        interaction_mode="SHADOWING",
        client_is_speaking=False,
        shadowing_index=0,
        shadowing_prompt=CLIPS[0],
        shadowing_clips=list(CLIPS),
        shadowing_awaiting_user=True,
        shadowing_clip_gen=1,
        shadowing_assess_index=0,
        shadowing_listen_pcm=[],
        shadowing_audio_saved_index=-1,
        asr_audio=[],
        config={"audio_archive": {"enabled": True, "dir": "data/recordings"}},
        last_asr_audio_snapshot=[],
        sentence_id=None,
        task_session=None,
        active_app_id="english_learning",
        current_task_id="shadowing",
        loop=None,
        prompt="",
    )
    begin_session(conn, "english_learning", "shadowing", "voice", "SHADOWING")
    return conn


def test_asr_does_not_play_next_only_timer_advances(tmp_path, monkeypatch):
    from core.app_runtime.runtime import (
        _on_shadowing_listen_window_closed,
        apply_shadowing_user_text,
    )

    played = []

    async def fake_play(conn, name):
        played.append(name)
        return True

    monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
    conn = _shadowing_conn(tmp_path, monkeypatch)
    outcome = apply_shadowing_user_text(conn, "today weather is bad")
    assert outcome == "recorded"
    assert conn.shadowing_index == 0
    assert played == []
    asyncio.run(_on_shadowing_listen_window_closed(conn))
    assert conn.shadowing_index == 1
    assert played == []
    assert conn.task_session.find_turn(0)["asr_text"] == "today weather is bad"


def test_skip_listen_stop_asr_while_shadowing_clip_plays():
    from core.app_runtime.runtime import should_skip_listen_stop_asr

    playing = SimpleNamespace(interaction_mode="SHADOWING", shadowing_awaiting_user=False)
    listening = SimpleNamespace(interaction_mode="SHADOWING", shadowing_awaiting_user=True)
    chat = SimpleNamespace(interaction_mode="CHAT", shadowing_awaiting_user=False)
    repeater = SimpleNamespace(interaction_mode="PLAY_AUDIO", shadowing_awaiting_user=False)
    assert should_skip_listen_stop_asr(playing) is True
    assert should_skip_listen_stop_asr(listening) is False
    assert should_skip_listen_stop_asr(chat) is False
    assert should_skip_listen_stop_asr(repeater) is True


def test_timer_advances_after_assess_and_upsert_errors(tmp_path, monkeypatch):
    from core.app_runtime.runtime import _on_shadowing_listen_window_closed

    played = []

    async def fake_play(conn, name):
        played.append(name)
        return True

    def boom_assess(*_args, **_kwargs):
        raise RuntimeError("assess exploded")

    monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
    monkeypatch.setattr("core.app_runtime.runtime.apply_deferred_assessment", boom_assess)
    conn = _shadowing_conn(tmp_path, monkeypatch)

    def boom_save():
        raise RuntimeError("upsert exploded")

    conn.task_session.save = boom_save
    asyncio.run(_on_shadowing_listen_window_closed(conn))
    assert conn.shadowing_index == 1
    assert played == []


def test_timer_advances_without_asr(tmp_path, monkeypatch):
    from core.app_runtime.runtime import _on_shadowing_listen_window_closed

    played = []

    async def fake_play(conn, name):
        played.append(name)
        return True

    monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
    conn = _shadowing_conn(tmp_path, monkeypatch)
    asyncio.run(_on_shadowing_listen_window_closed(conn))
    assert conn.shadowing_index == 1
    turn = conn.task_session.find_turn(0)
    assert turn is not None
    assert turn["asr_text"] == ""
    names = [event["name"] for event in conn.task_session.events]
    assert "listen_window_closed" in names
    assert played == []


def test_upsert_turn_keeps_last_asr_same_index(tmp_path, monkeypatch):
    monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
    session = TaskSession(task_session_id="dup", app_id="english_learning", task_id="shadowing")
    session.upsert_turn(0, CLIPS[0], "first try")
    session.upsert_turn(0, CLIPS[0], "second try")
    assert len(session.turns) == 1
    assert session.turns[0]["asr_text"] == "second try"


def test_deferred_assessment_writes_metrics_not_passed(tmp_path, monkeypatch):
    from core.app_runtime.runtime import apply_deferred_assessment, apply_shadowing_user_text

    conn = _shadowing_conn(tmp_path, monkeypatch)
    apply_shadowing_user_text(conn, "Hello, how are you today?")
    apply_deferred_assessment(conn, 0)
    turn = conn.task_session.find_turn(0)
    assert turn["expected_text"] == "Hello, how are you today?"
    assert not str(turn["expected_text"]).endswith(".mp3")
    assert turn["metrics"][0]["method"] == "text_similarity_v0"
    assert 0.0 <= turn["metrics"][0]["value"] <= 1.0
    assert "passed" not in turn
    assert "grade" not in turn
    assert "passed" not in turn["metrics"][0]


def test_upsert_empty_metrics_do_not_wipe_existing(tmp_path, monkeypatch):
    monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
    session = TaskSession(task_session_id="keep-m", app_id="english_learning", task_id="shadowing")
    session.upsert_turn(
        0,
        CLIPS[0],
        "hello",
        expected_text="Hello, how are you today?",
        metrics=[{"method": "text_similarity_v0", "value": 0.9}],
        assessed_at=1,
    )
    session.upsert_turn(0, CLIPS[0], "hello")
    assert session.turns[0]["metrics"][0]["value"] == 0.9
    assert "passed" not in session.turns[0]
    assert "passed" not in session.turns[0]["metrics"][0]


def test_skill_event_starts_listen_window_before_save(tmp_path, monkeypatch):
    from core.app_runtime.runtime import handle_skill_event, listen_window_ms

    scheduled = []

    def fake_schedule(conn, duration_ms):
        scheduled.append(duration_ms)

    monkeypatch.setattr(
        "core.app_runtime.runtime._schedule_shadowing_listen_timeout", fake_schedule
    )
    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_awaiting_user = False
    asyncio.run(
        handle_skill_event(
            conn, "[skill_event]audio.completed|english_learning|shadowing|2050"
        )
    )
    assert scheduled == [2050]
    assert listen_window_ms(2050, 1.2) == int(2050 * 1.2)
    assert conn.task_session.events[-1]["name"] == "audio.completed"


def test_repeater_completed_returns_to_chat_even_if_event_says_shadowing(monkeypatch):
    from core.app_runtime.runtime import handle_skill_event

    switched = []

    async def fake_start(conn, app_id, task_id, params, source="voice", **kwargs):
        switched.append(
            (app_id, task_id, kwargs.get("speak_switch"), kwargs.get("stop_skills"))
        )
        return None

    async def fake_continue(conn):
        return False

    monkeypatch.setattr("core.app_runtime.runtime.start_app_task", fake_start)
    monkeypatch.setattr("core.app_runtime.runtime._continue_repeater_if_needed", fake_continue)
    spoken = []
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    conn = SimpleNamespace(
        interaction_mode="PLAY_AUDIO",
        active_app_id="english_learning",
        current_task_id="repeater",
        task_session=None,
        repeater_speed_queue=[],
        repeater_play_name="",
        prompt="",
        config={},
        device_id="mac",
    )
    asyncio.run(
        handle_skill_event(
            conn, "[skill_event]audio.completed|english_learning|shadowing|6130"
        )
    )
    assert switched == [("chat", "conversation", False, False)]
    assert spoken == ["复读结束。我们来聊天吧。"]


def test_late_asr_after_listen_window_still_scores_previous_sentence(tmp_path, monkeypatch):
    from core.app_runtime.runtime import apply_shadowing_user_text

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_awaiting_user = False
    conn.shadowing_index = 1
    conn.shadowing_assess_index = 0
    conn.shadowing_prompt = CLIPS[1]
    conn.client_is_speaking = False
    outcome = apply_shadowing_user_text(conn, "hello how are you today")
    assert outcome == "recorded"
    assert conn.shadowing_index == 1
    turn = conn.task_session.find_turn(0)
    assert turn["asr_text"] == "hello how are you today"
    assert conn.task_session.find_turn(1) is None


def test_evaluate_turn_saves_wav_and_metrics(tmp_path, monkeypatch):
    from core.app_runtime.assessment import pcm_snapshot_to_wav
    from core.app_runtime.runtime import apply_deferred_assessment, apply_shadowing_user_text

    captured = {}

    def fake_save(session_id, step_index, wav_bytes, conn=None):
        captured["wav"] = wav_bytes
        captured["id"] = f"{session_id}_{step_index}.wav"
        return captured["id"]

    monkeypatch.setattr("core.app_runtime.runtime.save_turn_wav", fake_save)
    conn = _shadowing_conn(tmp_path, monkeypatch)
    # 16-bit silence ~ 20ms
    conn.last_asr_audio_snapshot = [b"\x00\x00" * 160]
    apply_shadowing_user_text(conn, "Hello, how are you today?")
    apply_deferred_assessment(conn, 0, audio_snapshot=conn.last_asr_audio_snapshot)
    turn = conn.task_session.find_turn(0)
    assert turn["metrics"][0]["method"] == "text_similarity_v0"
    assert turn["audio_id"] == captured["id"]
    wav = pcm_snapshot_to_wav(conn.last_asr_audio_snapshot)
    assert captured["wav"][:4] == b"RIFF"
    assert wav[8:12] == b"WAVE"
    assert captured["wav"] == wav


def test_shadowing_playlist_end_announces_then_chat(tmp_path, monkeypatch):
    from core.app_runtime.runtime import handle_skill_event

    switched = []
    spoken = []
    played = []

    async def fake_play(conn, name):
        played.append(name)
        return True

    async def fake_start(conn, app_id, task_id, params, source="voice", **kwargs):
        switched.append((app_id, task_id, kwargs.get("speak_switch")))
        conn.interaction_mode = "CHAT"
        conn.active_app_id = app_id
        conn.current_task_id = task_id
        return None

    monkeypatch.setattr("core.app_runtime.runtime._play_shadowing_clip", fake_play)
    monkeypatch.setattr("core.app_runtime.runtime.start_app_task", fake_start)
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_index = 1
    conn.shadowing_prompt = CLIPS[1]
    conn.shadowing_awaiting_user = False
    asyncio.run(
        handle_skill_event(
            conn, "[skill_event]audio.playlist_completed|english_learning|shadowing"
        )
    )
    assert played == []
    assert switched == [("chat", "conversation", False)]
    assert spoken == ["跟读结束。我们来聊天吧。"]


def test_playlist_completed_returns_chat_after_reconnect(monkeypatch):
    """WS 掉线后设备可能在新的 CHAT 连接上补发 playlist_completed。"""
    from core.app_runtime.runtime import handle_skill_event

    switched = []
    spoken = []

    async def fake_start(conn, app_id, task_id, params, source="voice", **kwargs):
        switched.append((app_id, task_id, kwargs.get("speak_switch")))
        conn.interaction_mode = "CHAT"
        return None

    monkeypatch.setattr("core.app_runtime.runtime.start_app_task", fake_start)
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    conn = SimpleNamespace(
        interaction_mode="CHAT",
        active_app_id="chat",
        current_task_id="conversation",
        task_session=None,
        repeater_speed_queue=[],
        repeater_playlist=[],
        shadowing_awaiting_user=False,
        prompt="",
        config={},
        device_id="mac",
        tts=None,
    )
    asyncio.run(
        handle_skill_event(
            conn, "[skill_event]audio.playlist_completed|english_learning|shadowing"
        )
    )
    assert switched == [("chat", "conversation", False)]
    assert spoken == ["跟读结束。我们来聊天吧。"]


def test_handle_shadowing_turn_does_not_enter_chat(tmp_path, monkeypatch):
    from core.app_runtime.runtime import handle_shadowing_turn

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.last_asr_audio_snapshot = [b"\x00\x00" * 80]
    asyncio.run(handle_shadowing_turn(conn, "Hello, how are you today?"))
    assert conn.interaction_mode == "SHADOWING"
    assert conn.shadowing_index == 0
    turn = conn.task_session.find_turn(0)
    assert turn["asr_text"] == "Hello, how are you today?"


def test_listen_window_saves_wav_even_if_user_silent(tmp_path, monkeypatch):
    from pathlib import Path
    from core.app_runtime.runtime import _on_shadowing_listen_window_closed

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_listen_pcm = []
    conn.asr_audio = []
    asyncio.run(_on_shadowing_listen_window_closed(conn))
    turn = conn.task_session.find_turn(0)
    assert turn is not None
    assert turn.get("audio_id")
    assert "mac/shadowing/" in turn["audio_id"]
    wav_path = tmp_path / "data" / "recordings" / Path(turn["audio_id"])
    assert wav_path.is_file()
    assert wav_path.stat().st_size > 44


def test_fifteen_listen_windows_write_fifteen_wavs(tmp_path, monkeypatch):
    from pathlib import Path
    from core.app_runtime.runtime import save_shadowing_listen_audio

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_clips = CLIPS[:1] * 15
    saved = []
    for i in range(15):
        conn.shadowing_assess_index = i
        conn.shadowing_index = i
        conn.shadowing_audio_saved_index = -1
        conn.shadowing_listen_pcm = [b"\x00\x00" * (80 + i)]
        save_shadowing_listen_audio(conn, i)
        turn = conn.task_session.find_turn(i)
        assert turn.get("audio_id")
        saved.append(turn["audio_id"])
        assert (tmp_path / "data" / "recordings" / Path(turn["audio_id"])).is_file()
    assert len(saved) == 15
    assert len(set(saved)) == 15


def test_shadowing_save_does_not_merge_previous_asr_audio(tmp_path, monkeypatch):
    from pathlib import Path
    import wave
    from core.app_runtime.runtime import save_shadowing_listen_audio

    conn = _shadowing_conn(tmp_path, monkeypatch)
    leftover = b"\x11\x00" * 8000
    current = b"\x22\x00" * 160
    conn.asr_audio = [leftover]
    conn.shadowing_listen_pcm = [current]
    conn.shadowing_assess_index = 0
    conn.shadowing_audio_saved_index = -1
    save_shadowing_listen_audio(conn, 0)
    turn = conn.task_session.find_turn(0)
    wav_path = tmp_path / "data" / "recordings" / Path(turn["audio_id"])
    with wave.open(str(wav_path), "rb") as handle:
        pcm = handle.readframes(handle.getnframes())
    assert leftover not in pcm
    assert current in pcm
    assert len(pcm) == len(current)


def test_one_step_writes_only_one_wav(tmp_path, monkeypatch):
    from core.app_runtime.runtime import save_shadowing_listen_audio

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_listen_pcm = [b"\x00\x00" * 80]
    save_shadowing_listen_audio(conn, 0)
    first = conn.task_session.find_turn(0)["audio_id"]
    conn.shadowing_listen_pcm = [b"\x00\x00" * 800]
    conn.asr_audio = [b"\x00\x00" * 8000]
    save_shadowing_listen_audio(conn, 0)
    assert conn.task_session.find_turn(0)["audio_id"] == first
    recordings = list((tmp_path / "data" / "recordings").rglob("*.wav"))
    assert len(recordings) == 1


def test_listen_stop_drains_queued_pcm_instead_of_silent_placeholder(tmp_path, monkeypatch):
    import queue as queue_mod
    import wave
    from pathlib import Path
    from core.app_runtime.runtime import close_shadowing_listen_window

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_awaiting_user = True
    conn.shadowing_index = 0
    conn.shadowing_listen_pcm = []
    conn.asr_audio_queue = queue_mod.Queue()
    real = b"\x22\x00" * 1600
    conn.asr_audio_queue.put(real)
    conn.asr_audio_queue.put(b"\x33\x00" * 1600)
    asyncio.run(close_shadowing_listen_window(conn))
    turn = conn.task_session.find_turn(0)
    wav_path = tmp_path / "data" / "recordings" / Path(turn["audio_id"])
    with wave.open(str(wav_path), "rb") as handle:
        pcm = handle.readframes(handle.getnframes())
    assert real in pcm
    assert len(pcm) == 1600 * 2 * 2
    assert conn.asr_audio_queue.empty()


def test_listen_stop_advances_index_for_next_wav(tmp_path, monkeypatch):
    from pathlib import Path
    from core.app_runtime.runtime import close_shadowing_listen_window, save_shadowing_listen_audio

    conn = _shadowing_conn(tmp_path, monkeypatch)
    conn.shadowing_clips = CLIPS[:1] * 15
    conn.shadowing_awaiting_user = True
    conn.shadowing_index = 0
    conn.shadowing_listen_pcm = [b"\x00\x00" * 80]
    asyncio.run(close_shadowing_listen_window(conn))
    assert conn.shadowing_index == 1
    assert conn.shadowing_awaiting_user is False
    first = conn.task_session.find_turn(0)["audio_id"]
    conn.shadowing_listen_pcm = [b"\x11\x00" * 80]
    conn.shadowing_audio_saved_index = -1
    save_shadowing_listen_audio(conn, 1)
    second = conn.task_session.find_turn(1)["audio_id"]
    assert first != second
    recordings = list((tmp_path / "data" / "recordings").rglob("*.wav"))
    assert len(recordings) == 2