services/protocol/conversation.py

from __future__ import annotations

import base64
import json
import mimetypes
import re
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field, replace
from pathlib import Path
from queue import Queue
from typing import Any, Iterable, Iterator
from urllib.parse import quote, unquote

import tiktoken

from services.account_service import account_service
from services.config import DATA_DIR, config
from services.image_conversation_state_service import image_conversation_state_service
from services.image_storage_service import image_storage_service
from services.openai_backend_api import (
    ImagePollTimeoutError,
    ImageTextOnlyAnswer,
    OpenAIBackendAPI,
    image_poll_timeout_message,
)
from utils.helper import IMAGE_MODELS, extract_image_from_message_content
from utils.image_tokens import count_image_content_tokens
from utils.log import logger
from utils.sensitive_data import sanitize_sensitive_text


class ImageGenerationError(Exception):
    def __init__(
        self,
        message: str,
        status_code: int = 502,
        error_type: str = "server_error",
        code: str | None = "upstream_error",
        param: str | None = None,
        account_email: str = "",
        conversation_id: str = "",
        terminal_no_image: bool = False,
        can_view_official: bool = True,
    ) -> None:
        super().__init__(message)
        self.status_code = status_code
        self.error_type = error_type
        self.code = code
        self.param = param
        self.account_email = account_email
        self.conversation_id = conversation_id
        # 上游已经给出终局结论（只回文字、本轮没生图）：继续等待也等不到图片，
        # 前台不该再给一个「继续等待」的假按钮。见 image_task_service 的 can_continue_waiting。
        self.terminal_no_image = terminal_no_image
        # 「查看官网对话」入口是否值得给。上游结论是**账号级配额/限流**时（例如
        # 「你已达到 Free 套餐的图像生成请求上限」），官网这一轮根本没生成任何东西，
        # 点进去只有同一句提示 → 失败卡片上不该摆这个入口。见 is_image_quota_exhausted_message。
        self.can_view_official = can_view_official

    def to_openai_error(self) -> dict[str, Any]:
        return {
            "error": {
                "message": sanitize_sensitive_text(str(self)),
                "type": self.error_type,
                "param": self.param,
                "code": self.code,
            }
        }


@dataclass
class TextConversationResult:
    text: str
    attachments: list[dict[str, Any]] = field(default_factory=list)


SANDBOX_ARTIFACT_RE = re.compile(r"sandbox:(/mnt/data/[^\s<>\]\)\"']+)", re.IGNORECASE)
CHAT_ARTIFACT_ROOT = DATA_DIR / "files" / "chat"


def extract_sandbox_artifact_paths(text: str) -> list[str]:
    paths: list[str] = []
    for match in SANDBOX_ARTIFACT_RE.finditer(str(text or "")):
        path = unquote(str(match.group(1) or "").strip()).rstrip(".,;:!?")
        if path.startswith("/mnt/data/") and path not in paths:
            paths.append(path)
    return paths


def is_token_invalid_error(message: str) -> bool:
    text = str(message or "").lower()
    return (
        "token_invalidated" in text
        or "token_revoked" in text
        or "authentication token has been invalidated" in text
        or "invalidated oauth token" in text
        # ChatGPT returns an HTML 403 from the bootstrap page for accounts
        # that are no longer accepted. Treat it like an unavailable account so
        # the pool can try another token instead of failing the whole request.
        or "bootstrap failed: status=403" in text
    )


def is_transient_image_connection_error(message: str) -> bool:
    """识别图片 SSE 中可通过短暂重试恢复的网络错误。

    curl 16/18 通常表示 HTTP/2 流或响应体被上游提前关闭，并不代表账号
    已失效；如果直接终止请求，号池会把一次瞬时网络抖动暴露给用户。
    """
    text = str(message or "").lower()
    return (
        "curl: (16)" in text
        or "curl: (18)" in text
        or "curl: (28)" in text
        or "curl: (35)" in text
        or "tls connect error" in text
        or "openssl_internal" in text
        or "remote disconnected" in text
        or "connection reset by peer" in text
        or "connection aborted" in text
        or "incomplete transfer" in text
        or "operation timed out" in text
        or "connection timed out" in text
        or "read timed out" in text
        or "connect timeout" in text
    )


_UPLOAD_THROTTLE_MARKERS = (
    # ChatGPT 的上传限流是一个账号维度的配额（upload_protection_limit_exceeded），
    # 换一个账号通常仍能上传，所以绝不能当成终态失败暴露给用户。
    "upload_protection_limit_exceeded",
    "throttled",
    "文件上传上限",
    "file upload limit",
    "upload limit",
)

_UPLOAD_PATHS = ("/backend-api/files", "image_upload failed", "file_upload failed", "image upload", "file upload")


def is_account_scoped_upload_error(message: str) -> bool:
    """识别「换个账号就能成功」的上传类失败。

    上传限流（429 throttled / upload_protection_limit_exceeded）是绑定在单个
    ChatGPT 账号上的配额，和网络抖动、token 失效不同：重试同一个账号没有意义，
    但换号几乎总能成功。因此这类错误必须进换号分支，而不是当作终态失败抛出。
    """
    text = str(message or "").lower()
    if not text:
        return False
    if not any(marker in text for marker in _UPLOAD_PATHS):
        return False
    if any(marker in text for marker in _UPLOAD_THROTTLE_MARKERS):
        return True
    # 未带 detail 的裸 429 上传响应同样按账号配额处理。
    return "status=429" in text


# 上游「本账号这轮生图配额用完了」的原话。和内容策略拒绝必须分开：
# 拒绝时官网会话里有原话可核对（用户点「查看官网对话」有价值），
# 而配额耗尽时官网这一轮**什么都没生成**，点进去只有同一句提示。
_IMAGE_QUOTA_MARKERS = (
    # 中文原话：你已达到 <套餐> 的图像生成请求上限。上限将在 N 小时 后重置，届时可创建更多图像。
    "图像生成请求上限",
    # 英文原话：You've reached the <plan> plan's limit for image generation requests.
    "limit for image generation",
    "image generation request limit",
)


def is_image_quota_exhausted_message(message: str) -> bool:
    """识别「本账号的生图配额已用完」这类终局结论（Free/Plus 套餐同款文案）。"""
    text = str(message or "").lower()
    if not text:
        return False
    if any(marker in text for marker in _IMAGE_QUOTA_MARKERS):
        return True
    # 中文原话的后半句（「上限将在 23小时 后重置」）兜底：套餐名/语序变化时仍能认出。
    return "上限将在" in text and "重置" in text


def is_retryable_multimodal_account_error(message: str) -> bool:
    """Treat account-scoped image-upload failures as eligible for failover."""
    text = str(message or "").lower()
    if is_token_invalid_error(text):
        return True
    if any(marker in text for marker in ("image_upload failed", "file_upload failed")) and any(
        marker in text
        for marker in ("status=401", "status=403", "authenticationfailed", "signature fields")
    ):
        return True
    if is_account_scoped_upload_error(text):
        return True
    return is_transient_image_connection_error(text)


def image_stream_error_message(message: str) -> str:
    text = str(message or "")
    lower = text.lower()
    if is_token_invalid_error(text):
        return "image generation failed"
    if is_account_scoped_upload_error(text):
        return "号池中所有账号的上传额度均已达到上限，请稍后重试或补充账号"
    if is_transient_image_connection_error(text) and (
        "curl: (35)" in lower
        or "tls connect error" in lower
        or "openssl_internal" in lower
        or "curl: (16)" in lower
        or "curl: (18)" in lower
        or "remote disconnected" in lower
        or "connection reset" in lower
        or "connection aborted" in lower
        or "incomplete transfer" in lower
    ):
        return "upstream image connection failed, please retry later"
    if is_transient_image_connection_error(text):
        return "upstream connection timed out, please retry later"
    return text or "image generation failed"


def encode_images(images: Iterable[tuple[bytes, str, str]]) -> list[str]:
    return [base64.b64encode(data).decode("ascii") for data, _, _ in images if data]


def save_image_bytes(image_data: bytes, base_url: str | None = None) -> str:
    return image_storage_service.save(image_data, base_url).url


def message_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for item in content:
            if isinstance(item, str):
                parts.append(item)
            elif isinstance(item, dict) and str(item.get("type") or "") in {"text", "input_text", "output_text"}:
                parts.append(str(item.get("text") or ""))
        return "".join(parts)
    return ""


def _decode_file_content_part(part: dict[str, Any]) -> tuple[bytes, str, str] | None:
    part_type = str(part.get("type") or "").strip()
    if part_type not in {"file", "input_file"}:
        return None
    file_payload = part.get("file") if part_type == "file" else part
    if not isinstance(file_payload, dict):
        return None
    file_data = str(file_payload.get("file_data") or "").strip()
    if not file_data:
        return None
    filename = str(file_payload.get("filename") or "attachment").strip() or "attachment"
    mime_type = str(file_payload.get("mime_type") or "").strip().lower()
    encoded = file_data
    data_url = re.match(r"^data:([^;,]+)?(?:;[^,]*)?;base64,(.*)$", file_data, re.IGNORECASE | re.DOTALL)
    if data_url:
        mime_type = mime_type or str(data_url.group(1) or "").strip().lower()
        encoded = str(data_url.group(2) or "").strip()
    try:
        data = base64.b64decode(encoded, validate=True)
    except (ValueError, TypeError) as exc:
        raise ValueError(f"invalid base64 file_data for {filename}") from exc
    if not data:
        raise ValueError(f"empty file_data for {filename}")
    guessed_mime = mimetypes.guess_type(filename)[0]
    if not mime_type or mime_type == "application/octet-stream":
        mime_type = guessed_mime or mime_type or "application/octet-stream"
    return data, filename, mime_type


def normalize_messages(messages: object, system: Any = None) -> list[dict[str, Any]]:
    normalized = []
    if config.global_system_prompt:
        normalized.append({"role": "system", "content": config.global_system_prompt})
    system_text = message_text(system)
    if system_text:
        normalized.append({"role": "system", "content": system_text})
    if isinstance(messages, list):
        for message in messages:
            if not isinstance(message, dict):
                continue
            role = message.get("role", "user")
            content = message.get("content", "")
            text = message_text(content)
            images: list[tuple[bytes, str]] = []
            files: list[tuple[bytes, str, str]] = []
            if role == "user":
                images.extend(extract_image_from_message_content(content))
                if isinstance(content, list):
                    for part in content:
                        if not isinstance(part, dict):
                            continue
                        if part.get("type") == "image":
                            data = part.get("data")
                            if isinstance(data, (bytes, bytearray)):
                                images.append((bytes(data), str(part.get("mime") or "image/png")))
                            continue
                        if part.get("type") == "file":
                            data = part.get("data")
                            if isinstance(data, (bytes, bytearray)):
                                files.append((
                                    bytes(data),
                                    str(part.get("name") or "attachment"),
                                    str(part.get("mime") or "application/octet-stream"),
                                ))
                                continue
                        decoded_file = _decode_file_content_part(part)
                        if decoded_file:
                            files.append(decoded_file)
            if images or files:
                parts: list[Any] = []
                if text:
                    parts.append({"type": "text", "text": text})
                for data, mime in images:
                    parts.append({"type": "image", "data": data, "mime": mime})
                for data, filename, mime in files:
                    parts.append({"type": "file", "data": data, "name": filename, "mime": mime})
                normalized.append({"role": role, "content": parts})
            else:
                normalized.append({"role": role, "content": text})
    return normalized


def prompt_with_global_system(prompt: str) -> str:
    return f"{config.global_system_prompt}\n\n{prompt}" if config.global_system_prompt else prompt


def assistant_history_text(messages: list[dict[str, Any]]) -> str:
    return "".join(str(item.get("content") or "") for item in messages if item.get("role") == "assistant")


def assistant_history_messages(messages: list[dict[str, Any]]) -> list[str]:
    return [str(item.get("content") or "") for item in messages if item.get("role") == "assistant" and item.get("content")]


_ASPECT_SIZE_PATTERN = re.compile(r"^(?:\d+(?:\.\d+)?:\d+(?:\.\d+)?|\d+[x×]\d+)$", re.IGNORECASE)


def normalize_aspect_size(size: str | None) -> str:
    """把 size 收紧成比例（16:9）或像素尺寸（1024x1024）写法。

    不是这两种形状的一律返回空串。`size` 有三个来源（对话页 LLM 抽取、画图页比例下拉、
    `/v1/images/*` 直调），其中直调**零校验**，实测出现过「高清 / optional / 原图比例 /
    宽幅长图 / vertical editorial poster」这类自然语言值。把它们拼进提示词只会白占 token，
    所以拼装前必须挡掉，而不是原样追加。
    """
    text = re.sub(r"\s+", "", str(size or "").strip())
    if not _ASPECT_SIZE_PATTERN.match(text):
        return ""
    return text


def build_image_prompt(prompt: str, size: str | None) -> str:
    # 比例后缀是可选行为：账号池走网页会话，没有接口级宽高比参数，
    # 需要时靠这段文本约束上游；但提示词通常已写明画幅，重复追加会干扰阅读。
    if not config.image_prompt_aspect_hint:
        return prompt
    aspect = normalize_aspect_size(size)
    if not aspect:
        return prompt
    # 实测上游会优先服从提示词里的第一条画幅描述，所以明确让它以这里为准，
    # 避免「用户正文写了 16:9、下拉选了 9:16」时按正文出图。
    return (
        f"{prompt.strip()}\n\n"
        f"输出宽高比：{aspect}。若上文中的宽高比或构图指令与此输出比例冲突，请以此比例为准。"
    )


def encoding_for_model(model: str):
    try:
        return tiktoken.encoding_for_model(model)
    except KeyError:
        try:
            return tiktoken.get_encoding("o200k_base")
        except KeyError:
            return tiktoken.get_encoding("cl100k_base")


def count_message_image_tokens(messages: list[dict[str, Any]], model: str) -> int:
    return sum(count_image_content_tokens(message.get("content"), model) for message in messages)


def count_message_text_tokens(messages: list[dict[str, Any]], model: str) -> int:
    encoding = encoding_for_model(model)
    total = 0
    for message in messages:
        total += 3
        for key, value in message.items():
            if key == "content" and isinstance(value, list):
                total += len(encoding.encode(message_text(value)))
            elif isinstance(value, str):
                total += len(encoding.encode(value))
            else:
                continue
            if key == "name":
                total += 1
    return total + 3


def count_message_tokens(messages: list[dict[str, Any]], model: str) -> int:
    return count_message_text_tokens(messages, model) + count_message_image_tokens(messages, model)


def count_text_tokens(text: str, model: str) -> int:
    return len(encoding_for_model(model).encode(text))


def format_image_result(
    items: list[dict[str, Any]],
    prompt: str,
    response_format: str,
    base_url: str | None = None,
    created: int | None = None,
    message: str = "",
) -> dict[str, Any]:
    data: list[dict[str, Any]] = []
    for item in items:
        revised_prompt = str(item.get("revised_prompt") or prompt).strip() or prompt
        image_url = str(item.get("url") or "").strip()
        if response_format != "b64_json" and image_url:
            data.append({
                "url": image_url,
                "revised_prompt": revised_prompt,
            })
            continue
        b64_json = str(item.get("b64_json") or "").strip()
        if not b64_json:
            continue
        if response_format == "b64_json":
            # A Base64 caller already receives the complete image payload.
            # Persisting a duplicate first adds disk I/O and can hold the HTTP
            # response until a large image has been written.  OpenAI-compatible
            # Base64 responses only require this field, so return it directly.
            data.append({
                "b64_json": b64_json,
                "revised_prompt": revised_prompt,
            })
        else:
            data.append({
                "url": save_image_bytes(base64.b64decode(b64_json), base_url),
                "revised_prompt": revised_prompt,
            })
    result: dict[str, Any] = {"created": created or int(time.time()), "data": data}
    if message and not data:
        result["message"] = message
    return result


@dataclass
class ConversationRequest:
    model: str = "auto"
    prompt: str = ""
    messages: list[dict[str, Any]] | None = None
    thinking_effort: str = ""
    images: list[str] | None = None
    n: int = 1
    size: str | None = None
    response_format: str = "url"
    base_url: str | None = None
    message_as_error: bool = False
    memory_enabled: bool = False
    memory_owner_id: str = ""
    memory_conversation_id: str = ""
    memory_reset: bool = False
    memory_mode: str = ""
    account_owner_id: str = ""
    account_pool_strategy: str = "admin_first"
    account_allow_shared: bool = True
    account_wait_for_slot: bool = True
    image_timeout_secs: float = 0.0
    progress_callback: Any = None


@dataclass
class ConversationState:
    text: str = ""
    raw_text: str = ""
    conversation_id: str = ""
    parent_message_id: str = ""
    file_ids: list[str] = field(default_factory=list)
    sediment_ids: list[str] = field(default_factory=list)
    blocked: bool = False
    tool_invoked: bool | None = None
    turn_use_case: str = ""


@dataclass
class ImageOutput:
    kind: str
    model: str
    index: int
    total: int
    created: int = field(default_factory=lambda: int(time.time()))
    text: str = ""
    upstream_event_type: str = ""
    conversation_id: str = ""
    parent_message_id: str = ""
    data: list[dict[str, Any]] = field(default_factory=list)
    account_email: str = ""

    def to_chunk(self) -> dict[str, Any]:
        chunk: dict[str, Any] = {
            "object": "image.generation.chunk",
            "created": self.created,
            "model": self.model,
            "index": self.index,
            "total": self.total,
            "progress_text": self.text,
            "upstream_event_type": self.upstream_event_type,
            "data": [],
        }
        if self.account_email:
            chunk["_account_email"] = self.account_email
        if self.conversation_id:
            chunk["_conversation_id"] = self.conversation_id
        if self.kind == "message":
            chunk.update({
                "object": "image.generation.message",
                "message": self.text,
            })
            chunk.pop("progress_text", None)
            chunk.pop("upstream_event_type", None)
        elif self.kind == "result":
            chunk.update({
                "object": "image.generation.result",
                "data": self.data,
            })
            chunk.pop("progress_text", None)
            chunk.pop("upstream_event_type", None)
        return chunk


def assistant_message_text(message: dict[str, Any]) -> str:
    content = message.get("content") or {}
    parts = content.get("parts") or []
    if not isinstance(parts, list):
        return ""
    return "".join(part for part in parts if isinstance(part, str))


def strip_history(text: str, history_text: str = "") -> str:
    text = str(text or "")
    history_text = str(history_text or "")
    while history_text and text.startswith(history_text):
        text = text[len(history_text):]
    return text


def is_visible_assistant_message(message: dict[str, Any]) -> bool:
    author = message.get("author")
    if not isinstance(author, dict) or str(author.get("role") or "").strip().lower() != "assistant":
        return False
    metadata = message.get("metadata") or {}
    if isinstance(metadata, dict) and metadata.get("is_visually_hidden_from_conversation") is True:
        return False
    recipient = str(message.get("recipient") or "").strip().lower()
    if recipient and recipient != "all":
        return False
    channel = str(message.get("channel") or "").strip().lower()
    return not channel or channel == "final"


def sanitize_output_text(text: str) -> str:
    text = str(text or "")

    def is_internal(part: str) -> bool:
        value = part.strip().lower()
        return not value or bool(re.fullmatch(r"turn\d+\w*", value)) or value.startswith(("source", "sources"))

    def annotation_text(payload: str) -> str:
        parts = [part.strip() for part in payload.split("\ue202")]
        kind = (parts[0] if parts else "").lower()
        data = parts[1:]
        if kind == "url":
            label = data[0] if data else ""
            url = data[1] if len(data) > 1 else ""
            if url.startswith("sandbox:/mnt/data/"):
                return f"[{label or Path(url).name}]({url})"
            return f"{label} ({url})" if label and url.startswith(("http://", "https://")) else label or url
        return next((part for part in data if not is_internal(part)), "")

    def replace_before_punctuation(match: re.Match[str]) -> str:
        replacement = annotation_text(match.group(2))
        return f"{match.group(1)}{replacement}" if replacement else ""

    text = re.sub(r"(\s*)\ue200([^\ue201]*)\ue201(?=[.,;:!?])", replace_before_punctuation, text)
    text = re.sub(r"\ue200([^\ue201]*)\ue201", lambda match: annotation_text(match.group(1)), text)
    text = re.sub(r"\ue200[^\ue201]*$", "", text)
    return text


def assistant_raw_text(event: dict[str, Any], current_text: str = "", history_text: str = "") -> str:
    for candidate in (event, event.get("v")):
        if not isinstance(candidate, dict):
            continue
        message = candidate.get("message")
        if not isinstance(message, dict) or not is_visible_assistant_message(message):
            continue
        text = assistant_message_text(message)
        if text:
            return strip_history(text, history_text)
    return apply_text_patch(event, current_text, history_text)


def assistant_text(event: dict[str, Any], current_text: str = "", history_text: str = "") -> str:
    return sanitize_output_text(assistant_raw_text(event, current_text, history_text))


def event_assistant_text(event: dict[str, Any], history_text: str = "") -> str:
    for candidate in (event, event.get("v")):
        if not isinstance(candidate, dict):
            continue
        message = candidate.get("message")
        if isinstance(message, dict) and is_visible_assistant_message(message):
            return strip_history(assistant_message_text(message), history_text)
    return ""


def apply_text_patch(event: dict[str, Any], current_text: str = "", history_text: str = "") -> str:
    if event.get("p") == "/message/content/parts/0":
        return apply_patch_op(event, current_text, history_text)

    operations = event.get("v")
    if isinstance(operations, str) and current_text and not event.get("p") and not event.get("o"):
        return current_text + operations

    if event.get("o") == "patch" and isinstance(operations, list):
        text = current_text
        for item in operations:
            if isinstance(item, dict):
                text = apply_text_patch(item, text, history_text)
        return text

    if not isinstance(operations, list):
        return current_text

    text = current_text
    for item in operations:
        if isinstance(item, dict):
            text = apply_text_patch(item, text, history_text)
    return text


def apply_patch_op(operation: dict[str, Any], current_text: str, history_text: str = "") -> str:
    op = operation.get("o")
    value = str(operation.get("v") or "")
    if op == "append":
        return current_text + value
    if op == "replace":
        return strip_history(value, history_text)
    return current_text


def add_unique(values: list[str], candidates: list[str]) -> None:
    for candidate in candidates:
        if candidate and candidate not in values:
            values.append(candidate)


FILE_SERVICE_ID_RE = re.compile(r"file-service://([A-Za-z0-9_-]+)")
# Direct generated image file ids are currently `file_00000000` + 24 hex chars.
# Do not accept loose `file_*` strings here; many user/input/upload ids share that prefix.
REAL_IMAGE_FILE_ID_RE = re.compile(r"\bfile_00000000[a-f0-9]{24}\b")
SEDIMENT_ID_RE = re.compile(r"sediment://([A-Za-z0-9_-]+)")


def extract_conversation_ids(payload: str) -> tuple[str, list[str], list[str]]:
    conversation_match = re.search(r'"conversation_id"\s*:\s*"([^"]+)"', payload)
    conversation_id = conversation_match.group(1) if conversation_match else ""
    file_ids: list[str] = []
    add_unique(file_ids, FILE_SERVICE_ID_RE.findall(payload))
    add_unique(file_ids, REAL_IMAGE_FILE_ID_RE.findall(payload))
    sediment_ids = SEDIMENT_ID_RE.findall(payload)
    return conversation_id, file_ids, sediment_ids


def is_image_tool_event(event: dict[str, Any]) -> bool:
    value = event.get("v")
    message = event.get("message") or (value.get("message") if isinstance(value, dict) else None)
    if not isinstance(message, dict):
        return False
    metadata = message.get("metadata") or {}
    author = message.get("author") or {}
    content = message.get("content") or {}
    if author.get("role") != "tool":
        return False
    if metadata.get("async_task_type") == "image_gen":
        return True
    if content.get("content_type") != "multimodal_text":
        return False
    return any(
        isinstance(part, dict) and (
                part.get("content_type") == "image_asset_pointer"
                or str(part.get("asset_pointer") or "").startswith(("file-service://", "sediment://"))
        )
        for part in content.get("parts") or []
    )


def _is_user_message_event(event: dict[str, Any]) -> bool:
    value = event.get("v")
    message = event.get("message") or (value.get("message") if isinstance(value, dict) else None)
    if not isinstance(message, dict):
        return False
    author = message.get("author") or {}
    return str(author.get("role") or "").strip().lower() == "user"


def event_message_id(event: dict[str, Any]) -> str:
    for candidate in (event, event.get("v")):
        if not isinstance(candidate, dict):
            continue
        message = candidate.get("message")
        if isinstance(message, dict):
            message_id = str(message.get("id") or "").strip()
            if message_id:
                return message_id
        message_id = str(candidate.get("message_id") or "").strip()
        if message_id:
            return message_id
    return ""


def update_conversation_state(state: ConversationState, payload: str, event: dict[str, Any] | None = None) -> None:
    conversation_id, file_ids, sediment_ids = extract_conversation_ids(payload)
    if conversation_id and not state.conversation_id:
        state.conversation_id = conversation_id
    # Accept file_id / sediment_id when any of:
    #   1) event is a complete image_gen tool message
    #   2) prior server_ste_metadata already flipped tool_invoked True (in an image_gen turn)
    #   3) patch event whose payload references asset_pointer / file-service://
    # User messages (type=conversation.message) never satisfy these, so attacker-controlled
    # substrings in user input cannot inject file ids into state.
    is_patch_event = isinstance(event, dict) and event.get("o") == "patch"
    is_user_msg = isinstance(event, dict) and _is_user_message_event(event)
    image_context = (
        (isinstance(event, dict) and is_image_tool_event(event))
        or (state.tool_invoked is True and not is_user_msg)
        or (is_patch_event and not is_user_msg and ("asset_pointer" in payload or "file-service://" in payload))
    )
    if image_context:
        add_unique(state.file_ids, file_ids)
        add_unique(state.sediment_ids, sediment_ids)
    if not isinstance(event, dict):
        return
    state.conversation_id = str(event.get("conversation_id") or state.conversation_id)
    message_id = event_message_id(event)
    if message_id:
        state.parent_message_id = message_id
    value = event.get("v")
    if isinstance(value, dict):
        state.conversation_id = str(value.get("conversation_id") or state.conversation_id)
    if event.get("type") == "moderation":
        moderation = event.get("moderation_response")
        if isinstance(moderation, dict) and moderation.get("blocked") is True:
            state.blocked = True
    if event.get("type") == "server_ste_metadata":
        metadata = event.get("metadata")
        if isinstance(metadata, dict):
            if isinstance(metadata.get("tool_invoked"), bool):
                state.tool_invoked = metadata["tool_invoked"]
            state.turn_use_case = str(metadata.get("turn_use_case") or state.turn_use_case)


def conversation_base_event(event_type: str, state: ConversationState, **extra: Any) -> dict[str, Any]:
    return {
        "type": event_type,
        "text": state.text,
        "conversation_id": state.conversation_id,
        "parent_message_id": state.parent_message_id,
        "file_ids": list(state.file_ids),
        "sediment_ids": list(state.sediment_ids),
        "blocked": state.blocked,
        "tool_invoked": state.tool_invoked,
        "turn_use_case": state.turn_use_case,
        **extra,
    }


def iter_conversation_payloads(payloads: Iterator[str], history_text: str = "",
                               history_messages: list[str] | None = None) -> Iterator[dict[str, Any]]:
    state = ConversationState()
    history_messages = history_messages or []
    history_index = 0
    for payload in payloads:
        # print(f"[upstream_sse] {payload}", flush=True)
        if not payload:
            continue
        if payload == "[DONE]":
            yield conversation_base_event("conversation.done", state, done=True)
            break
        try:
            event = json.loads(payload)
        except json.JSONDecodeError:
            update_conversation_state(state, payload)
            yield conversation_base_event("conversation.raw", state, payload=payload)
            continue
        if not isinstance(event, dict):
            yield conversation_base_event("conversation.event", state, raw=event)
            continue
        update_conversation_state(state, payload, event)
        if history_index < len(history_messages) and event_assistant_text(event, history_text) == history_messages[history_index]:
            history_index += 1
            state.raw_text = ""
            state.text = ""
            continue
        next_raw_text = assistant_raw_text(event, state.raw_text, history_text)
        next_text = sanitize_output_text(next_raw_text)
        state.raw_text = next_raw_text
        if next_text != state.text:
            delta = next_text[len(state.text):] if next_text.startswith(state.text) else next_text
            state.text = next_text
            yield conversation_base_event("conversation.delta", state, raw=event, delta=delta)
            continue
        yield conversation_base_event("conversation.event", state, raw=event)


def conversation_events(
    backend: OpenAIBackendAPI,
    messages: list[dict[str, Any]] | None = None,
    model: str = "auto",
    prompt: str = "",
    images: list[str] | None = None,
    size: str | None = None,
    upstream_conversation_id: str = "",
    upstream_parent_message_id: str = "",
    deadline_at: float = 0.0,
    timeout_secs: float = 0.0,
    thinking_effort: str = "",
    image_mode: bool = False,
) -> Iterator[dict[str, Any]]:
    normalized = normalize_messages(messages or ([{"role": "user", "content": prompt}] if prompt else []))
    image_model = image_mode or str(model or "").strip() in IMAGE_MODELS
    history_text = "" if image_model else assistant_history_text(normalized)
    history_messages = [] if image_model else assistant_history_messages(normalized)
    final_prompt = prompt_with_global_system(build_image_prompt(prompt, size)) if image_model else prompt
    payloads = backend.stream_conversation(
        messages=normalized,
        model=model,
        prompt=final_prompt,
        images=images if image_model else None,
        system_hints=["picture_v2"] if image_model else None,
        upstream_conversation_id=upstream_conversation_id,
        upstream_parent_message_id=upstream_parent_message_id,
        deadline_at=deadline_at if image_model else 0.0,
        timeout_secs=timeout_secs if image_model else 0.0,
        thinking_effort=thinking_effort if not image_model else "",
    )
    for event in iter_conversation_payloads(payloads, history_text, history_messages):
        if image_model and deadline_at > 0 and time.time() >= deadline_at:
            raise ImagePollTimeoutError(image_poll_timeout_message(timeout_secs or config.image_poll_timeout_secs))
        yield event


def text_backend(model: str = "auto") -> OpenAIBackendAPI:
    return OpenAIBackendAPI(access_token=account_service.get_text_access_token(model=model))


def stream_text_deltas(backend: OpenAIBackendAPI, request: ConversationRequest) -> Iterator[str]:
    attempted_tokens: set[str] = set()
    renewal_attempted_tokens: set[str] = set()
    token = getattr(backend, "access_token", "")
    emitted = False
    while True:
        if token and token in attempted_tokens:
            raise RuntimeError("no available text account")
        active_backend: OpenAIBackendAPI | None = None
        try:
            active_backend = OpenAIBackendAPI(access_token=token)
            for event in conversation_events(
                    active_backend,
                    messages=request.messages,
                    model=request.model,
                    prompt=request.prompt,
                    thinking_effort=request.thinking_effort,
            ):
                if event.get("type") != "conversation.delta":
                    continue
                delta = str(event.get("delta") or "")
                if delta:
                    emitted = True
                    yield delta
            account_service.mark_text_used(token)
            return
        except Exception as exc:
            error_message = str(exc)
            if token and not emitted and is_retryable_multimodal_account_error(error_message):
                if is_token_invalid_error(error_message) and token not in renewal_attempted_tokens:
                    recovery = account_service.recover_invalid_access_token(token, "text_stream")
                    recovered_token = str(recovery.get("access_token") or token).strip() or token
                    renewal_attempted_tokens.update({token, recovered_token})
                    if recovery.get("recovered"):
                        token = recovered_token
                        continue
                attempted_tokens.add(token)
                token = account_service.get_text_access_token(
                    excluded_tokens=set(attempted_tokens),
                    model=request.model,
                )
                if token:
                    continue
            raise
        finally:
            if active_backend is not None:
                active_backend.close()


def collect_text(backend: OpenAIBackendAPI, request: ConversationRequest) -> str:
    return "".join(stream_text_deltas(backend, request))


def collect_text_result(backend: OpenAIBackendAPI, request: ConversationRequest) -> TextConversationResult:
    """Collect assistant text and promote interpreter outputs to local artifacts.

    A ``sandbox:/mnt/data/...`` URL is meaningful only inside ChatGPT.  Resolve
    it while the account session and conversation are still available, then
    expose the durable local copy through the existing ``/files`` route.
    """
    attempted_tokens: set[str] = set()
    renewal_attempted_tokens: set[str] = set()
    token = getattr(backend, "access_token", "")
    emitted = False
    while True:
        if token and token in attempted_tokens:
            raise RuntimeError("no available text account")
        active_backend: OpenAIBackendAPI | None = None
        try:
            active_backend = OpenAIBackendAPI(access_token=token)
            request_started_at = time.time()
            text_parts: list[str] = []
            last_event: dict[str, Any] = {}
            active_request = request
            for empty_attempt in range(3):
                text_parts = []
                last_event = {}
                for event in conversation_events(
                        active_backend,
                        messages=active_request.messages,
                        model=active_request.model,
                        prompt=active_request.prompt,
                        thinking_effort=active_request.thinking_effort,
                ):
                    last_event = event
                    if event.get("type") != "conversation.delta":
                        continue
                    delta = str(event.get("delta") or "")
                    if delta:
                        emitted = True
                        text_parts.append(delta)
                attempt_text = str(last_event.get("text") or "".join(text_parts))
                if attempt_text.strip():
                    break
                if empty_attempt < 2:
                    logger.warning({
                        "event": "empty_text_response_retry",
                        "retry_count": empty_attempt + 1,
                    })
                    retry_messages = list(request.messages or [])
                    retry_messages.append({
                        "role": "user",
                        "content": (
                            "The previous attempt returned no visible answer. Inspect every attachment "
                            "and answer the latest user request now in complete plain text. Do not end "
                            "the turn with an empty response."
                        ),
                    })
                    active_request = replace(request, messages=retry_messages)
            account_service.mark_text_used(token)
            text = str(last_event.get("text") or "".join(text_parts))
            conversation_id = str(last_event.get("conversation_id") or "").strip()
            message_id = str(last_event.get("parent_message_id") or "").strip()
            attachments: list[dict[str, Any]] = []
            if conversation_id and message_id:
                artifact_paths = extract_sandbox_artifact_paths(text)
                discovered_artifacts = []
                try:
                    discovered_artifacts = active_backend.list_conversation_artifacts(
                        conversation_id,
                        min_create_time=request_started_at - 2.0,
                    )
                except Exception as exc:
                    logger.warning({
                        "event": "chat_artifact_discovery_failed",
                        "conversation_id": conversation_id,
                        "error": sanitize_sensitive_text(str(exc)),
                    })
                if artifact_paths or discovered_artifacts:
                    output_dir = CHAT_ARTIFACT_ROOT / time.strftime("%Y/%m/%d") / uuid.uuid4().hex
                    ordered_artifacts = []
                    discovered_by_path = {
                        str(item.sandbox_path or ""): item
                        for item in discovered_artifacts
                        if str(item.sandbox_path or "")
                    }
                    for sandbox_path in artifact_paths:
                        ordered_artifacts.append((sandbox_path, discovered_by_path.get(sandbox_path)))
                    # Prefer the files explicitly linked by the final assistant
                    # response.  Upstream tool messages can also repeat the
                    # user's input attachment; promoting every discovered item
                    # would then show the original upload as if it were a new
                    # assistant output.  Discovery remains the fallback for
                    # tool-only responses whose final text contains no link.
                    if not artifact_paths:
                        for item in discovered_artifacts:
                            ordered_artifacts.append((str(item.sandbox_path or ""), item))
                    promoted_keys: set[str] = set()
                    for sandbox_path, artifact in ordered_artifacts:
                        artifact_key = str(
                            getattr(artifact, "attachment_id", "")
                            or getattr(artifact, "file_id", "")
                            or sandbox_path
                            or getattr(artifact, "name", "")
                        )
                        if not artifact_key or artifact_key in promoted_keys:
                            continue
                        promoted_keys.add(artifact_key)
                        try:
                            saved_path = (
                                active_backend.download_conversation_artifact_record(conversation_id, artifact, output_dir)
                                if artifact is not None
                                else active_backend.download_conversation_artifact(
                                    conversation_id,
                                    message_id,
                                    sandbox_path,
                                    output_dir,
                                )
                            )
                        except Exception as exc:
                            logger.warning({
                                "event": "chat_artifact_download_failed",
                                "conversation_id": conversation_id,
                                "message_id": message_id,
                                "sandbox_path": sandbox_path,
                                "error": sanitize_sensitive_text(str(exc)),
                            })
                            continue
                        relative_path = saved_path.resolve().relative_to((DATA_DIR / "files").resolve()).as_posix()
                        relative_url = f"/files/{quote(relative_path, safe='/')}"
                        public_url = f"{str(request.base_url or '').rstrip('/')}{relative_url}" if request.base_url else relative_url
                        mime_type = mimetypes.guess_type(saved_path.name)[0] or "application/octet-stream"
                        attachments.append({
                            "id": f"artifact-{uuid.uuid4().hex}",
                            "name": saved_path.name,
                            "type": mime_type,
                            "size": saved_path.stat().st_size,
                            "url": public_url,
                            "preview_url": public_url,
                            "download_url": public_url,
                        })
                        if sandbox_path:
                            text = text.replace(f"sandbox:{sandbox_path}", public_url)
            if attachments and not text.strip():
                text = "文件已生成，可在下方预览或下载。"
            elif not text.strip():
                raise RuntimeError("upstream model returned no visible content after 3 attempts")
            return TextConversationResult(text=text, attachments=attachments)
        except Exception as exc:
            error_message = str(exc)
            if token and not emitted and is_retryable_multimodal_account_error(error_message):
                if is_token_invalid_error(error_message) and token not in renewal_attempted_tokens:
                    recovery = account_service.recover_invalid_access_token(token, "text_artifact")
                    recovered_token = str(recovery.get("access_token") or token).strip() or token
                    renewal_attempted_tokens.update({token, recovered_token})
                    if recovery.get("recovered"):
                        token = recovered_token
                        continue
                attempted_tokens.add(token)
                token = account_service.get_text_access_token(
                    excluded_tokens=set(attempted_tokens),
                    model=request.model,
                )
                if token:
                    continue
            raise
        finally:
            if active_backend is not None:
                active_backend.close()


def stream_image_outputs(
        backend: OpenAIBackendAPI,
        request: ConversationRequest,
        index: int = 1,
        total: int = 1,
        upstream_conversation_id: str = "",
        upstream_parent_message_id: str = "",
) -> Iterator[ImageOutput]:
    last: dict[str, Any] = {}
    request_started_at = time.time()
    timeout_secs = float(request.image_timeout_secs or config.image_poll_timeout_secs)
    deadline_at = request_started_at + max(1.0, timeout_secs)
    for event in conversation_events(
            backend,
            prompt=request.prompt,
            model=request.model,
            images=request.images or [],
            size=request.size,
            upstream_conversation_id=upstream_conversation_id,
            upstream_parent_message_id=upstream_parent_message_id,
            deadline_at=deadline_at,
            timeout_secs=timeout_secs,
            image_mode=True,
    ):
        last = event
        if event.get("type") == "conversation.delta":
            yield ImageOutput(
                kind="progress",
                model=request.model,
                index=index,
                total=total,
                text=str(event.get("delta") or ""),
                upstream_event_type="conversation.delta",
                conversation_id=str(event.get("conversation_id") or ""),
            )
            continue
        if event.get("type") == "conversation.event":
            raw = event.get("raw")
            raw_type = str(raw.get("type") or "") if isinstance(raw, dict) else ""
            yield ImageOutput(
                kind="progress",
                model=request.model,
                index=index,
                total=total,
                upstream_event_type=raw_type,
                conversation_id=str(event.get("conversation_id") or ""),
            )

    conversation_id = str(last.get("conversation_id") or "")
    parent_message_id = str(last.get("parent_message_id") or "")
    file_ids = [str(item) for item in last.get("file_ids") or []]
    sediment_ids = [str(item) for item in last.get("sediment_ids") or []]
    message = str(last.get("text") or "").strip()
    logger.info({
        "event": "image_stream_resolve_start",
        "conversation_id": conversation_id,
        "file_ids": file_ids,
        "sediment_ids": sediment_ids,
        "tool_invoked": last.get("tool_invoked"),
        "turn_use_case": last.get("turn_use_case"),
    })
    if request.progress_callback:
        request.progress_callback("image_stream_resolve_start")
    if message and not file_ids and not sediment_ids and last.get("blocked"):
        yield ImageOutput(kind="message", model=request.model, index=index, total=total, text=message)
        return
    should_poll_for_image = bool(request.images) or last.get("turn_use_case") == "image gen"
    if message and not file_ids and not sediment_ids and not should_poll_for_image:
        yield ImageOutput(kind="message", model=request.model, index=index, total=total, text=message)
        return

    # 上游这一轮可能已经用「纯文字」收尾了（`turn_use_case=="image gen"` 只是**请求侧**
    # 的提示位，不能当「一定会有图」的证据）。真机实测：用户只发「把图片上的文字改成英文」
    # 没带图时，上游 8 秒就回了「请上传需要修改的图片…」并结束回合，而我们按 image gen
    # 轮询了 111 秒才报超时。轮询里一旦识别出「叶子是终局的纯文字」就会抛
    # `ImageTextOnlyAnswer`，这里直接把它当结果返回 —— 前台看到的就是上游原话。
    remaining_timeout = deadline_at - time.time()
    if remaining_timeout <= 0:
        # 预算在进入解析前就耗光了，同样要带上会话 id，前台才能给出「查看官网」入口。
        raise ImagePollTimeoutError(image_poll_timeout_message(timeout_secs), conversation_id or "")

    try:
        resolution = backend.resolve_conversation_image_result(
            conversation_id,
            file_ids,
            sediment_ids,
            after_message_id=upstream_parent_message_id if upstream_conversation_id else "",
            min_create_time=(request_started_at - 2.0) if (upstream_conversation_id or request.memory_enabled) else 0.0,
            timeout_secs=remaining_timeout,
        )
    except ImageTextOnlyAnswer as exc:
        terminal_text = str(getattr(exc, "text", "") or "") or message
        if not terminal_text:
            # 理论上不会发生（异常自带文字才抛），但真出现时也不能把结论说成「有文字」。
            raise
        logger.info({
            "event": "image_stream_terminal_text_answer",
            "conversation_id": conversation_id,
            "text_chars": len(terminal_text),
            "elapsed_secs": round(time.time() - request_started_at, 1),
        })
        yield ImageOutput(
            kind="message",
            model=request.model,
            index=index,
            total=total,
            text=terminal_text,
            conversation_id=conversation_id,
            parent_message_id=parent_message_id,
        )
        return
    image_urls = resolution.urls
    if resolution.message_id:
        parent_message_id = resolution.message_id
    if image_urls:
        if request.progress_callback:
            request.progress_callback("receiving_image")
        download_started_at = time.monotonic()
        downloaded_images = backend.download_image_bytes(image_urls)
        logger.info({
            "event": "image_result_download_complete",
            "conversation_id": conversation_id,
            "image_count": len(downloaded_images),
            "image_bytes": sum(len(image_data) for image_data in downloaded_images),
            "duration_ms": round((time.monotonic() - download_started_at) * 1000),
        })
        if request.response_format == "b64_json":
            image_items = [
                {"b64_json": base64.b64encode(image_data).decode("ascii")}
                for image_data in downloaded_images
            ]
        else:
            image_items = [
                {"url": save_image_bytes(image_data, request.base_url)}
                for image_data in downloaded_images
            ]
        data = format_image_result(
            image_items,
            request.prompt,
            request.response_format,
            request.base_url,
            int(time.time()),
        )["data"]
        logger.info({
            "event": "image_result_payload_ready",
            "conversation_id": conversation_id,
            "response_format": request.response_format,
            "image_count": len(data),
            "duration_ms": round((time.monotonic() - download_started_at) * 1000),
        })
        if data:
            yield ImageOutput(
                kind="result",
                model=request.model,
                index=index,
                total=total,
                data=data,
                conversation_id=conversation_id,
                parent_message_id=parent_message_id,
            )
        return

    if message:
        yield ImageOutput(kind="message", model=request.model, index=index, total=total, text=message)


def stream_image_outputs_with_pool(request: ConversationRequest) -> Iterator[ImageOutput]:
    if request.memory_enabled and request.memory_owner_id:
        memory_key = str(request.memory_conversation_id or "default").strip() or "default"
        lock = image_conversation_state_service.lock_for(request.memory_owner_id, memory_key)
        if lock is not None:
            with lock:
                yield from _stream_image_outputs_with_pool(request, memory_enabled=True, memory_key=memory_key)
                return

    # Image memory relies on the previous upstream message as the parent of the
    # next one, so those requests must remain ordered. Regular multi-image
    # requests are independent and can safely reserve different pool slots.
    if (
        int(request.n or 1) > 1
        and bool(getattr(config, "image_parallel_generation", True))
        and not request.memory_enabled
    ):
        yield from _stream_image_outputs_in_parallel(request)
        return
    yield from _stream_image_outputs_with_pool(request, memory_enabled=False)


def _stream_image_outputs_in_parallel(request: ConversationRequest) -> Iterator[ImageOutput]:
    """Run each image in a multi-image request through an independent pool job.

    AccountService owns slot reservation, therefore workers automatically spread
    across ready accounts and still honour the configured per-account limit.
    Results are yielded as soon as their own upstream task finishes.
    """
    total = max(1, int(request.n or 1))
    events: Queue[tuple[str, int, ImageOutput | Exception | None]] = Queue()

    def run_image(index: int) -> None:
        image_request = replace(request, n=1)
        try:
            for output in _stream_image_outputs_with_pool(
                image_request,
                memory_enabled=False,
                index_offset=index - 1,
                total_count=total,
            ):
                events.put(("output", index, output))
        except Exception as exc:
            events.put(("error", index, exc))
        finally:
            events.put(("done", index, None))

    completed = 0
    emitted_result = False
    failures: list[Exception] = []
    with ThreadPoolExecutor(max_workers=total, thread_name_prefix="image-pool") as executor:
        for index in range(1, total + 1):
            executor.submit(run_image, index)

        while completed < total:
            event, _index, payload = events.get()
            if event == "output":
                output = payload
                if isinstance(output, ImageOutput):
                    emitted_result = emitted_result or output.kind in {"result", "message"}
                    yield output
            elif event == "error" and isinstance(payload, Exception):
                failures.append(payload)
            elif event == "done":
                completed += 1

    # Preserve the useful portion of a batch when the pool runs out of slots
    # partway through it. If none of the jobs produced a response, expose the
    # first upstream failure exactly as a normal single-image request would.
    if failures and not emitted_result:
        raise failures[0]

    if failures:
        logger.warning({
            "event": "image_parallel_partial_failure",
            "requested": total,
            "failed": len(failures),
            "errors": [str(error) for error in failures],
        })

    if not emitted_result:
        raise ImageGenerationError("no account in the pool could generate images")


def record_image_failure(
    access_token: str,
    reason: str,
    event: str,
    *,
    verify_free_account: bool = True,
    force_free_cleanup: bool = False,
) -> None:
    account_service.mark_image_result(access_token, False)
    if not verify_free_account:
        account_service.update_account(access_token, {"consecutive_image_failures": 0})
        return
    try:
        account_service.verify_free_account_after_image_failure(
            access_token,
            reason,
            event,
            force=force_free_cleanup,
        )
    except Exception as exc:
        logger.warning({
            "event": "free_account_cleanup_check_failed",
            "request_token": access_token,
            "error": str(exc),
        })


def _remove_image_conversation_later(
        backend: OpenAIBackendAPI,
        conversation_id: str,
        *,
        success: bool,
) -> None:
    if not conversation_id:
        return
    if not (config.image_remove_conversation_always or (success and config.image_remove_conversation_after_result)):
        return

    def _run() -> None:
        try:
            backend.delete_conversation(conversation_id)
            logger.info({"event": "image_conversation_removed", "conversation_id": conversation_id})
        except Exception as exc:
            logger.warning({
                "event": "image_conversation_remove_failed",
                "conversation_id": conversation_id,
                "error": str(exc),
            })

    threading.Thread(target=_run, name=f"remove-image-conversation-{conversation_id}", daemon=True).start()


def _stream_image_outputs_with_pool(
        request: ConversationRequest,
        *,
        memory_enabled: bool = False,
        memory_key: str = "default",
        index_offset: int = 0,
        total_count: int | None = None,
) -> Iterator[ImageOutput]:
    if not str(request.model or "").strip():
        raise ImageGenerationError("image model is required")

    emitted = False
    last_error = ""
    failed_tokens: set[str] = set()
    # 因上传限流被换掉的账号，用于在号池耗尽时给出准确原因。
    throttled_tokens: set[str] = set()
    renewal_attempted_tokens: set[str] = set()
    memory_key = str(memory_key or "default").strip() or "default"
    total = max(1, int(total_count or request.n or 1))
    for local_index in range(1, request.n + 1):
        index = int(index_offset) + local_index
        poll_timeout_attempts = 0
        connection_retry_attempts = 0
        # 因「本账号生图配额用完」换掉的次数（与轮询超时换号共用同一预算，见下方 except）。
        quota_switch_attempts = 0
        while True:
            memory_state = image_conversation_state_service.get(request.memory_owner_id, memory_key) if memory_enabled else {}
            preferred_token = str(memory_state.get("access_token") or "").strip()
            account_owner_id = str(request.account_owner_id or request.memory_owner_id or "").strip()
            allow_shared = bool(request.account_allow_shared) and not memory_enabled
            try:
                if request.progress_callback:
                    request.progress_callback("getting_account")
                token = account_service.get_available_access_token(
                    preferred_token if memory_enabled else "",
                    exclusive=memory_enabled,
                    owner_id=account_owner_id,
                    strategy=request.account_pool_strategy,
                    allow_shared=allow_shared,
                    excluded_tokens=failed_tokens,
                    wait_for_slot=bool(request.account_wait_for_slot),
                )
            except RuntimeError as exc:
                if emitted:
                    return
                # 如果本轮是因为上传限流把账号逐个排除到池空，直说原因，
                # 否则用户只会看到和「号池没号」一样的提示。
                if throttled_tokens:
                    raise ImageGenerationError(
                        "号池中所有账号的上传额度均已达到上限，请稍后重试或补充账号",
                        account_email="",
                    ) from exc
                # ⚠️ 号池被掏空时保留内部英文 "no available image quota" 原样抛出：
                # 站长配的「报错文案映射」就是按这串英文配的（→ 大白话），
                # 换成别的说法会让规则失配。见 services/error_message_service.py。
                raise ImageGenerationError(str(exc) or "image generation failed") from exc

            get_account = getattr(account_service, "get_account", None)
            account = get_account(token) if callable(get_account) else {}
            account = account or {}
            account_email = str(account.get("email") or "").strip()
            upstream_conversation_id = ""
            upstream_parent_message_id = ""
            if memory_enabled and not request.memory_reset and preferred_token and token == preferred_token:
                upstream_conversation_id = str(memory_state.get("upstream_conversation_id") or "").strip()
                upstream_parent_message_id = str(memory_state.get("upstream_parent_message_id") or "").strip()

            emitted_for_token = False
            returned_message = False
            returned_result = False
            next_conversation_id = ""
            next_parent_message_id = ""
            backend: OpenAIBackendAPI | None = None
            try:
                backend = OpenAIBackendAPI(access_token=token)
                if request.progress_callback:
                    backend.progress_callback = request.progress_callback
                for output in stream_image_outputs(
                    backend,
                    request,
                    index,
                    total,
                    upstream_conversation_id=upstream_conversation_id,
                    upstream_parent_message_id=upstream_parent_message_id,
                ):
                    if account_email and not output.account_email:
                        output.account_email = account_email
                    if output.kind == "message" and request.message_as_error:
                        # 上游只回文字 = 终局结论，不再有图。
                        # 但结论的**性质**要分开：账号生图配额用完（「你已达到 Free 套餐的
                        # 图像生成请求上限…」）是账号级问题，官网那边这一轮什么都没生成，
                        # 留在卡片上只会让人以为「官网有图没拿到」→ 不给「查看官网对话」入口，
                        # 也不谎称是内容违规（那会让这个账号逃过配额校验）。
                        terminal_text = output.text or "Image generation was rejected by upstream policy."
                        quota_exhausted = is_image_quota_exhausted_message(terminal_text)
                        raise ImageGenerationError(
                            terminal_text,
                            status_code=429 if quota_exhausted else 400,
                            error_type="rate_limit_error" if quota_exhausted else "invalid_request_error",
                            code="image_quota_exhausted" if quota_exhausted else "content_policy_violation",
                            account_email=account_email,
                            conversation_id=output.conversation_id,
                            terminal_no_image=True,
                            can_view_official=not quota_exhausted,
                        )
                    emitted = True
                    emitted_for_token = True
                    returned_message = output.kind == "message"
                    returned_result = returned_result or output.kind == "result"
                    if output.conversation_id:
                        next_conversation_id = output.conversation_id
                    if output.parent_message_id:
                        next_parent_message_id = output.parent_message_id
                    yield output
                if returned_message or not returned_result:
                    record_image_failure(token, "no image result returned", "image_no_result")
                    return
                account_service.mark_image_result(token, True)
                is_shared_account = account_service.is_shared_account_use(token, account_owner_id)
                if next_conversation_id and config.cleanup_shared_conversations:
                    # ⚠️ 「生成后清理官网对话」与「官网来源取证」天然冲突：隐藏会话 == 官方
                    # 删除，会让分享链接立刻 404。取证开启时改为登记到保留台账，隐藏推迟到
                    # 保留期结束由 _release 统一做（连隐藏带删链接），清理意图只是延后，不丢。
                    # 缺了这条分支，开着「生成后清理官网对话」的站点取证会**完全静默失效**：
                    # 出图时没登记，用户点「查看官网对话」只能走兜底登记（无账号信息）→ 报错。
                    evidence_deferred = False
                    if bool(getattr(config, "image_share_evidence_enabled", False)):
                        try:
                            from services.share_evidence_service import share_evidence_service

                            share_evidence_service.register_conversation(
                                conversation_id=next_conversation_id,
                                account_email=account_email,
                                owner_id=account_owner_id,
                                access_token=token,
                            )
                            evidence_deferred = True
                            logger.info({
                                "event": "image_conversation_deferred",
                                "conversation_id": next_conversation_id,
                                "reason": "share_evidence_cleanup",
                            })
                        except Exception as exc:
                            logger.warning({
                                "event": "image_conversation_defer_failed",
                                "conversation_id": next_conversation_id,
                                "error": str(exc),
                            })
                    if not evidence_deferred:
                        try:
                            backend.hide_conversation(next_conversation_id)
                            if is_shared_account:
                                account_service.mark_shared_cleanup_result(token, True)
                        except Exception as exc:
                            if is_shared_account:
                                account_service.mark_shared_cleanup_result(token, False)
                            logger.warning({
                                "event": "conversation_cleanup_fail",
                                "shared": is_shared_account,
                                "conversation_id": next_conversation_id,
                                "error": str(exc),
                            })
                if memory_enabled and next_conversation_id:
                    image_conversation_state_service.save(
                        request.memory_owner_id,
                        memory_key,
                        access_token=token,
                        upstream_conversation_id=next_conversation_id,
                        upstream_parent_message_id=next_parent_message_id,
                        model=request.model,
                        mode=request.memory_mode,
                    )
                break
            except ImagePollTimeoutError as exc:
                next_conversation_id = next_conversation_id or str(getattr(exc, "conversation_id", "") or "")
                record_image_failure(token, str(exc), "image_poll_timeout", verify_free_account=False)
                if account_email:
                    setattr(exc, "account_email", account_email)
                poll_timeout_attempts += 1
                max_poll_attempts = config.image_poll_max_attempts
                if not memory_enabled and poll_timeout_attempts < max_poll_attempts:
                    failed_tokens.add(token)
                    logger.warning({
                        "event": "image_poll_timeout_switch_account",
                        "attempt": poll_timeout_attempts,
                        "max_attempts": max_poll_attempts,
                        "account_email": account_email,
                        "conversation_id": next_conversation_id,
                    })
                    continue
                raise
            except ImageGenerationError as exc:
                next_conversation_id = next_conversation_id or str(getattr(exc, "conversation_id", "") or "")
                error_code = str(getattr(exc, "code", "") or "")
                record_image_failure(
                    token,
                    str(exc),
                    "image_generation_error",
                    verify_free_account=error_code != "content_policy_violation",
                )
                if account_email and not getattr(exc, "account_email", ""):
                    exc.account_email = account_email
                if error_code == "image_quota_exhausted" and not emitted_for_token:
                    # 上游明说「本账号生图配额用完了」→ 这是**账号级**问题，换个账号就能出图，
                    # 不该变成用户的失败：
                    # ① 立刻把该账号移出可用池 —— 只记账（mark_image_result 累加失败数）不够，
                    #    免费号 quota 未知时本地永远算不到 0，这个账号会被后续每个请求反复挑中；
                    # ② 本轮换号重试。预算与「轮询超时换号」同一配置（image_poll_max_attempts，
                    #    默认 2 → 允许换 1 次），避免整池都没额度时把用户拖在多次 20s 上游往返里。
                    account_service.mark_image_quota_exhausted(token, reason=str(exc))
                    quota_switch_attempts += 1
                    # 下限 2（至少换 1 次号）：轮询超时换号可能是白等（图也许马上就出来），
                    # 而配额耗尽时这个账号这一轮**确定**出不了图、原地重试 0 价值 —— 所以
                    # 即便管理员把「最多尝试账号数」设成 1（不因超时换号），这里也换 1 次。
                    max_quota_switches = max(2, min(10, int(config.image_poll_max_attempts or 2)))
                    if quota_switch_attempts < max_quota_switches:
                        failed_tokens.add(token)
                        logger.warning({
                            "event": "image_quota_exhausted_switch_account",
                            "attempt": quota_switch_attempts,
                            "max_attempts": max_quota_switches,
                            "account_email": account_email,
                            "error": str(exc),
                        })
                        continue
                raise
            except Exception as exc:
                next_conversation_id = next_conversation_id or str(getattr(exc, "conversation_id", "") or "")
                last_error = str(exc)
                logger.warning({
                    "event": "image_stream_fail",
                    "request_token": token,
                    "account_email": account_email,
                    "error": last_error,
                })
                if memory_enabled and upstream_conversation_id and not emitted_for_token:
                    image_conversation_state_service.clear(request.memory_owner_id, memory_key)
                    account_service.release_image_slot(token)
                    continue
                if not emitted_for_token and is_token_invalid_error(last_error):
                    if token not in renewal_attempted_tokens:
                        recovery = account_service.recover_invalid_access_token(
                            token,
                            "image_stream_invalid_token",
                        )
                        recovered_token = str(recovery.get("access_token") or token).strip() or token
                        renewal_attempted_tokens.update({token, recovered_token})
                        if recovery.get("recovered"):
                            account_service.release_image_slot(recovered_token)
                            continue
                    failed_tokens.add(token)
                    account_service.mark_image_result(token, False)
                    continue
                if not emitted_for_token and is_account_scoped_upload_error(last_error):
                    # 上传限流是单账号配额，换号即可，不应把账号判为不可用。
                    # 只把它加进本轮排除集合并归还并发槽，避免触发
                    # verify_free_account_after_image_failure 的账号清理。
                    failed_tokens.add(token)
                    throttled_tokens.add(token)
                    account_service.release_image_slot(token)
                    logger.warning({
                        "event": "image_upload_throttled_switch_account",
                        "account_email": account_email,
                        "excluded_accounts": len(failed_tokens),
                        "error": last_error,
                    })
                    continue
                if not emitted_for_token and is_transient_image_connection_error(last_error):
                    connection_retry_attempts += 1
                    if connection_retry_attempts <= 2:
                        wait_seconds = min(2.0 * connection_retry_attempts, 6.0)
                        logger.warning({
                            "event": "image_stream_connection_retry",
                            "account_email": account_email,
                            "retry_count": connection_retry_attempts,
                            "wait_seconds": wait_seconds,
                            "error": last_error,
                        })
                        # 只释放并发槽，不增加账号失败次数；这是网络抖动，
                        # 下一轮仍优先复用同一账号重试。
                        account_service.release_image_slot(token)
                        time.sleep(wait_seconds)
                        continue
                    # 同一账号多次传输失败后换下一个账号，避免单个上游
                    # 连接把整个号池请求拖住。
                    failed_tokens.add(token)
                    connection_retry_attempts = 0
                    account_service.mark_image_result(token, False)
                    continue
                record_image_failure(token, last_error, "image_stream_error")
                raise ImageGenerationError(
                    image_stream_error_message(last_error),
                    account_email=account_email,
                    conversation_id=next_conversation_id,
                ) from exc
            finally:
                if backend is not None:
                    should_hide = bool(
                        next_conversation_id
                        and not memory_enabled
                        and (
                            bool(getattr(config, "image_remove_conversation_always", False))
                            or (returned_result and bool(getattr(config, "image_remove_conversation_after_result", False)))
                        )
                        and not config.cleanup_shared_conversations
                    )
                    if should_hide:
                        if bool(getattr(config, "image_share_evidence_enabled", False)):
                            # 证据模式：隐藏会话 == 官方删除，会让分享链接立刻 404，
                            # 所以这里**不能**立刻隐藏。改为登记到保留台账，到期后
                            # 「隐藏会话 + 删分享」一起做。见 services/share_evidence_service.py
                            try:
                                from services.share_evidence_service import share_evidence_service

                                share_evidence_service.register_conversation(
                                    conversation_id=next_conversation_id,
                                    account_email=account_email,
                                    owner_id=account_owner_id,
                                    # 出图那一刻的热 token：账号之后被删出号池/轮换 token
                                    # 时，取证与回收仍可用它（见 share_evidence_service）。
                                    access_token=token,
                                )
                                logger.info({
                                    "event": "image_conversation_deferred",
                                    "conversation_id": next_conversation_id,
                                })
                            except Exception as exc:
                                logger.warning({
                                    "event": "image_conversation_defer_failed",
                                    "conversation_id": next_conversation_id,
                                    "error": str(exc),
                                })
                            should_hide = False

                    if should_hide:
                        conversation_id_to_hide = next_conversation_id

                        def _hide_conversation_later(backend_to_close: OpenAIBackendAPI = backend) -> None:
                            try:
                                backend_to_close.hide_conversation(conversation_id_to_hide)
                            except Exception as exc:
                                logger.warning({
                                    "event": "image_conversation_remove_failed",
                                    "conversation_id": conversation_id_to_hide,
                                    "error": str(exc),
                                })
                            finally:
                                backend_to_close.close()

                        threading.Thread(
                            target=_hide_conversation_later,
                            name=f"remove-image-conversation-{conversation_id_to_hide}",
                            daemon=True,
                        ).start()
                        backend = None
                    if backend is not None:
                        backend.close()

    if not emitted:
        if not last_error:
            last_error = "no account in the pool could generate images — check account quota and rate-limit status"
        raise ImageGenerationError(image_stream_error_message(last_error))


def stream_image_chunks(outputs: Iterable[ImageOutput]) -> Iterator[dict[str, Any]]:
    for output in outputs:
        yield output.to_chunk()


def collect_image_outputs(outputs: Iterable[ImageOutput]) -> dict[str, Any]:
    created = None
    data: list[dict[str, Any]] = []
    message = ""
    progress_parts: list[str] = []
    account_email = ""
    conversation_id = ""
    for output in outputs:
        created = created or output.created
        if output.account_email and not account_email:
            account_email = output.account_email
        if output.conversation_id and not conversation_id:
            conversation_id = output.conversation_id
        if output.kind == "progress" and output.text:
            progress_parts.append(output.text)
        elif output.kind == "message":
            message = output.text
        elif output.kind == "result":
            data.extend(output.data)

    result: dict[str, Any] = {"created": created or int(time.time()), "data": data}
    if not data:
        text = message or "".join(progress_parts).strip()
        if text:
            result["message"] = text
    if account_email:
        result["_account_email"] = account_email
    if conversation_id:
        result["_conversation_id"] = conversation_id
    return result