小智终端 5种 场景模式 OK

wangzhibo

2026-09-11 659c470eba542383987aad227b7f918100f131fa

 xiaozhi-esp32-server/main/xiaozhi-server/core/app_runtime/runtime.py

@@ -2,13 +2,48 @@

from __future__ import annotations

import asyncio
import json
import difflib
import queue
import random
import re
import uuid
from typing import Any, Dict, Optional, TYPE_CHECKING
from typing import Any, Dict, List, Optional, TYPE_CHECKING

from config.logger import setup_logging
from plugins_func.register import Action, ActionResponse

from .registry import default_app_id, get_app, get_task
from .assessment import (
    evaluate_turn,
    listen_window_ms,
    pcm_snapshot_to_wav,
    resolve_expected_text,
    resolve_listen_ratio,
    save_turn_wav,
    silent_pcm_snapshot,
    transcript_cache_dir,
)
from .playback import (
    clamp_play_speed,
    default_repeater_playlist,
    default_shadowing_playlist,
    expand_playlist_plays,
    lesson_day_index,
    mark_lesson_played_today,
    parse_playlist,
    resolve_play_speed,
    speeds_for_repeats,
)
from .registry import (
    default_app_id,
    get_app,
    get_registry,
    get_task,
    match_registry_voice,
    voice_menu_text,
)
from .session import begin_session, current_session, end_session, parse_skill_event

if TYPE_CHECKING:
    from core.connection import ConnectionHandler
@@ -28,19 +63,37 @@
    "未指定语言时：中文译成英文，英文译成中文；其它语言默认译成中文。"
)

BILINGUAL_PROMPT = (
    "你正在进行中英对照对话练习，不是翻译机，不要把用户的话原样译回去交差。"
    "严格执行：用户说中文时你只用英文完整回答；用户说英文时你只用中文完整回答。"
    "不要中英夹杂，不要同时给出两种语言，不要问要不要翻译。"
    "用户说退出中英对话、回到聊天时调用 switch_app，app_id=chat。"
)

CHAT_PROACTIVE_HINT = (
    "你是用户的语音伙伴。进入对话后可以先主动短句邀请聊天，例如「我们来聊天吧」。"
    "用户开口后按正常对话回答，不要反复催促。"
)

VOICE_MENU = "想聊天、听复读，还是打开翻译？也可以说我们来聊天吧。"
CHAT_RETURN_TOPICS = ("今天过得怎么样", "你最近在忙什么", "有什么新鲜事")

CHAT_PHRASES = ("我们来聊天吧", "聊会天吧", "回到聊天", "退出翻译", "结束翻译", "退出复读")
TRANSLATE_PHRASES = ("打开翻译", "开始翻译", "英语翻译", "翻译模式")
REPEATER_PHRASES = ("开始复读", "打开复读", "英语学习", "听复读")
DEFAULT_SHADOWING_SENTENCES = (
    "Hello, how are you today?",
    "I like to read English books.",
    "The weather is nice this morning.",
    "Can you say that again, please?",
    "Practice a little every day.",
)

DEVICE_PLAY_TOOL = "self_mp3_play"
DEVICE_STOP_TOOL = "self_mp3_stop"
DEVICE_LIST_TOOL = "self_mp3_get_list"
DEVICE_ENSURE_APP_TOOL = "self_sd_ensure_app"
SD_LAYOUT_MODES = ("PLAY_AUDIO", "SHADOWING")
SHADOWING_LISTEN_RATIO = 1.2
SHADOWING_LISTEN_MIN_MS = 800
REPEATER_END_PHRASE = "复读结束"
SHADOWING_END_PHRASE = "跟读结束"


def source_priority(source: str) -> int:
@@ -62,6 +115,150 @@
        conn.tool_allowlist = []
    if not hasattr(conn, "original_system_prompt"):
        conn.original_system_prompt = conn.prompt
    if not hasattr(conn, "shadowing_index"):
        conn.shadowing_index = 0
    if not hasattr(conn, "shadowing_prompt"):
        conn.shadowing_prompt = ""
    if not hasattr(conn, "shadowing_clips"):
        conn.shadowing_clips = []
    if not hasattr(conn, "shadowing_awaiting_user"):
        conn.shadowing_awaiting_user = False
    if not hasattr(conn, "shadowing_clip_gen"):
        conn.shadowing_clip_gen = 0
    if not hasattr(conn, "shadowing_timeout_task"):
        conn.shadowing_timeout_task = None
    if not hasattr(conn, "last_asr_language"):
        conn.last_asr_language = None
    if not hasattr(conn, "task_session"):
        conn.task_session = None
    if not hasattr(conn, "last_asr_audio_snapshot"):
        conn.last_asr_audio_snapshot = None
    if not hasattr(conn, "shadowing_listen_pcm"):
        conn.shadowing_listen_pcm = []
    if not hasattr(conn, "shadowing_audio_saved_index"):
        conn.shadowing_audio_saved_index = -1
    if not hasattr(conn, "ensured_app_dirs"):
        conn.ensured_app_dirs = set()
    if not hasattr(conn, "repeater_speed_queue"):
        conn.repeater_speed_queue = []
    if not hasattr(conn, "repeater_play_name"):
        conn.repeater_play_name = ""
    if not hasattr(conn, "repeater_power_off_after"):
        conn.repeater_power_off_after = False
    if not hasattr(conn, "repeater_playlist"):
        conn.repeater_playlist = []
    if not hasattr(conn, "shadowing_last_speed"):
        conn.shadowing_last_speed = 1.0
    if not hasattr(conn, "shadowing_assess_index"):
        conn.shadowing_assess_index = 0
    if not hasattr(conn, "chat_greeted"):
        conn.chat_greeted = False
    if not hasattr(conn, "pending_play_after_tts"):
        conn.pending_play_after_tts = None


def _task_needs_sd_layout(task: Optional[Dict[str, Any]]) -> bool:
    if not task:
        return False
    mode = task.get("interaction_mode") or ""
    if mode in SD_LAYOUT_MODES:
        return True
    for skill in task.get("skills") or []:
        name = (skill or {}).get("name") or ""
        if name in ("self.mp3.play", DEVICE_PLAY_TOOL):
            return True
    return False


def app_sd_layout_paths(app_id: str, app: Optional[Dict[str, Any]] = None) -> list:
    """Return SD lesson dirs for an App: /{app_id}/{task_id}/lessons."""
    if not app_id:
        return []
    if app is None:
        app = get_app(app_id)
    if not app:
        return []
    paths = []
    for task_id, task in (app.get("tasks") or {}).items():
        if _task_needs_sd_layout(task):
            paths.append(f"/{app_id}/{task_id}/lessons")
    return paths


def _mcp_layout_ok(result) -> bool:
    if result is None:
        return False
    if result.action in (Action.ERROR, Action.NOTFOUND):
        return False
    blob = str(getattr(result, "response", None) or getattr(result, "result", None) or "")
    compact = blob.replace(" ", "").lower()
    if '"ok":false' in compact:
        return False
    return True


async def ensure_app_sd_dirs(conn: "ConnectionHandler", app_id: str) -> bool:
    """Create /{app} and /{app}/{task}/lessons on device SD. Never raises; missing SD is ignored."""
    _ensure_runtime_fields(conn)
    app = get_app(app_id)
    paths = app_sd_layout_paths(app_id, app)
    if not paths:
        return True
    already = getattr(conn, "ensured_app_dirs", None)
    if already is None:
        conn.ensured_app_dirs = set()
        already = conn.ensured_app_dirs
    if app_id in already:
        return True
    try:
        root = await _execute_named_tool(conn, DEVICE_ENSURE_APP_TOOL, {"app_id": app_id, "task_id": ""})
        if not _mcp_layout_ok(root):
            logger.bind(tag=TAG).warning(f"创建设备 App 目录失败 {app_id}: {getattr(root, 'response', root)}")
            return False
        for task_id, task in (app.get("tasks") or {}).items():
            if not _task_needs_sd_layout(task):
                continue
            result = await _execute_named_tool(
                conn, DEVICE_ENSURE_APP_TOOL, {"app_id": app_id, "task_id": task_id}
            )
            if not _mcp_layout_ok(result):
                logger.bind(tag=TAG).warning(
                    f"创建设备 Task 目录失败 {app_id}/{task_id}: {getattr(result, 'response', result)}"
                )
                return False
        already.add(app_id)
        logger.bind(tag=TAG).info(f"已确保 SD 目录 {paths}")
        return True
    except Exception as e:
        logger.bind(tag=TAG).warning(f"创建设备 App 目录异常 {app_id}: {e}")
        return False


async def ensure_all_subscribed_app_sd_dirs(conn: "ConnectionHandler") -> None:
    """After MCP tools/list, mkdir for every registry App that needs SD. Failures do not drop WS."""
    try:
        apps = (get_registry().get("apps") or {})
        for app_id, app in apps.items():
            if app_sd_layout_paths(app_id, app):
                await ensure_app_sd_dirs(conn, app_id)
    except Exception as e:
        logger.bind(tag=TAG).warning(f"订阅 App 建目录失败（忽略）: {e}")


def schedule_ensure_subscribed_app_sd_dirs(conn: "ConnectionHandler") -> None:
    """Fire-and-forget so tools/list handling is not blocked waiting for MCP mkdir."""

    async def _run():
        await ensure_all_subscribed_app_sd_dirs(conn)

    try:
        loop = getattr(conn, "loop", None)
        if loop:
            loop.create_task(_run())
        else:
            asyncio.create_task(_run())
    except Exception as e:
        logger.bind(tag=TAG).warning(f"无法调度 App 建目录: {e}")


def filter_tools(conn: "ConnectionHandler", tools: Dict[str, Any]) -> Dict[str, Any]:
@@ -94,6 +291,8 @@
            "你现在是英语翻译机。用户说话后应调用 translate；用户说退出翻译、回到聊天时调用 switch_app，app_id=chat。"
            "不要讲笑话、不要查天气、不要闲聊。"
        )
    elif mode == "bilingual":
        conn.change_system_prompt(BILINGUAL_PROMPT)
    elif mode == "agent_proactive":
        base = conn.original_system_prompt or conn.prompt or ""
        if CHAT_PROACTIVE_HINT not in (base or ""):
@@ -118,13 +317,58 @@

async def _stop_current_skills(conn: "ConnectionHandler") -> None:
    mode = getattr(conn, "interaction_mode", "CHAT")
    if mode == "PLAY_AUDIO":
    _cancel_shadowing_timeout(conn)
    if mode in ("PLAY_AUDIO", "SHADOWING"):
        await _execute_named_tool(conn, DEVICE_STOP_TOOL, {})
    conn.awaiting_voice_app_choice = False
    conn.pending_play_after_tts = None
    conn.shadowing_index = 0
    conn.shadowing_prompt = ""
    conn.shadowing_clips = []
    conn.shadowing_awaiting_user = False
    conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
    conn.shadowing_assess_index = 0
    conn.shadowing_listen_pcm = []
    conn.shadowing_audio_saved_index = -1
    conn.repeater_speed_queue = []
    conn.repeater_play_name = ""
    conn.repeater_power_off_after = False
    conn.repeater_playlist = []


def tts_channels_ready(conn: "ConnectionHandler") -> bool:
    tts = getattr(conn, "tts", None)
    if tts is None:
        return False
    play = getattr(tts, "audio_play_priority_thread", None)
    text_th = getattr(tts, "tts_priority_thread", None)
    return bool(
        play is not None
        and play.is_alive()
        and text_th is not None
        and text_th.is_alive()
    )


async def wait_for_tts_channels(conn: "ConnectionHandler", timeout: float = 8.0) -> bool:
    loop = asyncio.get_running_loop()
    start = loop.time()
    while (loop.time() - start) < timeout:
        if tts_channels_ready(conn):
            return True
        await asyncio.sleep(0.1)
    return tts_channels_ready(conn)


def _speak(conn: "ConnectionHandler", text: str) -> None:
    if not text:
        return
    tts = getattr(conn, "tts", None)
    if tts is None:
        logger.bind(tag=TAG).warning("TTS未初始化，跳过播报")
        return
    if not tts_channels_ready(conn):
        logger.bind(tag=TAG).warning("TTS通道未就绪，跳过播报")
        return
    from core.handle.intentHandler import speak_txt

@@ -140,6 +384,877 @@
    return (conn.config or {}).get("proactive_chat_greeting") or "我们来聊天吧。"


def _display_name(app: Optional[Dict[str, Any]], task: Optional[Dict[str, Any]] = None) -> str:
    if task and task.get("display_name"):
        return str(task["display_name"])
    if app and app.get("display_name"):
        return str(app["display_name"])
    if app and app.get("name"):
        return str(app["name"])
    return "当前模式"


def _enter_prompt(app: Dict[str, Any], task: Dict[str, Any]) -> str:
    return (task.get("enter_prompt") or app.get("enter_prompt") or "").strip()


def _switch_announcement(
    prev_app: Optional[Dict[str, Any]],
    prev_task: Optional[Dict[str, Any]],
    app: Dict[str, Any],
    task: Dict[str, Any],
    app_id: str,
) -> str:
    if app_id == "chat":
        prev_name = _display_name(prev_app, prev_task)
        topic = random.choice(CHAT_RETURN_TOPICS)
        return f"已经退出{prev_name}，现在是对话模式，我们来聊点{topic}吧。"
    return _enter_prompt(app, task) or f"现在已经切换至{_display_name(app, task)}。"


def _shadowing_sentences(conn: "ConnectionHandler") -> list:
    cfg = (conn.config or {}).get("app_runtime") or {}
    sentences = cfg.get("shadowing_sentences")
    if not sentences:
        _, task = get_task("english_learning", "shadowing")
        sentences = (task or {}).get("sentences")
    if not sentences:
        sentences = DEFAULT_SHADOWING_SENTENCES
    return [str(s).strip() for s in sentences if str(s).strip()]


def lesson_mp3_clips(names, app_id: str, task_id: str, lesson_id: str = "") -> list:
    """Pick sentence MP3s under /{app}/{task}/lessons/{lesson}/..."""
    prefix = f"/{app_id}/{task_id}/lessons/"
    clips = []
    for raw in names or []:
        name = str(raw or "").replace("\\", "/")
        if not name.lower().endswith(".mp3"):
            continue
        if not name.startswith("/"):
            name = "/" + name.lstrip("/")
        if not name.startswith(prefix):
            continue
        rest = name[len(prefix) :]
        if not rest:
            continue
        base = rest.rsplit("/", 1)[-1].lower()
        if base == "full.mp3" or base.endswith("_full.mp3"):
            continue
        if lesson_id:
            if rest != lesson_id and not rest.startswith(str(lesson_id) + "/"):
                continue
        clips.append(name)
    clips.sort()
    if lesson_id or not clips:
        return clips
    first_lesson = clips[0][len(prefix) :].split("/", 1)[0]
    return [
        name
        for name in clips
        if name[len(prefix) :] == first_lesson
        or name[len(prefix) :].startswith(first_lesson + "/")
    ]


def repeater_lesson_mp3(names, app_id: str, task_id: str) -> str:
    """Pick the full-lesson MP3 under /{app}/{task}/lessons/, else the first file there."""
    prefix = f"/{app_id}/{task_id}/lessons/"
    fulls = []
    others = []
    for raw in names or []:
        name = str(raw or "").replace("\\", "/")
        if not name.lower().endswith(".mp3"):
            continue
        if not name.startswith("/"):
            name = "/" + name.lstrip("/")
        if not name.startswith(prefix):
            continue
        base = name.rsplit("/", 1)[-1].lower()
        if base == "full.mp3" or base.endswith("_full.mp3"):
            fulls.append(name)
        else:
            others.append(name)
    fulls.sort()
    others.sort()
    if fulls:
        return fulls[0]
    return others[0] if others else ""


def _is_clip_path(text: str) -> bool:
    t = (text or "").replace("\\", "/").lower()
    return t.endswith(".mp3") or t.startswith("/")


def _shadowing_items(conn: "ConnectionHandler") -> list:
    clips = getattr(conn, "shadowing_clips", None) or []
    if clips:
        return list(clips)
    return _shadowing_sentences(conn)


def _begin_shadowing(conn: "ConnectionHandler", enter_text: str) -> str:
    sentences = _shadowing_sentences(conn)
    conn.shadowing_index = 0
    if not sentences:
        conn.shadowing_prompt = ""
        return enter_text or "现在已经切换至跟读模式，但还没有跟读句子。"
    conn.shadowing_prompt = sentences[0]
    prefix = enter_text or "现在已经切换至跟读模式。"
    return f"{prefix}请跟我读：{sentences[0]}"


def _cancel_shadowing_timeout(conn: "ConnectionHandler") -> None:
    task = getattr(conn, "shadowing_timeout_task", None)
    if task is not None:
        try:
            if not task.done():
                task.cancel()
        except Exception:
            pass
    conn.shadowing_timeout_task = None


def _task_for(conn: "ConnectionHandler") -> Optional[Dict[str, Any]]:
    session = current_session(conn) or getattr(conn, "task_session", None)
    app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", None) or "english_learning"
    task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", None) or "shadowing"
    _, task = get_task(app_id, task_id)
    return task


def _expected_for_index(conn: "ConnectionHandler", index: int) -> Optional[str]:
    session = current_session(conn) or getattr(conn, "task_session", None)
    app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or ""
    task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or ""
    clips = list(getattr(conn, "shadowing_clips", None) or [])
    return resolve_expected_text(
        _task_for(conn),
        clips,
        index,
        transcript_cache_dir(app_id, task_id),
    )


def apply_deferred_assessment(
    conn: "ConnectionHandler",
    step_index: int,
    session: Any = None,
    clips: Optional[list] = None,
    app_id: str = "",
    task_id: str = "",
    audio_snapshot: Any = None,
    persist_audio: bool = True,
) -> None:
    """Write expected_text + metrics onto the captured session. Never raises into play."""
    import time as _time

    session = session or getattr(conn, "task_session", None)
    if not session:
        return
    app_id = app_id or getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or ""
    task_id = task_id or getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or ""
    _, task = get_task(app_id or "english_learning", task_id or "shadowing")
    cfg = (task or {}).get("assessment") or {}
    clip_list = list(clips if clips is not None else (getattr(conn, "shadowing_clips", None) or []))
    turn = session.find_turn(step_index)
    asr = _plain_user_text((turn or {}).get("asr_text") or "")
    expected = (turn or {}).get("expected_text") or resolve_expected_text(
        task, clip_list, step_index, transcript_cache_dir(app_id, task_id)
    )
    snapshot = audio_snapshot
    if snapshot is None:
        snapshot = getattr(conn, "last_asr_audio_snapshot", None)
    wav_bytes = pcm_snapshot_to_wav(snapshot)
    existing_audio = (turn or {}).get("audio_id")
    audio_id = existing_audio
    if persist_audio and not existing_audio:
        audio_id = save_turn_wav(
            getattr(session, "task_session_id", "") or "",
            step_index,
            wav_bytes,
            conn=conn,
        )
    try:
        from core.handle.reportHandle import enqueue_asr_report

        enqueue_asr_report(
            conn,
            asr or "",
            snapshot,
            getattr(session, "task_session_id", None),
            int(step_index),
        )
    except Exception:
        pass
    clip = ""
    if 0 <= int(step_index) < len(clip_list):
        clip = str(clip_list[step_index] or "")
    metrics = []
    if not (cfg and cfg.get("enabled") is False):
        metrics = evaluate_turn(
            asr,
            expected or "",
            wav_bytes,
            cfg,
            extra={"clip": clip, "step_index": step_index},
        )
    session.upsert_turn(
        index=step_index,
        prompt_text=(turn or {}).get("prompt_text") or getattr(conn, "shadowing_prompt", "") or "",
        expected_text=expected,
        metrics=metrics or None,
        assessed_at=int(_time.time() * 1000) if metrics else None,
        audio_id=audio_id,
    )
    session.save()


def _schedule_deferred_assessment(conn: "ConnectionHandler", step_index: int) -> None:
    session = getattr(conn, "task_session", None)
    clips = list(getattr(conn, "shadowing_clips", None) or [])
    app_id = getattr(conn, "active_app_id", None) or getattr(session, "app_id", "") or ""
    task_id = getattr(conn, "current_task_id", None) or getattr(session, "task_id", "") or ""
    audio_snapshot = getattr(conn, "last_asr_audio_snapshot", None)

    async def _run():
        try:
            apply_deferred_assessment(
                conn,
                step_index,
                session=session,
                clips=clips,
                app_id=app_id,
                task_id=task_id,
                audio_snapshot=audio_snapshot,
                persist_audio=False,
            )
        except Exception as exc:
            logger.bind(tag=TAG).warning(f"跟读后评估失败: {exc}")

    loop = getattr(conn, "loop", None)
    try:
        loop = loop or asyncio.get_running_loop()
        loop.create_task(_run())
    except RuntimeError:
        try:
            apply_deferred_assessment(
                conn,
                step_index,
                session=session,
                clips=clips,
                app_id=app_id,
                task_id=task_id,
                audio_snapshot=audio_snapshot,
                persist_audio=False,
            )
        except Exception as exc:
            logger.bind(tag=TAG).warning(f"跟读后评估失败: {exc}")


def _schedule_shadowing_listen_timeout(conn: "ConnectionHandler", duration_ms: int) -> None:
    _cancel_shadowing_timeout(conn)
    ratio = resolve_listen_ratio(conn, _task_for(conn))
    wait_ms = listen_window_ms(
        duration_ms or 0,
        ratio,
        SHADOWING_LISTEN_MIN_MS,
        speed=getattr(conn, "shadowing_last_speed", None),
        wall_clock=True,
    )
    conn.shadowing_awaiting_user = True
    conn.shadowing_assess_index = int(getattr(conn, "shadowing_index", 0) or 0)
    conn.shadowing_audio_saved_index = -1
    conn.shadowing_listen_pcm = []
    gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0)
    logger.bind(tag=TAG).info(
        f"跟读听窗 {wait_ms}ms = 实际播放 {duration_ms}ms × {ratio}"
        f"（倍速已含在实际播放时长里） gen={gen}"
    )

    async def _fire():
        try:
            await asyncio.sleep(wait_ms / 1000.0)
        except asyncio.CancelledError:
            return
        if int(getattr(conn, "shadowing_clip_gen", 0) or 0) != gen:
            return
        if getattr(conn, "interaction_mode", "") != "SHADOWING":
            return
        if not getattr(conn, "shadowing_awaiting_user", False):
            return
        await _on_shadowing_listen_window_closed(conn)

    loop = getattr(conn, "loop", None)
    if loop:
        conn.shadowing_timeout_task = loop.create_task(_fire())
    else:
        conn.shadowing_timeout_task = asyncio.create_task(_fire())


async def close_shadowing_listen_window(conn: "ConnectionHandler") -> None:
    """Device listen stop: archive this step once, cancel the clock, advance index."""
    _cancel_shadowing_timeout(conn)
    await _on_shadowing_listen_window_closed(conn)


def _drain_asr_queue_into_shadowing_pcm(conn: "ConnectionHandler") -> int:
    """Binary opus is queued; listen stop/timeout run on the WS loop first. Take pending PCM now."""
    q = getattr(conn, "asr_audio_queue", None)
    if q is None or not getattr(conn, "shadowing_awaiting_user", False):
        return 0
    drained = 0
    while True:
        try:
            frame = q.get_nowait()
        except (queue.Empty, EOFError, OSError):
            break
        if frame:
            append_shadowing_listen_pcm(conn, frame)
            drained += 1
    if drained:
        logger.bind(tag=TAG).info(f"跟读关窗前抽干 ASR 队列 frames={drained}")
    return drained


def append_shadowing_listen_pcm(conn: "ConnectionHandler", pcm_frame: Any) -> None:
    if getattr(conn, "interaction_mode", "") != "SHADOWING":
        return
    if not getattr(conn, "shadowing_awaiting_user", False):
        return
    if not pcm_frame:
        return
    buf = getattr(conn, "shadowing_listen_pcm", None)
    if buf is None:
        conn.shadowing_listen_pcm = []
        buf = conn.shadowing_listen_pcm
    buf.append(pcm_frame)


def _take_shadowing_listen_pcm(conn: "ConnectionHandler") -> list:
    frames = list(getattr(conn, "shadowing_listen_pcm", None) or [])
    conn.shadowing_listen_pcm = []
    return frames


def save_shadowing_listen_audio(conn: "ConnectionHandler", step_index: Optional[int] = None) -> None:
    """Persist exactly one wav per follow-read step. Never merge neighboring windows."""
    if getattr(conn, "interaction_mode", "") != "SHADOWING":
        return
    index = int(
        step_index
        if step_index is not None
        else (getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0)
    )
    if int(getattr(conn, "shadowing_audio_saved_index", -1)) == index:
        conn.shadowing_listen_pcm = []
        return
    snapshot = _take_shadowing_listen_pcm(conn)
    if not snapshot:
        logger.bind(tag=TAG).warning(f"跟读听窗无 PCM，写静音占位 step={index}")
        snapshot = silent_pcm_snapshot()
    conn.shadowing_audio_saved_index = index
    from core.app_runtime.session import stash_asr_audio

    stash_asr_audio(conn, snapshot)
    try:
        apply_deferred_assessment(conn, index, audio_snapshot=snapshot)
    except Exception as exc:
        logger.bind(tag=TAG).warning(f"跟读听窗录音落盘失败 step={index}: {exc}")


async def _on_shadowing_listen_window_closed(conn: "ConnectionHandler") -> None:
    """Clock or listen stop: archive this listen. Device already sequences play → record → play."""
    if not getattr(conn, "shadowing_awaiting_user", False):
        return
    _drain_asr_queue_into_shadowing_pcm(conn)
    conn.shadowing_awaiting_user = False
    listened_idx = int(getattr(conn, "shadowing_index", 0) or 0)
    listened_prompt = getattr(conn, "shadowing_prompt", "") or ""
    save_shadowing_listen_audio(conn, listened_idx)
    items = _shadowing_items(conn)
    nxt = listened_idx + 1
    conn.shadowing_index = nxt
    if nxt < len(items):
        conn.shadowing_prompt = items[nxt]
    logger.bind(tag=TAG).info(f"跟读：听窗结束 step={listened_idx}")
    try:
        session = current_session(conn) or getattr(conn, "task_session", None)
        if session:
            if session.find_turn(listened_idx) is None:
                session.upsert_turn(
                    index=listened_idx,
                    prompt_text=listened_prompt,
                    asr_text="",
                    expected_text=_expected_for_index(conn, listened_idx),
                )
            session.add_event("listen_window_closed")
            session.add_event("listen_timeout")
            session.save()
    except Exception as exc:
        logger.bind(tag=TAG).warning(f"跟读听窗归档失败: {exc}")


def _parse_mp3_list_names(raw: Any) -> list:
    text = ""
    if raw is None:
        return []
    if hasattr(raw, "response") or hasattr(raw, "result"):
        text = str(getattr(raw, "response", None) or getattr(raw, "result", None) or "")
    else:
        text = str(raw)
    text = text.strip()
    if not text:
        return []
    try:
        import json

        data = json.loads(text)
    except Exception:
        return []
    files = data.get("files") if isinstance(data, dict) else None
    names = []
    if isinstance(files, list):
        for item in files:
            if isinstance(item, dict) and item.get("name"):
                names.append(str(item["name"]))
            elif isinstance(item, str):
                names.append(item)
    return names


async def _list_sd_mp3_names(conn: "ConnectionHandler") -> list:
    result = await _execute_named_tool(conn, DEVICE_LIST_TOOL, {})
    return _parse_mp3_list_names(result)


def _is_full_lesson_mp3(name: str) -> bool:
    base = str(name or "").replace("\\", "/").rsplit("/", 1)[-1].lower()
    return base == "full.mp3" or base.endswith("_full.mp3")


def _compact_shadowing_playlist(
    conn: "ConnectionHandler",
    params: Optional[Dict[str, Any]],
) -> List[Dict[str, Any]]:
    """Keep {name, repeat, speed}. Device expands repeats locally with a listen window each pass."""
    items = parse_playlist(params)
    if not items:
        items = default_shadowing_playlist(_task_for(conn))
    compact: List[Dict[str, Any]] = []
    for raw in items:
        if not isinstance(raw, dict):
            continue
        name = str(raw.get("name") or "")
        if not name or _is_full_lesson_mp3(name):
            continue
        try:
            repeat_n = int(raw.get("repeat") or 1)
        except (TypeError, ValueError):
            repeat_n = 1
        if repeat_n < 1:
            repeat_n = 1
        compact.append(
            {
                "name": name,
                "repeat": repeat_n,
                "speed": clamp_play_speed(
                    raw["speed"] if raw.get("speed") is not None else 1.0
                ),
                "power_off_after": False,
                "finish_task": False,
            }
        )
    return compact


def _begin_shadowing_playlist(
    conn: "ConnectionHandler",
    params: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
    """One shot: send the whole shadowing playlist. Device plays, records, plays."""
    items = _compact_shadowing_playlist(conn, params)
    clips = expand_playlist_plays(items)
    conn.shadowing_clips = clips
    conn.shadowing_index = 0
    conn.shadowing_prompt = clips[0] if clips else ""
    conn.shadowing_awaiting_user = False
    first = items[0] if items else dict(params or {})
    play = dict(first)
    play["items"] = items
    play["finish_task"] = False
    play["listen_ratio"] = resolve_listen_ratio(conn, _task_for(conn))
    if items:
        conn.shadowing_last_speed = items[0].get("speed")
    return play


async def _play_shadowing_clip(conn: "ConnectionHandler", name: str) -> bool:
    if getattr(conn, "interaction_mode", "") != "SHADOWING":
        logger.bind(tag=TAG).info(f"跳过跟读播放，当前已离开跟读: {name}")
        return False
    conn.shadowing_awaiting_user = False
    conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
    speed = resolve_play_speed(_task_for(conn))
    conn.shadowing_last_speed = speed
    notice = await _execute_named_tool_play(
        conn,
        {
            "name": name,
            "repeat": 1,
            "power_off_after": False,
            "speed": speed,
            "finish_task": False,
            "listen_ratio": resolve_listen_ratio(conn, _task_for(conn)),
        },
    )
    if notice and notice.action in (Action.ERROR, Action.NOTFOUND):
        return False
    return True


async def _prepare_shadowing_sd(conn: "ConnectionHandler", params: Optional[Dict[str, Any]]) -> bool:
    params = params or {}
    items = _compact_shadowing_playlist(conn, params)
    clips = expand_playlist_plays(items)
    if not clips:
        names = await _list_sd_mp3_names(conn)
        lesson = str(params.get("lesson") or "").strip()
        single = str(params.get("name") or params.get("path") or params.get("url") or "")
        if not lesson and single and not single.lower().endswith(".mp3"):
            lesson = single
        if lesson.lower().endswith(".mp3"):
            lesson = ""
        clips = lesson_mp3_clips(
            names,
            getattr(conn, "active_app_id", "english_learning"),
            getattr(conn, "current_task_id", "shadowing"),
            lesson,
        )
        clips = [name for name in clips if not _is_full_lesson_mp3(name)]
    if not clips:
        conn.shadowing_clips = []
        conn.shadowing_prompt = ""
        return False
    conn.shadowing_clips = clips
    conn.shadowing_index = 0
    conn.shadowing_prompt = clips[0]
    conn.shadowing_awaiting_user = False
    return True


async def _start_shadowing_sd(conn: "ConnectionHandler", params: Optional[Dict[str, Any]]) -> bool:
    if not await _prepare_shadowing_sd(conn, params):
        return False
    play = _begin_shadowing_playlist(conn, params)
    conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
    notice = await _execute_named_tool_play(conn, play)
    if notice and notice.action in (Action.ERROR, Action.NOTFOUND):
        return False
    return True


async def _fill_repeater_play_name(
    conn: "ConnectionHandler", params: Optional[Dict[str, Any]]
) -> Dict[str, Any]:
    play = dict(params or {})
    if play.get("items"):
        return play
    if str(play.get("name") or "").strip():
        return play
    stub = default_repeater_playlist(_task_for(conn))
    if stub:
        first = stub[0]
        play["name"] = first.get("name") or ""
        if play.get("repeat") is None:
            play["repeat"] = first.get("repeat") or 1
        if play.get("speed") is None and first.get("speed") is not None:
            play["speed"] = first.get("speed")
        conn.repeater_play_name = play["name"]
        return play
    try:
        names = await _list_sd_mp3_names(conn)
    except Exception as exc:
        logger.bind(tag=TAG).warning(f"列举复读课文失败: {exc}")
        return play
    picked = repeater_lesson_mp3(
        names,
        getattr(conn, "active_app_id", "") or "english_learning",
        getattr(conn, "current_task_id", "") or "repeater",
    )
    if picked:
        play["name"] = picked
        conn.repeater_play_name = picked
    return play


async def maybe_flush_pending_play_after_tts(conn: "ConnectionHandler") -> None:
    """Start queued lesson audio only after enter-prompt TTS has finished on device."""
    _ensure_runtime_fields(conn)
    pending = getattr(conn, "pending_play_after_tts", None)
    if not pending:
        return
    conn.pending_play_after_tts = None
    kind = pending.get("kind") if isinstance(pending, dict) else None
    try:
        if kind == "repeater":
            play_params = await _fill_repeater_play_name(conn, pending.get("params") or {})
            await _execute_named_tool_play(conn, play_params)
        elif kind == "shadowing":
            play_params = pending.get("params") if isinstance(pending, dict) else None
            if not play_params:
                play_params = _begin_shadowing_playlist(conn, {})
            conn.shadowing_awaiting_user = False
            conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
            await _execute_named_tool_play(conn, play_params)
    except Exception as exc:
        logger.bind(tag=TAG).warning(f"延后播放课文失败: {exc}")


def _lesson_end_speech(conn: "ConnectionHandler", phrase: str) -> str:
    """Short lesson-end cue plus the same default chat greeting used after 翻译/中英对话."""
    greeting = (_greeting_text(conn) or "").strip()
    phrase = (phrase or "").strip()
    if phrase and greeting:
        if phrase.endswith("。") or phrase.endswith("."):
            return f"{phrase}{greeting}"
        return f"{phrase}。{greeting}"
    return phrase or greeting


async def _return_to_chat_after_lesson(
    conn: "ConnectionHandler",
    phrase: str,
    reason: str = "play_completed",
) -> None:
    """Playlist already finished on device: speak the end cue now, then chat. Do not MCP-stop."""
    conn.repeater_speed_queue = []
    conn.repeater_playlist = []
    end_session(conn, reason)
    if getattr(conn, "tts", None) is not None:
        await wait_for_tts_channels(conn, timeout=3)
    conn.sentence_id = uuid.uuid4().hex
    _speak(conn, _lesson_end_speech(conn, phrase))
    await start_app_task(
        conn,
        "chat",
        "conversation",
        {},
        source="voice",
        execute_skills=False,
        speak_switch=False,
        stop_skills=False,
    )


async def _shadowing_play_next_or_finish(conn: "ConnectionHandler") -> None:
    """Legacy helper: device now sequences clips. Only used if a caller still asks to finish."""
    conn.shadowing_index = 0
    conn.shadowing_prompt = ""
    conn.shadowing_clips = []
    await _return_to_chat_after_lesson(conn, SHADOWING_END_PHRASE, "finished")


def _mark_repeater_day(conn: "ConnectionHandler") -> None:
    try:
        mark_lesson_played_today(
            str(getattr(conn, "device_id", "") or ""),
            getattr(conn, "active_app_id", "") or "",
            getattr(conn, "current_task_id", "") or "",
            getattr(conn, "repeater_play_name", "") or "",
        )
    except Exception as exc:
        logger.bind(tag=TAG).warning(f"记录复读天数失败: {exc}")


def _apply_playback_speed(
    conn: "ConnectionHandler",
    task: Optional[Dict[str, Any]],
    params: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
    """Attach clamped speed. Multi-speed repeats are Server-driven (one pass per play)."""
    play = dict(params or {})
    try:
        repeat_n = int(play.get("repeat") or 1)
    except (TypeError, ValueError):
        repeat_n = 1
    if repeat_n < 1:
        repeat_n = 1
    device_id = str(getattr(conn, "device_id", "") or "")
    app_id = getattr(conn, "active_app_id", "") or ""
    task_id = getattr(conn, "current_task_id", "") or ""
    lesson = str(play.get("name") or "")
    day_index = lesson_day_index(device_id, app_id, task_id, lesson)
    if play.get("speed") is not None:
        speeds = [clamp_play_speed(play.get("speed"))] * repeat_n
    else:
        speeds = speeds_for_repeats(task, repeat_n, day_index)
    conn.repeater_play_name = lesson
    conn.repeater_power_off_after = bool(play.get("power_off_after") or False)
    varied = len(speeds) > 1 and any(abs(item - speeds[0]) > 0.001 for item in speeds[1:])
    if varied:
        conn.repeater_speed_queue = speeds[1:]
        play["speed"] = speeds[0]
        play["repeat"] = 1
        play["power_off_after"] = False
    else:
        conn.repeater_speed_queue = []
        play["speed"] = speeds[0]
        play["repeat"] = repeat_n
    # 复读清单结束由 Server 收 audio.completed 后切回对话。设备不要 finish_task 写 NVS。
    play["finish_task"] = False
    return play


def _begin_repeater_playlist(
    conn: "ConnectionHandler",
    task: Optional[Dict[str, Any]],
    params: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
    """One shot: send the whole playlist. Repeat is just the same clip again, no mode switch."""
    items = parse_playlist(params)
    if not items:
        items = default_repeater_playlist(task)
    normalized = []
    for raw in items:
        if not isinstance(raw, dict):
            continue
        name = str(raw.get("name") or "")
        if not name:
            continue
        try:
            repeat_n = int(raw.get("repeat") or 1)
        except (TypeError, ValueError):
            repeat_n = 1
        if repeat_n < 1:
            repeat_n = 1
        item = {
            "name": name,
            "repeat": repeat_n,
            "speed": clamp_play_speed(
                raw["speed"] if raw.get("speed") is not None else 1.0
            ),
            "power_off_after": bool(raw.get("power_off_after") or False),
            "finish_task": False,
        }
        normalized.append(item)
    conn.repeater_playlist = []
    conn.repeater_speed_queue = []
    first = normalized[0] if normalized else dict(params or {})
    conn.repeater_play_name = str(first.get("name") or "")
    conn.repeater_power_off_after = bool(first.get("power_off_after") or False)
    play = dict(first)
    play["items"] = normalized
    play["finish_task"] = False
    return play


async def _continue_repeater_if_needed(conn: "ConnectionHandler") -> bool:
    queue = list(getattr(conn, "repeater_speed_queue", None) or [])
    if queue:
        nxt = queue.pop(0)
        conn.repeater_speed_queue = queue
        last_speed = len(queue) == 0 and not list(getattr(conn, "repeater_playlist", None) or [])
        try:
            play_params = await _fill_repeater_play_name(
                conn,
                {
                    "name": getattr(conn, "repeater_play_name", "") or "",
                    "speed": nxt,
                    "repeat": 1,
                    "power_off_after": last_speed
                    and bool(getattr(conn, "repeater_power_off_after", False)),
                    "finish_task": False,
                },
            )
            await _execute_named_tool_play(conn, play_params)
        except Exception as exc:
            logger.bind(tag=TAG).warning(f"复读下一遍失败: {exc}")
            conn.repeater_speed_queue = []
            return False
        return True

    rest = list(getattr(conn, "repeater_playlist", None) or [])
    if not rest:
        return False
    nxt, conn.repeater_playlist = rest[0], rest[1:]
    play_params = _apply_playback_speed(conn, _task_for(conn), nxt)
    play_params["finish_task"] = False
    try:
        play_params = await _fill_repeater_play_name(conn, play_params)
        await _execute_named_tool_play(conn, play_params)
    except Exception as exc:
        logger.bind(tag=TAG).warning(f"复读下一首失败: {exc}")
        conn.repeater_playlist = []
        return False
    return True


async def _execute_named_tool_play(conn, params):
    items = params.get("items") if isinstance(params.get("items"), list) else []
    first = items[0] if items else params
    play_args = {
        "name": (first.get("name") if isinstance(first, dict) else None) or params.get("name") or "",
        "repeat": int((first.get("repeat") if isinstance(first, dict) else None) or params.get("repeat") or 1),
        "power_off_after": bool(params.get("power_off_after") or False),
        "speed": clamp_play_speed(
            (first.get("speed") if isinstance(first, dict) and first.get("speed") is not None else None)
            if items
            else (params["speed"] if params.get("speed") is not None else 1.0)
        ),
        "finish_task": bool(params.get("finish_task", True)),
        "app_id": str(getattr(conn, "active_app_id", "") or ""),
        "task_id": str(getattr(conn, "current_task_id", "") or ""),
        "force": True,
    }
    if items:
        play_args["items"] = json.dumps(items, ensure_ascii=False)
        play_args["speed"] = clamp_play_speed(
            first["speed"] if first.get("speed") is not None else 1.0
        )
        play_args["power_off_after"] = bool(
            items[-1].get("power_off_after") if items else False
        )
    if params.get("listen_ratio") is not None:
        try:
            play_args["listen_ratio"] = float(params["listen_ratio"])
        except (TypeError, ValueError):
            pass
    if play_args["task_id"] in ("repeater", "shadowing"):
        play_args["finish_task"] = False
    result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args)
    blob = ""
    if result:
        blob = str(result.response or result.result or "")
    compact = blob.replace(" ", "")
    if '"duplicate_ignored":true' in compact:
        logger.bind(tag=TAG).warning("设备忽略了同一课文的下一遍，强制再播一次")
        play_args["speed"] = clamp_play_speed(float(play_args["speed"]) + 0.02)
        play_args["force"] = True
        result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args)
        blob = ""
        if result:
            blob = str(result.response or result.result or "")
        compact = blob.replace(" ", "")
    rejected = (result and result.action in (Action.ERROR, Action.NOTFOUND)) or (
        "no_mp3" in blob
        or '"accepted":false' in compact
        or '"duplicate_ignored":true' in compact
    )
    if rejected:
        if not play_args["name"]:
            return ActionResponse(
                action=Action.RESPONSE,
                response="复读机已打开，但 SD 卡上还没有可播放的课文。",
            )
        return ActionResponse(
            action=Action.RESPONSE,
            response=(result.response if result else None) or "这台设备没有本地复读播放能力",
        )
    return None


async def start_app_task(
    conn: "ConnectionHandler",
    app_id: str,
@@ -147,6 +1262,8 @@
    params: Optional[Dict[str, Any]] = None,
    source: str = "voice",
    execute_skills: bool = True,
    speak_switch: bool = True,
    stop_skills: bool = True,
) -> ActionResponse:
    """Single entry for MQTT / schedule / voice / default chat."""
    _ensure_runtime_fields(conn)
@@ -176,8 +1293,16 @@
            response="当前有更高优先级的任务，先完成后再切换",
        )

    if not same_task or new_p >= cur_p:
        await _stop_current_skills(conn)
    prev_app_id = getattr(conn, "active_app_id", "chat")
    prev_task_id = getattr(conn, "current_task_id", "conversation")
    prev_app, prev_task = get_task(prev_app_id, prev_task_id)
    switching = (prev_app_id != app_id) or (prev_task_id != resolved_task_id)

    if not same_task:
        if stop_skills:
            await _stop_current_skills(conn)
        if app_id != "chat":
            conn.chat_greeted = False

    conn.active_app_id = app_id
    conn.current_task_id = resolved_task_id
@@ -192,51 +1317,92 @@
        f"start_app_task app={app_id} task={resolved_task_id} mode={conn.interaction_mode} source={source}"
    )

    if app_id == "chat":
        end_session(conn, "user_exit")
    elif switching:
        begin_session(conn, app_id, resolved_task_id, source, conn.interaction_mode)

    pending_play = None
    if execute_skills:
        if app_sd_layout_paths(app_id, app):
            await ensure_app_sd_dirs(conn, app_id)
        for skill in task.get("skills") or []:
            skill_name = (skill or {}).get("name") or ""
            if skill_name in ("self.mp3.play", DEVICE_PLAY_TOOL):
                play_args = {
                    "name": params.get("name") or "",
                    "repeat": int(params.get("repeat") or 1),
                    "power_off_after": bool(params.get("power_off_after") or False),
                }
                if params.get("speed") is not None:
                    play_args["speed"] = params.get("speed")
                if not play_args["name"]:
                    return ActionResponse(
                        action=Action.RESPONSE,
                        response="没有指定要复读的文件名，请先用 self.mp3.get_list 确认 files[].name",
                    )
                result = await _execute_named_tool(conn, DEVICE_PLAY_TOOL, play_args)
                if result and result.action in (Action.ERROR, Action.NOTFOUND):
                    return ActionResponse(
                        action=Action.RESPONSE,
                        response=result.response or "这台设备没有本地复读播放能力",
                    )
                play_params = _begin_repeater_playlist(conn, task, params)
                pending_play = {"kind": "repeater", "params": play_params}
            elif skill_name == "self.power_off":
                await _execute_named_tool(conn, "self_power_off", {})

    if conn.interaction_mode == "CHAT":
    announce = ""
    boot_chat = app_id == "chat" and source == "default_chat"
    if boot_chat:
        # 开机进 chat：必须播欢迎词。不要被 same_task / just_woken_up 跳过。
        enable = (conn.config or {}).get("enable_proactive_greeting", True)
        if enable and not getattr(conn, "chat_greeted", False):
            if not await wait_for_tts_channels(conn, timeout=10):
                logger.bind(tag=TAG).error("开机欢迎词跳过：TTS通道未就绪")
            else:
                conn.chat_greeted = True
                conn.sentence_id = uuid.uuid4().hex
                _speak(conn, _greeting_text(conn))
    elif switching:
        announce = _switch_announcement(prev_app, prev_task, app, task, app_id)
        if conn.interaction_mode == "SHADOWING":
            started = await _prepare_shadowing_sd(conn, params)
            if not started:
                prefix = announce or "现在已经切换至跟读模式。"
                announce = f"{prefix}跟读课文还没有放到SD卡。"
            else:
                announce = announce or "现在已经切换至跟读模式，请跟着课文读。"
                pending_play = {
                    "kind": "shadowing",
                    "params": _begin_shadowing_playlist(conn, params),
                }
        if speak_switch and announce:
            conn.sentence_id = uuid.uuid4().hex
            conn.pending_play_after_tts = pending_play
            _speak(conn, announce)
        elif pending_play:
            conn.pending_play_after_tts = pending_play
            await maybe_flush_pending_play_after_tts(conn)
        if app_id == "chat":
            conn.chat_greeted = True
    elif conn.interaction_mode == "CHAT":
        enable = (conn.config or {}).get("enable_proactive_greeting", True)
        woken = bool(getattr(conn, "just_woken_up", False))
        if enable and not woken:
        if enable and not woken and not getattr(conn, "chat_greeted", False):
            conn.chat_greeted = True
            conn.sentence_id = uuid.uuid4().hex
            _speak(conn, _greeting_text(conn))
    elif conn.interaction_mode == "SHADOWING" and not getattr(conn, "shadowing_clips", None):
        started = await _start_shadowing_sd(conn, params)
        if not started and speak_switch:
            conn.sentence_id = uuid.uuid4().hex
            _speak(conn, "跟读课文还没有放到SD卡。")
    elif pending_play:
        conn.pending_play_after_tts = pending_play
        await maybe_flush_pending_play_after_tts(conn)

    app_name = app.get("name") or app_id
    app_name = _display_name(app, task)
    return ActionResponse(
        action=Action.NONE,
        result=f"{app_id}/{resolved_task_id}",
        response=f"已打开{app_name}",
        response=announce or f"已打开{app_name}",
    )


async def switch_app(conn: "ConnectionHandler", app_id: str, source: str = "voice") -> ActionResponse:
async def switch_app(
    conn: "ConnectionHandler",
    app_id: str,
    source: str = "voice",
    task_id: str = "",
) -> ActionResponse:
    app = get_app(app_id)
    if not app:
        return ActionResponse(action=Action.ERROR, response=f"不支持的 App: {app_id}")
    task_id = app.get("default_task") or ""
    return await start_app_task(conn, app_id, task_id, {}, source)
    resolved = task_id or app.get("default_task") or ""
    return await start_app_task(conn, app_id, resolved, {}, source)


async def on_user_wakeup(conn: "ConnectionHandler") -> None:
@@ -248,24 +1414,11 @@


def match_voice_switch(text: str) -> Optional[tuple]:
    """Map a short phrase to (app_id, task_id) or ('__menu__', '')."""
    """Map a short phrase to (app_id, task_id) using apps.json voice_enter / voice_exit."""
    if not text:
        return None
    _, filtered = _strip(text)
    if filtered in ("切换模式", "qie huan mo shi"):
        return ("__menu__", "")
    for p in CHAT_PHRASES:
        if p in filtered:
            return ("chat", "conversation")
    for p in TRANSLATE_PHRASES:
        if p in filtered:
            return ("english_translation", "live_translate")
    for p in REPEATER_PHRASES:
        if p in filtered:
            return ("english_learning", "repeater")
    if any(k in filtered for k in ("数字背诵", "成语接龙", "口算", "加减法")):
        return ("__reserved__", "")
    return None
    return match_registry_voice(filtered)


def _strip(text: str) -> tuple:
@@ -274,9 +1427,23 @@
    return remove_punctuation_and_length(text)


def is_exit_to_chat(text: str) -> bool:
    mapped = match_voice_switch(text)
    return bool(mapped and mapped[0] == "chat")


def should_skip_listen_stop_asr(conn: Any) -> bool:
    """True when listen stop is closing the mic for lesson playback, not a follow-read."""
    mode = getattr(conn, "interaction_mode", "CHAT")
    if mode == "PLAY_AUDIO":
        return True
    if mode == "SHADOWING" and not getattr(conn, "shadowing_awaiting_user", False):
        return True
    return False


def is_exit_translation(text: str) -> bool:
    _, filtered = _strip(text)
    return any(p in filtered for p in ("退出翻译", "结束翻译", "回到聊天", "我们来聊天吧", "聊会天吧"))
    return is_exit_to_chat(text)


def _plain_user_text(text: str) -> str:
@@ -296,6 +1463,40 @@
    return raw


def _norm_for_echo(text: str) -> str:
    _, filtered = _strip(text or "")
    return re.sub(r"\s+", "", (filtered or "").lower())


def is_echo_of_prompt(user_text: str, prompt: str) -> bool:
    a = _norm_for_echo(user_text)
    b = _norm_for_echo(prompt)
    if not a or not b:
        return False
    if a == b or a in b or b in a:
        return True
    return difflib.SequenceMatcher(None, a, b).ratio() >= 0.82


def detect_user_language(text: str, hint: Optional[str] = None) -> str:
    if hint:
        h = str(hint).lower()
        if h.startswith("zh") or h in ("cn", "yue", "chinese"):
            return "zh"
        if h.startswith("en") or h in ("english",):
            return "en"
    body = _plain_user_text(text)
    cjk = len(re.findall(r"[\u4e00-\u9fff]", body))
    latin = len(re.findall(r"[A-Za-z]", body))
    return "zh" if cjk >= max(latin, 1) else "en"


def wrap_bilingual_user_text(text: str, user_lang: str) -> str:
    if user_lang == "zh":
        return "Respond in English only. Do not use Chinese characters.\n\n" + text
    return "请只用中文回答，不要出现英文单词或句子。\n\n" + text


def _looks_like_tool_error(text: str) -> bool:
    if not text:
        return True
@@ -307,6 +1508,180 @@
    return False


def _record_shadowing_turn(
    conn: "ConnectionHandler",
    text: str,
    echo_rejected: bool = False,
    index: Optional[int] = None,
) -> None:
    session = current_session(conn)
    if not session:
        return
    if index is None:
        index = int(getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0)
    session.upsert_turn(
        index=index,
        prompt_text=getattr(conn, "shadowing_prompt", "") or "",
        asr_text=_plain_user_text(text),
        echo_rejected=echo_rejected,
        expected_text=_expected_for_index(conn, index),
    )
    session.save()


def apply_shadowing_user_text(conn: "ConnectionHandler", text: str) -> str:
    """Apply one ASR utterance. Returns exit | ignore | echo | recorded.

    ASR never advances the clip index; the listen-window clock does.
    Late ASR after the window still scores the sentence that was just listened to.
    Device plays the next clip itself; Server does not MCP the next sentence.
    """
    if is_exit_to_chat(text):
        end_session(conn, "user_exit")
        return "exit"
    if getattr(conn, "client_is_speaking", False) and not getattr(
        conn, "shadowing_awaiting_user", False
    ):
        logger.bind(tag=TAG).info("跟读：播放中忽略拾音，避免喇叭回声触发对话")
        return "ignore"
    prompt = getattr(conn, "shadowing_prompt", "") or ""
    if (
        getattr(conn, "shadowing_awaiting_user", False)
        and prompt
        and not _is_clip_path(prompt)
        and is_echo_of_prompt(text, prompt)
    ):
        logger.bind(tag=TAG).info("跟读：忽略与范读过于相似的识别结果")
        session = current_session(conn)
        if session:
            session.add_event("echo_rejected")
            session.save()
        return "echo"
    _record_shadowing_turn(conn, text, echo_rejected=False)
    return "recorded"


async def handle_shadowing_turn(conn: "ConnectionHandler", text: str) -> None:
    """Score this utterance off the chat path. Next clip is sequenced on device."""
    outcome = apply_shadowing_user_text(conn, text)
    if outcome == "exit":
        _cancel_shadowing_timeout(conn)
        await start_app_task(conn, "chat", "conversation", {}, source="voice")
        return
    if outcome in ("ignore", "echo"):
        return

    step_index = int(
        getattr(conn, "shadowing_assess_index", getattr(conn, "shadowing_index", 0)) or 0
    )
    logger.bind(tag=TAG).info(f"跟读 ASR 仅评估不上对话 step={step_index}")
    _schedule_deferred_assessment(conn, step_index)


async def handle_skill_event(conn: "ConnectionHandler", text: str) -> bool:
    """Handle `[skill_event]audio.completed|app|task|duration_ms`. Returns True if consumed."""
    parsed = parse_skill_event(text)
    if parsed is None:
        return False
    _ensure_runtime_fields(conn)
    name = parsed.get("name") or ""
    app_id = parsed.get("app_id") or ""
    task_id = parsed.get("task_id") or ""
    try:
        duration_ms = int(parsed.get("duration_ms") or 0)
    except (TypeError, ValueError):
        duration_ms = 0
    logger.bind(tag=TAG).info(f"skill_event {name} {app_id}/{task_id} duration_ms={duration_ms}")

    mode = getattr(conn, "interaction_mode", "CHAT")
    # 听窗时钟必须先于 session.save()：磁盘/HTTP 不能挡住 1.2×duration 的热路径。
    shadowing_playlist_done = name in ("audio.playlist_completed", "playlist.completed") and (
        mode == "SHADOWING" or task_id == "shadowing"
    )
    if shadowing_playlist_done:
        _cancel_shadowing_timeout(conn)
        if getattr(conn, "shadowing_awaiting_user", False):
            await _on_shadowing_listen_window_closed(conn)
        else:
            save_shadowing_listen_audio(conn)
        session = getattr(conn, "task_session", None)
        if session:
            session.add_event(name, app_id, task_id)
        await _return_to_chat_after_lesson(conn, SHADOWING_END_PHRASE, "finished")
        return True
    if mode == "SHADOWING" and name == "audio.completed":
        if getattr(conn, "shadowing_awaiting_user", False):
            await _on_shadowing_listen_window_closed(conn)
        conn.shadowing_clip_gen = int(getattr(conn, "shadowing_clip_gen", 0) or 0) + 1
        _schedule_shadowing_listen_timeout(conn, duration_ms)
        session = getattr(conn, "task_session", None)
        if session:
            session.add_event(name, app_id, task_id)
        return True
    if mode == "SHADOWING" and name == "audio.stopped":
        session = getattr(conn, "task_session", None)
        if session:
            session.add_event(name, app_id, task_id)
        end_session(conn, "play_stopped")
        await start_app_task(
            conn,
            "chat",
            "conversation",
            {},
            source="voice",
            execute_skills=False,
            speak_switch=False,
        )
        return True

    if mode == "PLAY_AUDIO" and name == "audio.completed":
        continued = await _continue_repeater_if_needed(conn)
        session = getattr(conn, "task_session", None)
        if session:
            session.add_event(name, app_id, task_id)
        if continued:
            if session:
                session.save()
            return True
        _mark_repeater_day(conn)
        await _return_to_chat_after_lesson(conn, REPEATER_END_PHRASE, "play_completed")
        return True
    if mode == "PLAY_AUDIO" and name == "audio.stopped":
        conn.repeater_speed_queue = []

    session = getattr(conn, "task_session", None)
    if session:
        session.add_event(name, app_id, task_id)
        session.save()

    if name == "audio.completed":
        end_session(conn, "play_completed")
        await start_app_task(
            conn,
            "chat",
            "conversation",
            {},
            source="voice",
            execute_skills=False,
            speak_switch=True,
        )
    elif name == "audio.stopped":
        end_session(conn, "play_stopped")
        if getattr(conn, "active_app_id", "chat") != "chat":
            await start_app_task(
                conn,
                "chat",
                "conversation",
                {},
                source="voice",
                execute_skills=False,
                speak_switch=False,
            )
    else:
        logger.bind(tag=TAG).warning(f"未知 skill_event: {name}")
    return True


async def run_translation(conn: "ConnectionHandler", text: str) -> str:
    """Translate with the connection LLM first (智控台已配的模型). MCP is fallback only."""
    text = _plain_user_text(text)