app/upstream/parser.py

"""anuma.ai Responses SSE 事件 → IREvent（原生协议与 OpenAI Responses 一致）。

上游：``POST portal.anuma.ai/api/v1/responses``，返回标准 OpenAI Responses SSE：
``response.created`` / ``response.output_text.delta`` / ``response.completed`` 等。
正文增量由 ``response.output_text.delta`` 产出；思维链摘要增量由
``response.reasoning_summary_text.delta`` 产出（必须产出 kind="thinking"）。

**原生 function calling**：上游接受客户端 tools 并原生返回 function_call 事件
（``response.output_item.added`` type=function_call + ``response.function_call_arguments.delta/done``）。
本 parser 把 function_call 统一转成 ``<tool_call>{...}</tool_call>`` 围栏文本
（携带上游 call_id），走宿主既有解析链路（parse_tool_calls / ToolCallStreamParser）。
"""
from __future__ import annotations

import json
from typing import Any

from app.events import IREvent, Usage
from app.upstream.base import EventParser


class DefaultParser(EventParser):
    def __init__(self) -> None:
        # 流式 function_call 状态：call_id -> {name, arguments(字符串累积)}
        self._pending_calls: dict[str, dict[str, str]] = {}
        # 本轮是否已通过 output_text.delta 吐过正文（用于 done 去重）
        self._saw_text_delta = False

    def parse(self, raw: Any) -> list[IREvent]:
        """raw 为单个 SSE 事件 dict（``{"type": ..., ...}``），返回 0..n 个 IREvent。"""
        if not isinstance(raw, dict):
            return []
        etype = raw.get("type") or ""
        delta = raw.get("delta") or {}
        if not isinstance(delta, dict):
            delta = {}

        events: list[IREvent] = []

        # ---- 新响应开始：清本轮状态（parser 跨请求复用）----
        if etype == "response.created":
            self._pending_calls.clear()
            self._saw_text_delta = False
            return events

        # ---- function_call 开始（output_item.added 里 type=function_call）----
        if etype == "response.output_item.added":
            item = raw.get("item") or {}
            if isinstance(item, dict) and item.get("type") == "function_call":
                call_id = str(item.get("call_id") or item.get("id") or "")
                if call_id:
                    self._pending_calls[call_id] = {
                        "name": str(item.get("name") or ""),
                        "arguments": str(item.get("arguments") or ""),
                    }
            return events

        # ---- function_call arguments 增量 ----
        if etype == "response.function_call_arguments.delta":
            # delta 字段是 arguments 的字符串增量
            for call_id, pending in self._pending_calls.items():
                pending["arguments"] += str(delta if isinstance(delta, str) else delta.get("delta") or "")
            return events

        # ---- function_call arguments 完整 ----
        if etype == "response.function_call_arguments.done":
            for call_id, pending in self._pending_calls.items():
                fence = _function_call_fence(pending["name"], pending["arguments"], call_id)
                if fence:
                    events.append(IREvent(kind="text", text=fence))
            self._pending_calls.clear()
            return events

        # ---- 思维链摘要增量（独立事件，kimi 等推理模型的思考链）----
        # 2026-08-08 实测（HAR「聊天」kimi/kimi-k3）：kimi 的思考是**独立事件**
        # response.reasoning_summary_text.delta（×128），结构：
        #   {"delta": {"OfString": "We", "OfResponseReasoningSummaryDeltaEventDelta": "We"},
        #    "type": "response.reasoning_summary_text.delta", ...}
        # 两个字段都是**字符串**（GPT 系模型在 output_text.delta 里带的则是 dict）。
        # 不解析 → 思考全丢 → 客户端等待期空白 + 正文瞬间喷出 = "一大段突然出现"。
        if etype == "response.reasoning_summary_text.delta":
            t = delta.get("OfString") if isinstance(delta, dict) else None
            if not t:
                s = delta.get("OfResponseReasoningSummaryDeltaEventDelta") if isinstance(delta, dict) else None
                if isinstance(s, dict):
                    t = s.get("text") or ""
                else:
                    t = s or ""
            if t:
                events.append(IREvent(kind="thinking", thinking=str(t)))
            return events

        # ---- 正文增量 ----
        # 事件 type 是唯一语义来源：output_text.delta 只产出正文。
        # 某些上游 union 序列化会把同一个正文 token 同时写入 OfString 和
        # OfResponseReasoningSummaryDeltaEventDelta；后者无论字符串还是 dict 都不能在
        # 此事件中再解释成 thinking，否则客户端会在正文后生成逐 token 的碎片思考卡。
        if etype == "response.output_text.delta":
            text = delta.get("OfString") if isinstance(delta, dict) else ""
            if isinstance(text, dict):  # 部分模型带 {"text": ...} 结构
                text = text.get("text") or ""
            if text:
                self._saw_text_delta = True
                events.append(IREvent(kind="text", text=str(text)))
            return events

        # ---- 输出完整文本（生图时图片 markdown 只在此事件的**顶层 text** 里）----
        # 2026-08-08 实测（debug_parser.py）：生图响应里 /media/ 图片 URL（如
        # "![可爱的小猫咪写真](https://portal.anuma.ai/api/v1/media/media/...)"）只
        # 出现在 response.output_text.done 的**事件顶层 text 字段**（与 delta 同层，
        # 不是 delta 里）。delta 事件只带思维链摘要，没有图片。不处理顶层 text
        # 会把图片链接整个吞掉（此前生图只有 function_call 空壳、无媒体输出）。
        #
        # 聊天路径：delta 已流式吐完正文，done 再吐整段会重复（客户端先逐字再整段 dump）。
        # 去重：已见 text delta 且 done 不含 media URL → 跳过；生图 / 无 delta 兜底照发。
        if etype == "response.output_text.done":
            text = raw.get("text")
            if isinstance(text, str) and text:
                if self._saw_text_delta and "portal.anuma.ai/api/v1/media" not in text:
                    return events
                events.append(IREvent(kind="text", text=text))
            return events

        # ---- 完成事件：非流式 function_call 在 output 里；usage + finish ----
        # 生图时图片签名 URL 在 response.tool_call_events[].output（JSON 字符串，
        # 含 output_images[].url）——上游把生图结果当作工具执行结果回传。
        # 2026-08-08 实测（debug_parser.py）：流式响应的 output_text 是空壳
        # （{"type":"output_text"}），图片 markdown 只出现在 tool_call_events。
        if etype == "response.completed":
            resp = raw.get("response") or {}
            if isinstance(resp, dict):
                for item in resp.get("output") or []:
                    if isinstance(item, dict) and item.get("type") == "function_call":
                        fence = _function_call_fence(
                            str(item.get("name") or ""),
                            item.get("arguments") or "{}",
                            str(item.get("call_id") or item.get("id") or ""),
                        )
                        if fence:
                            events.append(IREvent(kind="text", text=fence))
                # 生图结果：tool_call_events 里带 output 的调用 → 把图片 URL 以
                # markdown 形式输出，客户端可见真实图片（不产出空壳调用围栏）。
                for tce in resp.get("tool_call_events") or []:
                    if not isinstance(tce, dict):
                        continue
                    output = tce.get("output")
                    if not isinstance(output, str) or not output:
                        continue
                    try:
                        out = json.loads(output)
                    except json.JSONDecodeError:
                        continue
                    urls = []
                    for img in out.get("output_images") or []:
                        u = img.get("url") if isinstance(img, dict) else None
                        if u:
                            urls.append(u)
                    if urls:
                        text = "\n".join(f"![generated image]({u})" for u in urls)
                        events.append(IREvent(kind="text", text=text))
            usage = self._parse_usage(raw)
            if usage:
                events.append(IREvent(kind="finish", usage_delta=usage, finish_reason="stop"))
            else:
                events.append(IREvent(kind="finish", finish_reason="stop"))
            return events

        # ---- 结尾独立 usage 事件（无 type 或 type 为 usage 的收尾事件）----
        if etype == "" or etype == "response.usage":
            usage = self._parse_usage(raw)
            if usage:
                events.append(IREvent(kind="finish", usage_delta=usage, finish_reason="stop"))
            return events

        # ---- 失败事件 ----
        if etype == "response.failed":
            err = raw.get("error") or raw.get("message") or "upstream response failed"
            msg = err.get("message") if isinstance(err, dict) else str(err)
            events.append(IREvent(kind="error", error=msg or "upstream response failed"))
            return events

        return events

    @staticmethod
    def _parse_usage(raw: dict[str, Any]) -> Usage | None:
        usage = raw.get("usage")
        if not isinstance(usage, dict):
            return None
        # anuma 返回 prompt_tokens/completion_tokens/total_tokens/cost_micro_usd/credits_used
        # （OpenAI 标准为 input_tokens/output_tokens）
        input_tokens = usage.get("input_tokens") or usage.get("prompt_tokens") or 0
        output_tokens = usage.get("output_tokens") or usage.get("completion_tokens") or 0
        if not input_tokens and not output_tokens:
            return None
        details = usage.get("output_tokens_details") or {}
        thinking = details.get("reasoning_tokens") if isinstance(details, dict) else 0
        cached = 0
        in_details = usage.get("input_tokens_details") or {}
        if isinstance(in_details, dict):
            cached = in_details.get("cached_tokens") or 0
        return Usage(
            input_tokens=int(input_tokens),
            output_tokens=int(output_tokens),
            thinking_tokens=int(thinking or 0),
            cached_tokens=int(cached or 0),
            model=str(usage.get("model") or "") or None,
            provider="anuma",
        )


def _function_call_fence(name: str, arguments: str, call_id: str) -> str | None:
    """function_call → ``<tool_call>{"name","arguments","id"}</tool_call>`` 围栏文本。"""
    if not name:
        return None
    try:
        args = json.loads(arguments) if arguments.strip() else {}
    except json.JSONDecodeError:
        args = {"value": arguments}
    if not isinstance(args, dict):
        args = {"value": args}
    return (
        f"<tool_call>{json.dumps({'name': name, 'arguments': args, 'id': call_id},
                                 ensure_ascii=False)}</tool_call>"
    )