app/upstream/client.py

"""anuma.ai UpstreamClient：把拍平 prompt 组装成 Responses 请求，SSE 解析产 IREvent。

上游：``POST https://portal.anuma.ai/api/v1/responses``
- 认证头由 AuthProvider 注入（authorization: Bearer <privy JWT> + x-anuma-* 头）。
- 请求体为标准 OpenAI Responses 格式（input/model/stream）。
- **原生 function calling**：客户端 tools 直接透传上游 body（上游原生执行并返回
  function_call 事件）；多轮 tool 历史（[tools] 块）解析回原生的
  function_call / function_call_output input 项。
- 响应为标准 OpenAI Responses SSE（response.created / output_text.delta / completed）。
"""
from __future__ import annotations

import json
import re
import time
import uuid
from collections.abc import AsyncIterator
from pathlib import Path
from typing import Any

import httpx

from app.events import IREvent
from app.tools import ToolDef
from app.upstream.base import UpstreamClient

RESPONSES_URL = "https://portal.anuma.ai/api/v1/responses"

# 2026-08-07 实测（probe 脚本 + HAR 对照）：portal /api/v1/responses 对 claude/kimi 等
# 模型要求 input 里带 **genuine ZetaChain system 文本** 才放行（否则 403）。
# 2026-08-08 决定性实验（verify_s500 / verify_sys_full_gateway_tool.py）：
# - 真实 500 字的精简 system → 生图请求返回 **fence**（tool_call 空壳+全参同 id，
#   服务端不执行生图、无图片、不扣积分）。
# - 完整 25896 字 ZetaChain 原版 system → **100% 出图**。
# 生图链路对 system 要求比普通对话更严；必须全程注入完整原版 system。
# 文件 scripts/har_system_full.txt 是 HAR entry 142 的完整 system（26118 字节）。
_SYSTEM_PROMPT = (
    Path(__file__).resolve().parent.parent.parent
    / "scripts" / "har_system_full.txt"
).read_text(encoding="utf-8")
# 2026-08-08 生图链路：system 长度是 fence/出图的分水岭（<500 字稳定 fence）。
# 启动时打一行确认加载的是完整版（25896 字），防止 cwd/pycache 加载错文件。
print(f"[client] _SYSTEM_PROMPT loaded {len(_SYSTEM_PROMPT)} chars "
      f"from {Path(__file__).resolve().parent.parent.parent / 'scripts' / 'har_system_full.txt'}", flush=True)

# 生图工具 schema（2026-08-08 从 chat.anuma.ai HAR 抓取）。上游没有独立图片 API：
# 在 /api/v1/responses body 带 image_model + tools 里含 AnumaMediaMCP-anuma_create_image，
# 模型输出 function_call 后由 **上游服务端自己执行生图**（无需客户端回传 output），
# 图片以签名 media URL 嵌入最终文本消息（[ ](https://portal.anuma.ai/api/v1/media/...)）。
_IMAGE_TOOL = {
    "type": "function",
    # 2026-08-08 实测对比实验（verify_img_tool.py）：名字必须是 **Anuma**MediaMCP-
    # anuma_create_image（Anumu 而非 Anima）——服务端按工具名匹配生图执行器，
    # 名字拼错时照常返回 tool_call 但**不执行**（无 media、不扣积分），
    # 三组对比只有 HAR 原版名出图。schema body 逐字一致也不管用，名字是唯一开关。
    "name": "AnumaMediaMCP-anuma_create_image",
    "description": "Turn a prompt into one or more images. Optionally provide input images to edit or combine them. The optional configuration parameters are best-effort and depend on the internal provider details of each model. Returns signed URLs pointing to the generated images.",
    "parameters": {
        "type": "object",
        "properties": {
            "aspect_ratio": {"type": "string", "enum": ["1:1", "3:4", "4:3", "16:9", "9:16"],
                             "description": "The aspect ratio of the generated image."},
            # 2026-08-08 实测：HAR 原版 schema 的 input_images.items 带 description
            # （"An http(s) URL or a base64-encoded data URI of the input image."），
            # 缺失时上游虽返回 tool_call 但不执行生图（不扣积分、无图片）——
            # 服务端按 schema 校验工具合法性。必须与 HAR 逐字一致。
            "input_images": {
                "type": "array",
                "items": {"type": "string", "format": "uri", "maxLength": 28000000,
                          "description": "An http(s) URL or a base64-encoded data URI of the input image."},
                "minItems": 1, "maxItems": 4,
                "description": "Input images to edit or combine. Provide between 1 and 4 images; omit to generate from the prompt alone.",
            },
            "model": {"type": "string",
                      "enum": ["auto", "anuma-flash", "anuma-pro", "anuma-flash-private",
                               "anuma-pro-private", "flux-2-klein-4b", "flux-2-pro",
                               "gpt-image-2", "grok-imagine", "nano-banana",
                               "nano-banana-pro", "nano-banana-2"],
                      "description": "The model used to generate the image. Use a tier alias (\"auto\", \"anuma-flash\", \"anuma-pro\", \"anuma-flash-private\", \"anuma-pro-private\") or a concrete model name."},
            "num_images": {"type": "integer", "minimum": 1, "maximum": 4,
                           "description": "The number of images to generate. Defaults to 1. The flux-2-pro model always generates a single image and ignores this value."},
            "output_format": {"type": "string", "enum": ["jpeg", "png"],
                              "description": "The format of the generated image."},
            "prompt": {"type": "string", "minLength": 1, "maxLength": 2000,
                       "description": "The text prompt describing the image to generate, or the edits to make when input images are provided. Resolution keywords such as 2k, 4k, 8k, or uhd select a higher output resolution for the nano-banana-pro and nano-banana-2 models and increase the cost accordingly."},
        },
        "required": ["prompt", "model"],
    },
}

# [system] / [user] / [assistant] / [tools] 角色标记（见 app/adapters/__init__.py）
_ROLE_RE = re.compile(r"^\[(system|user|assistant|tools)\]\n", re.MULTILINE)
# [tools] 块：\n[id]\nname: ...\narguments: {...}\n---\nresult:\n<正文>
_TOOLS_ENTRY_RE = re.compile(
    r"\[([^\]]+)\]\s*"
    r"(?:(?:^|\n)name:\s*([^\n]*))?"
    r"(?:(?:^|\n)arguments:\s*(\{.*?\}))?"
    r"(?:\n---\s*\n(?:result(?:\s*\(error\))?:\s*\n?)?(.*?))?(?=\n\[|\Z)",
    re.DOTALL,
)


def _split_role_blocks(prompt: str) -> list[dict[str, Any]]:
    """把拍平 prompt（带 [system]/[user]/[assistant]/[tools] 标记）切回 Responses input 数组。

    分段规则（文本区间不重叠、无遗漏）：
    - marker 之前的裸文本 → user 消息。
    - ``[system]`` / ``[user]`` / ``[assistant]`` → 对应 role 消息。
    - ``[tools]`` 块 → 解析为原生 function_call + function_call_output 项
      （多轮 tool 历史，供上游原生理解）。
    - marker 之后的剩余裸文本 → user 消息。
    """
    markers = [m for m in _ROLE_RE.finditer(prompt)]
    items: list[dict[str, Any]] = []

    def add_text(text: str) -> None:
        text = text.strip()
        if text:
            items.append({"role": "user", "content": [{"type": "text", "text": text}]})

    if not markers:
        add_text(prompt)
        return items
    add_text(prompt[:markers[0].start()])  # 开头裸文本
    for i, m in enumerate(markers):
        end = markers[i + 1].start() if i + 1 < len(markers) else len(prompt)
        body = prompt[m.end():end].strip()
        if m.group(1) == "tools":
            items.extend(_parse_tools_block(body))
        elif body:
            items.append({"role": m.group(1), "content": [{"type": "text", "text": body}]})
    return items


def _inject_time_before_user(input_items: list[dict[str, Any]], ts: str) -> None:
    """在第一条 user 消息前插入 `Current time (precise): <ts>` 文本块（对齐 HAR 生图请求）。

    HAR（连续同时生图）里 user 消息是两条独立 text：
      [{"type":"text","text":"Current time (precise): 2026-08-08T08:43:50Z"},
       {"type":"text","text":"一只小猫"}]
    无 user（纯 system+tool 场景）则追加在末尾。
    """
    time_item = {"role": "user", "content": [{"type": "text", "text": f"Current time (precise): {ts}"}]}
    for i, it in enumerate(input_items):
        if it.get("role") == "user":
            input_items.insert(i, time_item)
            return
    input_items.append(time_item)


def _flush_text(pending: list[str]) -> list[dict[str, Any]]:
    if not pending:
        return []
    return [{"role": "user", "content": [{"type": "text", "text": "\n\n".join(pending)}]}]


def _parse_tools_block(body: str) -> list[dict[str, Any]]:
    """[tools] 历史块 → Responses 原生 input 项（function_call + function_call_output 配对）。"""
    items: list[dict[str, Any]] = []
    for m in _TOOLS_ENTRY_RE.finditer(body):
        cid = (m.group(1) or "").strip()
        name = (m.group(2) or "").strip()
        args_str = (m.group(3) or "{}").strip()
        result = (m.group(4) or "").strip()
        if not cid:
            continue
        if name:
            items.append({
                "type": "function_call",
                "call_id": cid,
                "name": name,
                "arguments": args_str,
            })
        if result or True:
            items.append({"type": "function_call_output", "call_id": cid, "output": result})
    return items


class DefaultUpstreamClient(UpstreamClient):
    def __init__(self, account, settings, http_client, auth, parser,
                 account_file=None) -> None:
        self._account = account
        self._settings = settings
        self._http: httpx.AsyncClient = http_client
        self._auth = auth
        self._parser = parser
        self._account_file = account_file

    async def stream(
        self,
        prompt: str,
        model_id: str | None = None,
        tools: list[ToolDef] | None = None,
        **kw: Any,
    ) -> AsyncIterator[IREvent]:
        """发送 prompt 到上游。``tools`` 为客户端工具定义（native 模式透传上游）。

        ``kw`` 支持 ``image_model``（生图：body 加 image_model，tools 里自动注入
        AnumaMediaMCP-anuma_create_image，由**上游服务端**执行生图——客户端无需
        回传 function_call_output，图片以签名 URL 嵌入最终文本）。"""
        upstream_model = self._upstream_model(model_id)
        input_items = _split_role_blocks(prompt)
        # 2026-08-07 实测：portal 要求 input 首条带 **ZetaChain 真实 system 指令文本**
        # 才放行 claude/kimi（校验 system 内容特征，非仅角色存在；opencode 自带的 system
        # 不匹配 → 403）。故**无条件前置注入 _SYSTEM_PROMPT**，客户端自带 system 保留在后。
        input_items.insert(0, {"role": "system", "content": [{"type": "text", "text": _SYSTEM_PROMPT}]})
        # 2026-08-08 实测（HAR「连续同时生图」）：生图请求的 user 消息前置
        # `Current time (precise): <UTC>` 文本块（前端每次生图都带，独立 text 项）。
        # 上游可能用时间上下文做生图会话校验——缺它模型不触发工具、只回围栏。
        # 对齐 HAR：生图时在第一条 user 消息前插入时间文本块（不动用户原提示词）。
        if kw.get("image_model"):
            t = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
            _inject_time_before_user(input_items, t)
        body: dict[str, Any] = {
            "input": input_items,
            "model": upstream_model,
            "stream": True,
        }
        # 生图：image_model（如 'gpt-image-2'）进 body，上游识别后原生生图。
        image_model = kw.get("image_model")
        # 改图：http(s) URL 或 base64 data URI 都进 input_images。Cherry 传的本地
        # 二进制由 adapter 转 data URI（openai_images.image_edits），URL 由客户端
        # image_inputs 直传（网页 HAR 行为）。工具 schema 明说支持两者。
        image_inputs: list[str] = list(kw.get("image_inputs") or [])
        if image_model:
            body["image_model"] = image_model
            # 2026-08-08 实测（HAR 对照）：生图请求必须带 tool_choice:"required"，
            # 强制模型先调用生图工具（服务端才执行）；缺失时模型可自由不调工具，
            # 只输出空壳调用/围栏，生图不触发。
            body["tool_choice"] = "required"
            # 生图回复（图片 URL + 描述）很长，需较大输出预算（HAR 实测 max_output_tokens=32000）
            body.setdefault("max_output_tokens", 32000)
            # 2026-08-09 改图修复（HAR「图生图」铁证）：网页改图是**把上一张图的
            # 签名 URL 作为 assistant 历史消息喂给模型**，模型在工具调用里自动带
            # input_images 引用它去改图——不是我们注入 data URI/tool default
            # （2026-08-09 实测模型无视 default、data URI 不填 input_images）。
            # 改图时注入一条 assistant 历史消息：描述 + [ ](签名URL)，模型即可
            # 像网页一样自动填 input_images。仅 URL 注入（data URI 无效）。
            if image_inputs and all(s.startswith("http") for s in image_inputs):
                hist_text = ("[Previous generated image to edit] Edit the image(s) "
                             "referenced in this assistant message. Use the image URL "
                             "as an input image to the media tool, do not generate a "
                             "brand-new image from scratch.\n\n"
                             + "\n\n".join(f"[ ]({u})" for u in image_inputs[:4]))
                input_items.append({
                    "role": "assistant",
                    "content": [{"type": "text", "text": hist_text}],
                })
        if tools:
            body["tools"] = [
                {
                    "type": "function",
                    "name": t.name,
                    "description": t.description,
                    "parameters": t.parameters or {"type": "object", "properties": {}},
                    "strict": False,
                }
                for t in tools
            ]
        # 生图：上游靠 tools 里的 AnumaMediaMCP-anuma_create_image 触发服务端执行，
        # 客户端没传这个工具时自动注入（image_model 或 image_prompt 任一存在即注入）。
        # 已注入过则跳过（客户端显式传了就用它的）。
        if (image_model or kw.get("image_prompt")) and not any(
            t.name == _IMAGE_TOOL["name"] for t in (tools or [])
        ):
            tool_def = _IMAGE_TOOL
            body.setdefault("tools", []).append(tool_def)
        # 2026-08-08 实测（HAR「聊天」）：网页**普通聊天也带** conversation_id
        # （body + x-conversation-id 头同值）——上游可能按会话管理上下文/限额。
        # 生图已持久化复用；聊天复用同一会话（不新开，避免垃圾会话）。
        if image_model:
            conv_id = str(getattr(self._account, "conversation_id", "") or "")
            if not conv_id:
                conv_id = str(uuid.uuid4())
                self._account.conversation_id = conv_id
                self._persist_account()
            body["conversation_id"] = conv_id
        else:
            # 聊天：网页带会话 ID。复用账号里已有的（生图生成的）或新生成一个，
            # 但不落盘（聊天会话太多会开垃圾；每次请求用同一内存值即可）。
            conv_id = str(getattr(self._account, "conversation_id", "") or "")
            if not conv_id:
                conv_id = str(uuid.uuid4())
                self._account.conversation_id = conv_id
            body["conversation_id"] = conv_id
        # 2026-08-08 实测（HAR「聊天」kimi/kimi-k3 请求）：网页**所有请求**（含普通聊天）
        # 都带 max_output_tokens=32000 + reasoning 配置。缺省时上游 kimi-k3 默认只给
        # 1024 token 输出预算（completion_tokens=1024 实锤），思考+正文全被截断 → 空壳
        # 响应（usage 显示 1024 但正文为空）。对齐网页：**默认 32000**，客户端显式传
        # max_tokens 则覆盖。
        body.setdefault("max_output_tokens", 32000)
        # 网页聊天请求带 reasoning 配置（kimi 等推理模型靠它出思考链；缺省时上游可能
        # 不输出思考或直接空转）。对齐 HAR 原文 {'effort': 'low', 'summary': 'concise'}。
        # 仅普通聊天带（生图 HAR 无此字段，加了可能干扰生图执行）。
        if not image_model and "reasoning" not in body:
            body["reasoning"] = {"effort": "low", "summary": "concise"}

        headers = await self._auth.get_auth()
        headers["content-type"] = "application/json"

        # 2026-08-08 实测（HAR「同一会话连续生图gpt」）：生图**整轮**耗时 108-161s，
        # 但正文（output_text.delta）只占最后 1-2 秒，前面全部是服务端静默执行生图
        # （function_call_arguments.delta 也可能长时间无字节）。
        # 默认 request_timeout=120s 的 read timeout 会在**生图静默段**掐断流 → 只拿到
        # 第一轮 function_call 围栏、无 URL → 伪 fence 换号冷却（网页无此限制所以永不
        # fence）。修复：**生图请求用独立 300s 总超时**（HAR 实测最慢 161s < 300s），
        # 对齐网页无感等待。普通对话仍走默认 120s。
        stream_timeout = httpx.Timeout(300.0) if image_model else None
        async with self._http.stream(
            "POST", RESPONSES_URL, json=body, headers=headers,
            timeout=stream_timeout,
        ) as resp:
            if resp.status_code >= 400:
                body_text = (await resp.aread()).decode("utf-8", "replace")
                raise httpx.HTTPStatusError(
                    f"anuma responses failed: {resp.status_code} {body_text}",
                    request=resp.request,
                    response=resp,
                )
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                payload = line[6:].strip()
                if not payload or payload == "[DONE]":
                    continue
                try:
                    raw = json.loads(payload)
                except json.JSONDecodeError:
                    continue
                for ir in self._parser.parse(raw):
                    yield ir

    def _persist_account(self) -> None:
        """把生成的 conversation_id 写回账号文件（失败不阻断请求）。"""
        if self._account_file is None:
            return
        try:
            self._account_file.parent.mkdir(parents=True, exist_ok=True)
            tmp = self._account_file.with_suffix(".json.tmp")
            tmp.write_text(
                json.dumps(self._account.model_dump(mode="json"),
                           ensure_ascii=False, indent=2),
                encoding="utf-8",
            )
            tmp.replace(self._account_file)
        except Exception:  # noqa: BLE001
            pass

    def _upstream_model(self, model_id: str | None) -> str:
        if not model_id:
            return ""
        # 2026-08-07 实测：portal /api/v1/responses 只认 **catalog id**
        # （如 openai/gpt-5.6-luna、inclusionai/ling-2.6-flash），
        # 传 accounts/fireworks/models/<stable-id> 返回 400 model not available。
        # model_id 可能是 adapter 传的 upstream_id（Fireworks 格式），
        # 需把它 normalized 回 catalog id 再透传。
        low = model_id.lower()
        # 已是 catalog 形式：inclusionai/ling-2.6-flash、openai/gpt-5.6-luna
        if "/" in low and not low.startswith("accounts/"):
            return model_id
        # Fireworks 格式 accounts/fireworks/models/<短名> → 反查 catalog id
        short = low.rsplit("/", 1)[-1]
        from app.upstream.models import MODEL_CATALOG  # 延迟导入避免循环
        for m in MODEL_CATALOG:
            if m["id"].rsplit("/", 1)[-1] == short:
                return m["id"]
            if (m.get("upstream_id") or "").lower() == low:
                return m["id"]
        return model_id