From c39ed1e4ba493ac6dfb5794da8112d74133f890c Mon Sep 17 00:00:00 2001 From: JK <2081217232@qq.com> Date: Mon, 5 Oct 2026 17:19:53 +0800 Subject: [PATCH] =?UTF-8?q?feat(gateway):=20/v1/responses=E2=80=94?= =?UTF-8?q?=E2=80=94OpenAI=20Responses=20API=20=E5=8F=8C=E5=90=91=E7=BF=BB?= =?UTF-8?q?=E8=AF=91=EF=BC=88=E6=B5=81=E5=BC=8F/=E9=9D=9E=E6=B5=81?= =?UTF-8?q?=E5=BC=8F/previous=5Fresponse=5Fid=20LRU=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 9 +- app/response_store.py | 58 ++ app/responses_compat.py | 555 ++++++++++++++++++++ app/routes/gateway.py | 147 +++++- docs/development/04-api-spec.md | 37 +- docs/development/07-roadmap.md | 2 +- docs/testing/02-unit-tests.md | 20 + docs/testing/04-integration-e2e.md | 13 +- tests/integration/test_gateway_responses.py | 137 +++++ tests/unit/test_responses_compat.py | 527 +++++++++++++++++++ 10 files changed, 1497 insertions(+), 8 deletions(-) create mode 100644 app/response_store.py create mode 100644 app/responses_compat.py create mode 100644 tests/integration/test_gateway_responses.py create mode 100644 tests/unit/test_responses_compat.py diff --git a/README.md b/README.md index 0d9effa..4bb1739 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # zcode2api (zcode-hub) -**ZCode 账号运营 + 双协议网关一体机**：把池内 ZCode Coding Plan / Start Plan / API Key 账号额度，统一转换为标准的 -**Anthropic Messages API**（`/v1/messages`）与 **OpenAI Chat Completions API**（`/v1/chat/completions`），向 Claude Code、Cline、Codex CLI、Chatbox、NextChat 等各类 Agent 与客户端提供高可用 API 服务。 +**ZCode 账号运营 + 三协议网关一体机**：把池内 ZCode Coding Plan / Start Plan / API Key 账号额度，统一转换为标准的 +**Anthropic Messages API**（`/v1/messages`）、**OpenAI Chat Completions API**（`/v1/chat/completions`）与 **OpenAI Responses API**（`/v1/responses`），向 Claude Code、Cline、Codex CLI（原生走 `/v1/responses`）、Chatbox、NextChat 等各类 Agent 与客户端提供高可用 API 服务。 自带一套完整的账号运营控制台：多账号池轮询、高可信成套桌面指纹仿真、官方首启安装序仿真、单账号并发控制、429/5xx/3012 阶梯容灾、请求全景监控、限时套餐自动与手动领取、OAuth 免密登录入池，以及免浏览器的阿里云无痕验证求解——开箱即用，自托管部署。 @@ -9,7 +9,7 @@ ## 核心能力 -- **双协议兼容网关**：原生支持 `POST /v1/messages`（Anthropic Messages 协议，支持流式 SSE）与 `POST /v1/chat/completions`（OpenAI 格式双向翻译，支持流式与非流式），并提供 `GET /v1/models` 模型列表。 +- **三协议兼容网关**：原生支持 `POST /v1/messages`（Anthropic Messages 协议，支持流式 SSE）、`POST /v1/chat/completions`（OpenAI 格式双向翻译，支持流式与非流式）与 `POST /v1/responses`（OpenAI Responses 协议双向翻译，支持流式与非流式，Codex CLI 原生直连），并提供 `GET /v1/models` 模型列表。 - **高可信桌面指纹池（一号一台）**：入池自动生成加权桌面成套 SKU（Mac Sequoia/Tahoe、Win11 主流分辨率等），每账号独立全新 UUIDv4 `device_mid`，杜绝宿主机云 Linux 内核共享导致的账号集群关联风险。 - **官方首启安装序仿真**：入池与指纹轮换自动执行 `client/configs` 与激活事件上报（`app_launch` / `app_daily_active`），持久化 `install_id` 与 `installed_at`。 - **反代与 CDN 基础设施头深度清洗**：自动剔除 `x-forwarded-*`、`forwarded`、`cf-*`、`cdn-loop`、`x-real-ip`、`via`、`cookie`、`referer`、`origin`、`accept-language` 等网络中间件头部，杜绝部署网络拓扑泄露与客户端伪装冲突。 @@ -54,6 +54,7 @@ cp .env.example .env # 按需修改密钥、端口等 - 对话端点： - Anthropic 协议：`POST http://127.0.0.1:3000/v1/messages`（兼容 Claude Code、Cline） - OpenAI 协议：`POST http://127.0.0.1:3000/v1/chat/completions`（兼容 OpenAI 客户端） + - OpenAI Responses 协议：`POST http://127.0.0.1:3000/v1/responses`（Codex CLI 原生直连） - 模型列表：`GET http://127.0.0.1:3000/v1/models` - 探活端点：`GET http://127.0.0.1:3000/meta` @@ -140,7 +141,7 @@ JWT 账号调用上游时需携带阿里云无痕验证参数（请求头 `X-Ali ## 鉴权 - **后台鉴权**：所有 `/admin/api/*` 需 `Authorization: Bearer <后台密码>`（也支持 `?app_key=`），内置 IP 失败节流保护。 -- **网关鉴权（可选）**：在「设置」配置「网关 API Key」后，`/v1/messages` 与 `/v1/chat/completions` 须携带 +- **网关鉴权（可选）**：在「设置」配置「网关 API Key」后，`/v1/messages`、`/v1/chat/completions` 与 `/v1/responses` 须携带 `Authorization: Bearer ` 或 `x-api-key: `；留空则不校验。 ## 环境变量 diff --git a/app/response_store.py b/app/response_store.py new file mode 100644 index 0000000..592807a --- /dev/null +++ b/app/response_store.py @@ -0,0 +1,58 @@ +"""previous_response_id 状态存储 —— /v1/responses 专用内存 LRU。 + +OpenAI Responses 协议的多轮续接靠 previous_response_id 引用上一轮的 +input/output 条目；上游 Anthropic 协议是无状态的，网关在两次请求之间把 +历史条目暂存在这里，下一轮命中时合并成完整 messages 重放。 + +设计要点： +- 仅内存（OrderedDict 上限 256 条，超出淘汰最旧），进程重启即清零： + 定位是续接便利而非持久会话，客户端随时可重发全量 input 兜底 +- 与 reqlog 同一哲学：避免写库的 IO 放大，丢失可接受、绝不放大请求失败 +- 存取均为浅拷贝列表（防调用方原地改写污染库内条目），条目 dict 本体不深拷贝 +""" + +from __future__ import annotations + +import threading +from collections import OrderedDict + +_MAX = 256 + +_lock = threading.Lock() +_store: OrderedDict[str, dict] = OrderedDict() + + +def save(response_id: str, input_items: list, output_items: list) -> None: + """保存一轮响应的 input/output 条目（各存浅拷贝列表）。""" + with _lock: + if response_id in _store: + _store.move_to_end(response_id) + _store[response_id] = { + "input_items": list(input_items), + "output_items": list(output_items), + } + while len(_store) > _MAX: + _store.popitem(last=False) # 淘汰最旧 + + +def get(response_id: str) -> dict | None: + """取回条目（命中移到队尾刷新 LRU 位次）；未命中返回 None。""" + with _lock: + item = _store.get(response_id) + if item is None: + return None + _store.move_to_end(response_id) + return { + "input_items": list(item["input_items"]), + "output_items": list(item["output_items"]), + } + + +def clear() -> None: + with _lock: + _store.clear() + + +def size() -> int: + with _lock: + return len(_store) diff --git a/app/responses_compat.py b/app/responses_compat.py new file mode 100644 index 0000000..23c8576 --- /dev/null +++ b/app/responses_compat.py @@ -0,0 +1,555 @@ +"""OpenAI Responses 兼容层 —— /v1/responses ↔ Anthropic Messages 双向转换。 + +入站：Responses 请求体 → Anthropic messages 体（instructions→system、input +条目（message/reasoning/function_call/function_call_output）→ blocks、 +reasoning.effort→reasoning_effort、tools 扁平映射）。 +出站：Anthropic 响应（JSON 或 SSE 事件流）→ Responses 格式 +（output 条目 + response.created/.completed 事件帧）。 + +与 openai_compat 同一原则：对畸形输入保持宽容，映射不了的条目安静跳过， +绝不放大请求失败。previous_response_id 的历史合并由网关层（response_store） +完成后并入 input，本模块只做纯格式翻译。 +""" + +from __future__ import annotations + +import json +import time +import uuid + +# Responses reasoning.effort → Anthropic reasoning_effort +# 上游 GLM 原生接受 reasoning_effort（low/high/max），不支持关闭思考， +# medium 即上游默认档——不映射、不带该字段。 +_EFFORT_MAP = { + "minimal": "low", + "low": "low", + "high": "high", + "xhigh": "max", +} + + +def _as_int(value: object) -> int | None: + try: + return int(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return None + + +def _image_block(url: str) -> dict | None: + """data:image/...;base64,xxx → Anthropic image block；外链 URL 无法回填，跳过。""" + if not url.startswith("data:"): + return None + head, _, b64 = url.partition(",") + media_type = head[5:].split(";")[0] or "image/png" + if not b64: + return None + return {"type": "image", "source": {"type": "base64", "media_type": media_type, "data": b64}} + + +def _message_to_blocks(item: dict) -> dict | None: + """message 条目 → Anthropic 消息；无有效内容返回 None（整条丢弃）。 + + 兼容两种入参形态： + - Responses 规范：type="message"，分块用 input_text / output_text + - chat 简写（部分客户端的 input 数组就长这样）：无 type，分块用 text—— + 缺 type 的条目若带 role 同样按消息处理，否则会被当未知条目丢弃， + 最终发出空 messages 招致上游 400（实测 code 1214）。 + assistant 取 output_text/text，user（及其他角色按 user）取 + input_text/text 与 input_image（仅 data URL）。 + """ + role = item.get("role") + content = item.get("content") + if role == "assistant": + blocks: list[dict] = [] + if isinstance(content, str) and content: + blocks.append({"type": "text", "text": content}) + elif isinstance(content, list): + for part in content: + if (isinstance(part, dict) and part.get("type") in ("output_text", "text") + and isinstance(part.get("text"), str) and part["text"]): + blocks.append({"type": "text", "text": part["text"]}) + return {"role": "assistant", "content": blocks} if blocks else None + # user 及未知角色一律按 user 处理 + blocks = [] + if isinstance(content, str) and content: + blocks.append({"type": "text", "text": content}) + elif isinstance(content, list): + for part in content: + if not isinstance(part, dict): + continue + ptype = part.get("type") + if ptype in ("input_text", "text") and isinstance(part.get("text"), str) and part["text"]: + blocks.append({"type": "text", "text": part["text"]}) + elif ptype == "input_image": + url = part.get("image_url") + if isinstance(url, dict): + url = url.get("url") or "" + block = _image_block(str(url)) + if block: + blocks.append(block) + return {"role": "user", "content": blocks} if blocks else None + + +def _reasoning_to_thinking(item: dict) -> dict | None: + """type=="reasoning" 条目 → assistant 消息（thinking 块）。 + + 优先取 content[]（type reasoning_text）的原文，缺失退回 summary[]（type + summary_text）；两者皆空则整条跳过。thinking 块 signature 上游不校验 + （实测伪造值可通过），空串亦可——同 openai_compat 重放 reasoning 的先例。 + """ + texts: list[str] = [] + for part in item.get("content") or []: + if (isinstance(part, dict) and part.get("type") == "reasoning_text" + and isinstance(part.get("text"), str) and part["text"]): + texts.append(part["text"]) + if not texts: + for part in item.get("summary") or []: + if (isinstance(part, dict) and part.get("type") == "summary_text" + and isinstance(part.get("text"), str) and part["text"]): + texts.append(part["text"]) + text = "\n".join(texts) + if not text: + return None + return {"role": "assistant", "content": [{"type": "thinking", "thinking": text, "signature": ""}]} + + +def _function_call_to_msg(item: dict) -> dict: + """type=="function_call" 条目 → assistant 消息（tool_use 块）。 + + arguments 为 JSON 字符串，解析失败不丢数据，包成 {"_raw": ...} 上送 + （同 openai_compat 的 tool_calls 宽容策略）。 + """ + args = item.get("arguments") + if isinstance(args, str): + try: + args = json.loads(args or "{}") + except ValueError: + args = {"_raw": args} + elif not isinstance(args, dict): + args = {} + return {"role": "assistant", "content": [{ + "type": "tool_use", + "id": str(item.get("call_id") or ""), + "name": str(item.get("name") or ""), + "input": args if isinstance(args, dict) else {}, + }]} + + +def _function_output_to_msg(item: dict) -> dict: + """type=="function_call_output" 条目 → user 消息（tool_result 块）。 + + output 为 str 直接用；为 list 时拼接其中 output_text 的 text + （ Responses 规范的 content parts 形态）。 + """ + output = item.get("output") + if isinstance(output, list): + parts = [] + for part in output: + if (isinstance(part, dict) and part.get("type") == "output_text" + and isinstance(part.get("text"), str)): + parts.append(part["text"]) + content = "\n".join(p for p in parts if p) + elif isinstance(output, str): + content = output + else: + content = "" + return {"role": "user", "content": [{ + "type": "tool_result", + "tool_use_id": str(item.get("call_id") or ""), + "content": content, + }]} + + +def responses_to_anthropic(payload: dict) -> tuple[dict | None, str | None]: + """Responses 请求体 → Anthropic messages 体。非法时返回 (None, 错误信息)。""" + model = payload.get("model") + if not isinstance(model, str) or not model.strip(): + return None, "必须提供 model 参数" + + raw_input = payload.get("input") + if isinstance(raw_input, str): + # 字符串 input → 单条 user 消息 + items: list = [] + if raw_input: + items = [{"type": "message", "role": "user", + "content": [{"type": "input_text", "text": raw_input}]}] + elif isinstance(raw_input, list): + items = raw_input + else: + items = [] + # input 为空且无 previous_response_id 链入内容时才报错 + #（网关层命中历史后已合并进 input；直接调用时以 previous_response_id 字段为准） + if not items and not payload.get("previous_response_id"): + return None, "必须提供 input" + + out_msgs: list[dict] = [] + for item in items: + if not isinstance(item, dict): + continue # 非 dict 条目安静跳过 + itype = item.get("type") + # type 缺失但带 role 的条目按消息处理（客户端 chat 简写发法） + if itype == "message" or (itype is None and item.get("role")): + msg = _message_to_blocks(item) + if msg is not None: + out_msgs.append(msg) + elif itype == "reasoning": + msg = _reasoning_to_thinking(item) + if msg is not None: + out_msgs.append(msg) + elif itype == "function_call": + out_msgs.append(_function_call_to_msg(item)) + elif itype == "function_call_output": + out_msgs.append(_function_output_to_msg(item)) + # item_reference / 未知类型：安静跳过 + + # 相邻同 role 消息合并（blocks 拼接；Anthropic 要求 role 严格交替） + messages: list[dict] = [] + for msg in out_msgs: + if messages and messages[-1]["role"] == msg["role"]: + messages[-1]["content"] = messages[-1]["content"] + msg["content"] + else: + messages.append(msg) + # 全部条目都映射不出内容 → 本地 400 早失败：空 messages 发上游只会换来 + # 语焉不详的 1214「messages 参数非法」，客户端无法据此定位入参问题 + if not messages: + return None, ("input 未包含可用消息条目（支持 message/reasoning/" + "function_call/function_call_output，message 条目可省略 type 只留 role+content）") + + body: dict = { + "model": model, + "messages": messages, + "max_tokens": _as_int(payload.get("max_output_tokens")) or 4096, + } + instructions = payload.get("instructions") + if isinstance(instructions, str) and instructions.strip(): + body["system"] = instructions + elif isinstance(instructions, list): + parts = [p for p in instructions if isinstance(p, str) and p.strip()] + if parts: + body["system"] = "\n\n".join(parts) + try: + if payload.get("temperature") is not None: + body["temperature"] = float(payload["temperature"]) + if payload.get("top_p") is not None: + body["top_p"] = float(payload["top_p"]) + except (TypeError, ValueError): + pass + # 推理强度：上游 GLM 原生接受 reasoning_effort（low/high/max），不支持关思考 + # —— minimal/low 降到 low（低速档），xhigh 升到 max，medium 即默认档不带字段 + effort = (payload.get("reasoning") or {}).get("effort") if isinstance(payload.get("reasoning"), dict) else None + if isinstance(effort, str) and effort.strip().lower() in _EFFORT_MAP: + body["reasoning_effort"] = _EFFORT_MAP[effort.strip().lower()] + if payload.get("stream"): + body["stream"] = True + + tools = payload.get("tools") + if isinstance(tools, list) and tools: + mapped = [] + for t in tools: + # 仅 function 工具可映射；web_search 等内置类型上游无对应物，跳过 + if isinstance(t, dict) and t.get("type") == "function" and t.get("name"): + mapped.append({ + "name": str(t["name"]), + "description": str(t.get("description") or ""), + "input_schema": t.get("parameters") if isinstance(t.get("parameters"), dict) else {"type": "object"}, + }) + if mapped: + body["tools"] = mapped + + choice = payload.get("tool_choice") + if choice == "none": + body.pop("tools", None) + elif choice == "required": + body["tool_choice"] = {"type": "any"} + elif isinstance(choice, dict) and choice.get("type") == "function": + name = str(choice.get("name") or "") + if name: + body["tool_choice"] = {"type": "tool", "name": name} + # "auto"/缺省：Anthropic 默认即 auto，无需显式映射 + return body, None + + +def anthropic_to_responses(data: dict, model: str) -> dict: + """Anthropic message 响应 → OpenAI response 对象。""" + output: list[dict] = [] + for block in data.get("content") or []: + if not isinstance(block, dict): + continue + btype = block.get("type") + if btype == "thinking" and isinstance(block.get("thinking"), str): + output.append({ + "type": "reasoning", + "id": f"rs_{uuid.uuid4().hex[:12]}", + "summary": [{"type": "summary_text", "text": block["thinking"]}], + }) + elif btype == "text" and isinstance(block.get("text"), str): + output.append({ + "type": "message", + "id": f"msg_{uuid.uuid4().hex[:12]}", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": block["text"], "annotations": []}], + }) + elif btype == "tool_use": + output.append({ + "type": "function_call", + "id": f"fc_{uuid.uuid4().hex[:12]}", + "call_id": str(block.get("id") or ""), + "name": str(block.get("name") or ""), + "arguments": json.dumps(block.get("input") or {}, ensure_ascii=False), + }) + usage = data.get("usage") or {} + in_tok = _as_int(usage.get("input_tokens")) or 0 + out_tok = _as_int(usage.get("output_tokens")) or 0 + if data.get("stop_reason") == "max_tokens": + status = "incomplete" + incomplete_details: dict | None = {"reason": "max_output_tokens"} + else: + status = "completed" + incomplete_details = None + return { + "id": f"resp_{uuid.uuid4().hex[:24]}", + "object": "response", + "created_at": int(time.time()), + "status": status, + "model": str(data.get("model") or model), + "output": output, + "error": None, + "incomplete_details": incomplete_details, + "usage": { + "input_tokens": in_tok, + "output_tokens": out_tok, + "total_tokens": in_tok + out_tok, + "input_tokens_details": { + "cached_tokens": _as_int(usage.get("cache_read_input_tokens")) or 0, + }, + "output_tokens_details": {"reasoning_tokens": 0}, + }, + "parallel_tool_calls": True, + "tool_choice": "auto", + "tools": [], + } + + +class ResponsesStreamConverter: + """Anthropic SSE 事件流 → OpenAI Responses 事件流（有状态转换器）。 + + 用法：先 start() 产出 response.created + response.in_progress 两帧， + 逐条 feed(event_dict) 收输出帧，流结束后调用 done()。每帧形如 + `event: \\ndata: {json}\\n\\n`，data 内 sequence_number 严格递增。 + """ + + def __init__(self, model: str) -> None: + self.model = model + self.response_id = f"resp_{uuid.uuid4().hex[:24]}" + self.created_at = int(time.time()) + self.sequence_number = 0 + # None = 上游未上报（未知），0 = 真实为零；两者语义不同 + self.usage = {"input_tokens": None, "output_tokens": None, "total_tokens": None} + self.output_items: list[dict] = [] # 已完成的 output 条目（response.completed 引用） + self._stop_reason: str | None = None + self._output_index = -1 # 当前 content block 的 output_index + self._cur_type: str | None = None + self._cur_item_id: str | None = None + self._cur_text = "" # thinking / text 累积 + self._cur_args = "" # tool_use arguments 累积 + self._cur_call_id = "" + self._cur_name = "" + + def _sync_usage_total(self) -> None: + i, o = self.usage["input_tokens"], self.usage["output_tokens"] + self.usage["total_tokens"] = (i or 0) + (o or 0) if i is not None or o is not None else None + + def _reset_cur(self) -> None: + self._cur_type = None + self._cur_item_id = None + self._cur_text = "" + self._cur_args = "" + self._cur_call_id = "" + self._cur_name = "" + + # ── 帧构造 ──────────────────────────────────────────────────────────── + def _event(self, etype: str, payload: dict) -> str: + frame = {"type": etype, "sequence_number": self.sequence_number, **payload} + self.sequence_number += 1 + return f"event: {etype}\ndata: {json.dumps(frame, ensure_ascii=False)}\n\n" + + def _snapshot(self, status: str, output: list) -> dict: + return { + "id": self.response_id, + "object": "response", + "type": "response", + "created_at": self.created_at, + "status": status, + "model": self.model, + "output": output, + "error": None, + "incomplete_details": None, + "usage": dict(self.usage), + } + + def start(self) -> str: + snap = self._snapshot("in_progress", []) + return self._event("response.created", {"response": snap}) \ + + self._event("response.in_progress", {"response": snap}) + + def done(self) -> str: + # Responses 流没有 [DONE] 哨兵（OpenAI 规范以 response.completed 收尾）， + # 恒返回空串，网关据此不再 yield 任何收尾帧 + return "" + + # ── 上游事件处理 ────────────────────────────────────────────────────── + def feed(self, evt: dict) -> list[str]: + etype = evt.get("type") + if etype == "message_start": + msg = evt.get("message") or {} + if msg.get("id"): + self.response_id = f"resp_{msg['id']}" + in_tok = _as_int((msg.get("usage") or {}).get("input_tokens")) + if in_tok is not None: + self.usage["input_tokens"] = in_tok + self._sync_usage_total() + return [] + if etype == "content_block_start": + return self._on_block_start(evt.get("content_block") or {}) + if etype == "content_block_delta": + return self._on_block_delta(evt.get("delta") or {}) + if etype == "content_block_stop": + return self._on_block_stop() + if etype == "message_delta": + delta = evt.get("delta") or {} + if delta.get("stop_reason"): + self._stop_reason = str(delta["stop_reason"]) + out = _as_int((evt.get("usage") or {}).get("output_tokens")) + if out is not None: + self.usage["output_tokens"] = out + self._sync_usage_total() + return [] + if etype == "message_stop": + return [self._completed_frame()] + return [] # ping / error 等无需产出 + + def _on_block_start(self, block: dict) -> list[str]: + btype = block.get("type") + if btype not in ("thinking", "text", "tool_use"): + return [] # 未识别的块不占 output_index，安静跳过 + self._output_index += 1 + self._cur_type = btype + idx = self._output_index + if btype == "thinking": + self._cur_item_id = f"rs_{uuid.uuid4().hex[:12]}" + return [ + self._event("response.output_item.added", { + "output_index": idx, + "item": {"type": "reasoning", "id": self._cur_item_id, "summary": []}, + }), + self._event("response.reasoning_summary_part.added", { + "item_id": self._cur_item_id, "output_index": idx, "summary_index": 0, + "part": {"type": "summary_text", "text": ""}, + }), + ] + if btype == "text": + self._cur_item_id = f"msg_{uuid.uuid4().hex[:12]}" + return [ + self._event("response.output_item.added", { + "output_index": idx, + "item": {"type": "message", "id": self._cur_item_id, "role": "assistant", + "status": "in_progress", "content": []}, + }), + self._event("response.content_part.added", { + "item_id": self._cur_item_id, "output_index": idx, "content_index": 0, + "part": {"type": "output_text", "text": "", "annotations": []}, + }), + ] + # tool_use + self._cur_item_id = f"fc_{uuid.uuid4().hex[:12]}" + self._cur_call_id = str(block.get("id") or "") + self._cur_name = str(block.get("name") or "") + return [self._event("response.output_item.added", { + "output_index": idx, + "item": {"type": "function_call", "id": self._cur_item_id, + "call_id": self._cur_call_id, "name": self._cur_name, + "arguments": "", "status": "in_progress"}, + })] + + def _on_block_delta(self, delta: dict) -> list[str]: + dtype = delta.get("type") + idx = self._output_index + item_id = self._cur_item_id + if dtype == "thinking_delta" and isinstance(delta.get("thinking"), str): + self._cur_text += delta["thinking"] + return [self._event("response.reasoning_summary_text.delta", { + "item_id": item_id, "output_index": idx, "content_index": 0, + "delta": delta["thinking"], + })] + if dtype == "text_delta" and isinstance(delta.get("text"), str): + self._cur_text += delta["text"] + return [self._event("response.output_text.delta", { + "item_id": item_id, "output_index": idx, "content_index": 0, + "delta": delta["text"], + })] + if dtype == "input_json_delta" and isinstance(delta.get("partial_json"), str): + self._cur_args += delta["partial_json"] + return [self._event("response.function_call_arguments.delta", { + "item_id": item_id, "output_index": idx, "content_index": 0, + "delta": delta["partial_json"], + })] + return [] + + def _on_block_stop(self) -> list[str]: + idx = self._output_index + item_id = self._cur_item_id + if self._cur_type == "thinking": + item = {"type": "reasoning", "id": item_id, + "summary": [{"type": "summary_text", "text": self._cur_text}]} + frames = [ + self._event("response.reasoning_summary_text.done", { + "item_id": item_id, "output_index": idx, "content_index": 0, + "text": self._cur_text, + }), + self._event("response.reasoning_summary_part.done", { + "item_id": item_id, "output_index": idx, "summary_index": 0, + "part": {"type": "summary_text", "text": self._cur_text}, + }), + ] + elif self._cur_type == "text": + item = {"type": "message", "id": item_id, "role": "assistant", "status": "completed", + "content": [{"type": "output_text", "text": self._cur_text, "annotations": []}]} + frames = [ + self._event("response.output_text.done", { + "item_id": item_id, "output_index": idx, "content_index": 0, + "text": self._cur_text, + }), + self._event("response.content_part.done", { + "item_id": item_id, "output_index": idx, "content_index": 0, + "part": {"type": "output_text", "text": self._cur_text, "annotations": []}, + }), + ] + elif self._cur_type == "tool_use": + item = {"type": "function_call", "id": item_id, "call_id": self._cur_call_id, + "name": self._cur_name, "arguments": self._cur_args, "status": "completed"} + frames = [ + self._event("response.function_call_arguments.done", { + "item_id": item_id, "output_index": idx, + "arguments": self._cur_args, + }), + ] + else: + return [] + frames.append(self._event("response.output_item.done", { + "output_index": idx, "item": item, + })) + self.output_items.append(item) + self._reset_cur() + return frames + + def _completed_frame(self) -> str: + if self._stop_reason == "max_tokens": + status = "incomplete" + details: dict | None = {"reason": "max_output_tokens"} + else: + status = "completed" + details = None + snap = self._snapshot(status, list(self.output_items)) + snap["incomplete_details"] = details + return self._event("response.completed", {"response": snap}) diff --git a/app/routes/gateway.py b/app/routes/gateway.py index 714f951..7a803cb 100644 --- a/app/routes/gateway.py +++ b/app/routes/gateway.py @@ -15,13 +15,18 @@ from fastapi import APIRouter, Depends, Request from fastapi.responses import JSONResponse, StreamingResponse -from .. import constants, logs, reqlog, settings +from .. import constants, logs, reqlog, response_store, settings from ..agent import build_request from ..auth_admin import verify_gateway_key from ..captcha import captcha_manager from ..models import Account, Status from ..openai_compat import StreamConverter, anthropic_to_openai, openai_to_anthropic from ..quota import fetch_quota +from ..responses_compat import ( + ResponsesStreamConverter, + anthropic_to_responses, + responses_to_anthropic, +) from ..store import store _sleep = asyncio.sleep # 模块级引用：测试可 patch 此名而免污染全局 asyncio @@ -330,6 +335,146 @@ async def _iter(): headers={"Cache-Control": "no-cache"}) +@router.post("/v1/responses", dependencies=[Depends(verify_gateway_key)]) +async def responses_api(request: Request): + try: + payload = await request.json() + except (json.JSONDecodeError, ValueError): + return JSONResponse({"error": {"message": "请求体不是合法 JSON", "type": "invalid_request_error"}}, status_code=400) + if not isinstance(payload, dict): + return JSONResponse({"error": {"message": "请求体必须是 JSON 对象", "type": "invalid_request_error"}}, status_code=400) + + # previous_response_id 预处理：从内存库链入上一轮 input/output 条目， + # 合并成完整条目数组后按无状态请求转发（未命中即 400，客户端应重发全量 input）。 + # 当前请求的 input 无论是否链历史都归一为条目数组——首轮也要入库， + # 否则下一轮链入时缺第一轮 user 内容（OpenAI Responses 续接语义即全量重放）。 + pid = payload.get("previous_response_id") + raw_input = payload.get("input") + if isinstance(raw_input, str): + new_items = [{"type": "message", "role": "user", + "content": [{"type": "input_text", "text": raw_input}]}] if raw_input else [] + elif isinstance(raw_input, list): + new_items = raw_input + else: + new_items = [] + store_input_items: list = new_items + if pid: + st = response_store.get(str(pid)) + if st is None: + return JSONResponse( + {"error": {"message": "previous_response_id 未找到或已过期", "type": "invalid_request_error"}}, + status_code=400, + ) + merged = st["input_items"] + st["output_items"] + new_items + payload = {**payload, "input": merged} + payload.pop("previous_response_id", None) + store_input_items = merged # 保留引用：成功后原样入库供下一轮续接 + + body, err = responses_to_anthropic(payload) + if err or body is None: + return JSONResponse({"error": {"message": err or "请求体不合法", "type": "invalid_request_error"}}, status_code=400) + + incoming_headers = dict(request.headers) + provider = _detect_provider(body, request.headers) + body = _normalize_body(body) + port = request.url.port or settings.PORT + + req_id = secrets.token_hex(8) + logs.req(req_id, str(body.get("model") or "-"), bool(payload.get("stream")), _last_user_text(body)) + reqlog.begin(req_id, "responses", str(body.get("model") or "-"), + bool(payload.get("stream")), _last_user_text(body)) + + try: + result = await _dispatch(req_id, body, incoming_headers, port, provider) + except asyncio.CancelledError: + reqlog.finish_error(req_id, "客户端断开", status=499) + raise + except Exception as err: # noqa: BLE001 - 调度层意外异常也要收口监控条目 + reqlog.finish_error(req_id, f"网关内部错误: {err}", status=500) + return JSONResponse( + {"error": {"message": "网关内部错误", "type": "internal_error"}}, + status_code=500, + ) + if not isinstance(result, _Upstream): + return result + + model = str(body.get("model") or "") + if payload.get("stream"): + try: + return _responses_stream_response(result, model, req_id, store_input_items) + except asyncio.CancelledError: + await result.close() + raise + + try: + raw = await result.resp.aread() + logs.req_ok(req_id) + except asyncio.CancelledError: + reqlog.finish_error(req_id, "客户端断开", status=499, t_first=result.t_first) + raise + except Exception as err: # noqa: BLE001 + logs.req_err(req_id, f"读取上游响应失败: {err}") + reqlog.finish_error(req_id, f"读取上游响应失败: {err}", status=502) + return JSONResponse({"error": {"message": f"读取上游响应失败: {err}", "type": "upstream_error"}}, status_code=502) + finally: + await result.close() + data = _safe_json(raw.decode("utf-8", "ignore")) + if not isinstance(data, dict) or data.get("type") != "message": + reqlog.finish_error(req_id, "上游响应格式异常", status=502, t_first=result.t_first) + return JSONResponse({"error": {"message": "上游响应格式异常", "type": "upstream_error"}}, status_code=502) + usage = data.get("usage") or {} + reqlog.finish_ok(req_id, t_first=result.t_first, status=result.resp.status_code, + input_tokens=usage.get("input_tokens"), output_tokens=usage.get("output_tokens")) + resp = anthropic_to_responses(data, model) + response_store.save(resp["id"], store_input_items, resp["output"]) + return JSONResponse(resp) + + +def _responses_stream_response(up: _Upstream, model: str, req_id: str, + store_input_items: list) -> StreamingResponse: + """把上游 Anthropic SSE 事件流转换为 OpenAI Responses 事件流。 + + 成功收尾时把历史条目 + 本次 output 入库（response_store），供下一轮 + previous_response_id 续接；中断路径不入库（半截输出不可安全重放）。 + """ + conv = ResponsesStreamConverter(model) + + async def _iter(): + try: + first = conv.start() + if first: + yield first + async for line in up.resp.aiter_lines(): + if not line.startswith("data:"): + continue + data_str = line[5:].strip() + if not data_str: + continue + evt = _safe_json(data_str) + if isinstance(evt, dict): + for out in conv.feed(evt): + yield out + tail = conv.done() # Responses 流无 [DONE] 哨兵，done() 恒为空串 + if tail: + yield tail + response_store.save(conv.response_id, store_input_items, conv.output_items) + logs.req_ok(req_id) + reqlog.finish_ok(req_id, t_first=up.t_first, status=up.resp.status_code, + input_tokens=conv.usage.get("input_tokens"), + output_tokens=conv.usage.get("output_tokens")) + except asyncio.CancelledError: + reqlog.finish_error(req_id, "客户端断开", status=499, t_first=up.t_first) + raise + except Exception as err: # noqa: BLE001 + logs.req_err(req_id, f"流传输中断: {err}") + reqlog.finish_error(req_id, f"流传输中断: {err}", t_first=up.t_first) + finally: + await up.close() + + return StreamingResponse(_iter(), media_type="text/event-stream", + headers={"Cache-Control": "no-cache"}) + + async def _dispatch(req_id, body, incoming_headers, port, provider): """多账号轮询调度：_Upstream（成功）或 JSONResponse（错误）。 diff --git a/docs/development/04-api-spec.md b/docs/development/04-api-spec.md index b06b23a..00ca22a 100644 --- a/docs/development/04-api-spec.md +++ b/docs/development/04-api-spec.md @@ -1,6 +1,6 @@ # 04 — API 规范 -状态：与 2.5.11 实现对齐。对外网关只有 Anthropic Messages、OpenAI Chat Completions、`/v1/models`；**没有** `/v1/responses`。 +状态：与 2.5.11 实现对齐。对外网关为 Anthropic Messages、OpenAI Chat Completions、OpenAI Responses 三协议网关 + `/v1/models`，均已实现。 所有管理端点挂 `/admin/api/*`，需 `Authorization: Bearer <后台密码>`（连续失败达上限后 429 锁 5 分钟）；网关端点按「网关 Key」配置可选鉴权（`Authorization: Bearer` 或 `x-api-key`，未配置即放行——生产必须配置）。 @@ -26,6 +26,30 @@ - 入站 OpenAI Chat 格式 → 翻译为 Anthropic 上游 → 翻译回 OpenAI 响应；`stream:true` 逐块翻译。 - usage/tool_calls 映射规则以移植对照表为准（`openai_compat.py`）。 +### 1.4 `POST /v1/responses`（OpenAI Responses 协议） + +- 入站 OpenAI Responses 格式 → 翻译为 Anthropic 上游（`responses_to_anthropic`）→ 翻译回 Responses 对象（`anthropic_to_responses`）；`stream:true` 经 `ResponsesStreamConverter` 逐事件翻译。翻译器实现见 `app/responses_compat.py`，会话状态见 `app/response_store.py`。 +- 请求字段映射： + - `instructions` → `system`；`input` 条目 `message` / `reasoning` / `function_call` / `function_call_output` → Anthropic 对应消息与块（`reasoning` 回放为 thinking，工具调用对 → `tool_use` / `tool_result`）。 + - `tools` 仅接受 `function` 类型（`web_search` 等其他工具类型跳过）；`tool_choice` 映射同 chat/completions。 + - `reasoning.effort` → 上游思考档位：`minimal` / `low` → `low`，`high` → `high`，`xhigh` → `max`，`medium`（或缺省）→ 不携带该字段（上游默认档）。 + - `max_output_tokens` → `max_tokens`；`model` 规范化同 1.1。 + - 未支持字段（`store` / `include` / `metadata` / `prompt_cache_key` / `text` 等）安静忽略。 +- 会话续接：`previous_response_id` 命中 → 上一轮完整输入输出拼入上游上下文；存储为进程内存 LRU（256 条，重启清零），未命中返回 400。 +- 响应映射：thinking 思考以 `reasoning` 条目（`summary` 内 `summary_text`）回传；`stop_reason=max_tokens` → `status: "incomplete"`，其余 → `"completed"`。 + +```json +{ "id": "resp_…", "object": "response", "status": "completed", "model": "GLM-5.3-Flash", "output": [ { "type": "reasoning", "summary": [ { "type": "summary_text", "text": "…" } ] }, { "type": "message", "role": "assistant", "content": [ { "type": "output_text", "text": "…" } ] } ] } +``` + +- 流式为 Responses SSE（`event:` + `data:` 帧，帧内含 `sequence_number`），无 `[DONE]` 哨兵，以 `response.completed` 收尾。事件序： + - `response.created` → `response.in_progress` + - 思考：`response.output_item.added` → `response.reasoning_summary_part.added` → `response.reasoning_summary_text.delta` + - 正文：`response.output_item.added` → `response.content_part.added` → `response.output_text.delta` → `response.output_text.done` + - 工具：`response.function_call_arguments.delta` → `response.function_call_arguments.done` + - 收尾：`response.completed` +- 网关 Key 鉴权、故障转移与错误处理与 `/v1/messages` 一致（错误格式见 §1.6）。 + ### 1.6 错误格式 ```json @@ -35,6 +59,7 @@ | HTTP | type | 触发 | |------|------|------| | 400 | invalid_request / invalid_request_error | JSON 非法，或请求体不是对象 | +| 400 | invalid_request | `previous_response_id` 未命中（会话 LRU 无该记录） | | 401/403 | （FastAPI HTTPException） | 网关 key 缺失/不符 | | 500 | captcha_error | 验证码求解失败 | | 500 | internal_error | 调度层未捕获异常（监控条目会收口） | @@ -117,4 +142,14 @@ npx claude curl http://127.0.0.1:3000/v1/chat/completions \ -H "Authorization: Bearer " -H "Content-Type: application/json" \ -d '{"model":"glm-5.3","messages":[{"role":"user","content":"hi"}],"stream":true}' + +# OpenAI Responses 兼容（非流式） +curl http://127.0.0.1:3000/v1/responses \ + -H "Authorization: Bearer " -H "Content-Type: application/json" \ + -d '{"model":"glm-5.3","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}]}' + +# OpenAI Responses 兼容（流式） +curl http://127.0.0.1:3000/v1/responses \ + -H "Authorization: Bearer " -H "Content-Type: application/json" \ + -d '{"model":"glm-5.3","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"hi"}]}],"stream":true}' ``` diff --git a/docs/development/07-roadmap.md b/docs/development/07-roadmap.md index ec4d745..ccb0836 100644 --- a/docs/development/07-roadmap.md +++ b/docs/development/07-roadmap.md @@ -33,7 +33,7 @@ M0 骨架 ──▶ M1 账号运营层(P1) ──▶ M2 网关扩展(P2) - [ ] `translator/`：OpenAI↔Anthropic 请求/响应/usage/tool_calls 映射（对照表先写测试用例） - [ ] `gateway/openai.py`：`/v1/chat/completions`（流式 + 批量） - [ ] `translator/sse.py`：SSE 逐块双向翻译 -- [ ] `gateway/responses.py`：`/v1/responses` + previous_response_id LRU +- [x] `gateway/responses.py`：`/v1/responses` + previous_response_id LRU - [ ] 网关鉴权接入新端点；模型白名单统一 **出口标准**：Claude Code（Anthropic 透传）、OpenAI SDK（chat/completions 流式+批量）、Codex CLI（/v1/responses）三客户端在 Mock 上游回归全绿，真实上游冒烟各 1 次。 diff --git a/docs/testing/02-unit-tests.md b/docs/testing/02-unit-tests.md index 9959923..008d428 100644 --- a/docs/testing/02-unit-tests.md +++ b/docs/testing/02-unit-tests.md @@ -137,3 +137,23 @@ | ZC-007 | 切换前保全 | 未入库的当前登录自动先快照 → 不丢号 | P0 | | ZC-008 | 路径穿越防护 | id 含 `../`/非法字符 → 拒绝 | P0 | | ZC-009 | ZCode 运行中热切换保护 | 检测到客户端运行 → 按 behavior 配置拒绝/杀进程 | P2 | + +## 10. RS — Responses 协议翻译（responses_compat.py / response_store.py） + +| ID | 用例 | 输入 → 预期 | 优先级 | +|----|------|-------------|--------| +| RS-001 | 基本请求映射 | `instructions` → system；input message 条目 → user/assistant 消息（input_text 块）；model 规范化同网关 | P0 | +| RS-002 | reasoning.effort 映射 | 参数化 minimal/low/medium/high/xhigh → `low`/`low`/不带（上游默认档）/`high`/`max` | P0 | +| RS-003 | tools/tool_choice 映射 | function 工具 → Anthropic tools；tool_choice 各档同 chat/completions；非 function 类型（web_search 等）跳过 | P0 | +| RS-004 | function_call 条目映射 | input `function_call` / `function_call_output` 条目 → assistant `tool_use` / user `tool_result` 块 | P0 | +| RS-005 | reasoning 条目回放 | input `reasoning` 条目（summary_text）→ 上游 thinking 块 | P0 | +| RS-006 | max_output_tokens 映射 | `max_output_tokens: 512` → 上游 `max_tokens: 512` | P1 | +| RS-007 | 未支持字段安静忽略 | `store`/`include`/`metadata`/`prompt_cache_key`/`text` 等 → 不报错、不透传上游 | P1 | +| RS-008 | 非流式响应映射 | Anthropic content → output 数组：thinking → `reasoning`（summary_text）、text → `message`（output_text）；`stop_reason=max_tokens` → `status:"incomplete"`，end_turn → `"completed"` | P0 | +| RS-009 | response_id 生成 | 非流式 envelope 生成新 `resp_` id；流式沿用上游 message id（`resp_`）。入库由网关路由层完成（集成 INT-042 覆盖） | P0 | +| RS-010 | previous_response_id 链上下文 | store 命中 → 上一轮输入输出拼入上游 messages | P0 | +| RS-011 | previous_response_id 未命中 | 不存在 id → 400 invalid_request（见 api-spec §1.6 错误表） | P0 | +| RS-012 | 流式事件序 | 上游 message_start/thinking/text/tool_use 全链 → `response.created` → `response.in_progress` → 思考（`output_item.added` / `reasoning_summary_part.added` / `reasoning_summary_text.delta`）→ 正文（`output_item.added` / `content_part.added` / `output_text.delta` / `output_text.done`）→ 工具（`function_call_arguments.delta` / `.done`）→ `response.completed`；无 `[DONE]` 哨兵 | P0 | +| RS-013 | sequence_number 连续 | 全部 SSE 帧 `sequence_number` 严格递增、无跳变 | P1 | +| RS-014 | LRU 容量 256 | 连续写入 257 条 → 最早一条被淘汰，按其 id 再取 → `get` 返回 None | P0 | +| RS-015 | LRU 清零语义 | `clear()` 清空后旧 id 全部未命中；模块级存储无持久化，进程重启即失 | P1 | diff --git a/docs/testing/04-integration-e2e.md b/docs/testing/04-integration-e2e.md index 6e5c5d0..b100d5c 100644 --- a/docs/testing/04-integration-e2e.md +++ b/docs/testing/04-integration-e2e.md @@ -39,6 +39,17 @@ | INT-030 | 真实 solver 冒烟 | （可选，`@pytest.mark.real_captcha`）真实 jsdom 子进程求解一次 | 输出 verifyParam 结构合法（base64(JSON{certifyId,...})）；默认 CI 跳过 | | INT-031 | solver 缺失报错 | 未 npm install → 明确报错指引用，而非悬死 | 用户体验 | +### INT-E Responses 协议（/v1/responses，用例实现 `tests/integration/test_gateway_responses.py`） + +| ID | 场景 | 步骤 | 断言 | +|----|------|------|------| +| INT-040 | 非流式翻译 | `/v1/responses` 非流式一条（scenario=ok） | 200 且 Responses 对象形状完整（output 数组 / status / usage），文本与 /v1/messages 同源一致 | +| INT-041 | 流式事件序 | stream=true → Responses SSE | 帧序符合事件序（created → in_progress → … → completed），sequence_number 连续，无 `[DONE]` 哨兵 | +| INT-042 | previous_response_id 链 | 首请求取 `response_id` → 二次请求携带 → 200 | 第二轮上游请求含首轮上下文；response_store 命中 | +| INT-043 | previous_response_id 未命中 | 携带不存在的 id | 400 invalid_request（api-spec §1.6 错误表） | +| INT-044 | 503 全池耗尽 | 全部账号 402（`x-mock-sequence`）→ 请求 /v1/responses | 503 no_available_account，错误格式同 /v1/messages | +| INT-045 | 401/403 网关鉴权 | 配置网关 key 后缺 key / 错 key | 401/403，行为同 /v1/messages | + ## 2. 故障注入矩阵（Mock 上游 `x-mock-scenario`） | scenario | 行为 | 主要消费者 | @@ -67,7 +78,7 @@ | ID | 场景 | 步骤 | 断言 | |----|------|------|------| -| E2E-001 | 三客户端协议 | Claude Code 形态（Anthropic SSE）、OpenAI SDK 形态（chat/completions，Phase 2）、Codex 形态（/v1/responses，Phase 2）各发 1 流式请求 | 三协议均 200 且内容一致（同一 Mock 模型输出） | +| E2E-001 | 三客户端协议 | Claude Code 形态（Anthropic SSE）、OpenAI SDK 形态（chat/completions）、Codex 形态（/v1/responses，已落地，见 `tests/integration/test_gateway_responses.py`）各发 1 流式请求 | 三协议均 200 且内容一致（同一 Mock 模型输出） | | E2E-002 | 额度用尽自动降级 | 种子把号1/号2 置 exhausted → 请求自动走号3 → Mock 记录命中 | 故障转移跨进程 | | E2E-003 | 后台操作流 | UI 登录 → 加号 → 看额度面板 → 手动 claim → 导出 .zsb → 清库导入 | 管理面闭环 | | E2E-004 | 容器重建恢复 | compose down → up（卷保留）→ 池/设置/历史完好 | 持久化 | diff --git a/tests/integration/test_gateway_responses.py b/tests/integration/test_gateway_responses.py new file mode 100644 index 0000000..6ec64c2 --- /dev/null +++ b/tests/integration/test_gateway_responses.py @@ -0,0 +1,137 @@ +"""GW-012 /v1/responses（OpenAI Responses 风格）端点集成测试。 + +复用 /v1/messages 同一套 mock 上游（返回 Anthropic 格式），验证双向转换： +非流式 response 对象、流式 response.* 事件帧，以及 previous_response_id +跨轮续接（response_store 链入历史）。 +""" + +from __future__ import annotations + +import json + +import pytest + +# JWT 前缀是 mock 上游按 bind 键控 counters 的依据（session 级共享）， +# 不能与既有测试文件的凭证前缀撞车（connect_fail 的 n==0 判定会被破坏）。 +_GOOD_JWT = "hZ.eyJzdWIiOiJ6In0.sig" + + +@pytest.mark.integration +class TestResponsesEndpoint: + async def test_nonstream_basic(self, gateway_client, fresh_app): + client, mock = gateway_client + from tests.conftest import seed_account + + seed_account(fresh_app, _GOOD_JWT, name="resp") + res = await client.post( + "/v1/responses", + json={"model": "GLM-5.3", "input": "你好"}, + ) + assert res.status_code == 200 + data = res.json() + assert data["object"] == "response" + assert data["id"].startswith("resp_") + assert data["status"] == "completed" + assert data["model"] == "GLM-5.3" + msg_items = [o for o in data["output"] if o["type"] == "message"] + assert msg_items, "output 应含 message 条目" + texts = [p["text"] for p in msg_items[0]["content"] if p["type"] == "output_text"] + assert "Hello from mock upstream" in "".join(texts) + assert data["usage"]["input_tokens"] == 10 + assert data["usage"]["output_tokens"] == 5 + + # 上游收到的 body：messages[0] 是 user 文本、model 正确 + payload = json.loads(mock.state.calls[-1][3]) + assert payload["model"] == "GLM-5.3" + assert payload["messages"][0]["role"] == "user" + assert payload["messages"][0]["content"][0]["type"] == "text" + assert payload["messages"][0]["content"][0]["text"] == "你好" + + async def test_stream_emits_response_events(self, gateway_client, fresh_app): + client, mock = gateway_client + from tests.conftest import seed_account + + seed_account(fresh_app, _GOOD_JWT, name="resp-stream") + res = await client.post( + "/v1/responses", + json={"model": "glm-5.3", "stream": True, "input": "hi"}, + ) + assert res.status_code == 200 + assert res.headers["content-type"].startswith("text/event-stream") + assert "event: response.created" in res.text + assert "event: response.completed" in res.text + assert "[DONE]" not in res.text # Responses 流没有 [DONE] 哨兵 + # 上游收到流式标记 + payload = json.loads(mock.state.calls[-1][3]) + assert payload["stream"] is True + + async def test_previous_response_id_chain(self, gateway_client, fresh_app): + """两轮对话：第二轮带 previous_response_id，上游应收到完整历史重放。""" + client, mock = gateway_client + from tests.conftest import seed_account + + seed_account(fresh_app, _GOOD_JWT, name="resp-chain") + r1 = await client.post("/v1/responses", + json={"model": "GLM-5.3", "input": "第一轮问题"}) + assert r1.status_code == 200 + resp_id = r1.json()["id"] + + r2 = await client.post( + "/v1/responses", + json={"model": "GLM-5.3", "input": "第二轮问题", + "previous_response_id": resp_id}, + ) + assert r2.status_code == 200 + assert r2.json()["id"] != resp_id + payload = json.loads(mock.state.calls[-1][3]) + msgs = payload["messages"] + # 第一轮 user + 第一轮 assistant 输出 + 第二轮 user + assert len(msgs) == 3 + assert msgs[0]["role"] == "user" + assert msgs[0]["content"][0]["text"] == "第一轮问题" + assert msgs[1]["role"] == "assistant" + assert msgs[1]["content"][0]["text"] == "Hello from mock upstream" + assert msgs[2]["role"] == "user" + assert msgs[2]["content"][0]["text"] == "第二轮问题" + + async def test_invalid_payload_rejected(self, gateway_client, fresh_app): + client, _ = gateway_client + res = await client.post("/v1/responses", json={"input": "hi"}) # 缺 model + assert res.status_code == 400 + assert res.json()["error"]["type"] == "invalid_request_error" + + async def test_unknown_previous_response_id_rejected(self, gateway_client, fresh_app): + client, _ = gateway_client + res = await client.post( + "/v1/responses", + json={"model": "GLM-5.3", "input": "hi", + "previous_response_id": "resp_does_not_exist"}, + ) + assert res.status_code == 400 + assert res.json()["error"]["type"] == "invalid_request_error" + + async def test_no_account_returns_503(self, fresh_app): + from httpx import ASGITransport, AsyncClient + + from app.main import create_app + + app = create_app() + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + res = await client.post("/v1/responses", + json={"model": "glm-5.3", "input": "hi"}) + assert res.status_code == 503 + assert res.json()["error"]["type"] == "no_available_account" + + async def test_gateway_key_required(self, gateway_client, fresh_app): + client, _ = gateway_client + fresh_app.set_setting("gateway_key", "sk-gw-test") + try: + res = await client.post("/v1/responses", + json={"model": "glm-5.3", "input": "hi"}) + assert res.status_code == 401 + res = await client.post("/v1/responses", + headers={"x-api-key": "sk-gw-test"}, + json={"model": "glm-5.3", "input": "hi"}) + assert res.status_code == 503 # 鉴权通过，进入调度（无账号） + finally: + fresh_app.set_setting("gateway_key", "") diff --git a/tests/unit/test_responses_compat.py b/tests/unit/test_responses_compat.py new file mode 100644 index 0000000..34d6057 --- /dev/null +++ b/tests/unit/test_responses_compat.py @@ -0,0 +1,527 @@ +"""responses_compat / response_store 单元测试 —— Responses ↔ Anthropic 双向转换。""" + +from __future__ import annotations + +import json +import re + +from app import response_store +from app.responses_compat import ( + ResponsesStreamConverter, + anthropic_to_responses, + responses_to_anthropic, +) + + +class TestResponsesToAnthropic: + def test_model_required(self): + assert responses_to_anthropic({"input": "hi"})[1] is not None # 缺 model + assert responses_to_anthropic({"model": "", "input": "hi"})[1] is not None + assert responses_to_anthropic({"model": 123, "input": "hi"})[1] is not None + + def test_string_input(self): + body, err = responses_to_anthropic({"model": "GLM-5.3", "input": "你好"}) + assert err is None + assert body["model"] == "GLM-5.3" + assert body["messages"] == [{"role": "user", "content": [{"type": "text", "text": "你好"}]}] + assert body["max_tokens"] == 4096 # Responses 可缺省，Anthropic 必填 + + def test_instructions_to_system(self): + body, err = responses_to_anthropic({"model": "GLM-5.3", "instructions": "守则", "input": "hi"}) + assert err is None + assert body["system"] == "守则" + + def test_user_and_assistant_message_items(self): + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [ + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "问"}]}, + {"type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "答"}]}, + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "再问"}]}, + ], + }) + assert err is None + assert [m["role"] for m in body["messages"]] == ["user", "assistant", "user"] + assert body["messages"][0]["content"] == [{"type": "text", "text": "问"}] + assert body["messages"][1]["content"] == [{"type": "text", "text": "答"}] + + def test_unknown_role_treated_as_user(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "message", "role": "root", + "content": [{"type": "input_text", "text": "x"}]}], + }) + assert body["messages"][0]["role"] == "user" + + def test_input_image_data_url(self): + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "message", "role": "user", "content": [ + {"type": "input_text", "text": "看图"}, + {"type": "input_image", "image_url": "data:image/png;base64,AAAA"}, + ]}], + }) + assert err is None + blocks = body["messages"][0]["content"] + assert blocks[0] == {"type": "text", "text": "看图"} + assert blocks[1]["type"] == "image" + assert blocks[1]["source"] == {"type": "base64", "media_type": "image/png", "data": "AAAA"} + + def test_remote_image_url_skipped(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "message", "role": "user", "content": [ + {"type": "input_text", "text": "看图"}, + {"type": "input_image", "image_url": "https://example.com/a.png"}, + ]}], + }) + blocks = body["messages"][0]["content"] + assert len(blocks) == 1 # 外链图片无法回填，安静跳过 + + def test_function_call_item_to_tool_use(self): + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [ + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "天气如何"}]}, + {"type": "function_call", "call_id": "call_1", "name": "get_weather", + "arguments": "{\"city\":\"杭州\"}"}, + ], + }) + assert err is None + asst = body["messages"][1] + assert asst["role"] == "assistant" + assert asst["content"][0] == {"type": "tool_use", "id": "call_1", "name": "get_weather", + "input": {"city": "杭州"}} + + def test_function_call_invalid_arguments_kept_as_raw(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "function_call", "call_id": "c1", "name": "f", + "arguments": "{bad json"}], + }) + block = body["messages"][0]["content"][0] + assert block["input"] == {"_raw": "{bad json"} + + def test_function_call_output_str(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "function_call_output", "call_id": "call_1", "output": "晴 25 度"}], + }) + msg = body["messages"][0] + assert msg["role"] == "user" + assert msg["content"][0] == {"type": "tool_result", "tool_use_id": "call_1", "content": "晴 25 度"} + + def test_function_call_output_list_parts(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "function_call_output", "call_id": "call_1", "output": [ + {"type": "output_text", "text": "第一段"}, + {"type": "output_text", "text": "第二段"}, + ]}], + }) + assert body["messages"][0]["content"][0]["content"] == "第一段\n第二段" + + def test_reasoning_content_preferred_over_summary(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{ + "type": "reasoning", + "content": [{"type": "reasoning_text", "text": "完整思考"}], + "summary": [{"type": "summary_text", "text": "摘要"}], + }], + }) + block = body["messages"][0]["content"][0] + assert block == {"type": "thinking", "thinking": "完整思考", "signature": ""} + + def test_reasoning_summary_fallback(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{ + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "摘要思考"}], + }], + }) + assert body["messages"][0]["content"][0]["thinking"] == "摘要思考" + + def test_reasoning_empty_skipped(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [ + {"type": "reasoning", "summary": []}, + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "hi"}]}, + ], + }) + # 空 reasoning 不产生空 assistant 消息 + assert [m["role"] for m in body["messages"]] == ["user"] + + def test_adjacent_same_role_merged(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [ + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "一"}]}, + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "二"}]}, + ], + }) + # Anthropic 要求 role 交替，相邻同角色合并为一条消息 + assert len(body["messages"]) == 1 + assert body["messages"][0]["content"] == [{"type": "text", "text": "一"}, + {"type": "text", "text": "二"}] + + def test_tools_flat_mapping(self): + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": "hi", + "tools": [ + {"type": "function", "name": "get_weather", "description": "查天气", + "parameters": {"type": "object", "properties": {"city": {"type": "string"}}}}, + {"type": "web_search", "search_context_size": "low"}, # 内置类型跳过 + ], + }) + assert err is None + assert body["tools"] == [{"name": "get_weather", "description": "查天气", + "input_schema": {"type": "object", + "properties": {"city": {"type": "string"}}}}] + + def test_tools_parameters_default_object(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", "input": "hi", + "tools": [{"type": "function", "name": "f"}], + }) + assert body["tools"][0] == {"name": "f", "description": "", "input_schema": {"type": "object"}} + + def test_tool_choice_variants(self): + base = {"model": "GLM-5.3", "input": "hi", + "tools": [{"type": "function", "name": "f"}]} + body, _ = responses_to_anthropic({**base, "tool_choice": "required"}) + assert body["tool_choice"] == {"type": "any"} + body, _ = responses_to_anthropic({**base, "tool_choice": {"type": "function", "name": "f"}}) + assert body["tool_choice"] == {"type": "tool", "name": "f"} + body, _ = responses_to_anthropic({**base, "tool_choice": "none"}) + assert "tools" not in body and "tool_choice" not in body + body, _ = responses_to_anthropic({**base, "tool_choice": "auto"}) + assert "tool_choice" not in body + + def test_reasoning_effort_mapping(self): + base = {"model": "GLM-5.3", "input": "hi"} + for effort, expected in (("minimal", "low"), ("low", "low"), + ("high", "high"), ("xhigh", "max")): + body, _ = responses_to_anthropic({**base, "reasoning": {"effort": effort}}) + assert body["reasoning_effort"] == expected, effort + # medium 即上游默认档：不带该字段（上游不支持关思考） + body, _ = responses_to_anthropic({**base, "reasoning": {"effort": "medium"}}) + assert "reasoning_effort" not in body + body, _ = responses_to_anthropic(base) + assert "reasoning_effort" not in body + + def test_optional_params_mapping(self): + body, err = responses_to_anthropic({ + "model": "GLM-5.3", "input": "hi", + "max_output_tokens": 128, "temperature": 0.5, "top_p": 0.9, "stream": True, + }) + assert err is None + assert body["max_tokens"] == 128 + assert body["temperature"] == 0.5 + assert body["top_p"] == 0.9 + assert body["stream"] is True + + def test_unknown_fields_quietly_ignored(self): + body, err = responses_to_anthropic({ + "model": "GLM-5.3", "input": "hi", + "store": True, "include": [], "metadata": {"k": "v"}, + "prompt_cache_key": "ck", "text": {"format": {"type": "text"}}, + }) + assert err is None + assert set(body.keys()) == {"model", "messages", "max_tokens"} + + def test_empty_input_rejected(self): + assert responses_to_anthropic({"model": "GLM-5.3"})[1] == "必须提供 input" + assert responses_to_anthropic({"model": "GLM-5.3", "input": []})[1] == "必须提供 input" + + def test_empty_input_with_previous_response_id_rejected(self): + # 网关层命中历史后会把存储条目合并进 input，翻译器见到空 messages 一律早失败： + # 空数组发上游只会换来语焉不详的 1214「messages 参数非法」 + body, err = responses_to_anthropic({"model": "GLM-5.3", "input": [], + "previous_response_id": "resp_x"}) + assert err is not None and body is None + + def test_chat_shorthand_item_without_type(self): + # 部分客户端的 input 数组发法：{"role":"user","content":"hi"}（无 type）—— + # 曾被当未知条目丢弃、发出空 messages 招致上游 400（实测 code 1214） + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"role": "user", "content": "hi"}, + {"role": "assistant", "content": "yo"}], + }) + assert err is None + assert body["messages"] == [ + {"role": "user", "content": [{"type": "text", "text": "hi"}]}, + {"role": "assistant", "content": [{"type": "text", "text": "yo"}]}, + ] + + def test_chat_shorthand_parts_with_text_type(self): + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"role": "user", + "content": [{"type": "text", "text": "hi"}]}], + }) + assert err is None + assert body["messages"] == [ + {"role": "user", "content": [{"type": "text", "text": "hi"}]}, + ] + + def test_empty_content_message_dropped(self): + # 空内容消息无信息量，整条丢弃而非塞空 text 块（上游对空块同样敏感） + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "message", "role": "user", "content": ""}, + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "hi"}]}], + }) + assert err is None + assert body["messages"] == [ + {"role": "user", "content": [{"type": "text", "text": "hi"}]}, + ] + + def test_all_items_unusable_rejected(self): + # 条目都映射不出内容 → 本地 400，不把空 messages 发给上游 + body, err = responses_to_anthropic({ + "model": "GLM-5.3", + "input": [{"type": "message", "role": "user", "content": ""}, + {"type": "item_reference", "id": "msg_1"}], + }) + assert err is not None and body is None + + def test_non_dict_items_skipped(self): + body, _ = responses_to_anthropic({ + "model": "GLM-5.3", + "input": ["garbage", 42, {"type": "item_reference", "id": "msg_1"}, + {"type": "message", "role": "user", + "content": [{"type": "input_text", "text": "hi"}]}], + }) + assert [m["role"] for m in body["messages"]] == ["user"] + + +class TestAnthropicToResponses: + def test_blocks_to_output_items(self): + out = anthropic_to_responses({ + "id": "msg_1", "type": "message", "model": "GLM-5.3", + "content": [ + {"type": "thinking", "thinking": "先想", "signature": "sig"}, + {"type": "text", "text": "答案"}, + {"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "杭州"}}, + ], + "stop_reason": "tool_use", + "usage": {"input_tokens": 10, "output_tokens": 5}, + }, "GLM-5.3") + assert [o["type"] for o in out["output"]] == ["reasoning", "message", "function_call"] + assert out["output"][0]["summary"] == [{"type": "summary_text", "text": "先想"}] + assert out["output"][1]["role"] == "assistant" + assert out["output"][1]["content"] == [{"type": "output_text", "text": "答案", "annotations": []}] + assert out["output"][2]["call_id"] == "call_1" + assert out["output"][2]["name"] == "get_weather" + assert json.loads(out["output"][2]["arguments"]) == {"city": "杭州"} + + def test_envelope_shape(self): + out = anthropic_to_responses({ + "id": "msg_1", "type": "message", "model": "GLM-5.3", + "content": [{"type": "text", "text": "hi"}], + "stop_reason": "end_turn", "usage": {"input_tokens": 1, "output_tokens": 1}, + }, "GLM-5.3") + assert out["object"] == "response" + assert out["id"].startswith("resp_") + assert out["status"] == "completed" + assert out["model"] == "GLM-5.3" + assert out["error"] is None + assert out["incomplete_details"] is None + assert out["tool_choice"] == "auto" + assert out["tools"] == [] + + def test_max_tokens_incomplete(self): + out = anthropic_to_responses({ + "id": "m", "type": "message", "content": [{"type": "text", "text": "x"}], + "stop_reason": "max_tokens", "usage": {}, + }, "GLM-5.3") + assert out["status"] == "incomplete" + assert out["incomplete_details"] == {"reason": "max_output_tokens"} + + def test_usage_conversion(self): + out = anthropic_to_responses({ + "id": "m", "type": "message", "content": [{"type": "text", "text": "x"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 10, "output_tokens": 5, "cache_read_input_tokens": 3}, + }, "GLM-5.3") + assert out["usage"] == { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15, + "input_tokens_details": {"cached_tokens": 3}, + "output_tokens_details": {"reasoning_tokens": 0}, + } + + def test_usage_missing_cache_defaults_zero(self): + out = anthropic_to_responses({ + "id": "m", "type": "message", "content": [], "stop_reason": "end_turn", "usage": {}, + }, "GLM-5.3") + assert out["usage"]["input_tokens_details"] == {"cached_tokens": 0} + assert out["usage"]["total_tokens"] == 0 + + +class TestResponsesStream: + @staticmethod + def _events(): + """手工构造的完整 Anthropic SSE 事件序列（思考 → 文本 → 工具调用）。""" + return [ + {"type": "message_start", "message": {"id": "msg_x", "usage": {"input_tokens": 7}}}, + {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking"}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "想"}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "一下"}}, + {"type": "content_block_stop", "index": 0}, + {"type": "content_block_start", "index": 1, "content_block": {"type": "text"}}, + {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "你"}}, + {"type": "content_block_delta", "index": 1, "delta": {"type": "text_delta", "text": "好"}}, + {"type": "content_block_stop", "index": 1}, + {"type": "content_block_start", "index": 2, + "content_block": {"type": "tool_use", "id": "call_1", "name": "f"}}, + {"type": "content_block_delta", "index": 2, "delta": {"type": "input_json_delta", "partial_json": '{"a"'}}, + {"type": "content_block_delta", "index": 2, "delta": {"type": "input_json_delta", "partial_json": ":1}"}}, + {"type": "content_block_stop", "index": 2}, + {"type": "message_delta", "delta": {"stop_reason": "tool_use"}, "usage": {"output_tokens": 9}}, + {"type": "message_stop"}, + ] + + def test_full_event_stream(self): + conv = ResponsesStreamConverter("GLM-5.3") + frames = [conv.start()] + for evt in self._events(): + frames.extend(conv.feed(evt)) + assert conv.done() == "" # Responses 流没有 [DONE] 哨兵 + + types = [json.loads(f.split("data: ", 1)[1])["type"] + for f in "".join(frames).split("\n\n") if f] + assert types[0] == "response.created" + assert types[1] == "response.in_progress" + assert types[-1] == "response.completed" + assert "response.output_item.added" in types and "response.output_item.done" in types + + # sequence_number 从 0 起严格递增 + seqs = [json.loads(f.split("data: ", 1)[1])["sequence_number"] + for f in "".join(frames).split("\n\n") if f] + assert seqs == list(range(len(seqs))) + + def test_text_delta_content(self): + conv = ResponsesStreamConverter("GLM-5.3") + conv.start() + frames = [] + for evt in self._events(): + frames.extend(conv.feed(evt)) + raw = "".join(frames) + deltas = [json.loads(m.group(1))["delta"] + for m in re.finditer(r"data: (.*)", raw) + if json.loads(m.group(1))["type"] == "response.output_text.delta"] + assert "".join(deltas) == "你好" + + def test_function_call_arguments_done_accumulated(self): + conv = ResponsesStreamConverter("GLM-5.3") + conv.start() + raw = "".join(f for evt in self._events() for f in conv.feed(evt)) + done_frames = [json.loads(m.group(1)) for m in re.finditer(r"data: (.*)", raw) + if json.loads(m.group(1))["type"] == "response.function_call_arguments.done"] + assert len(done_frames) == 1 + assert done_frames[0]["arguments"] == '{"a":1}' + + def test_completed_snapshot(self): + conv = ResponsesStreamConverter("GLM-5.3") + conv.start() + raw = "".join(f for evt in self._events() for f in conv.feed(evt)) + completed = [json.loads(m.group(1)) for m in re.finditer(r"data: (.*)", raw) + if json.loads(m.group(1))["type"] == "response.completed"] + assert len(completed) == 1 + snap = completed[0]["response"] + assert snap["status"] == "completed" + assert snap["id"] == "resp_msg_x" # 沿用上游 message id + assert len(snap["output"]) == 3 + assert [o["type"] for o in snap["output"]] == ["reasoning", "message", "function_call"] + assert snap["usage"]["input_tokens"] == 7 + assert snap["usage"]["output_tokens"] == 9 + assert snap["usage"]["total_tokens"] == 16 + + def test_output_items_complete(self): + conv = ResponsesStreamConverter("GLM-5.3") + conv.start() + for evt in self._events(): + conv.feed(evt) + assert [o["type"] for o in conv.output_items] == ["reasoning", "message", "function_call"] + assert conv.output_items[0]["summary"][0]["text"] == "想一下" + assert conv.output_items[1]["content"][0]["text"] == "你好" + assert json.loads(conv.output_items[2]["arguments"]) == {"a": 1} + + def test_max_tokens_incomplete_snapshot(self): + conv = ResponsesStreamConverter("GLM-5.3") + conv.start() + events = [ + {"type": "message_start", "message": {"id": "msg_m", "usage": {"input_tokens": 1}}}, + {"type": "content_block_start", "index": 0, "content_block": {"type": "text"}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "x"}}, + {"type": "message_delta", "delta": {"stop_reason": "max_tokens"}, "usage": {"output_tokens": 1}}, + {"type": "message_stop"}, + ] + raw = "".join(f for evt in events for f in conv.feed(evt)) + completed = [json.loads(m.group(1)) for m in re.finditer(r"data: (.*)", raw) + if json.loads(m.group(1))["type"] == "response.completed"] + snap = completed[0]["response"] + assert snap["status"] == "incomplete" + assert snap["incomplete_details"] == {"reason": "max_output_tokens"} + + def test_unknown_events_ignored(self): + conv = ResponsesStreamConverter("GLM-5.3") + conv.start() + assert conv.feed({"type": "ping"}) == [] + assert conv.feed({"type": "content_block_stop", "index": 0}) == [] # 无在途块，安静忽略 + + +class TestResponseStore: + def setup_method(self): + response_store.clear() + + def test_save_get_roundtrip(self): + response_store.save("resp_a", [{"type": "message", "role": "user"}], + [{"type": "message", "role": "assistant"}]) + st = response_store.get("resp_a") + assert st is not None + assert st["input_items"] == [{"type": "message", "role": "user"}] + assert st["output_items"] == [{"type": "message", "role": "assistant"}] + assert response_store.size() == 1 + + def test_miss_returns_none(self): + assert response_store.get("resp_missing") is None + + def test_get_returns_shallow_copies(self): + inp: list = [{"a": 1}] + response_store.save("resp_b", inp, []) + inp.append({"a": 2}) # 调用方改写原列表不得污染库内条目 + assert response_store.get("resp_b")["input_items"] == [{"a": 1}] + + def test_eviction_after_257(self): + for i in range(257): + response_store.save(f"resp_{i}", [i], []) + assert response_store.size() == 256 + assert response_store.get("resp_0") is None # 最旧被淘汰 + assert response_store.get("resp_1") is not None + assert response_store.get("resp_256") is not None + + def test_get_promotes_lru_order(self): + for i in range(256): + response_store.save(f"resp_{i}", [i], []) + response_store.get("resp_0") # 命中移到队尾，脱离淘汰位次 + response_store.save("resp_new", ["new"], []) + assert response_store.size() == 256 + assert response_store.get("resp_0") is not None # 被提升，幸存 + assert response_store.get("resp_1") is None # 成为最旧，被淘汰 + assert response_store.get("resp_new") is not None