core/api/chat_handler.py

"""
聊天请求编排：解析 session_id、调度 browser/tab/session、调用插件流式补全，
并在响应末尾附加零宽字符编码的会话 ID。

当前调度模型：

- 一个浏览器对应一个代理组
- 一个浏览器内，一个 type 只有一个 tab
- 一个 tab 绑定一个 account，只有 drained 后才能切号
- 一个 session 绑定到某个 tab/account；复用成功时不传完整历史
- 无法复用时，新建会话并回放完整历史
"""

from __future__ import annotations

import asyncio
import json
import logging
import time
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Any, AsyncIterator, cast

from playwright.async_api import BrowserContext, Page

from core.account.pool import AccountPool
from core.config.repository import ConfigRepository
from core.config.schema import AccountConfig, ProxyGroupConfig
from core.config.settings import get
from core.constants import TIMEZONE
from core.plugin.base import (
    AccountFrozenError,
    BaseSitePlugin,
    BrowserResourceInvalidError,
    PluginRegistry,
)
from core.plugin.helpers import clear_cookies_for_domain
from core.runtime.browser_manager import BrowserManager, ClosedTabInfo, TabRuntime
from core.runtime.keys import ProxyKey
from core.runtime.local_proxy_forwarder import LocalProxyForwarder, UpstreamProxy, parse_proxy_server
from core.runtime.session_cache import SessionCache, SessionEntry

from core.api.conv_parser import parse_conv_uuid_from_messages, session_id_suffix
from core.api.fingerprint import compute_conversation_fingerprint
from core.api.react import format_react_prompt
from core.api.schemas import OpenAIChatRequest, extract_user_content
from core.hub.schemas import OpenAIStreamEvent
from core.runtime.conversation_index import ConversationIndex

logger = logging.getLogger(__name__)


def _request_messages_as_dicts(req: OpenAIChatRequest) -> list[dict[str, Any]]:
    """转为 conv_parser 需要的 list[dict]。"""
    out: list[dict[str, Any]] = []
    for m in req.messages:
        d: dict[str, Any] = {"role": m.role}
        if isinstance(m.content, list):
            d["content"] = [p.model_dump() for p in m.content]
        else:
            d["content"] = m.content
        out.append(d)
    return out


def _proxy_key_for_group(group: ProxyGroupConfig) -> ProxyKey:
    return ProxyKey(
        group.proxy_host,
        group.proxy_user,
        group.fingerprint_id,
        group.use_proxy,
        group.timezone or TIMEZONE,
    )


@dataclass
class _RequestTarget:
    proxy_key: ProxyKey
    group: ProxyGroupConfig
    account: AccountConfig
    context: BrowserContext
    page: Page
    session_id: str | None
    full_history: bool
    proxy_url: str | None = None
    proxy_auth: tuple[str, str] | None = None
    proxy_forwarder: LocalProxyForwarder | None = None


class ChatHandler:
    """编排一次 chat 请求：会话解析、tab 调度、插件调用。"""

    def __init__(
        self,
        pool: AccountPool,
        session_cache: SessionCache,
        browser_manager: BrowserManager,
        config_repo: ConfigRepository | None = None,
    ) -> None:
        self._pool = pool
        self._session_cache = session_cache
        self._browser_manager = browser_manager
        self._config_repo = config_repo
        self._conv_index = ConversationIndex()
        self._schedule_lock = asyncio.Lock()
        self._stop_event = asyncio.Event()
        self._busy_sessions: set[str] = set()
        self._tab_max_concurrent = int(get("scheduler", "tab_max_concurrent") or 5)
        self._gc_interval_seconds = float(
            get("scheduler", "browser_gc_interval_seconds") or 300
        )
        self._tab_idle_seconds = float(get("scheduler", "tab_idle_seconds") or 900)
        self._resident_browser_count = int(
            get("scheduler", "resident_browser_count", 1)
        )

    def reload_pool(
        self,
        groups: list[ProxyGroupConfig],
        config_repo: ConfigRepository | None = None,
    ) -> None:
        """配置热更新后替换账号池与 repository。"""
        self._pool.reload(groups)
        if config_repo is not None:
            self._config_repo = config_repo

    async def refresh_configuration(
        self,
        groups: list[ProxyGroupConfig],
        config_repo: ConfigRepository | None = None,
    ) -> None:
        """配置热更新：替换账号池、清理失效资源，并重新预热常驻浏览器。"""
        async with self._schedule_lock:
            self.reload_pool(groups, config_repo)
            await self._prune_invalid_resources_locked()
            await self._reconcile_tabs_locked()
        await self.prewarm_resident_browsers()

    async def prewarm_resident_browsers(self) -> None:
        """启动时预热常驻浏览器，并为其下可用 type 建立 tab。"""
        async with self._schedule_lock:
            warmed = 0
            for group in self._pool.groups():
                if warmed >= self._resident_browser_count:
                    break
                available_types = {
                    a.type
                    for a in group.accounts
                    if a.is_available() and PluginRegistry.get(a.type) is not None
                }
                if not available_types:
                    continue
                proxy_key = _proxy_key_for_group(group)
                await self._browser_manager.ensure_browser(proxy_key, group.proxy_pass)
                for type_name in sorted(available_types):
                    if self._browser_manager.get_tab(proxy_key, type_name) is not None:
                        continue
                    account = self._pool.available_accounts_in_group(group, type_name)
                    if not account:
                        continue
                    chosen = account[0]
                    plugin = PluginRegistry.get(type_name)
                    if plugin is None:
                        continue
                    await self._browser_manager.open_tab(
                        proxy_key,
                        group.proxy_pass,
                        type_name,
                        self._pool.account_id(group, chosen),
                        plugin.create_page,
                        self._make_apply_auth_fn(plugin, chosen),
                    )
                warmed += 1

    async def run_maintenance_loop(self) -> None:
        """周期性回收空闲浏览器，并收尾 drained/frozen tab。"""
        while not self._stop_event.is_set():
            try:
                await asyncio.wait_for(
                    self._stop_event.wait(),
                    timeout=self._gc_interval_seconds,
                )
                break
            except asyncio.TimeoutError:
                pass

            try:
                async with self._schedule_lock:
                    # Evict stale sessions to prevent unbounded accumulation.
                    stale_ids = self._session_cache.evict_stale()
                    # Evict stale fingerprint entries in sync.
                    stale_fp = self._conv_index.evict_stale(ttl=1800.0)
                    if stale_fp:
                        logger.info(
                            "[maintenance] evicted %d stale fingerprint entries",
                            len(stale_fp),
                        )
                    if stale_ids:
                        for sid in stale_ids:
                            plugin_type = None
                            for pk, entry in self._browser_manager.list_browser_entries():
                                for tn, tab in entry.tabs.items():
                                    if sid in tab.sessions:
                                        tab.sessions.discard(sid)
                                        plugin_type = tn
                                        break
                            if plugin_type:
                                plugin = PluginRegistry.get(plugin_type)
                                if plugin is not None:
                                    plugin.drop_session(sid)
                        logger.info(
                            "[maintenance] evicted %d stale sessions, cache size=%d",
                            len(stale_ids),
                            len(self._session_cache),
                        )
                    await self._reconcile_tabs_locked()
                    closed = await self._browser_manager.collect_idle_browsers(
                        idle_seconds=self._tab_idle_seconds,
                        resident_browser_count=self._resident_browser_count,
                    )
                    self._apply_closed_tabs_locked(closed)
            except Exception:
                logger.exception("维护循环执行失败")

    async def shutdown(self) -> None:
        """停止维护循环并关闭全部浏览器。"""
        self._stop_event.set()
        async with self._schedule_lock:
            closed = await self._browser_manager.close_all()
            self._apply_closed_tabs_locked(closed)

    def report_account_unfreeze(
        self,
        fingerprint_id: str,
        account_name: str,
        unfreeze_at: int,
    ) -> None:
        """记录账号解冻时间，并同步更新内存账号池。"""
        if self._config_repo is None:
            return
        self._config_repo.update_account_unfreeze_at(
            fingerprint_id, account_name, unfreeze_at
        )
        self._pool.update_account_unfreeze_at(
            fingerprint_id,
            account_name,
            unfreeze_at,
        )

    def get_account_runtime_status(self) -> dict[str, dict[str, Any]]:
        """返回当前账号运行时状态，供配置页展示角标。"""
        status: dict[str, dict[str, Any]] = {}
        for proxy_key, entry in self._browser_manager.list_browser_entries():
            for type_name, tab in entry.tabs.items():
                status[tab.account_id] = {
                    "fingerprint_id": proxy_key.fingerprint_id,
                    "type": type_name,
                    "is_active": True,
                    "tab_state": tab.state,
                    "accepting_new": tab.accepting_new,
                    "active_requests": tab.active_requests,
                    "frozen_until": tab.frozen_until,
                }
        return status

    def _make_apply_auth_fn(
        self,
        plugin: Any,
        account: AccountConfig,
    ) -> Any:
        async def _apply_auth(context: BrowserContext, page: Page) -> None:
            await plugin.apply_auth(context, page, account.auth)

        return _apply_auth

    def _apply_closed_tabs_locked(self, closed_tabs: list[ClosedTabInfo]) -> None:
        for info in closed_tabs:
            self._session_cache.delete_many(info.session_ids)
            plugin = PluginRegistry.get(info.type_name)
            if plugin is not None:
                plugin.drop_sessions(info.session_ids)

    def _stream_proxy_settings(
        self,
        target: _RequestTarget,
    ) -> tuple[str | None, tuple[str, str] | None, LocalProxyForwarder | None]:
        if not target.proxy_key.use_proxy:
            return (None, None, None)
        upstream_host, upstream_port = parse_proxy_server(target.proxy_key.proxy_host)
        forwarder = LocalProxyForwarder(
            UpstreamProxy(
                host=upstream_host,
                port=upstream_port,
                username=target.proxy_key.proxy_user,
                password=target.group.proxy_pass,
            ),
            listen_host="127.0.0.1",
            listen_port=0,
            on_log=lambda msg: logger.debug("[stream-proxy] %s", msg),
        )
        forwarder.start()
        return (
            forwarder.proxy_url,
            None,
            forwarder,
        )

    async def _clear_tab_domain_cookies_if_supported(
        self, proxy_key: ProxyKey, type_name: str
    ) -> None:
        """关 tab 前清该 type 对应域名的 cookie（仅支持带 site.cookie_domain 的插件）。"""
        entry = self._browser_manager.get_browser_entry(proxy_key)
        if entry is None:
            return
        plugin = PluginRegistry.get(type_name)
        if not isinstance(plugin, BaseSitePlugin) or not getattr(plugin, "site", None):
            return
        try:
            await clear_cookies_for_domain(entry.context, plugin.site.cookie_domain)
        except Exception as e:
            logger.debug("关 tab 前清 cookie 失败 type=%s: %s", type_name, e)

    async def _prune_invalid_resources_locked(self) -> None:
        """关闭配置中已不存在的浏览器/tab，避免热更新后继续使用失效资源。"""
        for proxy_key, entry in list(self._browser_manager.list_browser_entries()):
            group = self._pool.get_group_by_proxy_key(proxy_key)
            if group is None:
                self._apply_closed_tabs_locked(
                    await self._browser_manager.close_browser(proxy_key)
                )
                continue
            for type_name in list(entry.tabs.keys()):
                tab = entry.tabs[type_name]
                pair = self._pool.get_account_by_id(tab.account_id)
                if (
                    pair is None
                    or pair[0] is not group
                    or pair[1].type != type_name
                    or not pair[1].enabled
                ):
                    self._invalidate_tab_sessions_locked(proxy_key, type_name)
                    if tab.active_requests == 0:
                        # 与 reconcile 一致：优先同组同一页 re-auth，失败或无可用账号再关 tab
                        switched = False
                        group = self._pool.get_group_by_proxy_key(proxy_key)
                        if group is not None:
                            next_account = self._pool.next_available_account_in_group(
                                group,
                                type_name,
                                exclude_account_ids={tab.account_id},
                            )
                            if next_account is not None:
                                plugin = PluginRegistry.get(type_name)
                                if plugin is not None:
                                    switched = (
                                        await self._browser_manager.switch_tab_account(
                                            proxy_key,
                                            type_name,
                                            self._pool.account_id(group, next_account),
                                            self._make_apply_auth_fn(
                                                plugin,
                                                next_account,
                                            ),
                                        )
                                    )
                        if not switched:
                            await self._clear_tab_domain_cookies_if_supported(
                                proxy_key, type_name
                            )
                            closed = await self._browser_manager.close_tab(
                                proxy_key, type_name
                            )
                            if closed is not None:
                                self._apply_closed_tabs_locked([closed])
                    else:
                        self._browser_manager.mark_tab_draining(proxy_key, type_name)

    def _invalidate_session_locked(
        self,
        session_id: str,
        entry: SessionEntry | None = None,
    ) -> None:
        entry = entry or self._session_cache.get(session_id)
        if entry is None:
            return
        self._session_cache.delete(session_id)
        self._conv_index.remove_session(session_id)
        self._browser_manager.unregister_session(
            entry.proxy_key,
            entry.type_name,
            session_id,
        )
        plugin = PluginRegistry.get(entry.type_name)
        if plugin is not None:
            plugin.drop_session(session_id)

    def _invalidate_tab_sessions_locked(
        self,
        proxy_key: ProxyKey,
        type_name: str,
    ) -> None:
        tab = self._browser_manager.get_tab(proxy_key, type_name)
        if tab is None or not tab.sessions:
            return
        session_ids = list(tab.sessions)
        self._session_cache.delete_many(session_ids)
        plugin = PluginRegistry.get(type_name)
        if plugin is not None:
            plugin.drop_sessions(session_ids)
        tab.sessions.clear()

    async def _recover_browser_resource_invalid_locked(
        self,
        type_name: str,
        target: _RequestTarget,
        request_id: str,
        active_session_id: str | None,
        error: BrowserResourceInvalidError,
        attempt: int,
        max_retries: int,
    ) -> None:
        account_id = self._pool.account_id(target.group, target.account)
        diagnostics = self._browser_manager.browser_diagnostics(target.proxy_key)
        logger.warning(
            "[chat] browser resource invalid attempt=%s/%s type=%s proxy=%s account=%s session_id=%s request_id=%s resource=%s helper=%s stage=%s stream_phase=%s browser_present=%s proc_alive=%s cdp_listening=%s tab_count=%s active_requests=%s err=%s",
            attempt + 1,
            max_retries,
            type_name,
            target.proxy_key.fingerprint_id,
            account_id,
            active_session_id,
            request_id,
            error.resource_hint,
            error.helper_name,
            error.stage,
            error.stream_phase,
            diagnostics.get("browser_present"),
            diagnostics.get("proc_alive"),
            diagnostics.get("cdp_listening"),
            diagnostics.get("tab_count"),
            diagnostics.get("active_requests"),
            error,
        )
        stderr_tail = str(diagnostics.get("stderr_tail") or "").strip()
        if stderr_tail:
            logger.warning(
                "[chat] browser resource invalid stderr tail proxy=%s request_id=%s:\n%s",
                target.proxy_key.fingerprint_id,
                request_id,
                stderr_tail,
            )

        if active_session_id is not None:
            self._invalidate_session_locked(active_session_id)
        if error.resource_hint == "transport":
            logger.warning(
                "[chat] transport-level stream failure, keep tab/browser and retry proxy=%s request_id=%s",
                target.proxy_key.fingerprint_id,
                request_id,
            )
            return
        self._browser_manager.mark_tab_draining(target.proxy_key, type_name)

        browser_restart_reason: str | None = None
        if error.resource_hint == "browser":
            browser_restart_reason = "resource_hint"
        # Legacy: page_fetch transport is no longer used by Claude (context_request since v0.x).
        # Kept for potential future plugins that still use page_fetch transport.
        elif (
            error.helper_name == "stream_raw_via_page_fetch"
            and error.stage in {"read_timeout", "evaluate_timeout"}
        ):
            browser_restart_reason = f"{error.helper_name}:{error.stage}"

        if browser_restart_reason is not None:
            logger.warning(
                "[chat] escalating browser recovery to full restart proxy=%s request_id=%s reason=%s",
                target.proxy_key.fingerprint_id,
                request_id,
                browser_restart_reason,
            )
            closed = await self._browser_manager.close_browser(target.proxy_key)
            self._apply_closed_tabs_locked(closed)
            return

        self._invalidate_tab_sessions_locked(target.proxy_key, type_name)
        closed = await self._browser_manager.close_tab(target.proxy_key, type_name)
        if closed is not None:
            self._apply_closed_tabs_locked([closed])

    def _revive_tab_if_possible_locked(
        self,
        proxy_key: ProxyKey,
        type_name: str,
    ) -> bool:
        tab = self._browser_manager.get_tab(proxy_key, type_name)
        if tab is None or tab.active_requests != 0:
            return False
        if tab.accepting_new:
            return True

        pair = self._pool.get_account_by_id(tab.account_id)
        if pair is None:
            return False
        _, account = pair
        if not account.is_available():
            return False
        tab.accepting_new = True
        tab.state = "ready"
        tab.frozen_until = None
        tab.last_used_at = time.time()
        return True

    async def _reconcile_tabs_locked(self) -> None:
        """
        收尾所有 non-ready tab：

        - 若原账号已恢复可用，则恢复 tab
        - 否则若同组有其他可用账号，则在 drained 后切号
        - 否则关闭 tab
        """
        for proxy_key, entry in list(self._browser_manager.list_browser_entries()):
            for type_name in list(entry.tabs.keys()):
                tab = entry.tabs[type_name]
                if tab.accepting_new:
                    continue
                if tab.active_requests != 0:
                    continue
                if self._revive_tab_if_possible_locked(proxy_key, type_name):
                    continue

                group = self._pool.get_group_by_proxy_key(proxy_key)
                if group is None:
                    await self._clear_tab_domain_cookies_if_supported(
                        proxy_key, type_name
                    )
                    closed = await self._browser_manager.close_tab(proxy_key, type_name)
                    if closed is not None:
                        self._apply_closed_tabs_locked([closed])
                    continue

                next_account = self._pool.next_available_account_in_group(
                    group,
                    type_name,
                    exclude_account_ids={tab.account_id},
                )
                if next_account is not None:
                    plugin = PluginRegistry.get(type_name)
                    if plugin is None:
                        continue
                    self._invalidate_tab_sessions_locked(proxy_key, type_name)
                    switched = await self._browser_manager.switch_tab_account(
                        proxy_key,
                        type_name,
                        self._pool.account_id(group, next_account),
                        self._make_apply_auth_fn(plugin, next_account),
                    )
                    if switched:
                        continue

                await self._clear_tab_domain_cookies_if_supported(proxy_key, type_name)
                closed = await self._browser_manager.close_tab(proxy_key, type_name)
                if closed is not None:
                    self._apply_closed_tabs_locked([closed])

    async def _reuse_session_target_locked(
        self,
        plugin: Any,
        type_name: str,
        session_id: str,
    ) -> _RequestTarget | None:
        entry = self._session_cache.get(session_id)
        if entry is None or entry.type_name != type_name:
            return None

        pair = self._pool.get_account_by_id(entry.account_id)
        if pair is None:
            self._invalidate_session_locked(session_id, entry)
            return None
        group, account = pair

        tab = self._browser_manager.get_tab(entry.proxy_key, type_name)
        if (
            tab is None
            or tab.account_id != entry.account_id
            or not plugin.has_session(session_id)
        ):
            self._invalidate_session_locked(session_id, entry)
            return None

        if not tab.accepting_new:
            self._invalidate_session_locked(session_id, entry)
            return None
        if session_id in self._busy_sessions:
            raise RuntimeError("当前会话正在处理中，请稍后再试")
        if tab.active_requests >= self._tab_max_concurrent:
            raise RuntimeError("当前会话所在 tab 繁忙，请稍后再试")

        page = self._browser_manager.acquire_tab(
            entry.proxy_key,
            type_name,
            self._tab_max_concurrent,
        )
        if page is None:
            raise RuntimeError("当前会话暂不可复用，请稍后再试")

        self._session_cache.touch(session_id)
        self._busy_sessions.add(session_id)
        context = await self._browser_manager.ensure_browser(
            entry.proxy_key,
            group.proxy_pass,
        )
        return _RequestTarget(
            proxy_key=entry.proxy_key,
            group=group,
            account=account,
            context=context,
            page=page,
            session_id=session_id,
            full_history=False,
        )

    async def _allocate_new_target_locked(
        self,
        type_name: str,
    ) -> _RequestTarget:
        # 1. 已打开浏览器里已有该 type 的可服务 tab，直接复用。
        existing_tabs: list[tuple[int, float, ProxyKey, TabRuntime]] = []
        for proxy_key, entry in self._browser_manager.list_browser_entries():
            tab = entry.tabs.get(type_name)
            if (
                tab is not None
                and tab.accepting_new
                and tab.active_requests < self._tab_max_concurrent
            ):
                existing_tabs.append(
                    (tab.active_requests, tab.last_used_at, proxy_key, tab)
                )
        if existing_tabs:
            _, _, proxy_key, tab = min(existing_tabs, key=lambda item: item[:2])
            pair = self._pool.get_account_by_id(tab.account_id)
            if pair is None:
                self._invalidate_tab_sessions_locked(proxy_key, type_name)
                closed = await self._browser_manager.close_tab(proxy_key, type_name)
                if closed is not None:
                    self._apply_closed_tabs_locked([closed])
            else:
                group, account = pair
                page = self._browser_manager.acquire_tab(
                    proxy_key,
                    type_name,
                    self._tab_max_concurrent,
                )
                if page is not None:
                    context = await self._browser_manager.ensure_browser(
                        proxy_key,
                        group.proxy_pass,
                    )
                    return _RequestTarget(
                        proxy_key=proxy_key,
                        group=group,
                        account=account,
                        context=context,
                        page=page,
                        session_id=None,
                        full_history=True,
                    )

        # 2. 已打开浏览器里还没有该 type tab，但该组有可用账号，直接建新 tab。
        open_browser_candidates: list[
            tuple[int, float, ProxyKey, ProxyGroupConfig]
        ] = []
        for proxy_key, entry in self._browser_manager.list_browser_entries():
            if type_name in entry.tabs:
                continue
            group = self._pool.get_group_by_proxy_key(proxy_key)
            if group is None:
                continue
            if not self._pool.has_available_account_in_group(group, type_name):
                continue
            open_browser_candidates.append(
                (
                    self._browser_manager.browser_load(proxy_key),
                    entry.last_used_at,
                    proxy_key,
                    group,
                )
            )
        if open_browser_candidates:
            _, _, proxy_key, group = min(
                open_browser_candidates, key=lambda item: item[:2]
            )
            account = self._pool.next_available_account_in_group(group, type_name)
            if account is not None:
                plugin = PluginRegistry.get(type_name)
                if plugin is None:
                    raise ValueError(f"未注册的 type: {type_name}")
                await self._browser_manager.open_tab(
                    proxy_key,
                    group.proxy_pass,
                    type_name,
                    self._pool.account_id(group, account),
                    plugin.create_page,
                    self._make_apply_auth_fn(plugin, account),
                )
                page = self._browser_manager.acquire_tab(
                    proxy_key,
                    type_name,
                    self._tab_max_concurrent,
                )
                if page is None:
                    raise RuntimeError("新建 tab 后仍无法占用请求槽位")
                context = await self._browser_manager.ensure_browser(
                    proxy_key,
                    group.proxy_pass,
                )
                return _RequestTarget(
                    proxy_key=proxy_key,
                    group=group,
                    account=account,
                    context=context,
                    page=page,
                    session_id=None,
                    full_history=True,
                )

        # 3. 已打开浏览器里该 type tab 已 drained，且同组有备用账号，可在当前 tab 切号。
        switch_candidates: list[tuple[float, ProxyKey, ProxyGroupConfig]] = []
        for proxy_key, entry in self._browser_manager.list_browser_entries():
            tab = entry.tabs.get(type_name)
            if tab is None or tab.active_requests != 0:
                continue
            group = self._pool.get_group_by_proxy_key(proxy_key)
            if group is None:
                continue
            if not self._pool.has_available_account_in_group(
                group,
                type_name,
                exclude_account_ids={tab.account_id},
            ):
                continue
            switch_candidates.append((tab.last_used_at, proxy_key, group))
        if switch_candidates:
            _, proxy_key, group = min(switch_candidates, key=lambda item: item[0])
            tab = self._browser_manager.get_tab(proxy_key, type_name)
            plugin = PluginRegistry.get(type_name)
            if tab is not None and plugin is not None:
                next_account = self._pool.next_available_account_in_group(
                    group,
                    type_name,
                    exclude_account_ids={tab.account_id},
                )
                if next_account is not None:
                    self._invalidate_tab_sessions_locked(proxy_key, type_name)
                    switched = await self._browser_manager.switch_tab_account(
                        proxy_key,
                        type_name,
                        self._pool.account_id(group, next_account),
                        self._make_apply_auth_fn(plugin, next_account),
                    )
                    if switched:
                        page = self._browser_manager.acquire_tab(
                            proxy_key,
                            type_name,
                            self._tab_max_concurrent,
                        )
                        if page is None:
                            raise RuntimeError("切号后仍无法占用请求槽位")
                        context = await self._browser_manager.ensure_browser(
                            proxy_key,
                            group.proxy_pass,
                        )
                        return _RequestTarget(
                            proxy_key=proxy_key,
                            group=group,
                            account=next_account,
                            context=context,
                            page=page,
                            session_id=None,
                            full_history=True,
                        )

        # 4. 开新浏览器。
        open_groups = {
            proxy_key.fingerprint_id
            for proxy_key in self._browser_manager.current_proxy_keys()
        }
        pair = self._pool.next_available_pair(
            type_name,
            exclude_fingerprint_ids=open_groups,
        )
        if pair is None:
            raise ValueError(f"没有类别为 {type_name!r} 的可用账号，请稍后再试")
        group, account = pair
        proxy_key = _proxy_key_for_group(group)
        plugin = PluginRegistry.get(type_name)
        if plugin is None:
            raise ValueError(f"未注册的 type: {type_name}")
        await self._browser_manager.open_tab(
            proxy_key,
            group.proxy_pass,
            type_name,
            self._pool.account_id(group, account),
            plugin.create_page,
            self._make_apply_auth_fn(plugin, account),
        )
        page = self._browser_manager.acquire_tab(
            proxy_key,
            type_name,
            self._tab_max_concurrent,
        )
        if page is None:
            raise RuntimeError("新浏览器建 tab 后仍无法占用请求槽位")
        context = await self._browser_manager.ensure_browser(
            proxy_key, group.proxy_pass
        )
        return _RequestTarget(
            proxy_key=proxy_key,
            group=group,
            account=account,
            context=context,
            page=page,
            session_id=None,
            full_history=True,
        )

    async def _stream_completion(
        self,
        type_name: str,
        req: OpenAIChatRequest,
    ) -> AsyncIterator[str]:
        """
        内部实现：调度 + 插件 stream_completion 字符串流，末尾附加 session_id 零宽编码。
        对外仅通过 stream_openai_events() 暴露事件流。
        """
        plugin = PluginRegistry.get(type_name)
        if plugin is None:
            raise ValueError(f"未注册的 type: {type_name}")

        raw_messages = _request_messages_as_dicts(req)
        conv_uuid = req.resume_session_id or parse_conv_uuid_from_messages(raw_messages)

        # Fingerprint matching: when the client doesn't preserve the zero-width
        # session marker (conv_uuid is None), compute a fingerprint from
        # system prompt + first user message and look up the matching session.
        # This replaces sticky session and prevents context pollution.
        fingerprint = ""
        if not conv_uuid:
            fingerprint = compute_conversation_fingerprint(req.messages)
            if fingerprint:
                entry = self._conv_index.lookup(fingerprint)
                if entry is not None:
                    conv_uuid = entry.session_id

        logger.info("[chat] type=%s parsed conv_uuid=%s fingerprint=%s", type_name, conv_uuid, fingerprint or "n/a")

        has_tools = bool(req.tools)
        react_prompt_prefix = format_react_prompt(req.tools or []) if has_tools else ""

        debug_path = (
            Path(__file__).resolve().parent.parent.parent
            / "debug"
            / "chat_prompt_debug.json"
        )

        max_retries = 3
        for attempt in range(max_retries):
            target: _RequestTarget | None = None
            active_session_id: str | None = None
            request_id = uuid.uuid4().hex
            try:
                async with self._schedule_lock:
                    if conv_uuid:
                        target = await self._reuse_session_target_locked(
                            plugin,
                            type_name,
                            conv_uuid,
                        )
                    if target is None:
                        target = await self._allocate_new_target_locked(type_name)
                    if target.session_id is not None:
                        active_session_id = target.session_id

                content = extract_user_content(
                    req.messages,
                    has_tools=has_tools,
                    react_prompt_prefix=react_prompt_prefix,
                    full_history=target.full_history,
                )
                if not content.strip() and req.attachment_files:
                    content = "Please analyze the attached image."
                if not content.strip():
                    raise ValueError("messages 中需至少有一条带 content 的 user 消息")

                debug_path.parent.mkdir(parents=True, exist_ok=True)
                debug_path.write_text(
                    json.dumps(
                        {
                            "prompt": content,
                            "full_history": target.full_history,
                            "type": type_name,
                        },
                        ensure_ascii=False,
                        indent=2,
                    ),
                    encoding="utf-8",
                )

                account_id = self._pool.account_id(target.group, target.account)
                session_id = target.session_id
                if session_id is None:
                    await plugin.ensure_request_ready(
                        target.context,
                        target.page,
                        request_id=request_id,
                        session_id=None,
                        phase="create_conversation",
                        account_id=account_id,
                    )
                    logger.info(
                        "[chat] create_conversation type=%s proxy=%s account=%s",
                        type_name,
                        target.proxy_key.fingerprint_id,
                        account_id,
                    )
                    session_id = await plugin.create_conversation(
                        target.context,
                        target.page,
                        timezone=target.group.timezone
                        or getattr(target.proxy_key, "timezone", None)
                        or TIMEZONE,
                        public_model=str(getattr(req, "model", "") or ""),
                        upstream_model=str(getattr(req, "upstream_model", "") or ""),
                        request_id=request_id,
                    )
                    if not session_id:
                        raise RuntimeError("插件创建会话失败")
                    async with self._schedule_lock:
                        self._session_cache.put(
                            session_id,
                            target.proxy_key,
                            type_name,
                            account_id,
                        )
                        self._browser_manager.register_session(
                            target.proxy_key,
                            type_name,
                            session_id,
                        )
                        self._busy_sessions.add(session_id)
                        # Register fingerprint for future matching
                        if fingerprint:
                            self._conv_index.register(
                                fingerprint,
                                session_id,
                                len(req.messages),
                                account_id,
                            )
                active_session_id = session_id

                # Skip pre-stream probe for newly created sessions:
                # create_conversation already validated page health.
                if target.session_id is not None:
                    await plugin.ensure_request_ready(
                        target.context,
                        target.page,
                        request_id=request_id,
                        session_id=session_id,
                        phase="stream_completion",
                        account_id=account_id,
                    )
                logger.info(
                    "[chat] stream_completion type=%s session_id=%s proxy=%s account=%s full_history=%s",
                    type_name,
                    session_id,
                    target.proxy_key.fingerprint_id,
                    account_id,
                    target.full_history,
                )
                # 根据是否 full_history 选择附件来源：
                # - 复用会话（full_history=False）：仅最后一条 user 的图片（可能为空，则本轮不带图）
                # - 新建/重建会话（full_history=True）：所有历史 user 的图片
                attachments = (
                    req.attachment_files_all_users
                    if target.full_history
                    else req.attachment_files_last_user
                )

                proxy_url = None
                proxy_auth = None
                proxy_forwarder = None
                if plugin.stream_transport() == "context_request":
                    proxy_url, proxy_auth, proxy_forwarder = self._stream_proxy_settings(target)
                    target.proxy_url = proxy_url
                    target.proxy_auth = proxy_auth
                    target.proxy_forwarder = proxy_forwarder
                try:
                    stream = cast(
                        AsyncIterator[str],
                        plugin.stream_completion(
                            target.context,
                            target.page,
                            session_id,
                            content,
                            request_id=request_id,
                            attachments=attachments,
                            proxy_url=proxy_url,
                            proxy_auth=proxy_auth,
                        ),
                    )
                    async for chunk in stream:
                        yield chunk
                finally:
                    if proxy_forwarder is not None:
                        try:
                            proxy_forwarder.stop()
                        except Exception:
                            pass
                        target.proxy_forwarder = None

                yield session_id_suffix(session_id)
                return
            except AccountFrozenError as e:
                logger.warning(
                    "账号限流/额度用尽（插件上报），切换资源重试: type=%s proxy=%s err=%s",
                    type_name,
                    target.proxy_key.fingerprint_id if target else None,
                    e,
                )
                async with self._schedule_lock:
                    if target is not None:
                        self.report_account_unfreeze(
                            target.group.fingerprint_id,
                            target.account.name,
                            e.unfreeze_at,
                        )
                        self._browser_manager.mark_tab_draining(
                            target.proxy_key,
                            type_name,
                            frozen_until=e.unfreeze_at,
                        )
                        self._invalidate_tab_sessions_locked(
                            target.proxy_key, type_name
                        )
                if attempt == max_retries - 1:
                    raise RuntimeError(
                        f"已重试 {max_retries} 次仍限流/过载，请稍后再试: {e}"
                    ) from e
                continue
            except BrowserResourceInvalidError as e:
                proxy_for_log = (
                    target.proxy_key.fingerprint_id
                    if target is not None
                    else getattr(getattr(e, "proxy_key", None), "fingerprint_id", None)
                )
                logger.warning(
                    "[chat] browser resource invalid bubbled type=%s request_id=%s proxy=%s session_id=%s helper=%s stage=%s resource=%s err=%s",
                    type_name,
                    request_id,
                    proxy_for_log,
                    active_session_id,
                    e.helper_name,
                    e.stage,
                    e.resource_hint,
                    e,
                )
                async with self._schedule_lock:
                    if target is not None:
                        await self._recover_browser_resource_invalid_locked(
                            type_name,
                            target,
                            request_id,
                            active_session_id,
                            e,
                            attempt,
                            max_retries,
                        )
                    elif getattr(e, "proxy_key", None) is not None:
                        closed = await self._browser_manager.close_browser(e.proxy_key)
                        self._apply_closed_tabs_locked(closed)
                if attempt == max_retries - 1:
                    raise RuntimeError(
                        f"浏览器资源已失效，重试 {max_retries} 次后仍失败: {e}"
                    ) from e
                continue
            finally:
                if target is not None:
                    async with self._schedule_lock:
                        if active_session_id is not None:
                            self._busy_sessions.discard(active_session_id)
                        self._browser_manager.release_tab(target.proxy_key, type_name)

    async def stream_openai_events(
        self,
        type_name: str,
        req: OpenAIChatRequest,
    ) -> AsyncIterator[OpenAIStreamEvent]:
        """
        唯一流式出口：以 OpenAIStreamEvent 为中间态。插件产出字符串流，
        在此包装为 content_delta + finish，供协议适配层编码为各协议 SSE。
        """
        async for chunk in self._stream_completion(type_name, req):
            # session marker 也作为 content_delta 透传（对事件消费者而言是普通文本片段）
            yield OpenAIStreamEvent(type="content_delta", content=chunk)
        yield OpenAIStreamEvent(type="finish", finish_reason="stop")