src/api/vcore_client.py

"""Vcore AI客户端"""

import asyncio
import codecs
import contextlib
import json
import os
import subprocess
import tempfile
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast, AsyncGenerator, Awaitable, Callable
from src.core.config import load_config
from src.transport.port_allocator import PortLease, port_allocator
from src.transport.codec import build_config, needs_worker
from src.transport.worker import worker
from src.utils.node_store import (
    DIRECT_NODE_KEY,
    NodeCandidateRef,
    ProxyRuntimePlan,
    get_candidate_queue,
    get_node_config,
    load_enabled_nodes,
    record_node_failure,
    record_node_success,
    record_node_stream_complete,
    record_node_stream_failure,
    record_node_stream_stall,
    resolve_proxy_runtime_plan,
)

from src.core.errors import (
    VcoreError,
    AuthenticationError,
    RateLimitError,
    InternalError,
    InvalidArgumentError,
    NotFoundError,
    PermissionDeniedError,
    RequestPoolTimeoutError,
    UpstreamResponseTimeoutError,
    parse_error_response,
    raise_for_status,
    UpstreamResponseIncompleteError,
)
from src.utils.logger import get_logger

# 从拆分的模块导入
from .model_config import ModelConfigBuilder
from .transform import RequestTransformer, ResponseAggregator
from .network import NetworkClient

# 初始化日志
logger = get_logger(__name__)

_INTERNAL_STREAM_PROGRESS_KEY = "_vcore_proxy_stream_progress"
_STREAM_TASK_CANCEL_TIMEOUT_SECONDS = 1.0
_JSON_THREAD_OFFLOAD_THRESHOLD_CHARS = 256 * 1024


def _consume_background_task_result(task: asyncio.Task[Any]) -> None:
    try:
        task.result()
    except asyncio.CancelledError:
        pass
    except Exception as e:
        logger.debug(f"后台任务结束时出现异常: {e}")


async def _cancel_tasks_bounded(
    tasks: list[asyncio.Task[Any]],
    timeout: float = _STREAM_TASK_CANCEL_TIMEOUT_SECONDS,
    owner: Any | None = None,
    reason: str = "",
) -> None:
    """取消任务但只等待有限时间，避免 loser 清理阻塞 winner 响应转发。"""
    pending_tasks = [task for task in tasks if not task.done()]
    if not pending_tasks:
        return
    for task in pending_tasks:
        task.cancel()
    done, pending = await asyncio.wait(pending_tasks, timeout=timeout)
    if done:
        await asyncio.gather(*done, return_exceptions=True)
    if pending:
        for task in pending:
            if owner is not None and hasattr(owner, "untrack_task"):
                with contextlib.suppress(Exception):
                    owner.untrack_task(task)
            task.add_done_callback(_consume_background_task_result)
        label = f"，原因={reason}" if reason else ""
        message = f"取消后台任务超时，已转后台清理: tasks={len(pending)}, timeout={timeout:.1f}s{label}"
        if timeout <= 0:
            logger.debug(message)
        else:
            logger.warning(message)


async def _aclose_async_generator_bounded(
    generator: AsyncGenerator[Any, None],
    timeout: float = _STREAM_TASK_CANCEL_TIMEOUT_SECONDS,
    reason: str = "",
) -> None:
    """限时关闭异步生成器，避免关闭上游流时卡住响应链路。"""
    task = asyncio.create_task(generator.aclose())
    done, pending = await asyncio.wait({task}, timeout=timeout)
    if done:
        await asyncio.gather(*done, return_exceptions=True)
        return
    task.add_done_callback(_consume_background_task_result)
    label = f"，原因={reason}" if reason else ""
    logger.warning(f"关闭异步生成器超时，已转后台继续关闭: timeout={timeout:.1f}s{label}")


async def _json_loads_maybe_thread(json_str: str) -> Any:
    if len(json_str) >= _JSON_THREAD_OFFLOAD_THRESHOLD_CHARS:
        return await asyncio.to_thread(json.loads, json_str)
    return json.loads(json_str)


def _run_sync_background(func: Callable[..., Any], *args: Any, reason: str = "") -> None:
    async def runner() -> None:
        try:
            await asyncio.to_thread(func, *args)
        except Exception as e:
            label = f"，原因={reason}" if reason else ""
            logger.debug(f"后台同步任务失败: {e}{label}")

    task = asyncio.create_task(runner())
    task.add_done_callback(_consume_background_task_result)


def _stream_winner_stall_timeout_seconds(cfg: dict[str, Any]) -> float:
    try:
        return max(0.0, float(cfg.get("stream_winner_stall_timeout_seconds", 0) or 0))
    except Exception:
        return 0.0


def _make_stream_stall_error(timeout_seconds: float, details: dict[str, Any] | None = None) -> UpstreamResponseTimeoutError:
    return UpstreamResponseTimeoutError(
        message=f"上游响应超时：winner 首包后 {timeout_seconds:.1f}s 内没有收到新的 raw chunk",
        details=details or {},
    )


async def _anext_with_stream_stall_guard(
    generator: AsyncGenerator[dict[str, Any], None],
    stall_guard: dict[str, Any] | None,
    timeout_seconds: float,
    request_id: str,
    winner_label: str,
) -> dict[str, Any]:
    if timeout_seconds <= 0 or stall_guard is None:
        return await anext(generator)

    next_task = asyncio.create_task(anext(generator))
    try:
        while True:
            if bool(stall_guard.get("completed")):
                return await next_task

            now = time.monotonic()
            last_raw_at = float(stall_guard.get("last_raw_at") or stall_guard.get("started_at") or now)
            remaining = max(0.0, last_raw_at + timeout_seconds - now)
            if remaining <= 0:
                gap_ms = max(0.0, (now - last_raw_at) * 1000)
                details = {
                    "requestId": request_id,
                    "winner": winner_label,
                    "timeoutSeconds": timeout_seconds,
                    "rawGapMs": round(gap_ms, 1),
                    "rawChunkCount": int(stall_guard.get("raw_chunk_count") or 0),
                    "rawBytesTotal": int(stall_guard.get("raw_bytes_total") or 0),
                }
                logger.warning(
                    f"会话 {request_id} winner {winner_label} 上游 raw chunk 停顿超时: "
                    f"timeout={timeout_seconds:.1f}s, gap={gap_ms:.0f}ms, "
                    f"raw_chunks={details['rawChunkCount']}, bytes={details['rawBytesTotal']}"
                )
                raise _make_stream_stall_error(timeout_seconds, details)

            done, _ = await asyncio.wait({next_task}, timeout=min(remaining, 0.5))
            if next_task in done:
                return await next_task
    except BaseException:
        if not next_task.done():
            await _cancel_tasks_bounded([next_task], reason="winner raw chunk 停顿/取消，停止等待后续 chunk")
        raise


@dataclass
class _ParallelNodeResult:
    """并行节点尝试的结果。"""
    node: dict[str, Any]
    index: int
    name: str
    candidate: NodeCandidateRef | None = None
    first_chunk: dict[str, Any] | None = None
    error: Exception | None = None
    generator: AsyncGenerator[dict[str, Any], None] | None = None
    elapsed_ms: float = 0.0
    first_chunk_is_internal: bool = False
    attempt_no: int = 0
    stall_guard: dict[str, Any] | None = None


@dataclass
class _ParallelValueResult:
    """并行请求池中单个非流式上游尝试的结果。"""
    node: dict[str, Any]
    index: int
    name: str
    candidate: NodeCandidateRef | None = None
    value: Any = None
    error: Exception | None = None
    elapsed_ms: float = 0.0
    attempt_no: int = 0


class _StreamingJsonObjectParser:
    """跨网络 chunk 维护状态的 JSON 对象解析器。

    上游 GraphQL 流会把文本、工具调用、图片 base64 等内容包装成连续 JSON
    对象。这里按字符状态机提取完整对象，只扫描新增 chunk，并用分段缓存避免
    大对象反复 ``buffer += chunk`` / 切片造成的 O(N²) 拷贝。
    """

    def __init__(self) -> None:
        self._object_parts: list[str] = []
        self._completed_objects: list[str] = []
        self._buffer_length = 0
        self._object_started = False
        self._brace_count = 0
        self._in_string = False
        self._escape = False

    @property
    def buffer_length(self) -> int:
        return self._buffer_length

    def feed(self, text: str) -> None:
        if not text:
            return

        part_start = 0 if self._object_started else None

        for idx, char in enumerate(text):
            if not self._object_started:
                if char != '{':
                    continue
                self._object_started = True
                self._brace_count = 1
                self._in_string = False
                self._escape = False
                part_start = idx
                continue

            if self._in_string:
                if self._escape:
                    self._escape = False
                    continue
                if char == '\\':
                    self._escape = True
                    continue
                if char == '"':
                    self._in_string = False
                continue

            if char == '"':
                self._in_string = True
            elif char == '{':
                self._brace_count += 1
            elif char == '}':
                self._brace_count -= 1
                if self._brace_count == 0:
                    if part_start is not None:
                        part = text[part_start:idx + 1]
                        if part:
                            self._object_parts.append(part)
                            self._buffer_length += len(part)
                    self._completed_objects.append(''.join(self._object_parts))
                    self._object_parts = []
                    self._buffer_length = 0
                    self._object_started = False
                    self._in_string = False
                    self._escape = False
                    part_start = None

        if self._object_started and part_start is not None:
            part = text[part_start:]
            if part:
                self._object_parts.append(part)
                self._buffer_length += len(part)

    def pop_complete_objects(self) -> list[str]:
        objects = self._completed_objects
        self._completed_objects = []
        return objects


class _ParallelNodeWorker:
    """并行节点专用临时 worker，避免多个 task 争抢全局 worker。"""

    def __init__(self, uri: str, name: str, request_id: str, node_index: int) -> None:
        self.uri = uri
        self.name = name
        self.port: int | None = None
        self.proxy_url: str | None = None
        self.request_id = request_id
        self.node_index = node_index
        self.lease: PortLease | None = None
        safe_id = f"{request_id}-{node_index}"
        temp_dir = Path(tempfile.gettempdir())
        self.config_path = temp_dir / f"parallel-worker-{safe_id}.json"
        self.log_path = temp_dir / f"parallel-worker-{safe_id}.log"
        self.proc: subprocess.Popen[bytes] | None = None

    async def start(self) -> str:
        binary = worker.ensure_binary()
        self.lease = await port_allocator.acquire()
        self.port = self.lease.port
        self.proxy_url = f"socks5://127.0.0.1:{self.lease.port}"
        cfg = build_config(self.uri, socks_port=self.lease.port)
        self.config_path.parent.mkdir(parents=True, exist_ok=True)
        with open(self.config_path, "w", encoding="utf-8") as f:
            json.dump(cfg, f, ensure_ascii=False, indent=2)

        log_f = open(self.log_path, "ab")
        try:
            self.proc = subprocess.Popen(
                [binary, "run", "-c", str(self.config_path)],
                stdout=log_f,
                stderr=log_f,
                start_new_session=True,
            )
        except Exception:
            log_f.close()
            await self.stop()
            raise
        else:
            log_f.close()

        await asyncio.sleep(0.8)
        if self.proc.poll() is not None:
            error = RuntimeError(f"并行 worker 启动后退出，exit code={self.proc.returncode}")
            await self.stop()
            raise error

        return self.proxy_url

    async def stop(self) -> None:
        proc = self.proc
        if proc is not None:
            try:
                if proc.poll() is None:
                    proc.terminate()
                    try:
                        await asyncio.to_thread(proc.wait, 3)
                    except subprocess.TimeoutExpired:
                        proc.kill()
                        await asyncio.to_thread(proc.wait, 2)
            except Exception as e:
                logger.debug(f"并行 worker 停止失败: {e}")
            finally:
                self.proc = None
        for path in (self.config_path, self.log_path):
            with contextlib.suppress(Exception):
                os.remove(path)
        if self.lease is not None:
            await port_allocator.release(self.lease)
            self.lease = None
            self.port = None
            self.proxy_url = None


class VcoreAIClient:
    """Vcore AI API客户端 (Anonymous 模式)"""
    
    def __init__(self):
        logger.info("初始化 Vcore AI 客户端")
        
        # 加载配置
        self.config = load_config()
        self.node_retry_count = int(self.config.get("node_retry_count", 0) or 0)
        
        # 初始化组件
        self.model_builder = ModelConfigBuilder()
        self.transformer = RequestTransformer(self.model_builder)
        self.aggregator = ResponseAggregator()
        self.network = NetworkClient()
        
        # 匿名接口基础 URL
        self.vcore_ai_anonymous_base_api = "https://cloudconsole-pa.clients6.google.com"
        
        logger.success("Vcore AI 客户端初始化完成")

    def _format_node_label(self, index: int, name: str) -> str:
        return f"[{index+1}] {name}"

    def _format_node_error(self, error: Exception) -> str:
        text = str(error)
        if "Could not fetch recaptcha token" in text:
            cause = getattr(error, "__cause__", None)
            cause_text = str(cause) if cause else ""
            return f"获取 recaptcha_token 失败{f': {cause_text}' if cause_text else ''}"
        if text.startswith("Internal error: Could not fetch recaptcha token"):
            cause = getattr(error, "__cause__", None)
            cause_text = str(cause) if cause else text.removeprefix("Internal error: ")
            return f"获取 recaptcha_token 失败: {cause_text}"
        return text

    async def close(self):
        """关闭客户端并释放资源"""
        await self.network.close()

    async def complete_chat(self, model: str, gemini_payload: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
        """聚合同一个 winner 的流式响应为非流式 ChatCompletion 对象。"""
        _raw_image_response = kwargs.pop('_raw_image_response', False)
        _expected_image_count = kwargs.pop('_expected_image_count', None)

        is_image_or_audio_request = False
        gen_config = gemini_payload.get("generationConfig") or gemini_payload.get("generation_config") or {}
        if isinstance(gen_config, dict):
            modalities = gen_config.get("responseModalities") or gen_config.get("response_modalities")
            if isinstance(modalities, list) and any(str(m).upper() in ("IMAGE", "AUDIO") for m in modalities):
                is_image_or_audio_request = True
        elif "image" in model.lower() or "audio" in model.lower():
            is_image_or_audio_request = True

        expected_count = 1
        if is_image_or_audio_request:
            if isinstance(gen_config, dict):
                image_config = gen_config.get("imageConfig") or gen_config.get("image_config") or {}
                if isinstance(image_config, dict):
                    expected_count = int(image_config.get("numberOfImages") or image_config.get("number_of_images") or 0)
                if expected_count <= 0:
                    expected_count = int(gen_config.get("candidateCount") or gen_config.get("candidate_count") or 1)
            if expected_count <= 1:
                expected_count = int(_expected_image_count or 1)

        if is_image_or_audio_request and expected_count > 1:
            import copy
            payload_copy = copy.deepcopy(gemini_payload)
            if "generationConfig" in payload_copy:
                payload_copy["generationConfig"]["candidateCount"] = 1
                if "candidate_count" in payload_copy["generationConfig"]:
                    payload_copy["generationConfig"]["candidate_count"] = 1
            elif "generation_config" in payload_copy:
                payload_copy["generation_config"]["candidateCount"] = 1
                if "candidate_count" in payload_copy["generation_config"]:
                    payload_copy["generation_config"]["candidate_count"] = 1

            async def _run_single() -> dict[str, Any]:
                cfg = load_config()
                progress_context: dict[str, str] = {}
                local_kwargs = dict(kwargs)
                local_kwargs["progress_context"] = progress_context
                generator = self._stream_realtime_parallel_pool(model, payload_copy, cfg, **local_kwargs)
                try:
                    return await self.aggregator.aggregate_stream(
                        generator,
                        _raw_image_response=_raw_image_response,
                        progress_context=progress_context,
                    )
                finally:
                    await _aclose_async_generator_bounded(generator, reason="非流式聚合结束清理")

            tasks = [_run_single() for _ in range(expected_count)]
            results = await asyncio.gather(*tasks, return_exceptions=True)

            merged_data = []
            merged_candidates = []
            base_response = None
            
            for res in results:
                if isinstance(res, Exception):
                    logger.error(f"并发多模态请求失败: {res}")
                    continue
                if not base_response:
                    base_response = res
                
                if _raw_image_response and "data" in res:
                    merged_data.extend(res["data"])
                if "candidates" in res:
                    merged_candidates.extend(res["candidates"])
            
            if not base_response:
                for res in results:
                    if isinstance(res, Exception):
                        raise res
            
            result = dict(base_response)
            if _raw_image_response and merged_data:
                result["data"] = merged_data
            if merged_candidates:
                for i, candidate in enumerate(merged_candidates):
                    candidate["index"] = i
                result["candidates"] = merged_candidates
                
            return result

        cfg = load_config()
        progress_context: dict[str, str] = {}
        kwargs["progress_context"] = progress_context
        generator = self._stream_realtime_parallel_pool(model, gemini_payload, cfg, **kwargs)
        try:
            return await self.aggregator.aggregate_stream(
                generator,
                _raw_image_response=_raw_image_response,
                progress_context=progress_context,
            )
        finally:
            await _aclose_async_generator_bounded(generator, reason="非流式聚合结束清理")

    def _should_remove_pool_node(self, error: Exception) -> bool:
        """判断是否为代理节点本身不可用，需要从节点池移除。"""
        text = str(error).lower()
        return any(marker in text for marker in (
            "couldn't connect",
            "could not connect",
            "connection refused",
            "connection reset",
            "connection timed out",
            "connect timeout",
            "proxy connect",
            "failed to connect",
            "no route to host",
            "network is unreachable",
        ))

    def _should_rotate_pool_node(self, error: Exception) -> bool:
        """判断是否应切换下一个节点但保留当前节点。"""
        text = str(error).lower()
        return any(marker in text for marker in (
            "could not fetch recaptcha token",
            "failed to verify action",
            "the caller does not have permission",
            "wrong_version_number",
            "tls connect error",
            "ssl routines",
            "timed out",
            "timeout",
            "curl",
        ))

    def _is_fatal_request_error(self, error: Exception) -> bool:
        """判断是否为换节点也无法修复的请求错误，应立即终止请求池。"""
        if isinstance(error, (InvalidArgumentError, NotFoundError, PermissionDeniedError)):
            return True
        if isinstance(error, VcoreError):
            if error.status in {"INVALID_ARGUMENT", "NOT_FOUND", "PERMISSION_DENIED", "FAILED_PRECONDITION", "UNIMPLEMENTED"}:
                return True
            return False

        text = str(error).lower()
        fatal_markers = (
            "invalid argument",
            "request contains an invalid argument",
            "model not found",
            "not found",
            "unsupported",
            "unimplemented",
            "bad request",
            "failed_precondition",
        )
        return any(marker in text for marker in fatal_markers)

    def _is_retryable_node_failure(self, error: Exception) -> bool:
        """判断是否为可通过换节点/补位继续等待成功的失败。"""
        if self._is_fatal_request_error(error):
            return False
        if isinstance(error, UpstreamResponseIncompleteError):
            return True
        if isinstance(error, RateLimitError):
            return True
        if isinstance(error, AuthenticationError):
            return True
        if isinstance(error, VcoreError):
            return error.is_retryable or self._should_rotate_pool_node(error) or self._should_remove_pool_node(error)
        return True

    def _runtime_plan(self, cfg: dict[str, Any]) -> ProxyRuntimePlan:
        """按启用节点数量解析直连/固定/动态代理运行计划。"""
        nodes = load_enabled_nodes()
        return resolve_proxy_runtime_plan(cfg, len(nodes))

    def _select_parallel_candidates(
        self,
        cfg: dict[str, Any],
        plan: ProxyRuntimePlan,
    ) -> list[NodeCandidateRef]:
        """生成请求池候选队列；无启用节点时直连，启用节点按候选队列调度。"""
        return get_candidate_queue(cfg, plan)

    def _node_config_for_candidate(self, candidate: NodeCandidateRef) -> dict[str, Any] | None:
        """候选接口只给标识，这里单独按标识取配置。"""
        return get_node_config(candidate.node_key)

    def _node_retry_limit(self, value: Any | None = None) -> int:
        try:
            source = self.node_retry_count if value is None else value
            return max(0, int(source or 0))
        except (TypeError, ValueError):
            return max(0, self.node_retry_count)

    def _direct_proxy_url_from_node(self, node: dict[str, Any]) -> str | None:
        """只对无需 worker 的代理节点返回可直接使用的代理地址。"""
        raw_uri = str(node.get("raw_uri", "")).strip()
        if raw_uri.startswith(("http://", "https://", "socks5://", "socks://")):
            return raw_uri
        return None

    async def _run_with_parallel_request_pool(
        self,
        operation_name: str,
        node_operation: Callable[[Any, str | None], Awaitable[Any]],
        cfg: dict[str, Any],
        business_session_id: str | None = None,
        gateway_session: Any | None = None,
    ) -> Any:
        """统一非流式请求池：并行选择代理节点，失败补位，首个成功返回。"""
        plan = self._runtime_plan(cfg)
        parallel_size = plan.request_pool_size
        request_id = business_session_id or f"pool-{int(time.time() * 1000) % 1000000}"
        max_rounds = plan.candidate_queue_rounds
        deadline_seconds = plan.deadline_seconds
        deadline_at = time.monotonic() + deadline_seconds if deadline_seconds > 0 else 0.0
        logger.info(
            f"业务请求池：启动 operation={operation_name}, session={request_id}, "
            f"模式={plan.mode}, 并发={parallel_size}, 总节点={plan.enabled_node_count}, "
            f"候选长度={plan.candidate_queue_length}, 最大轮次={'不限' if max_rounds <= 0 else max_rounds}, "
            f"首包 winner 超时={'底层网络超时' if deadline_seconds <= 0 else f'{deadline_seconds:.0f}s'}"
        )

        pending_nodes: list[NodeCandidateRef] = []
        running: dict[asyncio.Task[_ParallelValueResult], NodeCandidateRef] = {}
        active_keys: set[str] = set()
        failures: list[_ParallelValueResult] = []
        attempt_round = 0
        attempted_count = 0

        async def cancel_running_tasks(timeout: float = _STREAM_TASK_CANCEL_TIMEOUT_SECONDS, reason: str = "") -> None:
            if not running:
                return
            tasks = list(running.keys())
            await _cancel_tasks_bounded(tasks, timeout=timeout, owner=gateway_session, reason=reason)
            running.clear()

        def expired() -> bool:
            return bool(deadline_at and time.monotonic() >= deadline_at)

        def refill_candidates() -> None:
            nonlocal attempt_round, pending_nodes
            if pending_nodes or expired() or (max_rounds > 0 and attempt_round >= max_rounds):
                return
            attempt_round += 1
            selected = self._select_parallel_candidates(cfg, plan)
            pending_nodes = [candidate for candidate in selected if candidate.node_key not in active_keys]
            logger.info(
                f"业务请求池：生成候选 operation={operation_name}, session={request_id}, "
                f"轮次={attempt_round}, 候选={len(pending_nodes)}, 运行中={len(running)}"
            )

        async def run_node(candidate: NodeCandidateRef) -> _ParallelValueResult:
            node = self._node_config_for_candidate(candidate)
            if node is None:
                return _ParallelValueResult(node={}, index=candidate.index, name=candidate.name or candidate.node_key, candidate=candidate, error=InternalError(message="节点配置不存在，可能已被删除"))
            node_name = str(node.get("name") or candidate.name or node.get("raw_uri", "")[:40] or f"node-{candidate.index+1}")
            raw_uri = str(node.get("raw_uri", "")).strip()
            proxy_url = self._direct_proxy_url_from_node(node)
            temp_worker: _ParallelNodeWorker | None = None
            session: Any | None = None
            started_at = time.perf_counter()
            try:
                if candidate.node_key == DIRECT_NODE_KEY or candidate.mode == "direct":
                    session = self.network.create_session()
                    value = await node_operation(session, None)
                    elapsed_ms = (time.perf_counter() - started_at) * 1000
                    return _ParallelValueResult(node=node, index=candidate.index, name=node_name, candidate=candidate, value=value, elapsed_ms=elapsed_ms)

                if not proxy_url and raw_uri and needs_worker(raw_uri):
                    temp_worker = _ParallelNodeWorker(
                        uri=raw_uri,
                        name=node_name,
                        request_id=request_id,
                        node_index=candidate.index,
                    )
                    proxy_url = await temp_worker.start()
                if not proxy_url:
                    raise InternalError(message="节点 URI 不是可用代理地址，也不是支持的订阅节点格式")

                session = self.network.create_session_with_proxy(proxy_url)
                value = await node_operation(session, proxy_url)
                elapsed_ms = (time.perf_counter() - started_at) * 1000
                return _ParallelValueResult(node=node, index=candidate.index, name=node_name, candidate=candidate, value=value, elapsed_ms=elapsed_ms)
            except asyncio.CancelledError:
                raise
            except Exception as e:
                elapsed_ms = (time.perf_counter() - started_at) * 1000
                return _ParallelValueResult(node=node, index=candidate.index, name=node_name, candidate=candidate, error=e, elapsed_ms=elapsed_ms)
            finally:
                if session is not None:
                    with contextlib.suppress(Exception):
                        await session.close()
                if temp_worker is not None:
                    await temp_worker.stop()

        async def start_next(reason: str = "启动") -> None:
            nonlocal attempted_count
            refill_candidates()
            if not pending_nodes:
                return
            candidate = pending_nodes.pop(0)
            active_keys.add(candidate.node_key)
            attempted_count += 1
            node_name = candidate.name or candidate.node_key
            logger.info(
                f"业务请求池：{reason}槽位 operation={operation_name}, session={request_id}, "
                f"[{candidate.index+1}] {node_name}, 已尝试={attempted_count}, 运行中={len(running)+1}/{parallel_size}"
            )
            task = gateway_session.create_task(run_node(candidate)) if gateway_session is not None else asyncio.create_task(run_node(candidate))
            running[task] = candidate

        try:
            for _ in range(parallel_size):
                await start_next()

            while running:
                wait_timeout = max(0.0, deadline_at - time.monotonic()) if deadline_at else None
                done, _ = await asyncio.wait(running.keys(), timeout=wait_timeout, return_when=asyncio.FIRST_COMPLETED)
                if not done:
                    raise RequestPoolTimeoutError(message=f"请求池在 {deadline_seconds:.0f}s 内未收到上游响应首包，未能选出 winner")

                for task in done:
                    finished_candidate = running.pop(task, None)
                    if finished_candidate is not None:
                        active_keys.discard(finished_candidate.node_key)
                    try:
                        result = await task
                    except asyncio.CancelledError:
                        raise
                    except Exception as e:
                        result = _ParallelValueResult(
                            node={},
                            index=finished_candidate.index if finished_candidate else -1,
                            name=finished_candidate.name if finished_candidate else "unknown",
                            candidate=finished_candidate,
                            error=e,
                        )

                    if result.error is None:
                        if result.candidate and result.candidate.node_key != DIRECT_NODE_KEY:
                            _run_sync_background(record_node_success, result.node, result.elapsed_ms, reason="非流式 winner 成功记录")
                        logger.success(
                            f"业务请求池：winner operation={operation_name}, session={request_id}, "
                            f"[{result.index+1}] {result.name}, 耗时={result.elapsed_ms:.0f}ms"
                        )
                        await cancel_running_tasks(timeout=0.0, reason="非流式 winner 已返回，取消其它节点")
                        return result.value

                    failures.append(result)
                    err = result.error or InternalError(message="节点未知失败")
                    if not self._is_retryable_node_failure(err):
                        logger.error(
                            f"业务请求池：检测到不可重试错误，终止 operation={operation_name}, "
                            f"session={request_id}, error={err}"
                        )
                        await cancel_running_tasks(reason="非流式不可重试错误")
                        raise err
                    if result.candidate and result.candidate.node_key != DIRECT_NODE_KEY:
                        _run_sync_background(record_node_failure, result.node, err, reason="非流式节点失败记录")
                    logger.warning(
                        f"业务请求池：槽位失败 operation={operation_name}, session={request_id}, "
                        f"[{result.index+1}] {result.name}: {err}"
                    )
                    await start_next("补位")

                while len(running) < parallel_size and not expired():
                    before = len(running)
                    await start_next("补位")
                    if len(running) == before:
                        break

            last_error = failures[-1].error if failures and failures[-1].error else None
            if last_error:
                raise last_error
            raise InternalError(message="业务请求池所有节点均不可用")
        finally:
            await cancel_running_tasks()

    async def _prime_realtime_node(
        self,
        candidate: NodeCandidateRef,
        model: str,
        gemini_payload: dict[str, Any],
        kwargs: dict[str, Any],
        cfg: dict[str, Any],
        request_id: str,
        attempt_no: int,
    ) -> _ParallelNodeResult:
        """启动单个节点尝试并读取首个有效 chunk，成功后把生成器交给 winner 继续消费。"""
        node = self._node_config_for_candidate(candidate)
        if node is None:
            return _ParallelNodeResult(node={}, index=candidate.index, name=candidate.name or candidate.node_key, candidate=candidate, error=InternalError(message="节点配置不存在，可能已被删除"), attempt_no=attempt_no)
        node_name = str(node.get("name") or candidate.name or node.get("raw_uri", "")[:40] or f"node-{candidate.index+1}")
        node_kwargs = dict(kwargs)
        node_kwargs["node_retry_count_override"] = self._node_retry_limit(cfg.get("node_retry_count", 0))
        stall_timeout = _stream_winner_stall_timeout_seconds(cfg)
        stall_guard: dict[str, Any] | None = None
        if stall_timeout > 0:
            started_at_mono = time.monotonic()
            stall_guard = {
                "started_at": started_at_mono,
                "last_raw_at": started_at_mono,
                "raw_chunk_count": 0,
                "raw_bytes_total": 0,
                "completed": False,
            }
            node_kwargs["stream_stall_guard"] = stall_guard
        raw_uri = str(node.get("raw_uri", "")).strip()
        proxy_url = self._direct_proxy_url_from_node(node)
        temp_worker: _ParallelNodeWorker | None = None
        generator: AsyncGenerator[dict[str, Any], None] | None = None
        started_at = time.perf_counter()

        try:
            if candidate.node_key == DIRECT_NODE_KEY or candidate.mode == "direct":
                generator = self._stream_realtime_inner(
                    model,
                    gemini_payload=gemini_payload,
                    session_override=self.network.create_session(),
                    session_proxy_override=None,
                    worker_override=None,
                    **node_kwargs,
                )
                first_chunk = await anext(generator)
                elapsed_ms = (time.perf_counter() - started_at) * 1000
                return _ParallelNodeResult(
                    node=node,
                    index=candidate.index,
                    name=node_name,
                    candidate=candidate,
                    first_chunk=first_chunk,
                    generator=generator,
                    elapsed_ms=elapsed_ms,
                    first_chunk_is_internal=bool(first_chunk.get(_INTERNAL_STREAM_PROGRESS_KEY)) if isinstance(first_chunk, dict) else False,
                    attempt_no=attempt_no,
                    stall_guard=stall_guard,
                )

            if not proxy_url and raw_uri and needs_worker(raw_uri):
                temp_worker = _ParallelNodeWorker(
                    uri=raw_uri,
                    name=node_name,
                    request_id=request_id,
                    node_index=candidate.index,
                )
                proxy_url = await temp_worker.start()

            if not proxy_url:
                raise InternalError(message="节点 URI 不是可用代理地址，也不是支持的订阅节点格式")

            generator = self._stream_realtime_inner(
                model,
                gemini_payload=gemini_payload,
                session_override=self.network.create_session_with_proxy(proxy_url),
                session_proxy_override=proxy_url,
                worker_override=temp_worker,
                **node_kwargs,
            )

            first_chunk = await anext(generator)
            elapsed_ms = (time.perf_counter() - started_at) * 1000
            return _ParallelNodeResult(
                node=node,
                index=candidate.index,
                name=node_name,
                candidate=candidate,
                first_chunk=first_chunk,
                generator=generator,
                elapsed_ms=elapsed_ms,
                first_chunk_is_internal=bool(first_chunk.get(_INTERNAL_STREAM_PROGRESS_KEY)) if isinstance(first_chunk, dict) else False,
                attempt_no=attempt_no,
                stall_guard=stall_guard,
            )
        except UpstreamResponseIncompleteError as e:
            if generator is not None:
                await _aclose_async_generator_bounded(generator, reason="流式节点响应不完整")
            if temp_worker:
                await temp_worker.stop()
            return _ParallelNodeResult(node=node, index=candidate.index, name=node_name, candidate=candidate, error=e, attempt_no=attempt_no)
        except StopAsyncIteration:
            if generator is not None:
                await _aclose_async_generator_bounded(generator, reason="流式节点无首包")
            if temp_worker:
                await temp_worker.stop()
            return _ParallelNodeResult(node=node, index=candidate.index, name=node_name, candidate=candidate, error=UpstreamResponseIncompleteError(message="节点未返回任何有效响应结构"), attempt_no=attempt_no)
        except asyncio.CancelledError:
            if generator is not None:
                await _aclose_async_generator_bounded(generator, timeout=0.0, reason="流式节点任务取消")
            if temp_worker:
                await temp_worker.stop()
            raise
        except Exception as e:
            if generator is not None:
                await _aclose_async_generator_bounded(generator, reason="流式节点异常")
            if temp_worker:
                await temp_worker.stop()
            return _ParallelNodeResult(node=node, index=candidate.index, name=node_name, candidate=candidate, error=e, attempt_no=attempt_no)

    async def _stream_realtime_parallel_pool(
        self,
        model: str,
        gemini_payload: dict[str, Any],
        cfg: dict[str, Any],
        **kwargs: Any,
    ) -> AsyncGenerator[dict[str, Any], None]:
        """真流式滚动并行节点池：固定 n 个探测位，失败即补位，首包成功即清理其它请求。"""
        gateway_session = kwargs.pop("gateway_session", None)
        progress_context = kwargs.get("progress_context")
        plan = self._runtime_plan(cfg)
        parallel_size = plan.request_pool_size
        request_id = f"parallel-{int(time.time() * 1000) % 1000000}"
        max_rounds = plan.candidate_queue_rounds
        deadline_seconds = plan.deadline_seconds
        deadline_at = time.monotonic() + deadline_seconds if deadline_seconds > 0 else 0.0
        logger.info(
            f"会话 {request_id} 启动请求池: 并发={parallel_size}, 模式={plan.mode}, "
            f"总节点={plan.enabled_node_count}, 候选长度={plan.candidate_queue_length}, "
            f"最大轮次={'不限' if max_rounds <= 0 else max_rounds}, "
            f"winner超时={'底层网络超时' if deadline_seconds <= 0 else f'{deadline_seconds:.0f}s'}"
        )

        pending_nodes: list[NodeCandidateRef] = []
        running: dict[asyncio.Task[_ParallelNodeResult], NodeCandidateRef] = {}
        failures: list[_ParallelNodeResult] = []
        winner: _ParallelNodeResult | None = None
        active_keys: set[str] = set()
        attempt_round = 0
        attempted_count = 0

        async def cancel_running_tasks(timeout: float = _STREAM_TASK_CANCEL_TIMEOUT_SECONDS, reason: str = "") -> None:
            if not running:
                return
            tasks = list(running.keys())
            await _cancel_tasks_bounded(tasks, timeout=timeout, owner=gateway_session, reason=reason)
            running.clear()

        def expired() -> bool:
            return bool(deadline_at and time.monotonic() >= deadline_at)

        def refill_candidates() -> None:
            nonlocal attempt_round, pending_nodes
            if pending_nodes or expired() or (max_rounds > 0 and attempt_round >= max_rounds):
                return

            attempt_round += 1
            selected = self._select_parallel_candidates(cfg, plan)
            pending_nodes = [candidate for candidate in selected if candidate.node_key not in active_keys]
            logger.debug(
                f"会话 {request_id} 生成候选: 轮次={attempt_round}, "
                f"候选={len(pending_nodes)}, 运行中={len(running)}"
            )

        async def start_next(reason: str = "启动") -> None:
            nonlocal attempted_count
            refill_candidates()
            if not pending_nodes:
                return
            candidate = pending_nodes.pop(0)
            node_name = candidate.name or candidate.node_key
            active_keys.add(candidate.node_key)
            attempted_count += 1
            attempt_no = attempted_count
            node_label = self._format_node_label(candidate.index, node_name)
            action = "补位节点请求" if reason == "补位" else "启动节点请求"
            logger.info(
                f"会话 {request_id} 协程#{attempt_no} {action}: {node_label}, "
                f"轮次={attempt_round}, 已尝试={attempt_no}, 剩余补位={len(pending_nodes)}, 运行中={len(running)+1}/{parallel_size}"
            )
            node_cfg = dict(cfg)
            node_cfg["node_retry_count"] = plan.node_retry_count
            coro = self._prime_realtime_node(candidate, model, gemini_payload, kwargs, node_cfg, request_id, attempt_no)
            task = gateway_session.create_task(coro) if gateway_session is not None else asyncio.create_task(coro)
            running[task] = candidate

        try:
            for _ in range(parallel_size):
                await start_next()

            while running and winner is None:
                wait_timeout = max(0.0, deadline_at - time.monotonic()) if deadline_at else None
                done, _ = await asyncio.wait(running.keys(), timeout=wait_timeout, return_when=asyncio.FIRST_COMPLETED)
                if not done:
                    raise RequestPoolTimeoutError(message=f"请求池在 {deadline_seconds:.0f}s 内未收到上游响应首包，未能选出 winner")

                for task in done:
                    finished_candidate = running.pop(task, None)
                    if finished_candidate is not None:
                        active_keys.discard(finished_candidate.node_key)
                    try:
                        result = await task
                    except asyncio.CancelledError:
                        raise
                    except Exception as e:
                        result = _ParallelNodeResult(
                            node={},
                            index=finished_candidate.index if finished_candidate else -1,
                            name=finished_candidate.name if finished_candidate else "unknown",
                            candidate=finished_candidate,
                            error=e,
                        )
                    if result.first_chunk is not None and result.generator is not None:
                        winner = result
                        if result.candidate and result.candidate.node_key != DIRECT_NODE_KEY:
                            _run_sync_background(record_node_success, result.node, result.elapsed_ms, reason="流式 winner 首包成功记录")
                        logger.success(f"会话 {request_id} 协程#{result.attempt_no or '?'} {self._format_node_label(result.index, result.name)} winner，首包={result.elapsed_ms:.0f}ms")
                        break

                    failures.append(result)
                    err = result.error or InternalError(message="节点未知失败")
                    if not self._is_retryable_node_failure(err):
                        logger.error(f"会话 {request_id} 检测到不可重试错误，终止请求池: {self._format_node_error(err)}")
                        await cancel_running_tasks(reason="流式不可重试错误")
                        if generator := result.generator:
                            await _aclose_async_generator_bounded(generator, reason="流式不可重试错误")
                        raise err
                    if result.candidate and result.candidate.node_key != DIRECT_NODE_KEY:
                        _run_sync_background(record_node_failure, result.node, err, reason="流式节点失败记录")
                    logger.warning(
                        f"会话 {request_id} 协程#{result.attempt_no or '?'} 节点请求失败: "
                        f"{self._format_node_label(result.index, result.name)}, 原因={self._format_node_error(err)}"
                    )
                    await start_next("补位")

                while winner is None and len(running) < parallel_size and not expired():
                    before = len(running)
                    await start_next("补位")
                    if len(running) == before:
                        break

            if winner is None:
                last_error = failures[-1].error if failures and failures[-1].error else None
                if last_error:
                    raise last_error
                raise InternalError(message="节点池所有节点均不可用")

            await cancel_running_tasks(timeout=0.0, reason="流式 winner 已选出，取消其它节点")

            winner_label = self._format_node_label(winner.index, winner.name)
            progress_prefix = f"会话 {request_id} winner {winner_label}"
            if isinstance(progress_context, dict):
                progress_context["prefix"] = progress_prefix
            logger.info(f"会话 {request_id} winner {winner_label} 后续响应开始转发")
            forwarded_count = 0
            stall_timeout = _stream_winner_stall_timeout_seconds(cfg)
            try:
                if winner.first_chunk_is_internal:
                    logger.debug(
                        f"会话 {request_id} winner {winner_label} 首包为内部进度信号，跳过下游发送: "
                        f"keys={list(winner.first_chunk.keys()) if isinstance(winner.first_chunk, dict) else type(winner.first_chunk)}"
                    )
                else:
                    forwarded_count = 1
                    logger.debug(
                        f"会话 {request_id} winner {winner_label} 转发首包: "
                        f"chunk={forwarded_count}, keys={list(winner.first_chunk.keys()) if isinstance(winner.first_chunk, dict) else type(winner.first_chunk)}"
                    )
                    yield winner.first_chunk
                    logger.debug(f"会话 {request_id} winner {winner_label} 首包已交给下游生成器: chunk={forwarded_count}")
                while True:
                    try:
                        chunk = await _anext_with_stream_stall_guard(
                            winner.generator,
                            winner.stall_guard,
                            stall_timeout,
                            request_id,
                            winner_label,
                        )
                    except StopAsyncIteration:
                        break

                    if isinstance(chunk, dict) and chunk.get(_INTERNAL_STREAM_PROGRESS_KEY):
                        logger.debug(
                            f"会话 {request_id} winner {winner_label} 后续内部进度信号，跳过下游发送: "
                            f"raw_chunks={chunk.get('rawChunkCount')}, buffer={chunk.get('bufferSize')}"
                        )
                        continue
                    forwarded_count += 1
                    logger.debug(
                        f"会话 {request_id} winner {winner_label} 后续chunk准备转发: "
                        f"chunk={forwarded_count}, keys={list(chunk.keys()) if isinstance(chunk, dict) else type(chunk)}"
                    )
                    yield chunk
                    logger.debug(f"会话 {request_id} winner {winner_label} 后续chunk已交给下游生成器: chunk={forwarded_count}")
                if winner.candidate and winner.candidate.node_key != DIRECT_NODE_KEY:
                    _run_sync_background(record_node_stream_complete, winner.node, reason="流式完成记录")
            except asyncio.CancelledError:
                raise
            except UpstreamResponseTimeoutError as e:
                logger.warning(f"会话 {request_id} winner {winner_label} 上游响应超时，断开当前请求并清理资源: {e.message}")
                if winner.candidate and winner.candidate.node_key != DIRECT_NODE_KEY:
                    gap_ms = 0.0
                    if isinstance(e.details, dict):
                        try:
                            gap_ms = float(e.details.get("rawGapMs") or 0)
                        except (TypeError, ValueError):
                            gap_ms = 0.0
                    _run_sync_background(record_node_stream_stall, winner.node, gap_ms, e, reason="winner raw chunk 停顿降权")
                raise
            except Exception as e:
                logger.debug(f"会话 {request_id} winner {winner_label} 后续转发异常: chunk={forwarded_count}, error={e}")
                if winner.candidate and winner.candidate.node_key != DIRECT_NODE_KEY:
                    _run_sync_background(record_node_stream_failure, winner.node, e, reason="流式失败记录")
                raise
        finally:
            await cancel_running_tasks()
            if winner and winner.generator:
                await _aclose_async_generator_bounded(winner.generator, reason="流式 winner 结束清理")

    async def stream_chat_realtime(self, model: str, gemini_payload: dict[str, Any], **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]:
        """真流式聊天，统一走业务请求池。"""
        is_image_or_audio_request = False
        gen_config = gemini_payload.get("generationConfig") or gemini_payload.get("generation_config") or {}
        if isinstance(gen_config, dict):
            modalities = gen_config.get("responseModalities") or gen_config.get("response_modalities")
            if isinstance(modalities, list) and any(str(m).upper() in ("IMAGE", "AUDIO") for m in modalities):
                is_image_or_audio_request = True
        elif "image" in model.lower() or "audio" in model.lower():
            is_image_or_audio_request = True

        expected_count = 1
        if is_image_or_audio_request:
            if isinstance(gen_config, dict):
                image_config = gen_config.get("imageConfig") or gen_config.get("image_config") or {}
                if isinstance(image_config, dict):
                    expected_count = int(image_config.get("numberOfImages") or image_config.get("number_of_images") or 0)
                if expected_count <= 0:
                    expected_count = int(gen_config.get("candidateCount") or gen_config.get("candidate_count") or 1)

        if is_image_or_audio_request and expected_count > 1:
            try:
                logger.info(f"检测到流式多候选多模态生成请求 (n={expected_count})，自动在服务端降级为并发聚合")
                result = await self.complete_chat(model, gemini_payload, **kwargs)
                yield result
                return
            except Exception as e:
                logger.error(f"流式接口并发降级处理失败: {e}")
                raise

        cfg = load_config()
        generator = self._stream_realtime_parallel_pool(model, gemini_payload, cfg, **kwargs)
        try:
            async for chunk in generator:
                yield chunk
        finally:
            await _aclose_async_generator_bounded(generator, reason="流式入口结束清理")

    def _build_request_payload(self, model: str, gemini_payload: dict[str, Any], recaptcha_token: str, kwargs: dict[str, Any]) -> dict[str, Any]:
        """构建上游请求体（共用逻辑）"""
        dummy_original_body = {"variables": {}}
        new_variables = self.transformer.build_vcore_payload(
            model=model, gemini_payload=gemini_payload,
            original_body=dummy_original_body, kwargs=kwargs
        )['variables']
        new_variables["region"] = "global"
        new_variables["recaptchaToken"] = recaptcha_token
        payload = {
            "requestContext": self._build_request_context(),
            "querySignature": "2/l8eCsMMY49imcDQ/lwwXyL8cYtTjxZBF2dNqy69LodY=",
            "operationName": "StreamGenerateContentAnonymous",
            "variables": new_variables,
        }
        self._log_upstream_payload_summary(model, payload)
        return payload

    def _log_upstream_payload_summary(self, model: str, payload: dict[str, Any]) -> None:
        variables = payload.get("variables") if isinstance(payload, dict) else {}
        variables = variables if isinstance(variables, dict) else {}
        contents = variables.get("contents") if isinstance(variables.get("contents"), list) else []
        generation_config = variables.get("generationConfig") if isinstance(variables.get("generationConfig"), dict) else {}
        tools = variables.get("tools") if isinstance(variables.get("tools"), list) else []
        image_config = generation_config.get("imageConfig") if isinstance(generation_config, dict) and isinstance(generation_config.get("imageConfig"), dict) else {}
        logger.debug(
            "上游匿名接口请求已构建: "
            f"operation={payload.get('operationName')}, model={variables.get('model') or model}, "
            f"region={variables.get('region')}, contents={len(contents)}, tools={len(tools)}, "
            f"modalities={generation_config.get('responseModalities') if isinstance(generation_config, dict) else None}, "
            f"images={image_config.get('numberOfImages') if isinstance(image_config, dict) else None}"
        )
        logger.debug_json("上游匿名接口标准请求体", payload)

    def _build_request_context(self) -> dict[str, Any]:
        """构建 AI Studio 浏览器端常见的 GraphQL requestContext。"""
        return {
            "clientVersion": "boq_cloud-boq-clientweb-vcoreaistudio_20260402.09_p0",
            "pagePath": "/vcore-ai/studio/multimodal",
            "jurisdiction": "global",
            "localizationData": {
                "locale": "zh_CN",
                "timezone": "Asia/Shanghai",
            },
        }

    def _build_browser_headers(self) -> dict[str, str]:
        """构建更贴近 console.cloud.google.com 浏览器请求的头。"""
        return {
            "accept": "*/*",
            "accept-language": "zh-CN,zh;q=0.9,en;q=0.8",
            "content-type": "application/json",
            "origin": "https://console.cloud.google.com",
            "referer": "https://console.cloud.google.com/vcore-ai/studio/multimodal",
            "x-goog-authuser": "0",
        }

    async def _execute_streaming_attempt(
        self, session: Any, model: str, gemini_payload: dict[str, Any],
        recaptcha_token: str, kwargs: dict[str, Any], is_first_auth_attempt: bool = False,
    ) -> AsyncGenerator[dict[str, Any], None]:
        """真流式：解析上游响应，yield 增量 Gemini dict"""
        new_body = self._build_request_payload(model, gemini_payload, recaptcha_token, kwargs)
        headers = self._build_browser_headers()
        url = f"{self.vcore_ai_anonymous_base_api}/v3/entityServices/AiplatformEntityService/schemas/AIPLATFORM_GRAPHQL:batchGraphql?key=AIzaSyCI-zsRP85UVOi0DjtiCwWBwQ1djDy741g&prettyPrint=false"

        async for response in self.network.stream_request(
            session,
            'POST',
            url,
            headers=headers,
            json_data=new_body,
        ):
            if response.status_code != 200:
                error_bytes = await response.aread()
                error_text_str = error_bytes.decode('utf-8') if isinstance(error_bytes, bytes) else str(error_bytes)
                if response.status_code in [401, 403] or "Failed to verify action" in error_text_str or "The caller does not have permission" in error_text_str:
                    raise AuthenticationError(message=f"Authentication/Recaptcha failed: {error_text_str}", upstream_response=error_text_str)
                parsed_error = parse_error_response(error_text_str)
                if parsed_error:
                    raise parsed_error
                raise raise_for_status(code=response.status_code, message=f"Upstream Error: {error_text_str}", upstream_response=error_text_str)

            logger.debug(f"上游流式响应已建立: status={response.status_code}, model={model}")
            progress_context = kwargs.get("progress_context")
            stall_guard = kwargs.get("stream_stall_guard")
            parser = _StreamingJsonObjectParser()
            utf8_decoder = codecs.getincrementaldecoder("utf-8")()
            raw_chunk_count = 0
            raw_bytes_total = 0
            object_count = 0
            gemini_chunk_count = 0
            progress_signal_sent = False
            raw_started_at = time.monotonic()
            last_raw_progress_at = raw_started_at
            last_raw_chunk_at = raw_started_at
            max_raw_gap_ms = 0.0

            async def emit_completed_objects() -> AsyncGenerator[dict[str, Any], None]:
                nonlocal object_count, gemini_chunk_count
                for json_str in parser.pop_complete_objects():
                    object_count += 1
                    logger.debug(
                        f"上游JSON对象解析完成: model={model}, object={object_count}, "
                        f"json_chars={len(json_str)}, buffer_after={parser.buffer_length}, "
                        f"gemini_chunks={gemini_chunk_count}"
                    )
                    try:
                        obj = await _json_loads_maybe_thread(json_str)
                        async for chunk_data in self._process_streaming_object(obj):
                            gemini_chunk_count += 1
                            logger.debug(
                                f"上游Gemini chunk产出: model={model}, gemini_chunk={gemini_chunk_count}, "
                                f"keys={list(chunk_data.keys())}"
                            )
                            yield chunk_data
                    except json.JSONDecodeError:
                        logger.warning(f"上游JSON对象解析失败: model={model}, object={object_count}, json_chars={len(json_str)}")

            async for chunk in self._iter_response_content(response):
                if not chunk: continue
                raw_chunk_count += 1
                now = time.monotonic()
                raw_gap_ms = max(0.0, (now - last_raw_chunk_at) * 1000)
                max_raw_gap_ms = max(max_raw_gap_ms, raw_gap_ms)
                last_raw_chunk_at = now
                if isinstance(chunk, bytes):
                    chunk_bytes = len(chunk)
                    text_chunk = utf8_decoder.decode(chunk, final=False)
                else:
                    text_chunk = chunk
                    chunk_bytes = len(text_chunk.encode('utf-8'))
                raw_bytes_total += chunk_bytes
                if isinstance(stall_guard, dict):
                    stall_guard["last_raw_at"] = time.monotonic()
                    stall_guard["raw_chunk_count"] = raw_chunk_count
                    stall_guard["raw_bytes_total"] = raw_bytes_total
                if now - last_raw_progress_at >= 10.0:
                    progress_prefix = progress_context.get("prefix") if isinstance(progress_context, dict) else ""
                    prefix = f"{progress_prefix} " if progress_prefix else ""
                    logger.info(
                        f"{prefix}上游原始流块接收进度: raw_chunks={raw_chunk_count}, "
                        f"bytes={raw_bytes_total}, buffer={parser.buffer_length}, "
                        f"objects={object_count}, gemini_chunks={gemini_chunk_count}, "
                        f"raw_gap={raw_gap_ms:.0f}ms, max_raw_gap={max_raw_gap_ms:.0f}ms, "
                        f"elapsed={now - raw_started_at:.1f}s"
                    )
                    last_raw_progress_at = now
                logger.debug(
                    f"上游原始流块: model={model}, raw_chunk={raw_chunk_count}, "
                    f"bytes={chunk_bytes}, raw_gap={raw_gap_ms:.0f}ms, max_raw_gap={max_raw_gap_ms:.0f}ms, "
                    f"buffer_before={parser.buffer_length}"
                )
                parser.feed(text_chunk)
                if not progress_signal_sent and raw_chunk_count >= 2 and parser.buffer_length >= 4096:
                    progress_signal_sent = True
                    logger.debug(
                        f"上游大响应进度信号: model={model}, raw_chunk={raw_chunk_count}, "
                        f"buffer={parser.buffer_length}，用于提前选出winner并取消其它节点"
                    )
                    yield {
                        _INTERNAL_STREAM_PROGRESS_KEY: True,
                        "rawChunkCount": raw_chunk_count,
                        "bufferSize": parser.buffer_length,
                    }
                
                async for chunk_data in emit_completed_objects():
                    yield chunk_data

            trailing_text = utf8_decoder.decode(b"", final=True)
            if trailing_text:
                parser.feed(trailing_text)
                async for chunk_data in emit_completed_objects():
                    yield chunk_data
            logger.debug(
                f"上游流式读取结束: model={model}, raw_chunks={raw_chunk_count}, "
                f"objects={object_count}, gemini_chunks={gemini_chunk_count}, "
                f"max_raw_gap={max_raw_gap_ms:.0f}ms, "
                f"remaining_buffer={parser.buffer_length}"
            )
            if isinstance(stall_guard, dict):
                stall_guard["completed"] = True

    async def _iter_response_content(
        self,
        response: Any,
    ) -> AsyncGenerator[Any, None]:
        """顺序读取上游响应体，原样传播底层读取错误。"""
        iterator = response.aiter_content().__aiter__()
        try:
            while True:
                try:
                    chunk = await anext(iterator)
                except StopAsyncIteration:
                    break
                yield chunk
        finally:
            aclose = getattr(iterator, "aclose", None)
            if aclose is not None:
                close_task = asyncio.create_task(aclose())
                done, pending = await asyncio.wait({close_task}, timeout=_STREAM_TASK_CANCEL_TIMEOUT_SECONDS)
                if done:
                    await asyncio.gather(*done, return_exceptions=True)
                else:
                    close_task.add_done_callback(_consume_background_task_result)
                    logger.warning(
                        f"关闭上游响应迭代器超时，已转后台继续关闭: "
                        f"timeout={_STREAM_TASK_CANCEL_TIMEOUT_SECONDS:.1f}s"
                    )

    async def _process_streaming_object(self, obj: dict[str, Any]) -> AsyncGenerator[dict[str, Any], None]:
        """从单个上游 JSON 对象中提取增量 chunk"""
        results = obj.get("results", [])
        logger.debug(f"_process_streaming_object: results 数量={len(results)}")
        for result in results:
            # 错误检测
            errors = result.get("errors")
            if errors and isinstance(errors, list) and len(errors) > 0:
                err_msg = errors[0].get("message", "") if isinstance(errors[0], dict) else str(errors[0])
                # "Failed to verify action" 是匿名接口首次必败的预期错误
                if "Failed to verify action" in err_msg or "The caller does not have permission" in err_msg:
                    raise AuthenticationError(message=err_msg, upstream_response=err_msg)
                parsed = parse_error_response({"errors": errors})
                if parsed:
                    raise parsed

            data = result.get("data")
            if not isinstance(data, dict):
                logger.debug(f"result.data 不是 dict: type={type(data)}")
                continue

            # 展开 ui.streamGenerateContentAnonymous 包装
            ui = data.get("ui", {})
            if isinstance(ui, dict) and "streamGenerateContentAnonymous" in ui:
                inner = ui["streamGenerateContentAnonymous"]
                logger.debug(f"展开 ui 包装: inner type={type(inner)}, len={len(inner) if isinstance(inner, list) else 'N/A'}")
                if isinstance(inner, dict):
                    data = inner
                elif isinstance(inner, list):
                    for item in inner:
                        if isinstance(item, dict):
                            logger.debug(f"yield list item: keys={list(item.keys())}")
                            yield self._sanitize_downstream_chunk(item)
                    continue
                else:
                    continue

            candidates = data.get("candidates", [])
            chunk: dict[str, Any] = {}
            if candidates:
                chunk["candidates"] = candidates
            if data.get("usageMetadata"):
                chunk["usageMetadata"] = data["usageMetadata"]
            if data.get("modelVersion"):
                chunk["modelVersion"] = data["modelVersion"]
            if data.get("responseId"):
                chunk["responseId"] = data["responseId"]
            if data.get("promptFeedback"):
                chunk["promptFeedback"] = data["promptFeedback"]
            
            if chunk:
                yield self._sanitize_downstream_chunk(chunk)

    def _sanitize_downstream_chunk(self, chunk: dict[str, Any]) -> dict[str, Any]:
        """清理下发给 Gemini 客户端的空壳 part 字段，避免客户端写入坏历史。"""
        sanitized = dict(chunk)
        candidates = sanitized.get("candidates")
        if not isinstance(candidates, list):
            return sanitized

        new_candidates: list[Any] = []
        for candidate in cast(list[Any], candidates):
            if not isinstance(candidate, dict):
                new_candidates.append(candidate)
                continue
            candidate_dict = cast(dict[str, Any], candidate).copy()
            content = candidate_dict.get("content")
            if isinstance(content, dict):
                content_dict = cast(dict[str, Any], content).copy()
                parts = content_dict.get("parts")
                if isinstance(parts, list):
                    content_dict["parts"] = [
                        self._sanitize_downstream_part(cast(dict[str, Any], part)) if isinstance(part, dict) else part
                        for part in cast(list[Any], parts)
                    ]
                candidate_dict["content"] = content_dict
            new_candidates.append(candidate_dict)
        sanitized["candidates"] = new_candidates
        return sanitized

    def _sanitize_downstream_part(self, part: dict[str, Any]) -> dict[str, Any]:
        cleaned = dict(part)
        if cleaned.get("data") == "text":
            cleaned.pop("data", None)
        if cleaned.get("type") == "text":
            cleaned.pop("type", None)

        for key in ("inlineData", "inline_data", "fileData", "file_data", "functionCall", "function_call", "functionResponse", "function_response"):
            value = cleaned.get(key)
            if not self._has_meaningful_downstream_part_value(value):
                cleaned.pop(key, None)
        return cleaned

    @staticmethod
    def _has_meaningful_downstream_part_value(value: Any) -> bool:
        if value is None or value is False:
            return False
        if isinstance(value, str):
            return value != ""
        if isinstance(value, dict):
            return any(VcoreAIClient._has_meaningful_downstream_part_value(v) for v in value.values())
        if isinstance(value, (list, tuple, set)):
            return any(VcoreAIClient._has_meaningful_downstream_part_value(v) for v in value)
        return True

    async def _execute_count_tokens_attempt(
        self,
        session: Any,
        model: str,
        contents: list[dict[str, Any]],
        recaptcha_token: str,
    ) -> int:
        """执行一次 CountTokens 上游请求。"""
        target_model = self.model_builder.parse_model_name(model)
        if target_model.startswith("models/"):
            target_model = target_model[7:]

        payload = {
            "requestContext": self._build_request_context(),
            "querySignature": "2/mENOSldfC+HZM+tGhVuJLrl8M6gEyK3HRjUKuA5AM58=",
            "operationName": "CountTokens",
            "variables": {
                "contents": contents,
                "endpoint": "",
                "model": target_model,
                "region": "global",
                "recaptchaToken": recaptcha_token,
            },
        }
        headers = self._build_browser_headers()
        url = f"{self.vcore_ai_anonymous_base_api}/v3/entityServices/AiplatformEntityService/schemas/AIPLATFORM_GRAPHQL:batchGraphql?key=AIzaSyCI-zsRP85UVOi0DjtiCwWBwQ1djDy741g&prettyPrint=false"

        response = await self.network.post_request(session, url, headers, payload)
        if response.status_code != 200:
            text = response.text if hasattr(response, "text") else ""
            if response.status_code in [401, 403] or "Failed to verify action" in text or "The caller does not have permission" in text:
                raise AuthenticationError(message=f"Authentication/Recaptcha failed: {text}", upstream_response=text)
            parsed_error = parse_error_response(text)
            if parsed_error:
                raise parsed_error
            raise raise_for_status(code=response.status_code, message=f"Upstream Error: {text}", upstream_response=text)

        data = response.json()
        items = data if isinstance(data, list) else [data]
        for entry in items:
            if not isinstance(entry, dict):
                continue
            parsed_error = parse_error_response(entry)
            if parsed_error:
                if "Failed to verify action" in parsed_error.message or "The caller does not have permission" in parsed_error.message:
                    raise AuthenticationError(message=parsed_error.message, upstream_response=str(entry))
                raise parsed_error
            for result in entry.get("results", []) or []:
                if not isinstance(result, dict):
                    continue
                parsed_result_error = parse_error_response(result)
                if parsed_result_error:
                    raise parsed_result_error
                data_obj = result.get("data", {})
                if not isinstance(data_obj, dict):
                    continue
                ui_data = data_obj.get("ui", {}) if isinstance(data_obj.get("ui"), dict) else {}
                count_data = ui_data.get("countTokensV2") or data_obj.get("countTokensV2") or data_obj.get("countTokens")
                if isinstance(count_data, dict) and "totalTokens" in count_data:
                    return int(count_data["totalTokens"])
        raise InternalError(message="CountTokens response did not contain totalTokens")

    async def _count_tokens_inner(
        self,
        session: Any,
        model: str,
        contents: list[dict[str, Any]],
        retry_limit_override: int | None = None,
    ) -> int:
        retry_limit = self._node_retry_limit(retry_limit_override)
        retries_used = 0
        recaptcha_token = None
        is_first_auth_attempt = True

        async def consume_retry(reason: str) -> bool:
            nonlocal retries_used
            if retries_used >= retry_limit:
                return False
            retries_used += 1
            logger.debug(f"CountTokens 单节点重试 {retries_used}/{retry_limit}: {reason}")
            await asyncio.sleep(0)
            return True

        while True:
            if not recaptcha_token:
                recaptcha_token = await self.network.fetch_recaptcha_token(session)
                is_first_auth_attempt = True
            if not recaptcha_token:
                if await consume_retry("获取 recaptcha token 失败"):
                    continue
                raise AuthenticationError("Could not fetch recaptcha token.")
            try:
                return await self._execute_count_tokens_attempt(session, model, contents, recaptcha_token)
            except AuthenticationError:
                if is_first_auth_attempt:
                    is_first_auth_attempt = False
                    if await consume_retry("首次认证失败"):
                        continue
                    raise
                recaptcha_token = None
                if await consume_retry("认证失败"):
                    continue
                raise
            except RateLimitError:
                recaptcha_token = None
                if await consume_retry("429 限流"):
                    continue
                raise
            except VcoreError as e:
                if not e.is_retryable:
                    raise
                if await consume_retry(f"可重试上游错误: {e.message}"):
                    continue
                raise
            except Exception as e:
                recaptcha_token = None
                if await consume_retry(f"CountTokens 网络/内部异常: {e}"):
                    continue
                raise InternalError(message=f"CountTokens error: {e}") from e

    async def count_tokens(self, model: str, contents: list[dict[str, Any]], **kwargs: Any) -> int:
        """通过统一业务请求池执行 CountTokens。"""
        cfg = load_config()
        business_session_id = str(kwargs.get("business_session_id") or "") or None
        retry_limit = self._node_retry_limit(cfg.get("node_retry_count", self.node_retry_count))

        async def operation(session: Any, proxy_url: str | None) -> int:
            return await self._count_tokens_inner(session, model, contents, retry_limit_override=retry_limit)

        return cast(int, await self._run_with_parallel_request_pool(
            "CountTokens",
            operation,
            cfg,
            business_session_id=business_session_id,
            gateway_session=kwargs.get("gateway_session"),
        ))

    async def _stream_realtime_inner(self, model: str, gemini_payload: dict[str, Any], **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]:
        """真流式内部方法（含重试逻辑）"""
        retry_limit = self._node_retry_limit(kwargs.pop("node_retry_count_override", self.node_retry_count))
        session_override = kwargs.pop("session_override", None)
        session_proxy_override = kwargs.pop("session_proxy_override", None)
        worker_override = kwargs.pop("worker_override", None)
        content_yielded = False
        recaptcha_token = None
        is_first_auth_attempt = True
        retries_used = 0

        async def consume_retry(reason: str) -> bool:
            nonlocal retries_used
            if retries_used >= retry_limit:
                return False
            retries_used += 1
            logger.debug(f"真流式单节点重试 {retries_used}/{retry_limit}: {reason}")
            await asyncio.sleep(0)
            return True

        session = session_override or self.network.create_session()
        try:
            while True:
                if not recaptcha_token:
                    recaptcha_token = await self.network.fetch_recaptcha_token(session)
                    is_first_auth_attempt = True

                if not recaptcha_token:
                    last_error = getattr(session, "_vcore_proxy_last_recaptcha_error", "")
                    if await consume_retry("获取 recaptcha token 失败"):
                        continue
                    error = AuthenticationError("Could not fetch recaptcha token.")
                    if last_error:
                        raise error from RuntimeError(last_error)
                    raise error

                try:
                    emitted_count = 0
                    actual_chunk_count = 0
                    async for chunk in self._execute_streaming_attempt(
                        session, model, gemini_payload, recaptcha_token, kwargs,
                        is_first_auth_attempt=is_first_auth_attempt,
                    ):
                        yield chunk
                        emitted_count += 1
                        is_internal_progress = bool(chunk.get(_INTERNAL_STREAM_PROGRESS_KEY)) if isinstance(chunk, dict) else False
                        if not is_internal_progress:
                            content_yielded = True
                            actual_chunk_count += 1

                    if actual_chunk_count == 0 and is_first_auth_attempt:
                        logger.debug("真流式首次请求返回空数据，触发认证重试")
                        is_first_auth_attempt = False
                        if await consume_retry("首次请求返回空数据"):
                            continue
                        raise UpstreamResponseIncompleteError(message="节点未返回任何有效响应结构")
                    if actual_chunk_count == 0 and emitted_count > 0:
                        raise UpstreamResponseIncompleteError(message="节点只返回了内部进度信号，未返回任何有效响应结构")
                    break

                except AuthenticationError:
                    if content_yielded:
                        raise
                    if is_first_auth_attempt:
                        is_first_auth_attempt = False
                        if await consume_retry("首次认证失败"):
                            continue
                        raise
                    recaptcha_token = None
                    if await consume_retry("认证失败"):
                        continue
                    raise

                except RateLimitError as e:
                    if content_yielded:
                        raise
                    if not await consume_retry("429 限流"):
                        raise
                    logger.info("429 限流，销毁当前 session 并重建以切换出口 IP")
                    await session.close()
                    if session_override is not None:
                        session = self.network.create_session_with_proxy(session_proxy_override)
                    else:
                        session = self.network.create_session()
                    recaptcha_token = None

                except VcoreError as e:
                    if not e.is_retryable or content_yielded:
                        raise
                    if await consume_retry(f"可重试上游错误: {e.message}"):
                        continue
                    raise

                except Exception as e:
                    if content_yielded:
                        raise InternalError(message=f"Internal error: {e}") from e
                    if await consume_retry(f"网络/内部异常: {e}"):
                        continue
                    raise InternalError(message=f"Internal error: {e}") from e
        finally:
            logger.debug(
                f"真流式内部资源清理: model={model}, proxy={session_proxy_override or 'direct'}, "
                f"content_yielded={content_yielded}"
            )
            await session.close()
            if worker_override is not None:
                await worker_override.stop()