Add chat completion cache and request dedupe

README.md

@@ -115,6 +115,7 @@ environment:
 ### 实验性 / 规划中
 
 - `/v1/complete` 文本补全与流式输出已实现，但仍在测试，目前会出现对话重复的问题，请谨慎测试使用
+- `/v1/chat/completions` 文本链路支持短 TTL 缓存、重复请求合并与相邻重复消息清理，可通过 `chat_completion_cache` 配置调整
 - 详细状态说明见：[功能清单](./docs/feature-status.en.md)
 
 ## 效果展示

docs/feature-status.en.md

@@ -29,6 +29,7 @@
 | Render / Vercel 等部署表述 | ⚠️ | 当前主要以 Docker 部署为主，其他平台部署方式暂未重点说明。 |
 | `/v1/complete` 文本补全与流式输出 | ✅  | 已实现。 |
 | 流式输出支持 | ✅  | 已实现。 |
+| 文本补全缓存与重复请求合并 | ✅  | `/v1/chat/completions` 文本链路默认启用 60 秒短缓存、流式结果回放、in-flight 请求合并和相邻重复消息清理；可通过 `chat_completion_cache` 配置关闭或调整。 |
 | 图片尺寸参数 | ❌  | 待实现。 |
 | 服务端图片 URL 缓存 | ✅  | 已实现。 |
 | `rt_token` 刷新 | ❌  | 待实现。 |

services/config.py

@@ -37,6 +37,17 @@ DEFAULT_IMAGE_STORAGE = {
     "public_base_url": "",
 }
 
+DEFAULT_CHAT_COMPLETION_CACHE = {
+    "enabled": True,
+    "ttl_seconds": 60,
+    "max_entries": 256,
+    "dedupe_inflight": True,
+    "stream_cache": True,
+    "normalize_messages": True,
+    "drop_adjacent_duplicates": True,
+    "drop_assistant_history": False,
+}
+
 
 def _normalize_bool(value: object, default: bool = False) -> bool:
     if isinstance(value, str):
@@ -115,6 +126,43 @@ def _normalize_image_storage_settings(value: object) -> dict[str, object]:
     }
 
 
+def _normalize_chat_completion_cache_settings(value: object) -> dict[str, object]:
+    source = value if isinstance(value, dict) else {}
+    return {
+        "enabled": _normalize_bool(source.get("enabled"), DEFAULT_CHAT_COMPLETION_CACHE["enabled"]),
+        "ttl_seconds": _normalize_positive_int(
+            source.get("ttl_seconds"),
+            int(DEFAULT_CHAT_COMPLETION_CACHE["ttl_seconds"]),
+            0,
+        ),
+        "max_entries": _normalize_positive_int(
+            source.get("max_entries"),
+            int(DEFAULT_CHAT_COMPLETION_CACHE["max_entries"]),
+            1,
+        ),
+        "dedupe_inflight": _normalize_bool(
+            source.get("dedupe_inflight"),
+            bool(DEFAULT_CHAT_COMPLETION_CACHE["dedupe_inflight"]),
+        ),
+        "stream_cache": _normalize_bool(
+            source.get("stream_cache"),
+            bool(DEFAULT_CHAT_COMPLETION_CACHE["stream_cache"]),
+        ),
+        "normalize_messages": _normalize_bool(
+            source.get("normalize_messages"),
+            bool(DEFAULT_CHAT_COMPLETION_CACHE["normalize_messages"]),
+        ),
+        "drop_adjacent_duplicates": _normalize_bool(
+            source.get("drop_adjacent_duplicates"),
+            bool(DEFAULT_CHAT_COMPLETION_CACHE["drop_adjacent_duplicates"]),
+        ),
+        "drop_assistant_history": _normalize_bool(
+            source.get("drop_assistant_history"),
+            bool(DEFAULT_CHAT_COMPLETION_CACHE["drop_assistant_history"]),
+        ),
+    }
+
+
 def _validate_image_storage_settings(settings: dict[str, object]) -> None:
     if not _normalize_bool(settings.get("enabled"), False):
         return
@@ -344,6 +392,7 @@ class ConfigStore:
         data["global_system_prompt"] = self.global_system_prompt
         data["backup"] = self.get_backup_settings()
         data["image_storage"] = self.get_image_storage_settings()
+        data["chat_completion_cache"] = self.get_chat_completion_cache_settings()
         data.pop("auth-key", None)
         return data
 
@@ -358,6 +407,10 @@ class ConfigStore:
         if "image_storage" in next_data:
             next_data["image_storage"] = _normalize_image_storage_settings(next_data.get("image_storage"))
             _validate_image_storage_settings(next_data["image_storage"])
+        if "chat_completion_cache" in next_data:
+            next_data["chat_completion_cache"] = _normalize_chat_completion_cache_settings(
+                next_data.get("chat_completion_cache")
+            )
         next_data.pop("backup_state", None)
         self.data = next_data
         self._save()
@@ -369,6 +422,9 @@ class ConfigStore:
     def get_image_storage_settings(self) -> dict[str, object]:
         return _normalize_image_storage_settings(self.data.get("image_storage"))
 
+    def get_chat_completion_cache_settings(self) -> dict[str, object]:
+        return _normalize_chat_completion_cache_settings(self.data.get("chat_completion_cache"))
+
     def get_storage_backend(self) -> StorageBackend:
         """获取存储后端实例（单例）"""
         if self._storage_backend is None:

services/protocol/chat_completion_cache.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any, Callable, Iterable, Iterator
+
+from services.config import config
+
+CACHEABLE_TEXT_KEYS = {
+    "frequency_penalty",
+    "max_completion_tokens",
+    "max_tokens",
+    "metadata",
+    "model",
+    "presence_penalty",
+    "reasoning_effort",
+    "response_format",
+    "seed",
+    "stop",
+    "temperature",
+    "tool_choice",
+    "tools",
+    "top_p",
+    "user",
+}
+
+
+@dataclass
+class CacheEntry:
+    expires_at: float
+    value: Any
+
+
+@dataclass
+class InflightCall:
+    condition: threading.Condition = field(default_factory=lambda: threading.Condition(threading.RLock()))
+    done: bool = False
+    value: Any = None
+    error: BaseException | None = None
+
+
+def _json_safe(value: Any) -> Any:
+    if isinstance(value, bytes):
+        return {"__bytes_sha256__": hashlib.sha256(value).hexdigest(), "length": len(value)}
+    if isinstance(value, bytearray):
+        data = bytes(value)
+        return {"__bytes_sha256__": hashlib.sha256(data).hexdigest(), "length": len(data)}
+    if isinstance(value, dict):
+        return {str(key): _json_safe(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_json_safe(item) for item in value]
+    return value
+
+
+def canonical_body(body: dict[str, Any], messages: list[dict[str, Any]], *, stream: bool) -> dict[str, Any]:
+    payload = {key: body.get(key) for key in CACHEABLE_TEXT_KEYS if key in body}
+    payload["messages"] = messages
+    payload["stream"] = bool(stream)
+    return payload
+
+
+def cache_key(body: dict[str, Any], messages: list[dict[str, Any]], *, stream: bool) -> str:
+    encoded = json.dumps(
+        _json_safe(canonical_body(body, messages, stream=stream)),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def _message_signature(message: dict[str, Any]) -> str:
+    return json.dumps(_json_safe(message), ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+
+
+def normalize_text_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    settings = config.get_chat_completion_cache_settings()
+    if not settings.get("normalize_messages"):
+        return messages
+
+    normalized: list[dict[str, Any]] = []
+    previous_signature = ""
+    for message in messages:
+        if settings.get("drop_assistant_history") and str(message.get("role") or "") == "assistant":
+            continue
+        signature = _message_signature(message)
+        if settings.get("drop_adjacent_duplicates") and signature == previous_signature:
+            continue
+        normalized.append(message)
+        previous_signature = signature
+    return normalized
+
+
+class ChatCompletionCache:
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self._entries: dict[str, CacheEntry] = {}
+        self._inflight: dict[str, InflightCall] = {}
+
+    def clear(self) -> None:
+        with self._lock:
+            self._entries.clear()
+            self._inflight.clear()
+
+    def _settings(self) -> dict[str, object]:
+        return config.get_chat_completion_cache_settings()
+
+    def _prune_locked(self, now: float, max_entries: int) -> None:
+        expired = [key for key, item in self._entries.items() if item.expires_at <= now]
+        for key in expired:
+            self._entries.pop(key, None)
+        while len(self._entries) > max_entries:
+            oldest_key = min(self._entries, key=lambda key: self._entries[key].expires_at)
+            self._entries.pop(oldest_key, None)
+
+    @staticmethod
+    def _copy(value: Any) -> Any:
+        return copy.deepcopy(value)
+
+    def get_or_compute_response(self, key: str, compute: Callable[[], dict[str, Any]]) -> dict[str, Any]:
+        settings = self._settings()
+        if not settings.get("enabled") or int(settings.get("ttl_seconds") or 0) <= 0:
+            return compute()
+
+        now = time.time()
+        max_entries = int(settings.get("max_entries") or 1)
+        with self._lock:
+            self._prune_locked(now, max_entries)
+            entry = self._entries.get(key)
+            if entry and entry.expires_at > now:
+                return self._copy(entry.value)
+            inflight = self._inflight.get(key) if settings.get("dedupe_inflight") else None
+            if inflight is None:
+                inflight = InflightCall()
+                if settings.get("dedupe_inflight"):
+                    self._inflight[key] = inflight
+                owner = True
+            else:
+                owner = False
+
+        if not owner:
+            with inflight.condition:
+                while not inflight.done:
+                    inflight.condition.wait()
+                if inflight.error:
+                    raise inflight.error
+                return self._copy(inflight.value)
+
+        try:
+            value = compute()
+        except BaseException as exc:
+            with self._lock:
+                self._inflight.pop(key, None)
+            with inflight.condition:
+                inflight.error = exc
+                inflight.done = True
+                inflight.condition.notify_all()
+            raise
+
+        expires_at = time.time() + int(settings.get("ttl_seconds") or 0)
+        with self._lock:
+            self._entries[key] = CacheEntry(expires_at=expires_at, value=self._copy(value))
+            self._prune_locked(time.time(), max_entries)
+            self._inflight.pop(key, None)
+        with inflight.condition:
+            inflight.value = self._copy(value)
+            inflight.done = True
+            inflight.condition.notify_all()
+        return value
+
+    def get_or_compute_stream(self, key: str, compute: Callable[[], Iterable[dict[str, Any]]]) -> Iterator[dict[str, Any]]:
+        settings = self._settings()
+        if (
+            not settings.get("enabled")
+            or not settings.get("stream_cache")
+            or int(settings.get("ttl_seconds") or 0) <= 0
+        ):
+            yield from compute()
+            return
+
+        now = time.time()
+        max_entries = int(settings.get("max_entries") or 1)
+        with self._lock:
+            self._prune_locked(now, max_entries)
+            entry = self._entries.get(key)
+            if entry and entry.expires_at > now:
+                yield from self._copy(entry.value)
+                return
+            inflight = self._inflight.get(key) if settings.get("dedupe_inflight") else None
+            if inflight is None:
+                inflight = InflightCall()
+                if settings.get("dedupe_inflight"):
+                    self._inflight[key] = inflight
+                owner = True
+            else:
+                owner = False
+
+        if not owner:
+            with inflight.condition:
+                while not inflight.done:
+                    inflight.condition.wait()
+                if inflight.error:
+                    raise inflight.error
+                yield from self._copy(inflight.value)
+                return
+
+        chunks: list[dict[str, Any]] = []
+        try:
+            for chunk in compute():
+                chunks.append(self._copy(chunk))
+                yield chunk
+        except BaseException as exc:
+            with self._lock:
+                self._inflight.pop(key, None)
+            with inflight.condition:
+                inflight.error = exc
+                inflight.done = True
+                inflight.condition.notify_all()
+            raise
+
+        expires_at = time.time() + int(settings.get("ttl_seconds") or 0)
+        with self._lock:
+            self._entries[key] = CacheEntry(expires_at=expires_at, value=self._copy(chunks))
+            self._prune_locked(time.time(), max_entries)
+            self._inflight.pop(key, None)
+        with inflight.condition:
+            inflight.value = self._copy(chunks)
+            inflight.done = True
+            inflight.condition.notify_all()
+
+
+chat_completion_cache = ChatCompletionCache()

services/protocol/conversation.py

@@ -246,6 +246,7 @@ class ConversationRequest:
 @dataclass
 class ConversationState:
     text: str = ""
+    raw_text: str = ""
     conversation_id: str = ""
     file_ids: list[str] = field(default_factory=list)
     sediment_ids: list[str] = field(default_factory=list)
@@ -312,7 +313,27 @@ def strip_history(text: str, history_text: str = "") -> str:
     return text
 
 
-def assistant_text(event: dict[str, Any], current_text: str = "", history_text: str = "") -> str:
+def sanitize_output_text(text: str) -> str:
+    text = str(text or "")
+
+    def replace_url(match: re.Match[str]) -> str:
+        label = match.group(1).strip()
+        url = match.group(2).strip()
+        if label and url.startswith(("http://", "https://")):
+            return f"{label} ({url})"
+        return label or url
+
+    # ChatGPT web sometimes returns rich annotation markers using private-use
+    # characters. API clients cannot render those, so convert links and drop
+    # citations before emitting OpenAI-compatible text.
+    text = re.sub(r"\ue200url\ue202([^\ue202\ue201]*)\ue202([^\ue201]*)\ue201", replace_url, text)
+    text = re.sub(r"\ue200cite\ue202[^\ue201]*\ue201", "", text)
+    text = re.sub(r"\ue200[^\ue201]*\ue201", "", text)
+    text = re.sub(r"\ue200[^\ue201]*$", "", text)
+    return text
+
+
+def assistant_raw_text(event: dict[str, Any], current_text: str = "", history_text: str = "") -> str:
     for candidate in (event, event.get("v")):
         if not isinstance(candidate, dict):
             continue
@@ -328,6 +349,10 @@ def assistant_text(event: dict[str, Any], current_text: str = "", history_text:
     return apply_text_patch(event, current_text, history_text)
 
 
+def assistant_text(event: dict[str, Any], current_text: str = "", history_text: str = "") -> str:
+    return sanitize_output_text(assistant_raw_text(event, current_text, history_text))
+
+
 def event_assistant_text(event: dict[str, Any], history_text: str = "") -> str:
     for candidate in (event, event.get("v")):
         if not isinstance(candidate, dict):
@@ -486,9 +511,12 @@ def iter_conversation_payloads(payloads: Iterator[str], history_text: str = "",
         update_conversation_state(state, payload, event)
         if history_index < len(history_messages) and event_assistant_text(event, history_text) == history_messages[history_index]:
             history_index += 1
+            state.raw_text = ""
             state.text = ""
             continue
-        next_text = assistant_text(event, state.text, history_text)
+        next_raw_text = assistant_raw_text(event, state.raw_text, history_text)
+        next_text = sanitize_output_text(next_raw_text)
+        state.raw_text = next_raw_text
         if next_text != state.text:
             delta = next_text[len(state.text):] if next_text.startswith(state.text) else next_text
             state.text = next_text

services/protocol/openai_v1_chat_complete.py

@@ -6,6 +6,7 @@ from typing import Any, Iterable, Iterator
 
 from fastapi import HTTPException
 
+from services.protocol.chat_completion_cache import cache_key, chat_completion_cache, normalize_text_messages
 from services.protocol.conversation import (
     ConversationRequest,
     ImageOutput,
@@ -28,6 +29,12 @@ from utils.image_tokens import (
     image_usage,
 )
 
+TOOL_UNAVAILABLE_SYSTEM_MESSAGE = (
+    "This compatibility backend cannot execute local tools, shell commands, web searches, "
+    "or file operations. Do not claim to have run tools or inspected external resources. "
+    "If a user asks you to use a tool, say that tool execution is unavailable through this backend."
+)
+
 
 def completion_chunk(model: str, delta: dict[str, Any], finish_reason: str | None = None, completion_id: str = "", created: int | None = None) -> dict[str, Any]:
     return {
@@ -66,10 +73,12 @@ def completion_response(
             "prompt_tokens_details": {
                 "text_tokens": prompt_text_tokens,
                 "image_tokens": prompt_image_tokens,
+                "cached_tokens": 0,
             },
             "completion_tokens_details": {
                 "text_tokens": completion_tokens,
                 "image_tokens": 0,
+                "reasoning_tokens": 0,
             },
         },
     }
@@ -127,7 +136,10 @@ def chat_image_args(body: dict[str, Any]) -> tuple[str, str, int, list[tuple[byt
 
 def text_chat_parts(body: dict[str, Any]) -> tuple[str, list[dict[str, Any]]]:
     model = str(body.get("model") or "auto").strip() or "auto"
-    messages = normalize_messages(chat_messages_from_body(body))
+    messages = normalize_text_messages(normalize_messages(chat_messages_from_body(body)))
+    tools = body.get("tools")
+    if isinstance(tools, list) and tools:
+        messages.insert(0, {"role": "system", "content": TOOL_UNAVAILABLE_SYSTEM_MESSAGE})
     return model, messages
 
 
@@ -200,9 +212,20 @@ def handle(body: dict[str, Any]) -> dict[str, Any] | Iterator[dict[str, Any]]:
         if is_image_chat_request(body):
             return image_chat_events(body)
         model, messages = text_chat_parts(body)
-        return stream_text_chat_completion(text_backend(), messages, model)
+        key = cache_key(body, messages, stream=True)
+        return chat_completion_cache.get_or_compute_stream(
+            key,
+            lambda: stream_text_chat_completion(text_backend(), messages, model),
+        )
     if is_image_chat_request(body):
         return image_chat_response(body)
     model, messages = text_chat_parts(body)
-    request = ConversationRequest(model=model, messages=messages)
-    return completion_response(model, collect_text(text_backend(), request), messages=messages)
+    key = cache_key(body, messages, stream=False)
+    return chat_completion_cache.get_or_compute_response(
+        key,
+        lambda: completion_response(
+            model,
+            collect_text(text_backend(), ConversationRequest(model=model, messages=messages)),
+            messages=messages,
+        ),
+    )

services/protocol/openai_v1_response.py

@@ -7,6 +7,7 @@ from typing import Any, Iterable, Iterator
 
 from fastapi import HTTPException
 
+from services.protocol.chat_completion_cache import cache_key, chat_completion_cache, normalize_text_messages
 from services.protocol.conversation import (
     ConversationRequest,
     ImageOutput,
@@ -25,11 +26,27 @@ from utils.image_tokens import (
     token_usage,
 )
 
+TOOL_UNAVAILABLE_SYSTEM_MESSAGE = (
+    "This compatibility backend cannot execute local tools, shell commands, web searches, "
+    "or file operations. Do not claim to have run tools or inspected external resources. "
+    "If a user asks you to use a tool, say that tool execution is unavailable through this backend."
+)
+
 
 def is_text_response_request(body: dict[str, Any]) -> bool:
     return not has_response_image_generation_tool(body)
 
 
+def has_non_image_tools(body: dict[str, Any]) -> bool:
+    tools = body.get("tools")
+    if not isinstance(tools, list):
+        return False
+    return any(
+        isinstance(tool, dict) and str(tool.get("type") or "").strip() != "image_generation"
+        for tool in tools
+    )
+
+
 def response_image_tool(body: dict[str, Any]) -> dict[str, object]:
     for tool in body.get("tools") or []:
         if isinstance(tool, dict) and tool.get("type") == "image_generation":
@@ -174,9 +191,17 @@ def response_completed(
     return response
 
 
-def stream_text_response(backend, body: dict[str, Any]) -> Iterator[dict[str, Any]]:
+def text_response_parts(body: dict[str, Any]) -> tuple[str, list[dict[str, Any]]]:
+    model = str(body.get("model") or "auto").strip() or "auto"
+    messages = normalize_text_messages(messages_from_input(body.get("input"), body.get("instructions")))
+    if has_non_image_tools(body):
+        messages.insert(0, {"role": "system", "content": TOOL_UNAVAILABLE_SYSTEM_MESSAGE})
+    return model, messages
+
+
+def stream_text_response(backend, body: dict[str, Any], messages: list[dict[str, Any]] | None = None) -> Iterator[dict[str, Any]]:
     model = str(body.get("model") or "auto").strip() or "auto"
-    messages = messages_from_input(body.get("input"), body.get("instructions"))
+    messages = messages if messages is not None else messages_from_input(body.get("input"), body.get("instructions"))
     response_id = f"resp_{uuid.uuid4().hex}"
     item_id = f"msg_{uuid.uuid4().hex}"
     created = int(time.time())
@@ -250,7 +275,12 @@ def collect_response(events: Iterable[dict[str, Any]]) -> dict[str, Any]:
 
 def response_events(body: dict[str, Any]) -> Iterator[dict[str, Any]]:
     if is_text_response_request(body):
-        yield from stream_text_response(text_backend(), body)
+        model, messages = text_response_parts(body)
+        key = cache_key(body, messages, stream=bool(body.get("stream")))
+        yield from chat_completion_cache.get_or_compute_stream(
+            key,
+            lambda: stream_text_response(text_backend(), body, messages),
+        )
         return
 
     prompt = extract_response_prompt(body.get("input"))

test/test_chat_completion_cache.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import unittest
+from unittest import mock
+import json
+
+from services.config import config
+from services.protocol import openai_v1_chat_complete, openai_v1_response
+from services.protocol.chat_completion_cache import chat_completion_cache
+from services.protocol.conversation import iter_conversation_payloads, sanitize_output_text
+
+
+class ChatCompletionCacheTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.old_cache_settings = config.data.get("chat_completion_cache")
+        config.data["chat_completion_cache"] = {
+            "enabled": True,
+            "ttl_seconds": 60,
+            "max_entries": 32,
+            "dedupe_inflight": True,
+            "stream_cache": True,
+            "normalize_messages": True,
+            "drop_adjacent_duplicates": True,
+            "drop_assistant_history": False,
+        }
+        chat_completion_cache.clear()
+
+    def tearDown(self) -> None:
+        if self.old_cache_settings is None:
+            config.data.pop("chat_completion_cache", None)
+        else:
+            config.data["chat_completion_cache"] = self.old_cache_settings
+        chat_completion_cache.clear()
+
+    def test_repeated_non_stream_text_completion_uses_cache(self) -> None:
+        calls = 0
+
+        def fake_collect_text(_backend, _request):
+            nonlocal calls
+            calls += 1
+            return f"cached answer {calls}"
+
+        body = {
+            "model": "auto",
+            "messages": [{"role": "user", "content": "cache this exact prompt"}],
+        }
+
+        with (
+            mock.patch("services.protocol.openai_v1_chat_complete.text_backend", return_value=object()),
+            mock.patch("services.protocol.openai_v1_chat_complete.collect_text", side_effect=fake_collect_text),
+        ):
+            first = openai_v1_chat_complete.handle(body)
+            second = openai_v1_chat_complete.handle(body)
+
+        self.assertEqual(calls, 1)
+        self.assertEqual(
+            first["choices"][0]["message"]["content"],
+            second["choices"][0]["message"]["content"],
+        )
+
+    def test_repeated_stream_text_completion_replays_cached_chunks(self) -> None:
+        calls = 0
+
+        def fake_stream_text_deltas(_backend, _request):
+            nonlocal calls
+            calls += 1
+            yield "streamed"
+            yield " answer"
+
+        body = {
+            "model": "auto",
+            "stream": True,
+            "messages": [{"role": "user", "content": "stream cache this exact prompt"}],
+        }
+
+        with (
+            mock.patch("services.protocol.openai_v1_chat_complete.text_backend", return_value=object()),
+            mock.patch(
+                "services.protocol.openai_v1_chat_complete.stream_text_deltas",
+                side_effect=fake_stream_text_deltas,
+            ),
+        ):
+            first = list(openai_v1_chat_complete.handle(body))
+            second = list(openai_v1_chat_complete.handle(body))
+
+        self.assertEqual(calls, 1)
+        self.assertEqual(first, second)
+        content = "".join(str(chunk["choices"][0]["delta"].get("content") or "") for chunk in second)
+        self.assertEqual(content, "streamed answer")
+
+    def test_adjacent_duplicate_messages_are_removed_before_upstream_call(self) -> None:
+        captured_messages = []
+
+        def fake_collect_text(_backend, request):
+            captured_messages.extend(request.messages or [])
+            return "ok"
+
+        body = {
+            "model": "auto",
+            "messages": [
+                {"role": "user", "content": "repeat me"},
+                {"role": "user", "content": "repeat me"},
+                {"role": "assistant", "content": "old answer"},
+                {"role": "user", "content": "next prompt"},
+            ],
+        }
+
+        with (
+            mock.patch("services.protocol.openai_v1_chat_complete.text_backend", return_value=object()),
+            mock.patch("services.protocol.openai_v1_chat_complete.collect_text", side_effect=fake_collect_text),
+        ):
+            openai_v1_chat_complete.handle(body)
+
+        self.assertEqual(
+            captured_messages,
+            [
+                {"role": "user", "content": "repeat me"},
+                {"role": "assistant", "content": "old answer"},
+                {"role": "user", "content": "next prompt"},
+            ],
+        )
+
+    def test_chat_completion_usage_includes_cached_tokens(self) -> None:
+        with (
+            mock.patch("services.protocol.openai_v1_chat_complete.text_backend", return_value=object()),
+            mock.patch("services.protocol.openai_v1_chat_complete.collect_text", return_value="ok"),
+        ):
+            response = openai_v1_chat_complete.handle({
+                "model": "auto",
+                "messages": [{"role": "user", "content": "usage shape"}],
+            })
+
+        details = response["usage"]["prompt_tokens_details"]
+        self.assertEqual(details["cached_tokens"], 0)
+        output_details = response["usage"]["completion_tokens_details"]
+        self.assertEqual(output_details["reasoning_tokens"], 0)
+
+    def test_responses_completed_usage_includes_cached_tokens(self) -> None:
+        with (
+            mock.patch("services.protocol.openai_v1_response.text_backend", return_value=object()),
+            mock.patch("services.protocol.openai_v1_response.stream_text_deltas", return_value=iter(["ok"])),
+        ):
+            response = openai_v1_response.handle({
+                "model": "auto",
+                "input": "usage shape",
+            })
+
+        details = response["usage"]["input_tokens_details"]
+        self.assertEqual(details["cached_tokens"], 0)
+        output_details = response["usage"]["output_tokens_details"]
+        self.assertEqual(output_details["reasoning_tokens"], 0)
+
+    def test_repeated_responses_text_request_uses_cache(self) -> None:
+        calls = 0
+
+        def fake_stream_text_deltas(_backend, _request):
+            nonlocal calls
+            calls += 1
+            yield f"response cache {calls}"
+
+        body = {
+            "model": "auto",
+            "input": "cache this responses prompt",
+            "stream": True,
+        }
+
+        with (
+            mock.patch("services.protocol.openai_v1_response.text_backend", return_value=object()),
+            mock.patch("services.protocol.openai_v1_response.stream_text_deltas", side_effect=fake_stream_text_deltas),
+        ):
+            first = list(openai_v1_response.handle(body))
+            second = list(openai_v1_response.handle(body))
+
+        self.assertEqual(calls, 1)
+        self.assertEqual(first, second)
+
+    def test_output_sanitizer_removes_chatgpt_annotation_markup(self) -> None:
+        text = (
+            "Repo: \ue200url\ue202basketikun/chatgpt2api"
+            "\ue202https://github.com/basketikun/chatgpt2api\ue201 "
+            "details \ue200cite\ue202turn0search0\ue201."
+        )
+
+        self.assertEqual(
+            sanitize_output_text(text),
+            "Repo: basketikun/chatgpt2api (https://github.com/basketikun/chatgpt2api) details .",
+        )
+
+    def test_stream_sanitizer_does_not_emit_partial_annotation_or_repeat_prefix(self) -> None:
+        events = [
+            {"p": "/message/content/parts/0", "o": "append", "v": "Repo: \ue200url\ue202chat"},
+            {"p": "/message/content/parts/0", "o": "append", "v": "gpt2api\ue202turn0search0\ue201 done \ue200cite\ue202turn0\ue201."},
+            "[DONE]",
+        ]
+        payloads = [json.dumps(event, ensure_ascii=False) if isinstance(event, dict) else event for event in events]
+        deltas = [
+            str(event.get("delta") or "")
+            for event in iter_conversation_payloads(iter(payloads))
+            if event.get("type") == "conversation.delta"
+        ]
+
+        self.assertEqual("".join(deltas), "Repo: chatgpt2api done .")
+        self.assertFalse(any("\ue200" in delta or "\ue202" in delta or "\ue201" in delta for delta in deltas))
+
+    def test_responses_tools_add_honest_no_tool_guard(self) -> None:
+        model, messages = openai_v1_response.text_response_parts({
+            "model": "auto",
+            "input": "run echo hi",
+            "tools": [{"type": "function", "name": "shell"}],
+        })
+
+        self.assertEqual(model, "auto")
+        self.assertEqual(messages[0]["role"], "system")
+        self.assertIn("cannot execute local tools", str(messages[0]["content"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

utils/image_tokens.py

@@ -302,10 +302,12 @@ def token_usage(
         "input_tokens_details": {
             "text_tokens": max(0, int(input_text_tokens or 0)),
             "image_tokens": max(0, int(input_image_tokens or 0)),
+            "cached_tokens": 0,
         },
         "output_tokens_details": {
             "text_tokens": max(0, int(output_text_tokens or 0)),
             "image_tokens": max(0, int(output_image_tokens or 0)),
+            "reasoning_tokens": 0,
         },
     }
 
@@ -334,9 +336,11 @@ def chat_usage_from_image_usage(usage: dict[str, Any]) -> dict[str, Any]:
         "prompt_tokens_details": {
             "text_tokens": int(input_details.get("text_tokens") or 0),
             "image_tokens": int(input_details.get("image_tokens") or 0),
+            "cached_tokens": int(input_details.get("cached_tokens") or 0),
         },
         "completion_tokens_details": {
             "text_tokens": int(output_details.get("text_tokens") or 0),
             "image_tokens": int(output_details.get("image_tokens") or 0),
+            "reasoning_tokens": int(output_details.get("reasoning_tokens") or 0),
         },
     }