5种场景模式家长端ok

wangzhibo

10 天以前 220e4fab27b7f419b3e89e2f2d28ae789aac09d7

 xiaozhi-esp32-server/main/xiaozhi-server/tests/test_app_runtime.py

@@ -33,7 +33,7 @@
    assert match_voice_switch("退出翻译") == ("chat", "conversation")
    assert match_voice_switch("返回聊天") == ("chat", "conversation")
    assert match_voice_switch("退出跟读") == ("chat", "conversation")
    assert match_voice_switch("退出复读") == ("chat", "conversation")
    assert match_voice_switch("退出泛听") == ("chat", "conversation")


def test_enter_prompts_announce_mode_without_exit_echo():
@@ -43,8 +43,8 @@
    _, shadowing = get_task("english_learning", "shadowing")
    repeater_prompt = (repeater or {}).get("enter_prompt") or ""
    shadowing_prompt = (shadowing or {}).get("enter_prompt") or ""
    assert "现在已经切换至复读机模式" in repeater_prompt
    assert "退出复读" not in repeater_prompt
    assert "现在已经切换至泛听模式" in repeater_prompt
    assert "退出泛听" not in repeater_prompt
    assert "现在已经切换至跟读模式" in shadowing_prompt
    assert "退出跟读" not in shadowing_prompt

@@ -53,7 +53,7 @@
    from core.app_runtime.runtime import voice_menu_text

    menu = voice_menu_text()
    assert "听复读机" in menu
    assert "泛听" in menu
    assert "跟读" in menu
    assert "翻译" in menu
    assert "中英对话" in menu
@@ -64,11 +64,11 @@
    from core.app_runtime.runtime import is_exit_to_chat

    assert match_voice_switch("跟读") == ("english_learning", "shadowing")
    assert match_voice_switch("听读") == ("english_learning", "shadowing")
    assert match_voice_switch("泛听") == ("english_learning", "repeater")
    assert match_voice_switch("我想退出跟读") == ("chat", "conversation")
    assert is_exit_to_chat("退出翻译") is True
    assert is_exit_to_chat("开始跟读") is False
    assert is_exit_to_chat("退出听读") is True
    assert is_exit_to_chat("退出泛听") is True


def test_new_task_phrases_come_from_registry_only(monkeypatch):
@@ -207,27 +207,27 @@

def test_lesson_mp3_clips_picks_first_lesson_folder():
    names = [
        "/english_learning/repeater/lessons/unit01/full.mp3",
        "/english_learning/shadowing/lessons/unit02/02.mp3",
        "/english_learning/shadowing/lessons/unit01/02.mp3",
        "/english_learning/shadowing/lessons/unit01/01.mp3",
        "/english_learning/shadowing/lessons/unit01/full.mp3",
        "/english_learning/shadowing/lessons/unit01/audio/03.mp3",
        "/english_learning/repeater/lessons/l1/001/cn/01.mp3",
        "/english_learning/shadowing/lessons/l2/002/en/02.mp3",
        "/english_learning/shadowing/lessons/l1/001/en/02.mp3",
        "/english_learning/shadowing/lessons/l1/001/en/01.mp3",
        "/english_learning/shadowing/lessons/l1/001/full.mp3",
        "/english_learning/shadowing/lessons/l1/001/en/03.mp3",
        "loose.mp3",
    ]
    clips = lesson_mp3_clips(names, "english_learning", "shadowing")
    assert clips == [
        "/english_learning/shadowing/lessons/unit01/01.mp3",
        "/english_learning/shadowing/lessons/unit01/02.mp3",
        "/english_learning/shadowing/lessons/unit01/audio/03.mp3",
        "/english_learning/shadowing/lessons/l1/001/en/01.mp3",
        "/english_learning/shadowing/lessons/l1/001/en/02.mp3",
        "/english_learning/shadowing/lessons/l1/001/en/03.mp3",
    ]
    only_unit2 = lesson_mp3_clips(names, "english_learning", "shadowing", "unit02")
    assert only_unit2 == ["/english_learning/shadowing/lessons/unit02/02.mp3"]
    only_l2 = lesson_mp3_clips(names, "english_learning", "shadowing", "l2")
    assert only_l2 == ["/english_learning/shadowing/lessons/l2/002/en/02.mp3"]
    assert repeater_lesson_mp3(names, "english_learning", "repeater") == (
        "/english_learning/repeater/lessons/unit01/full.mp3"
        "/english_learning/repeater/lessons/l1/001/cn/01.mp3"
    )
    assert repeater_lesson_mp3(names, "english_learning", "shadowing") == (
        "/english_learning/shadowing/lessons/unit01/full.mp3"
        "/english_learning/shadowing/lessons/l1/001/full.mp3"
    )


@@ -244,7 +244,7 @@
        started_at=1,
        status="running",
    )
    session.add_turn(0, "/english_learning/shadowing/lessons/unit01/01.mp3", "hello", False)
    session.add_turn(0, "/english_learning/shadowing/lessons/l1/001/en/01.mp3", "hello", False)
    session.add_event("audio.completed", "english_learning", "shadowing")
    data = session.to_dict()
    assert data["task_session_id"] == "abc123"
@@ -252,6 +252,19 @@
    assert "agent_id" in data
    assert data["turns"][0]["prompt_text"].endswith("01.mp3")
    assert data["events"][0]["name"] == "audio.completed"


def test_turn_audio_id_keeps_local_recording_path():
    session = TaskSession(
        task_session_id="abc123",
        app_id="english_learning",
        task_id="shadowing",
        status="running",
    )
    long_path = "14-c1-9f-49-43-94/shadowing/2026-09-18/091728893_asr_b5a60ebacb9544bfba4782a5cc978fb9_0.wav"
    session.upsert_turn(0, prompt_text="01.mp3", asr_text="hello", audio_id=long_path)
    assert session.to_dict()["turns"][0]["audio_id"] == long_path
    assert len(long_path) > 32


def test_save_survives_upsert_scheduler_error(tmp_path, monkeypatch):
@@ -281,7 +294,7 @@
    assert len(loaded.turns) == 1


def test_repeater_queues_play_until_listen_start(monkeypatch):
def test_repeater_queues_play_until_listen_start(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import maybe_flush_pending_play_after_tts, start_app_task

@@ -296,34 +309,86 @@

    async def fake_list(conn):
        return [
            "/english_learning/shadowing/lessons/unit01/full.mp3",
            "/english_learning/repeater/lessons/unit01/full.mp3",
            "/english_learning/shadowing/lessons/unit01/01.mp3",
            "/english_learning/shadowing/lessons/l1/001/full.mp3",
            "/english_learning/repeater/lessons/l1/001/cn/01.mp3",
            "/english_learning/shadowing/lessons/l1/001/en/01.mp3",
        ]

    monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play)
    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs)
    monkeypatch.setattr("core.app_runtime.runtime._list_sd_mp3_names", fake_list)
    monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path))

    async def fake_ensure(*args, **kwargs):
        return True

    monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure)
    spoken = []
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    conn = SimpleNamespace(config={}, prompt="", func_handler=None)
    asyncio.run(start_app_task(conn, "english_learning", "repeater", {}, source="voice"))
    assert spoken and "复读机" in spoken[0]
    conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-queue")
    asyncio.run(start_app_task(conn, "english_learning", "repeater", {"lesson_id": "l1.001"}, source="voice"))
    assert spoken and "泛听" in spoken[0]
    assert plays == []
    assert conn.pending_play_after_tts and conn.pending_play_after_tts["kind"] == "repeater"
    asyncio.run(maybe_flush_pending_play_after_tts(conn))
    assert len(plays) == 1
    assert plays[0]["name"] == "/english_learning/shadowing/lessons/unit01/01.mp3"
    assert plays[0]["repeat"] == 10
    assert plays[0]["name"] == "/english_learning/repeater/lessons/l1/001/cn/01.mp3"
    assert plays[0]["repeat"] == 1
    assert plays[0]["speed"] == 1.0
    assert plays[0].get("finish_task") is False
    assert len(plays[0]["items"]) == 2
    assert plays[0]["items"][1]["name"].endswith("/02.mp3")
    assert plays[0]["items"][1]["repeat"] == 5
    assert len(plays[0]["items"]) == 4
    assert plays[0]["items"][1]["name"].endswith("/l1/001/en/01.mp3")
    assert plays[0]["items"][1]["speed"] == 0.8
    assert plays[0]["items"][3]["repeat"] == 5
    assert plays[0]["items"][3]["speed"] == 1.0
    assert conn.repeater_playlist == []
    assert conn.pending_play_after_tts is None


def test_shadowing_queues_whole_playlist_until_listen_start(monkeypatch):
def test_sd_ready_skips_sync_and_queued_play(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    plays = []
    spoken = []
    dirs = []
    ensures = []

    async def fake_play(conn, params):
        plays.append(params)
        return None

    async def fake_dirs(conn, app_id):
        dirs.append(app_id)
        return None

    async def fake_ensure(*args, **kwargs):
        ensures.append(args)
        return True

    monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play)
    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs)
    monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure)
    monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path))
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-sd-ready")
    asyncio.run(start_app_task(
        conn,
        "english_learning",
        "repeater",
        {"lesson_id": "l1.001", "sd_ready": True},
        source="mqtt_immediate",
    ))
    assert plays == []
    assert dirs == []
    assert ensures == []
    assert getattr(conn, "pending_play_after_tts", None) is None
    assert conn.interaction_mode == "PLAY_AUDIO"
    assert spoken == ["开始泛听任务"]
    assert not any("同步" in str(text) for text in spoken)


def test_shadowing_queues_whole_playlist_until_listen_start(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import maybe_flush_pending_play_after_tts, start_app_task

@@ -336,47 +401,192 @@
    async def fake_dirs(conn, app_id):
        return None

    async def fake_ensure(*args, **kwargs):
        return True

    monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play)
    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs)
    monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure)
    monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None)
    spoken = []
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    conn = SimpleNamespace(config={}, prompt="", func_handler=None)
    asyncio.run(start_app_task(conn, "english_learning", "shadowing", {}, source="voice"))
    conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-shadow-queue")
    asyncio.run(start_app_task(conn, "english_learning", "shadowing", {"lesson_id": "l1.001"}, source="voice"))
    assert spoken and "跟读" in spoken[0]
    assert plays == []
    assert conn.pending_play_after_tts and conn.pending_play_after_tts["kind"] == "shadowing"
    queued = conn.pending_play_after_tts["params"]["items"]
    assert len(queued) == 2
    assert queued[0]["repeat"] == 5
    assert queued[1]["name"].endswith("/02.mp3")
    assert queued[1]["repeat"] == 10
    assert len(queued) == 58
    assert queued[0]["repeat"] == 1
    assert queued[0]["speed"] == 0.8
    assert queued[-1]["name"].endswith("/58.mp3")
    assert queued[-1]["repeat"] == 1
    asyncio.run(maybe_flush_pending_play_after_tts(conn))
    assert len(plays) == 1
    assert len(plays[0]["items"]) == 2
    assert plays[0]["items"][0]["repeat"] == 5
    assert plays[0]["items"][1]["repeat"] == 10
    assert len(plays[0]["items"]) == 58
    assert plays[0]["items"][0]["repeat"] == 1
    assert plays[0]["items"][-1]["repeat"] == 1
    assert plays[0].get("finish_task") is False
    assert plays[0].get("listen_ratio") == 1.2
    assert conn.pending_play_after_tts is None


def test_shadowing_expands_default_playlist_into_listen_turns():
def test_shadowing_ensures_mp3s_even_when_sd_empty(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    ensure_calls = []

    async def fake_play(conn, params):
        return None

    async def fake_dirs(conn, app_id):
        return None

    async def fake_list(conn):
        return []

    async def fake_ensure(conn, names, task_id, params):
        ensure_calls.append((list(names), task_id, dict(params or {})))
        return True

    monkeypatch.setattr("core.app_runtime.runtime._execute_named_tool_play", fake_play)
    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs)
    monkeypatch.setattr("core.app_runtime.runtime._list_sd_mp3_names", fake_list)
    monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure)
    monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None)
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: None)
    conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-shadow-sync")
    asyncio.run(start_app_task(
        conn,
        "english_learning",
        "shadowing",
        {"lesson_id": "l1.001"},
        source="mqtt_immediate",
    ))
    assert len(ensure_calls) == 1
    names, task_id, params = ensure_calls[0]
    assert task_id == "shadowing"
    assert params.get("lesson_id") == "l1.001"
    assert len(names) == 58
    assert names[0].endswith("/en/01.mp3")
    assert names[-1].endswith("/58.mp3")
    assert conn.pending_play_after_tts and conn.pending_play_after_tts["kind"] == "shadowing"


def test_shadowing_expands_default_playlist_into_listen_turns(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import _prepare_shadowing_sd

    monkeypatch.setattr("core.app_runtime.playback.list_lesson_clips", lambda *_args, **_kwargs: None)
    conn = SimpleNamespace(
        config={},
        active_app_id="english_learning",
        current_task_id="shadowing",
    )
    ok = asyncio.run(_prepare_shadowing_sd(conn, {}))
    ok = asyncio.run(_prepare_shadowing_sd(conn, {"lesson_id": "l1.001"}))
    assert ok is True
    unit01_01 = "/english_learning/shadowing/lessons/unit01/01.mp3"
    unit01_02 = "/english_learning/shadowing/lessons/unit01/02.mp3"
    assert conn.shadowing_clips.count(unit01_01) == 5
    assert conn.shadowing_clips.count(unit01_02) == 10
    assert len(conn.shadowing_clips) == 15
    assert conn.shadowing_prompt == unit01_01
    clip_01 = "/english_learning/shadowing/lessons/l1/001/en/01.mp3"
    clip_58 = "/english_learning/shadowing/lessons/l1/001/en/58.mp3"
    assert conn.shadowing_clips.count(clip_01) == 1
    assert conn.shadowing_clips[-1] == clip_58
    assert len(conn.shadowing_clips) == 58
    assert conn.shadowing_prompt == clip_01


def test_voice_repeater_without_last_lesson_prompts_parent(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path))

    async def fake_dirs(conn, app_id):
        return None

    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs)
    conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-empty")
    asyncio.run(start_app_task(conn, "english_learning", "repeater", {}, source="voice"))
    assert spoken and "还没有下发过泛听课文" in spoken[0]
    assert getattr(conn, "pending_play_after_tts", None) is None


def test_voice_repeater_replays_last_parent_task(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path))

    async def fake_dirs(conn, app_id):
        return None

    async def fake_ensure(*args, **kwargs):
        return True

    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs)
    monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure)
    conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-last")
    asyncio.run(
        start_app_task(
            conn,
            "english_learning",
            "repeater",
            {"lesson_id": "l2.010", "en_passes": [{"speed": 0.8, "repeat": 1}]},
            source="mqtt_immediate",
        )
    )
    conn.active_app_id = "chat"
    conn.current_task_id = "conversation"
    conn.current_source = "default_chat"
    conn.interaction_mode = "CHAT"
    spoken.clear()
    asyncio.run(start_app_task(conn, "english_learning", "repeater", {}, source="voice"))
    pending = conn.pending_play_after_tts
    assert pending and pending["kind"] == "repeater"
    assert pending["params"]["name"].endswith("/l2/010/cn/01.mp3")
    assert spoken and "泛听" in spoken[0]


def test_voice_shadowing_replays_last_parent_task(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    monkeypatch.setattr("core.app_runtime.playback.get_project_dir", lambda: str(tmp_path))

    async def fake_dirs(conn, app_id):
        return None

    async def fake_ensure(*args, **kwargs):
        return True

    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", fake_dirs)
    monkeypatch.setattr("core.app_runtime.runtime._ensure_lesson_mp3s", fake_ensure)
    conn = SimpleNamespace(config={}, prompt="", func_handler=None, device_id="mac-last-shd")
    asyncio.run(
        start_app_task(
            conn,
            "english_learning",
            "shadowing",
            {"lesson_id": "l1.001", "language": "en", "speed": 0.8},
            source="schedule",
        )
    )
    conn.active_app_id = "chat"
    conn.current_task_id = "conversation"
    conn.current_source = "default_chat"
    conn.interaction_mode = "CHAT"
    conn.shadowing_clips = []
    spoken.clear()
    asyncio.run(start_app_task(conn, "english_learning", "shadowing", {}, source="voice"))
    pending = conn.pending_play_after_tts
    assert pending and pending["kind"] == "shadowing"
    assert pending["params"]["items"][0]["name"].endswith("/l1/001/en/01.mp3")
    assert spoken and "跟读" in spoken[0] and "还没有下发过" not in spoken[0]


def test_app_sd_layout_paths_from_registry():
@@ -422,6 +632,48 @@
    assert conn.chat_greeted is True


def test_boot_unbound_skips_greeting_and_plays_bind_code(monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []
    bind_codes = []
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))

    async def _ready(conn, timeout=8):
        return True

    monkeypatch.setattr("core.app_runtime.runtime.wait_for_tts_channels", _ready)

    async def fake_bind(conn):
        bind_codes.append(conn.bind_code)

    monkeypatch.setattr("core.app_runtime.runtime._play_bind_prompt", fake_bind)
    conn = SimpleNamespace(
        config={
            "enable_proactive_greeting": True,
            "proactive_chat_greeting": "我们来聊天吧。",
        },
        prompt="",
        func_handler=None,
        just_woken_up=False,
        chat_greeted=False,
        active_app_id="chat",
        current_task_id="conversation",
        current_source="default_chat",
        interaction_mode="CHAT",
        need_bind=True,
        bind_code="123456",
        bind_completed_event=asyncio.Event(),
    )
    conn.bind_completed_event.set()
    conn.change_system_prompt = lambda text: setattr(conn, "prompt", text)
    asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="default_chat"))
    assert spoken == []
    assert bind_codes == ["123456"]
    assert conn.chat_greeted is False


def test_client_hello_is_recognized_before_bind():
    from core.websocket_handshake import is_client_hello

@@ -449,6 +701,9 @@
        '{"type":"listen","state":"detect","text":"[app_task]chat|conversation|default_chat"}'
    ) is True
    assert is_priority_listen_detect(
        '{"type":"listen","state":"detect","text":"[lesson_sync]"}'
    ) is True
    assert is_priority_listen_detect(
        '{"type":"listen","state":"detect","text":"你好小智"}'
    ) is False

@@ -472,3 +727,195 @@
        }
    }
    assert app_sd_layout_paths("story_time", app) == ["/story_time/listen/lessons"]


def test_language_pair_uses_bilingual_prompt_and_unknown_falls_back():
    from core.app_runtime.runtime import BILINGUAL_PROMPT, _apply_prompt

    conn = SimpleNamespace(prompt="base", original_system_prompt=None)
    conn.change_system_prompt = lambda text: setattr(conn, "prompt", text)
    app = {"system_prompt_mode": "bilingual"}
    _apply_prompt(conn, app, {"language_pair": "zh-en"})
    assert conn.prompt == BILINGUAL_PROMPT
    assert conn.language_pair_fallback is False

    conn2 = SimpleNamespace(prompt="base", original_system_prompt=None)
    conn2.change_system_prompt = lambda text: setattr(conn2, "prompt", text)
    _apply_prompt(conn2, app, {"language_pair": "zh-ko"})
    assert conn2.prompt == BILINGUAL_PROMPT
    assert conn2.language_pair_fallback is True
    assert conn2.language_pair == "zh-ko"


def test_unknown_repeater_lesson_does_not_queue_play(monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []

    async def _true(*args, **kwargs):
        return True

    async def _false(*args, **kwargs):
        return False

    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    monkeypatch.setattr("core.app_runtime.runtime.wait_for_tts_channels", _true)
    monkeypatch.setattr("core.app_runtime.runtime.ensure_app_sd_dirs", _true)
    monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false)

    conn = SimpleNamespace(
        config={},
        prompt="",
        func_handler=None,
        just_woken_up=False,
        chat_greeted=True,
        active_app_id="chat",
        current_task_id="conversation",
        current_source="default_chat",
        interaction_mode="CHAT",
        pending_play_after_tts=None,
    )
    conn.change_system_prompt = lambda text: setattr(conn, "prompt", text)
    asyncio.run(
        start_app_task(
            conn,
            "english_learning",
            "repeater",
            {"lesson_id": "unit99"},
            source="mqtt_immediate",
        )
    )
    assert conn.pending_play_after_tts is None
    assert any("SD卡" in text for text in spoken)


def test_parent_translation_clears_abort_and_announces(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []

    async def _false(*args, **kwargs):
        return False

    monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false)
    monkeypatch.setattr("core.app_runtime.runtime.schedule_close_parent_except", lambda *a, **k: None)
    monkeypatch.setattr("core.app_runtime.session._schedule_close_parent", lambda *a, **k: None)

    conn = SimpleNamespace(
        config={},
        prompt="base",
        func_handler=None,
        device_id="mac-translate",
        client_abort=True,
        tts=SimpleNamespace(tts_stop_request=True),
    )
    conn.change_system_prompt = lambda text: setattr(conn, "prompt", text)
    asyncio.run(start_app_task(
        conn,
        "english_translation",
        "live_translate",
        {"language_pair": "zh-en"},
        source="mqtt_immediate",
    ))
    assert conn.client_abort is False
    assert conn.tts.tts_stop_request is False
    assert spoken and "翻译模式" in spoken[0]
    assert conn.interaction_mode == "TRANSLATION"


def test_schedule_talk_announces_start_phrase(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []

    async def _false(*args, **kwargs):
        return False

    monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false)
    monkeypatch.setattr("core.app_runtime.runtime.schedule_close_parent_except", lambda *a, **k: None)
    monkeypatch.setattr("core.app_runtime.session._schedule_close_parent", lambda *a, **k: None)

    conn = SimpleNamespace(
        config={"enable_proactive_greeting": True},
        prompt="base",
        func_handler=None,
        device_id="mac-schedule-talk",
        client_abort=False,
        tts=SimpleNamespace(tts_stop_request=False),
        active_app_id="chat",
        current_task_id="conversation",
        current_source="default_chat",
        interaction_mode="CHAT",
        chat_greeted=True,
    )
    conn.change_system_prompt = lambda text: setattr(conn, "prompt", text)

    asyncio.run(start_app_task(conn, "english_translation", "live_translate", {}, source="schedule"))
    assert spoken == ["开始翻译任务"]
    assert conn.interaction_mode == "TRANSLATION"

    spoken.clear()
    conn.active_app_id = "chat"
    conn.current_task_id = "conversation"
    conn.interaction_mode = "CHAT"
    asyncio.run(start_app_task(conn, "english_conversation", "bilingual_chat", {}, source="schedule"))
    assert spoken == ["开始中英对话任务"]
    assert conn.interaction_mode == "BILINGUAL"

    spoken.clear()
    asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="schedule"))
    assert spoken == ["开始对话任务"]
    assert conn.chat_greeted is True
    assert conn.interaction_mode == "CHAT"


def test_voice_exit_overrides_parent_bilingual_and_translation(tmp_path, monkeypatch):
    import asyncio
    from core.app_runtime.runtime import start_app_task

    spoken = []

    async def _false(*args, **kwargs):
        return False

    monkeypatch.setattr("core.app_runtime.session.session_dir", lambda: str(tmp_path))
    monkeypatch.setattr("core.app_runtime.runtime._speak", lambda conn, text: spoken.append(text))
    monkeypatch.setattr("core.app_runtime.runtime.prompt_bind_if_needed", _false)
    monkeypatch.setattr("core.app_runtime.runtime.schedule_close_parent_except", lambda *a, **k: None)
    monkeypatch.setattr("core.app_runtime.session._schedule_close_parent", lambda *a, **k: None)

    conn = SimpleNamespace(
        config={},
        prompt="base",
        func_handler=None,
        device_id="mac-exit",
        client_abort=False,
        tts=SimpleNamespace(tts_stop_request=False),
    )
    conn.change_system_prompt = lambda text: setattr(conn, "prompt", text)

    asyncio.run(start_app_task(conn, "english_conversation", "bilingual_chat", {}, source="schedule"))
    assert conn.current_source == "schedule"
    assert conn.interaction_mode == "BILINGUAL"
    spoken.clear()
    asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="voice"))
    assert conn.interaction_mode == "CHAT"
    assert conn.current_source == "voice"
    assert spoken and "对话模式" in spoken[0]

    asyncio.run(start_app_task(conn, "english_translation", "live_translate", {}, source="mqtt_immediate"))
    assert conn.interaction_mode == "TRANSLATION"
    spoken.clear()
    asyncio.run(start_app_task(conn, "english_learning", "repeater", {}, source="voice"))
    assert conn.interaction_mode == "TRANSLATION"
    spoken.clear()
    asyncio.run(start_app_task(conn, "chat", "conversation", {}, source="voice"))
    assert conn.interaction_mode == "CHAT"
    assert spoken and "对话模式" in spoken[0]