test_proxy.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""proxy.py 的自测:用本地 mock 上游验证改写与透传逻辑。"""

from __future__ import annotations

import http.client
import json
import logging
import socket
import threading
import time
import unittest
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import proxy as proxy_module
from proxy import (
    ProxyHandler,
    ProxyServer,
    _BadRequest,
    _log_safe,
    _redact_secrets,
    build_chat_path,
    build_upstream_path,
    extract_upstream_error,
    infer_request_agent,
    normalize_agent_messages,
)
from responses_chat import (
    ChatUpstreamError,
    ChatToResponsesStream,
    chat_json_to_sse_events,
    chat_to_responses,
    model_needs_chat_conversion,
    responses_to_chat_request,
)


class MockUpstream(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True

    def __init__(self, addr):
        super().__init__(addr, MockUpstreamHandler)
        self.status = 200
        self.response_payload = {"ok": True}
        self.extra_headers = {}
        self.mode = "json"
        self.last = None
        self.sse_chunks = []
        self.sse_done = True


class MockUpstreamHandler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def do_POST(self):
        length = int(self.headers.get("Content-Length") or 0)
        body = self.rfile.read(length)
        server = self.server
        server.last = {
            "method": self.command,
            "path": self.path,
            "headers": dict(self.headers.items()),
            "body": body,
        }

        if server.mode == "sse":
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Transfer-Encoding", "chunked")
            self.end_headers()
            event = b'data: {"type":"response.output_text.delta","delta":"hello"}\n\n'
            parts = [event[:13], event[13:]]
            for part in parts:
                self.wfile.write(("%x\r\n" % len(part)).encode("ascii") + part + b"\r\n")
                self.wfile.flush()
            self.wfile.write(b"0\r\n\r\n")
            self.wfile.flush()
            return

        if server.mode == "close_sse":
            # close-delimited 流式响应:无 Content-Length、无 chunked,
            # 发送一个事件后保持连接一段时间再关闭。
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Connection", "close")
            self.end_headers()
            event = b'data: {"type":"response.output_text.delta","delta":"hi"}\n\n'
            self.wfile.write(event)
            self.wfile.flush()
            time.sleep(3)
            return

        if server.mode == "chat_sse":
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Transfer-Encoding", "chunked")
            self.end_headers()
            for chunk in server.sse_chunks:
                event = ("data: " + json.dumps(chunk) + "\n\n").encode("utf-8")
                self.wfile.write(("%x\r\n" % len(event)).encode("ascii") + event + b"\r\n")
                self.wfile.flush()
            if server.sse_done:
                done = b"data: [DONE]\n\n"
                self.wfile.write(("%x\r\n" % len(done)).encode("ascii") + done + b"\r\n")
            self.wfile.write(b"0\r\n\r\n")
            self.wfile.flush()
            return

        if server.mode == "chat_bad_json":
            payload = b"this is not json"
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)
            return

        if server.mode == "chat_sse_invalid":
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Transfer-Encoding", "chunked")
            self.end_headers()
            event = b"data: not-json\n\n"
            self.wfile.write(("%x\r\n" % len(event)).encode("ascii") + event + b"\r\n")
            self.wfile.write(b"0\r\n\r\n")
            self.wfile.flush()
            return

        if server.mode == "binary_error":
            payload = b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03binary-garbage-\xff\xfe"
            self.send_response(server.status)
            self.send_header("Content-Type", "application/octet-stream")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)
            return

        payload = json.dumps(server.response_payload).encode("utf-8")
        self.send_response(server.status)
        for key, value in server.extra_headers.items():
            self.send_header(key, value)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(payload)))
        self.end_headers()
        self.wfile.write(payload)

    def log_message(self, fmt, *args):
        pass


class ProxyTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.upstream = MockUpstream(("127.0.0.1", 0))
        cls.upstream_port = cls.upstream.server_address[1]
        threading.Thread(target=cls.upstream.serve_forever, daemon=True).start()

        cls.upstream_url = f"http://127.0.0.1:{cls.upstream_port}"
        cls.proxy = ProxyServer(
            ("127.0.0.1", 0),
            argparse_namespace(
                upstream=cls.upstream_url,
                chat_models=[],
                chat_upstream=None,
                fix_agent_message=False,
                timeout=10,
                client_timeout=10,
                max_connections=64,
                log_body=False,
            ),
        )
        cls.proxy_port = cls.proxy.server_address[1]
        threading.Thread(target=cls.proxy.serve_forever, daemon=True).start()

        cls.fix_proxy = ProxyServer(
            ("127.0.0.1", 0),
            argparse_namespace(
                upstream=cls.upstream_url,
                chat_models=[],
                chat_upstream=None,
                fix_agent_message=True,
                timeout=10,
                client_timeout=10,
                max_connections=64,
                log_body=False,
            ),
        )
        cls.fix_proxy_port = cls.fix_proxy.server_address[1]
        threading.Thread(target=cls.fix_proxy.serve_forever, daemon=True).start()

        cls.chat_proxy = ProxyServer(
            ("127.0.0.1", 0),
            argparse_namespace(
                upstream=cls.upstream_url,
                chat_models=["deepseek", "mimo"],
                chat_upstream=None,
                fix_agent_message=False,
                timeout=10,
                client_timeout=10,
                max_connections=64,
                log_body=False,
            ),
        )
        cls.chat_proxy_port = cls.chat_proxy.server_address[1]
        threading.Thread(target=cls.chat_proxy.serve_forever, daemon=True).start()

        cls.mimo_proxy = ProxyServer(
            ("127.0.0.1", 0),
            argparse_namespace(
                upstream=cls.upstream_url,
                chat_models=["mimo"],
                chat_upstream=None,
                fix_agent_message=True,
                timeout=10,
                client_timeout=10,
                max_connections=64,
                log_body=False,
            ),
        )
        cls.mimo_proxy_port = cls.mimo_proxy.server_address[1]
        threading.Thread(target=cls.mimo_proxy.serve_forever, daemon=True).start()

    @classmethod
    def tearDownClass(cls):
        cls.proxy.shutdown()
        cls.proxy.server_close()
        cls.fix_proxy.shutdown()
        cls.fix_proxy.server_close()
        cls.chat_proxy.shutdown()
        cls.chat_proxy.server_close()
        cls.mimo_proxy.shutdown()
        cls.mimo_proxy.server_close()
        cls.upstream.shutdown()
        cls.upstream.server_close()

    def post(self, path, payload, headers=None, proxy_port=None):
        conn = http.client.HTTPConnection(
            "127.0.0.1", proxy_port or self.proxy_port, timeout=10
        )
        body = json.dumps(payload).encode("utf-8")
        req_headers = {"Authorization": "Bearer test-key-123"}
        if headers:
            req_headers.update(headers)
        if not any(key.lower() == "content-type" for key in req_headers):
            req_headers["Content-Type"] = "application/json"
        conn.request("POST", path, body=body, headers=req_headers)
        resp = conn.getresponse()
        data = resp.read()
        conn.close()
        return resp.status, data

    def test_agent_message_rewritten(self):
        payload = {
            "model": "deepseek-v4-flash",
            "input": [
                {"type": "message", "role": "user",
                 "content": [{"type": "input_text", "text": "hello"}]},
                {
                    "type": "agent_message",
                    "id": "amsg_123",
                    "author": "/root",
                    "recipient": "/root/worker",
                    "content": [
                        {"type": "input_text",
                         "text": "Message Type: NEW_TASK\nTask name: /root/worker\nSender: /root\nPayload:\n"},
                        {"type": "encrypted_content",
                         "encrypted_content": "REPRO-TOKEN-1234"},
                    ],
                },
            ],
            "stream": False,
        }
        status, data = self.post(
            "/v1/responses", payload, proxy_port=self.fix_proxy_port
        )
        self.assertEqual(status, 200)

        last = self.upstream.last
        self.assertIsNotNone(last)
        self.assertEqual(last["path"], "/responses")
        self.assertEqual(last["headers"].get("Authorization"), "Bearer test-key-123")

        body = json.loads(last["body"])
        items = body["input"]
        self.assertEqual(items[0], payload["input"][0])
        self.assertEqual(items[1]["type"], "message")
        self.assertEqual(items[1]["role"], "user")
        self.assertEqual(items[1]["id"], "amsg_123")
        self.assertNotIn("author", items[1])
        self.assertNotIn("recipient", items[1])
        self.assertEqual(
            items[1]["content"],
            [
                {"type": "input_text",
                 "text": "Message Type: NEW_TASK\nTask name: /root/worker\nSender: /root\nPayload:\n"},
                {"type": "input_text", "text": "REPRO-TOKEN-1234"},
            ],
        )
        raw = last["body"].decode()
        self.assertNotIn("agent_message", raw)
        self.assertNotIn("encrypted_content", raw)

    def test_default_passthrough_keeps_agent_message(self):
        """裸启动(无任何开关)时 agent_message 原样透传,不做改写。"""
        payload = {
            "model": "deepseek-v4-flash",
            "input": [
                {
                    "type": "agent_message",
                    "id": "amsg_keep",
                    "author": "/root",
                    "recipient": "/root/worker",
                    "content": [
                        {"type": "encrypted_content", "encrypted_content": "SECRET-TASK"},
                    ],
                }
            ],
        }
        status, data = self.post("/v1/responses", payload)
        self.assertEqual(status, 200)
        raw = self.upstream.last["body"].decode()
        self.assertIn("agent_message", raw)
        self.assertIn("SECRET-TASK", raw)

    def test_chat_conversion_implicitly_normalizes_agent_message(self):
        """只开 chat 转换、不开 --fix-agent-message 时,agent_message 仍被改写。"""
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "id": "chatcmpl-1",
            "choices": [{"finish_reason": "stop",
                         "message": {"role": "assistant", "content": "ok"}}],
        }
        try:
            payload = {
                "model": "deepseek-v4-flash",
                "input": [
                    {
                        "type": "agent_message",
                        "id": "amsg_1",
                        "content": [
                            {"type": "input_text", "text": "task header"},
                            {"type": "encrypted_content", "encrypted_content": "task body"},
                        ],
                    }
                ],
            }
            status, data = self.post(
                "/v1/responses", payload, proxy_port=self.chat_proxy_port
            )
            self.assertEqual(status, 200)
            last = self.upstream.last
            self.assertEqual(last["path"], "/chat/completions")
            sent = json.loads(last["body"])
            self.assertEqual(sent["messages"][0]["role"], "user")
            self.assertEqual(sent["messages"][0]["content"], "task headertask body")
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_mimo_only_chat_conversion(self):
        """--chat-models mimo + --fix-agent-message:mimo 转换,deepseek 仅改写透传。"""
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "id": "chatcmpl-mimo",
            "created": 1700000000,
            "model": "mimo-v2.5",
            "choices": [{"finish_reason": "stop",
                         "message": {"role": "assistant", "content": "hi from mimo"}}],
        }
        try:
            status, data = self.post(
                "/v1/responses",
                {"model": "mimo-v2.5", "input": "hello"},
                proxy_port=self.mimo_proxy_port,
            )
            self.assertEqual(status, 200)
            self.assertEqual(self.upstream.last["path"], "/chat/completions")

            deepseek_payload = {
                "model": "deepseek-v4-flash",
                "input": [
                    {"type": "agent_message", "content": [
                        {"type": "encrypted_content", "encrypted_content": "ds-task"},
                    ]},
                ],
            }
            status, data = self.post(
                "/v1/responses", deepseek_payload, proxy_port=self.mimo_proxy_port
            )
            self.assertEqual(status, 200)
            self.assertEqual(self.upstream.last["path"], "/responses")
            raw = self.upstream.last["body"].decode()
            self.assertNotIn("agent_message", raw)
            self.assertIn("ds-task", raw)
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_request_log_shows_rewrite_and_proto(self):
        request_logger = logging.getLogger("responses-proxy")

        with self.assertLogs(request_logger, level="DEBUG") as captured:
            self.post(
                "/v1/responses",
                {
                    "model": "m",
                    "input": [
                        {"type": "agent_message", "author": "/root",
                         "recipient": "/root/worker", "content": [
                            {"type": "encrypted_content", "encrypted_content": "task"},
                        ]},
                    ],
                },
                proxy_port=self.fix_proxy_port,
            )
        lines = "\n".join(captured.output)
        self.assertIn("agent=/root/worker", lines)
        self.assertIn("proto=responses", lines)

        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "id": "chatcmpl-log",
            "created": 1700000000,
            "model": "deepseek-v4-flash",
            "choices": [{"finish_reason": "stop",
                         "message": {"role": "assistant", "content": "ok"}}],
        }
        try:
            with self.assertLogs(request_logger, level="DEBUG") as captured:
                self.post(
                    "/v1/responses",
                    {"model": "deepseek-v4-flash", "input": "hi"},
                    proxy_port=self.chat_proxy_port,
                )
            lines = "\n".join(captured.output)
            self.assertIn("agent=main", lines)
            self.assertIn("proto=chat", lines)
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_infer_request_agent(self):
        self.assertEqual(
            infer_request_agent({"input": [
                {"type": "agent_message", "author": "/root/worker",
                 "recipient": "/root"},
            ]}),
            "main",
        )
        self.assertEqual(
            infer_request_agent({"input": [
                {"type": "agent_message", "author": "/root",
                 "recipient": "/root/worker"},
            ]}),
            "/root/worker",
        )
        self.assertEqual(
            infer_request_agent({"input": [
                {"type": "agent_message", "author": "/root",
                 "recipient": "/root/worker"},
                {"type": "agent_message", "author": "/root/worker",
                 "recipient": "/root/worker/deep"},
            ]}),
            "/root/worker/deep",
        )
        self.assertEqual(infer_request_agent({"input": []}), "main")

    def test_path_without_v1_prefix(self):
        self.post("/responses", {"model": "m", "input": []})
        self.assertEqual(self.upstream.last["path"], "/responses")

    def test_normal_request_unchanged(self):
        payload = {
            "model": "deepseek-v4-flash",
            "input": [
                {"type": "message", "role": "user",
                 "content": [{"type": "input_text", "text": "ping"}]}
            ],
        }
        self.post("/v1/responses", payload)
        self.assertEqual(json.loads(self.upstream.last["body"]), payload)

    def test_lowercase_content_type_not_duplicated(self):
        payload = {"model": "m", "input": [{"type": "message", "role": "user",
                                            "content": [{"type": "input_text", "text": "hi"}]}]}
        self.post("/v1/responses", payload, headers={"content-type": "application/json"})

        content_types = [
            value
            for key, value in self.upstream.last["headers"].items()
            if key.lower() == "content-type"
        ]
        self.assertEqual(content_types, ["application/json"])

    def test_sse_stream_passthrough(self):
        self.upstream.mode = "sse"
        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.proxy_port, timeout=10)
            conn.request(
                "POST",
                "/v1/responses",
                body=json.dumps({"model": "m", "input": [], "stream": True}).encode(),
                headers={"Content-Type": "application/json"},
            )
            resp = conn.getresponse()
            data = resp.read()
            conn.close()
            self.assertEqual(resp.status, 200)
            self.assertEqual(resp.getheader("Content-Type"), "text/event-stream")
            self.assertIn(b"hello", data)
        finally:
            self.upstream.mode = "json"

    def test_upstream_error_passthrough(self):
        self.upstream.status = 400
        self.upstream.response_payload = {
            "error": {"message": "bad request", "type": "invalid_request_error"}
        }
        try:
            status, data = self.post("/v1/responses", {"model": "m", "input": []})
            self.assertEqual(status, 400)
            self.assertIn(b"invalid_request_error", data)
        finally:
            self.upstream.status = 200
            self.upstream.response_payload = {"ok": True}

    def test_build_upstream_path(self):
        self.assertEqual(build_upstream_path("/zen/go/v1", "/v1/responses"), "/zen/go/v1/responses")
        self.assertEqual(build_upstream_path("/zen/go/v1", "/responses"), "/zen/go/v1/responses")
        self.assertEqual(build_upstream_path("/v1", "/v1/responses"), "/v1/responses")
        self.assertEqual(build_upstream_path("/v1", "/responses"), "/v1/responses")
        self.assertEqual(build_upstream_path("/zen/go/v1", "/zen/go/v1/responses"), "/zen/go/v1/responses")

    def test_build_upstream_path_prefix_boundary(self):
        """base 前缀必须按路径段匹配,不能把 /v10 误当成 /v1。"""
        self.assertEqual(
            build_upstream_path("/v1", "/v10/responses"), "/v1/v10/responses"
        )

    def test_build_chat_path(self):
        self.assertEqual(build_chat_path("/zen/go/v1"), "/zen/go/v1/chat/completions")
        self.assertEqual(build_chat_path("/v1"), "/v1/chat/completions")
        self.assertEqual(build_chat_path(""), "/chat/completions")
        self.assertEqual(
            build_chat_path("https://api.deepseek.com/v1/chat/completions"),
            "https://api.deepseek.com/v1/chat/completions",
        )

    def test_model_needs_chat_conversion(self):
        self.assertTrue(model_needs_chat_conversion("deepseek-v4-flash", ["deepseek", "mimo"]))
        self.assertTrue(model_needs_chat_conversion("MIMO-v2.5", ["deepseek", "mimo"]))
        self.assertFalse(model_needs_chat_conversion("gpt-5.6", ["deepseek", "mimo"]))
        self.assertFalse(model_needs_chat_conversion(None, ["deepseek"]))
        self.assertFalse(model_needs_chat_conversion("deepseek-v4-flash", []))

    def test_normalize_unit(self):
        original = {
            "type": "agent_message",
            "id": "amsg_1",
            "author": "a",
            "recipient": "b",
            "content": [{"type": "encrypted_content", "encrypted_content": "task"}],
        }
        normalized = normalize_agent_messages(original)
        self.assertEqual(
            normalized,
            {
                "type": "message",
                "role": "user",
                "id": "amsg_1",
                "content": [{"type": "input_text", "text": "task"}],
            },
        )

    def test_responses_to_chat_request(self):
        payload = {
            "model": "deepseek-v4-flash",
            "instructions": [{"type": "input_text", "text": "be helpful"}],
            "input": [
                {"type": "message", "role": "user",
                 "content": [{"type": "input_text", "text": "hi"}]},
                {"type": "reasoning",
                 "content": [{"type": "reasoning_text", "text": "think first"}]},
                {"type": "function_call", "call_id": "call_1", "name": "get_weather",
                 "arguments": '{"city":"shanghai"}'},
                {"type": "function_call", "call_id": "call_2", "name": "get_time",
                 "arguments": '{"city":"shanghai"}'},
                {"type": "function_call_output", "call_id": "call_1", "output": "22c"},
                {"type": "message", "role": "developer",
                 "content": [{"type": "input_text", "text": "dev rule"}]},
            ],
            "tools": [
                {"type": "function", "function": {"name": "get_weather",
                                                  "description": "weather", "parameters": {"type": "object"}}},
                {"type": "custom", "name": "apply_patch", "description": "patch"},
                {"type": "web_search", "name": "web_search"},
            ],
            "tool_choice": {"type": "function", "name": "get_weather"},
            "max_output_tokens": 2048,
            "reasoning": {"effort": "high"},
            "stream": True,
            "store": True,
            "metadata": {"k": "v"},
        }
        out = responses_to_chat_request(payload)

        self.assertEqual(out["model"], "deepseek-v4-flash")
        self.assertEqual(out["max_tokens"], 2048)
        self.assertEqual(out["reasoning_effort"], "high")
        self.assertEqual(out["thinking"], {"type": "enabled"})
        self.assertEqual(out["stream_options"], {"include_usage": True})
        self.assertNotIn("store", out)
        self.assertNotIn("metadata", out)
        self.assertNotIn("max_output_tokens", out)
        self.assertEqual(out["tool_choice"], {"type": "function", "function": {"name": "get_weather"}})
        self.assertEqual(len(out["tools"]), 2)
        self.assertEqual(out["tools"][0]["function"]["name"], "get_weather")
        self.assertEqual(out["tools"][1]["function"]["name"], "apply_patch")

        self.assertEqual(out["messages"][0], {"role": "system", "content": "be helpful"})
        self.assertEqual(out["messages"][1], {"role": "user", "content": "hi"})
        assistant = out["messages"][2]
        self.assertEqual(assistant["role"], "assistant")
        self.assertEqual(assistant["reasoning_content"], "think first")
        self.assertEqual(len(assistant["tool_calls"]), 2)
        self.assertEqual(assistant["tool_calls"][0]["id"], "call_1")
        self.assertEqual(assistant["tool_calls"][1]["id"], "call_2")
        self.assertEqual(out["messages"][3], {"role": "tool", "tool_call_id": "call_1", "content": "22c"})
        self.assertEqual(out["messages"][4], {"role": "system", "content": "dev rule"})
        self.assertNotIn("n", out)

    def test_n_dropped_and_input_file_mapped(self):
        out = responses_to_chat_request({
            "model": "m",
            "n": 2,
            "input": [
                {"type": "message", "role": "user", "content": [
                    {"type": "input_text", "text": "look at "},
                    {"type": "input_file", "file_id": "file-abc", "filename": "a.txt"},
                ]},
            ],
        })
        self.assertNotIn("n", out)
        self.assertEqual(
            out["messages"][0]["content"],
            [
                {"type": "text", "text": "look at "},
                {"type": "file", "file": {"file_id": "file-abc", "filename": "a.txt"}},
            ],
        )

    def test_empty_input_gets_fallback_message(self):
        out = responses_to_chat_request({"model": "m", "input": []})
        self.assertEqual(out["messages"], [{"role": "user", "content": ""}])

    def test_flat_responses_function_tools(self):
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": "hi",
            "tools": [
                {"type": "function", "name": "get_weather",
                 "description": "weather",
                 "parameters": {"type": "object",
                                "properties": {"q": {"type": "string"}}},
                 "strict": False},
                {"type": "function", "name": "get_time"},
            ],
            "tool_choice": {"type": "function", "name": "get_weather"},
        })
        self.assertEqual(len(out["tools"]), 2)
        self.assertEqual(out["tools"][0]["function"]["name"], "get_weather")
        self.assertEqual(out["tools"][0]["function"]["strict"], False)
        self.assertEqual(out["tools"][1]["function"]["name"], "get_time")
        self.assertEqual(
            out["tools"][1]["function"]["parameters"],
            {"type": "object", "properties": {}},
        )
        self.assertEqual(
            out["tool_choice"],
            {"type": "function", "function": {"name": "get_weather"}},
        )

    def test_reasoning_content_merged_into_tool_call_assistant(self):
        """DeepSeek 同回合 reasoning + content + tool_calls 必须合并成一条 assistant。"""
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": [
                {"type": "message", "role": "user",
                 "content": [{"type": "input_text", "text": "hi"}]},
                {"type": "reasoning",
                 "content": [{"type": "reasoning_text", "text": "think A"}]},
                {"type": "message", "role": "assistant",
                 "content": [{"type": "output_text", "text": "I will check"}]},
                {"type": "function_call", "call_id": "call_1",
                 "name": "get_weather", "arguments": "{}"},
                {"type": "function_call_output", "call_id": "call_1", "output": "22c"},
            ],
        })
        self.assertEqual(
            [m["role"] for m in out["messages"]], ["user", "assistant", "tool"]
        )
        assistant = out["messages"][1]
        self.assertEqual(assistant["content"], "I will check")
        self.assertEqual(assistant["reasoning_content"], "think A")
        self.assertEqual(len(assistant["tool_calls"]), 1)
        self.assertEqual(assistant["tool_calls"][0]["id"], "call_1")

    def test_reasoning_blocks_accumulate(self):
        """多个 reasoning 块必须累加回传,不能覆盖丢失。"""
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": [
                {"type": "message", "role": "user", "content": "hi"},
                {"type": "reasoning", "content": [{"type": "reasoning_text", "text": "r1"}]},
                {"type": "function_call", "call_id": "c1", "name": "f1", "arguments": "{}"},
                {"type": "reasoning", "content": [{"type": "reasoning_text", "text": "r2"}]},
                {"type": "function_call", "call_id": "c2", "name": "f2", "arguments": "{}"},
                {"type": "function_call_output", "call_id": "c1", "output": "x"},
            ],
        })
        assistants = [m for m in out["messages"] if m["role"] == "assistant"]
        self.assertEqual(len(assistants), 1)
        self.assertEqual(assistants[0]["reasoning_content"], "r1r2")
        self.assertEqual(
            [c["function"]["name"] for c in assistants[0]["tool_calls"]], ["f1", "f2"]
        )

    def test_arguments_dict_encoded(self):
        """function_call 的 arguments 为对象时直接序列化本体,不能包成 {"input": null}。"""
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": [
                {"type": "function_call", "call_id": "c1", "name": "apply_patch",
                 "arguments": {"patch": "x"}},
                {"type": "function_call_output", "call_id": "c1", "output": "ok"},
            ],
        })
        self.assertEqual(
            out["messages"][0]["tool_calls"][0]["function"]["arguments"],
            '{"patch": "x"}',
        )

    def test_tool_search_items_dropped(self):
        """tool_search 工具被丢弃,输入里的 call/output 也应跳过,不能变空名函数。"""
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": [
                {"type": "tool_search_call", "call_id": "ts1",
                 "name": "tool_search", "arguments": "{}"},
                {"type": "tool_search_output", "call_id": "ts1", "output": "x"},
            ],
        })
        self.assertEqual(out["messages"], [{"role": "user", "content": ""}])
        self.assertNotIn("tool_calls", out["messages"][0])

    def test_dropped_tools_and_fields_aggregated_log(self):
        """同一请求里被丢弃的工具和字段各自聚合,不逐条刷屏。"""
        request_logger = logging.getLogger("responses-proxy")
        with self.assertLogs(request_logger, level="DEBUG") as captured:
            responses_to_chat_request({
                "model": "deepseek-v4-flash",
                "input": "hi",
                "tools": [
                    {"type": "namespace", "name": "n"},
                    {"type": "tool_search"},
                    {"type": "namespace", "name": "n2"},
                    {"type": "function", "name": "f", "parameters": {}},
                    {"type": "web_search"},
                ],
                "include": ["reasoning"],
                "store": False,
            })
        tool_lines = [
            line for line in captured.output
            if "丢弃 Chat 不支持的 tool" in line
        ]
        field_lines = [
            line for line in captured.output
            if "丢弃 Chat 协议不支持的字段" in line
        ]
        self.assertEqual(len(tool_lines), 1)
        self.assertEqual(len(field_lines), 1)
        self.assertIn("namespace、tool_search、web_search", tool_lines[0])
        self.assertIn("include、store", field_lines[0])

    def test_custom_tool_without_name_skipped(self):
        """custom 工具没有 name 时与 function 分支一致跳过,不能生成空名函数。"""
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": "hi",
            "tools": [
                {"type": "custom", "description": "no name"},
                {"type": "custom", "name": "apply_patch", "description": "patch"},
            ],
        })
        self.assertEqual(len(out["tools"]), 1)
        self.assertEqual(out["tools"][0]["function"]["name"], "apply_patch")

    def test_text_format_strict_only_when_explicit(self):
        """text.format 未显式给 strict 时不能静默强制 strict=True。"""
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": "hi",
            "text": {"format": {"type": "json_schema", "name": "s", "schema": {}}},
        })
        json_schema = out["response_format"]["json_schema"]
        self.assertNotIn("strict", json_schema)

        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": "hi",
            "text": {"format": {"type": "json_schema", "name": "s",
                                "schema": {}, "strict": False}},
        })
        self.assertIs(out["response_format"]["json_schema"]["strict"], False)

    def test_empty_choices_not_masked_as_completed(self):
        """200 但无 choices 是异常响应,不能伪装成 completed。"""
        with self.assertRaises(ChatUpstreamError):
            chat_to_responses({"id": "x", "choices": []}, request_model="deepseek-v4-flash")

        events = chat_json_to_sse_events(
            {"id": "x", "choices": []}, request_model="deepseek-v4-flash"
        )
        last = json.loads(events[-1].decode("utf-8").split("data: ", 1)[1])
        self.assertEqual(last["type"], "response.failed")
        self.assertEqual(last["response"]["error"]["code"], "invalid_upstream_response")

    def test_normalize_string_content_and_encrypted_content(self):
        """agent_message 字符串 content 不能丢;encrypted_content 非字符串也要处理。"""
        normalized = normalize_agent_messages({
            "type": "agent_message",
            "id": "a",
            "content": "直接字符串任务内容",
        })
        self.assertEqual(
            normalized["content"],
            [{"type": "input_text", "text": "直接字符串任务内容"}],
        )

        normalized = normalize_agent_messages(
            {"type": "encrypted_content", "encrypted_content": 12345}
        )
        self.assertEqual(normalized, {"type": "input_text", "text": "12345"})

        normalized = normalize_agent_messages({"type": "encrypted_content"})
        self.assertEqual(normalized, {"type": "input_text", "text": ""})

    def test_normalize_nesting_too_deep(self):
        """深嵌套请求不能触发 RecursionError,应显式 400。"""
        nested = {"leaf": True}
        for _ in range(150):
            nested = {"child": nested}
        with self.assertRaises(_BadRequest):
            normalize_agent_messages(nested)

    def test_log_safe_and_redact_secrets(self):
        self.assertEqual(_log_safe("a\nb\rc"), "a\\nb\\rc")
        self.assertEqual(_log_safe("bad \ufffd byte"), "bad \\ufffd byte")
        self.assertEqual(
            _redact_secrets("error: Bearer sk-abcdef123456"), "error: Bearer ***"
        )
        self.assertNotIn("sk-abcdef123456", _redact_secrets("token=sk-abcdef123456"))

    def test_extract_upstream_error_binary_body(self):
        raw = b"\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\x03\xff\x01\x02\x03"
        message = extract_upstream_error(404, raw)
        self.assertIn("HTTP 404", message)
        self.assertIn("gzip-compressed body", message)
        self.assertIn("14 bytes", message)
        self.assertNotIn("\ufffd", message)

    def test_binary_error_body_logged_as_summary(self):
        request_logger = logging.getLogger("responses-proxy")
        self.upstream.mode = "binary_error"
        self.upstream.status = 404
        try:
            with self.assertLogs(request_logger, level="DEBUG") as captured:
                self.post(
                    "/robots.txt",
                    {"model": "m", "input": []},
                )
            lines = "\n".join(captured.output)
            self.assertIn("error=upstream returned HTTP 404 (gzip-compressed body", lines)
            self.assertNotIn("\ufffd", lines)
        finally:
            self.upstream.mode = "json"
            self.upstream.status = 200
            self.upstream.response_payload = {"ok": True}

    def test_chat_json_fallback_keeps_multiple_tool_calls(self):
        """上游忽略 stream=true 返回整段 JSON 时,多个 tool_calls 不能塌陷成一个。"""
        chat = {
            "id": "chatcmpl-multi",
            "model": "deepseek-v4-flash",
            "choices": [{
                "finish_reason": "tool_calls",
                "message": {"role": "assistant", "content": "", "tool_calls": [
                    {"id": "call_1", "type": "function",
                     "function": {"name": "get_weather", "arguments": '{"q":"sh"}'}},
                    {"id": "call_2", "type": "function",
                     "function": {"name": "get_time", "arguments": '{"city":"sh"}'}},
                ]},
            }],
        }
        events = chat_json_to_sse_events(chat, request_model="deepseek-v4-flash")
        calls = []
        for event in events:
            data = json.loads(event.decode("utf-8").split("data: ", 1)[1])
            if data["type"] == "response.output_item.done" and \
                    data.get("item", {}).get("type") == "function_call":
                item = data["item"]
                calls.append((item["name"], item["arguments"]))
        self.assertEqual(
            calls,
            [("get_weather", '{"q":"sh"}'), ("get_time", '{"city":"sh"}')],
        )

    def test_stream_tool_calls_without_index_continue_single_tool(self):
        """流式 tool delta 无 index 时,无标识续片接续最近打开的调用。"""
        t = ChatToResponsesStream("deepseek-v4-flash")
        events = t.feed_chunk({
            "choices": [{"delta": {"tool_calls": [
                {"id": "call_a", "type": "function",
                 "function": {"name": "f1", "arguments": "{"}}
            ]}, "finish_reason": None}]
        })
        events += t.feed_chunk({
            "choices": [{"delta": {"tool_calls": [
                {"type": "function", "function": {"arguments": '"a":1}'}}
            ]}, "finish_reason": "tool_calls"}]
        })
        events += t.finish()
        calls = []
        for event in events:
            data = json.loads(event.decode("utf-8").split("data: ", 1)[1])
            if data["type"] == "response.output_item.done" and \
                    data.get("item", {}).get("type") == "function_call":
                item = data["item"]
                calls.append((item["name"], item["arguments"]))
        self.assertEqual(calls, [("f1", '{"a":1}')])

    def test_translator_output_cap_fails(self):
        """累积输出超过上限时发 response.failed,不伪装成功。"""
        t = ChatToResponsesStream("deepseek-v4-flash", max_output_bytes=10)
        events = t.feed_chunk({
            "choices": [{"delta": {"content": "x" * 11}, "finish_reason": None}]
        })
        types = [
            json.loads(event.decode("utf-8").split("data: ", 1)[1])["type"]
            for event in events
        ]
        self.assertEqual(types[-1], "response.failed")
        self.assertIn("output_too_large", events[-1].decode("utf-8"))
        self.assertNotIn("response.completed", events[-1].decode("utf-8"))

    def test_stream_tool_call_id_preserved(self):
        """流式 tool delta 显式给出 id 时,function_call 的 call_id 应沿用上游 id。"""
        t = ChatToResponsesStream("deepseek-v4-flash")
        events = t.feed_chunk({
            "choices": [{"delta": {"tool_calls": [
                {"index": 0, "id": "call_upstream_1", "type": "function",
                 "function": {"name": "f1", "arguments": "{}"}}
            ]}, "finish_reason": "tool_calls"}]
        })
        events += t.finish()
        call_ids = []
        for event in events:
            data = json.loads(event.decode("utf-8").split("data: ", 1)[1])
            if data["type"] == "response.output_item.done" and \
                    data.get("item", {}).get("type") == "function_call":
                call_ids.append(data["item"]["call_id"])
        self.assertEqual(call_ids, ["call_upstream_1"])

    def test_message_after_tool_closes_tool(self):
        """content 出现在 tool_calls 之后时,先关闭 tool 再开 message,不能同时 in_progress。"""
        t = ChatToResponsesStream("deepseek-v4-flash")
        events = t.feed_chunk({
            "choices": [{"delta": {"tool_calls": [
                {"index": 0, "id": "call_a", "function": {"name": "f1", "arguments": "{}"}}
            ]}, "finish_reason": None}]
        })
        events += t.feed_chunk({
            "choices": [{"delta": {"content": "after tool"}, "finish_reason": None}]
        })
        opened = []
        done = []
        for event in events:
            data = json.loads(event.decode("utf-8").split("data: ", 1)[1])
            if data["type"] == "response.output_item.added":
                opened.append(data["item"]["type"])
            elif data["type"] == "response.output_item.done":
                done.append(data["item"]["type"])
        self.assertEqual(opened, ["function_call", "message"])
        self.assertEqual(done, ["function_call"])

    def test_failed_event_includes_partial_output(self):
        """流中途 fail 时,已产生的部分输出要进入 failed 快照。"""
        t = ChatToResponsesStream("deepseek-v4-flash")
        t.feed_chunk({
            "choices": [{"delta": {"content": "partial"}, "finish_reason": None}]
        })
        events = t.fail("proxy_error", "boom")
        last = json.loads(events[-1].decode("utf-8").split("data: ", 1)[1])
        self.assertEqual(last["type"], "response.failed")
        self.assertEqual(last["response"]["output"][0]["type"], "message")
        self.assertEqual(
            last["response"]["output"][0]["content"][0]["text"], "partial"
        )

    def test_extract_sse_block_empty_data_skipped(self):
        """空 data: 事件是合法 keepalive,应跳过而不是让整条流失败。"""
        buffer = bytearray(b"data:\n\ndata: {\"x\":1}\n\n")
        self.assertEqual(
            ProxyHandler._extract_sse_block(buffer), b'{"x":1}'
        )
        self.assertEqual(buffer, bytearray())

    def test_mimo_reasoning_uses_thinking_not_effort(self):
        out = responses_to_chat_request({
            "model": "mimo-v2.5",
            "input": "hi",
            "reasoning": {"effort": "max"},
        })
        self.assertEqual(out["thinking"], {"type": "enabled"})
        self.assertNotIn("reasoning_effort", out)

        out = responses_to_chat_request({
            "model": "mimo-v2.5",
            "input": "hi",
            "reasoning": {"effort": "none"},
        })
        self.assertEqual(out["thinking"], {"type": "disabled"})
        self.assertNotIn("reasoning_effort", out)

    def test_deepseek_reasoning_effort_mapping(self):
        out = responses_to_chat_request({
            "model": "deepseek-v4-pro",
            "input": "hi",
            "reasoning": {"effort": "low"},
        })
        self.assertEqual(out["reasoning_effort"], "high")
        self.assertEqual(out["thinking"], {"type": "enabled"})

        out = responses_to_chat_request({
            "model": "deepseek-v4-pro",
            "input": "hi",
            "reasoning": {"effort": "xhigh"},
        })
        self.assertEqual(out["reasoning_effort"], "max")

    def test_deepseek_flash_reasoning_effort_mapping(self):
        """flash 与 pro 的 effort 映射不同:flash low 保持 low、xhigh 降到 high。"""
        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": "hi",
            "reasoning": {"effort": "low"},
        })
        self.assertEqual(out["reasoning_effort"], "low")

        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": "hi",
            "reasoning": {"effort": "xhigh"},
        })
        self.assertEqual(out["reasoning_effort"], "high")

        out = responses_to_chat_request({
            "model": "deepseek-v4-flash",
            "input": "hi",
            "reasoning": {"effort": "max"},
        })
        self.assertEqual(out["reasoning_effort"], "max")

    def test_tool_choice_dropped_tool_falls_back_auto(self):
        out = responses_to_chat_request({
            "model": "mimo-v2.5",
            "input": "hi",
            "tools": [
                {"type": "function", "function": {"name": "get_weather",
                                                  "description": "", "parameters": {}}},
                {"type": "web_search", "name": "web_search"},
            ],
            "tool_choice": {"type": "function", "name": "web_search"},
        })
        self.assertEqual(len(out["tools"]), 1)
        self.assertEqual(out["tool_choice"], "auto")

        out = responses_to_chat_request({
            "model": "mimo-v2.5",
            "input": "hi",
            "tools": [{"type": "web_search", "name": "web_search"}],
            "tool_choice": {"type": "function", "name": "web_search"},
        })
        self.assertNotIn("tools", out)
        self.assertNotIn("tool_choice", out)

    def test_chat_to_responses(self):
        chat = {
            "id": "chatcmpl-abc",
            "created": 1700000000,
            "model": "deepseek-v4-flash",
            "choices": [{
                "finish_reason": "length",
                "message": {
                    "role": "assistant",
                    "reasoning_content": "let me think",
                    "content": "partial answer",
                    "tool_calls": [
                        {"id": "call_x", "type": "function",
                         "function": {"name": "get_weather", "arguments": '{"q":"sh"}'}}
                    ],
                },
            }],
            "usage": {
                "prompt_tokens": 10,
                "completion_tokens": 20,
                "total_tokens": 30,
                "prompt_tokens_details": {"cached_tokens": 4},
                "completion_tokens_details": {"reasoning_tokens": 8},
            },
        }
        resp = chat_to_responses(chat, request_model="deepseek-v4-flash")
        self.assertEqual(resp["id"], "chatcmpl-abc")
        self.assertEqual(resp["object"], "response")
        self.assertEqual(resp["status"], "incomplete")
        self.assertEqual(resp["incomplete_details"], {"reason": "max_output_tokens"})
        self.assertEqual(resp["model"], "deepseek-v4-flash")
        self.assertEqual([item["type"] for item in resp["output"]],
                         ["reasoning", "message", "function_call"])
        self.assertEqual(resp["output"][0]["content"][0]["text"], "let me think")
        self.assertEqual(resp["output"][1]["content"][0]["text"], "partial answer")
        self.assertEqual(resp["output"][2]["name"], "get_weather")
        self.assertEqual(resp["output"][2]["call_id"], "call_x")
        self.assertEqual(resp["usage"]["input_tokens"], 10)
        self.assertEqual(resp["usage"]["output_tokens"], 20)
        self.assertEqual(resp["usage"]["input_tokens_details"]["cached_tokens"], 4)
        self.assertEqual(resp["usage"]["output_tokens_details"]["reasoning_tokens"], 8)

    def test_stream_translator(self):
        t = ChatToResponsesStream("deepseek-v4-flash")
        chunks = [
            {"choices": [{"delta": {"role": "assistant", "content": ""}, "finish_reason": None}]},
            {"choices": [{"delta": {"reasoning_content": "think "}, "finish_reason": None}]},
            {"choices": [{"delta": {"reasoning_content": "more"}, "finish_reason": None}]},
            {"choices": [{"delta": {"content": "Hello"}, "finish_reason": None}]},
            {"choices": [{"delta": {"content": " world"}, "finish_reason": None}]},
            {"choices": [{"delta": {"tool_calls": [
                {"index": 0, "id": "call_9", "type": "function",
                 "function": {"name": "get_weather", "arguments": '{"q":'}}
            ]}, "finish_reason": None}]},
            {"choices": [{"delta": {"tool_calls": [
                {"index": 0, "id": "call_9", "type": "function",
                 "function": {"arguments": '"sh"}'}}
            ]}, "finish_reason": None}]},
            {"choices": [{"delta": {}, "finish_reason": "tool_calls"}]},
            {"choices": [], "usage": {"prompt_tokens": 5, "completion_tokens": 7,
                                      "total_tokens": 12,
                                      "completion_tokens_details": {"reasoning_tokens": 3}}},
        ]
        events = []
        for chunk in chunks:
            events.extend(t.feed_chunk(chunk))
        events.extend(t.finish())

        parsed = []
        for ev in events:
            text = ev.decode("utf-8")
            event_line = [line for line in text.splitlines() if line.startswith("event: ")][0]
            parsed.append(event_line.split("event: ")[1].strip())

        self.assertEqual(parsed[0], "response.created")
        self.assertEqual(parsed[1], "response.in_progress")
        self.assertIn("response.output_item.added", parsed)
        self.assertIn("response.reasoning_text.delta", parsed)
        self.assertIn("response.output_text.delta", parsed)
        self.assertIn("response.function_call_arguments.delta", parsed)
        self.assertEqual(parsed[-1], "response.completed")

        # created 与 completed 的 response.id 必须一致
        created = json.loads(events[0].decode("utf-8").split("data: ", 1)[1].strip())
        completed = json.loads(events[-1].decode("utf-8").split("data: ", 1)[1].strip())
        self.assertEqual(created["response"]["id"], completed["response"]["id"])
        self.assertEqual(completed["response"]["status"], "completed")
        self.assertEqual([item["type"] for item in completed["response"]["output"]],
                         ["reasoning", "message", "function_call"])
        self.assertEqual(completed["response"]["usage"]["input_tokens"], 5)
        self.assertEqual(completed["response"]["usage"]["output_tokens_details"]["reasoning_tokens"], 3)
        self.assertEqual(completed["response"]["output"][2]["arguments"], '{"q":"sh"}')

    def test_chat_json_integration(self):
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "id": "chatcmpl-123",
            "created": 1700000000,
            "model": "deepseek-v4-flash",
            "choices": [{
                "finish_reason": "stop",
                "message": {"role": "assistant", "content": "converted answer"},
            }],
            "usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
        }
        try:
            status, data = self.post(
                "/v1/responses",
                {"model": "deepseek-v4-flash", "input": "hi"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(status, 200)
            body = json.loads(data)
            self.assertEqual(body["object"], "response")
            self.assertEqual(body["status"], "completed")
            self.assertEqual(body["output"][0]["content"][0]["text"], "converted answer")
            self.assertEqual(body["usage"]["input_tokens"], 1)

            last = self.upstream.last
            self.assertEqual(last["path"], "/chat/completions")
            sent = json.loads(last["body"])
            self.assertNotIn("input", sent)
            self.assertEqual(sent["messages"], [{"role": "user", "content": "hi"}])
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_chat_sse_integration(self):
        self.upstream.mode = "chat_sse"
        self.upstream.sse_chunks = [
            {"choices": [{"delta": {"role": "assistant", "content": ""}, "finish_reason": None}]},
            {"choices": [{"delta": {"content": "streamed "}, "finish_reason": None}]},
            {"choices": [{"delta": {"content": "answer"}, "finish_reason": None}]},
            {"choices": [{"delta": {}, "finish_reason": "stop"}]},
            {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3, "total_tokens": 5}},
        ]
        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.chat_proxy_port, timeout=10)
            conn.request(
                "POST",
                "/v1/responses",
                body=json.dumps({
                    "model": "deepseek-v4-flash",
                    "input": [{"type": "message", "role": "user", "content": "hi"}],
                    "stream": True,
                }).encode(),
                headers={"Content-Type": "application/json"},
            )
            resp = conn.getresponse()
            data = resp.read()
            conn.close()
            self.assertEqual(resp.status, 200)
            self.assertEqual(resp.getheader("Content-Type"), "text/event-stream")
            text = data.decode("utf-8")
            self.assertIn("event: response.created", text)
            self.assertIn("event: response.in_progress", text)
            self.assertIn("event: response.output_text.delta", text)
            self.assertIn('"delta": "streamed "', text)
            self.assertIn("event: response.completed", text)
            self.assertIn('"status": "completed"', text)
            self.assertNotIn("[DONE]", text)
        finally:
            self.upstream.mode = "json"
            self.upstream.sse_chunks = []

    def test_chat_json_as_sse_fallback(self):
        """上游忽略 stream=true 返回整段 JSON 时,代理合成 SSE 事件。"""
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "id": "chatcmpl-fallback",
            "created": 1700000000,
            "model": "deepseek-v4-flash",
            "choices": [{
                "finish_reason": "stop",
                "message": {"role": "assistant", "content": "whole json answer"},
            }],
            "usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
        }
        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.chat_proxy_port, timeout=10)
            conn.request(
                "POST",
                "/v1/responses",
                body=json.dumps({
                    "model": "deepseek-v4-flash",
                    "input": "hi",
                    "stream": True,
                }).encode(),
                headers={"Content-Type": "application/json"},
            )
            resp = conn.getresponse()
            data = resp.read()
            conn.close()
            self.assertEqual(resp.status, 200)
            self.assertEqual(resp.getheader("Content-Type"), "text/event-stream")
            text = data.decode("utf-8")
            self.assertIn("event: response.created", text)
            self.assertIn("event: response.output_text.delta", text)
            self.assertIn('"delta": "whole json answer"', text)
            self.assertIn("event: response.completed", text)
            self.assertNotIn("[DONE]", text)
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_chat_json_error_as_failed_sse(self):
        """stream=true 时上游返回 200 + error envelope,必须转成 failed。"""
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "error": {"message": "insufficient balance", "type": "insufficient_quota"}
        }
        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.chat_proxy_port, timeout=10)
            conn.request(
                "POST",
                "/v1/responses",
                body=json.dumps({
                    "model": "deepseek-v4-flash",
                    "input": "hi",
                    "stream": True,
                }).encode(),
                headers={"Content-Type": "application/json"},
            )
            resp = conn.getresponse()
            data = resp.read()
            conn.close()
            text = data.decode("utf-8")
            self.assertIn("event: response.failed", text)
            self.assertIn('"status": "failed"', text)
            self.assertIn("insufficient balance", text)
            self.assertNotIn("response.completed", text)
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_chat_non_json_fallback_passthrough(self):
        self.upstream.mode = "chat_bad_json"
        try:
            status, data = self.post(
                "/v1/responses",
                {"model": "deepseek-v4-flash", "input": "hi"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(status, 200)
            self.assertEqual(data, b"this is not json")
        finally:
            self.upstream.mode = "json"

    def test_chat_stream_truncation_reports_incomplete(self):
        self.upstream.mode = "chat_sse"
        self.upstream.sse_chunks = [
            {"choices": [{"delta": {"content": "partial"}, "finish_reason": None}]},
        ]
        self.upstream.sse_done = False
        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.chat_proxy_port, timeout=10)
            conn.request(
                "POST",
                "/v1/responses",
                body=json.dumps({
                    "model": "deepseek-v4-flash",
                    "input": "hi",
                    "stream": True,
                }).encode(),
                headers={"Content-Type": "application/json"},
            )
            resp = conn.getresponse()
            data = resp.read()
            conn.close()
            text = data.decode("utf-8")
            self.assertIn("event: response.incomplete", text)
            self.assertIn('"reason": "max_output_tokens"', text)
            self.assertIn("partial", text)
        finally:
            self.upstream.mode = "json"
            self.upstream.sse_chunks = []
            self.upstream.sse_done = True

    def test_chat_accept_encoding_identity(self):
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "id": "chatcmpl-1",
            "created": 1700000000,
            "model": "deepseek-v4-flash",
            "choices": [{"finish_reason": "stop",
                         "message": {"role": "assistant", "content": "ok"}}],
        }
        try:
            self.post(
                "/v1/responses",
                {"model": "deepseek-v4-flash", "input": "hi"},
                headers={"Accept-Encoding": "gzip"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(self.upstream.last["headers"].get("Accept-Encoding"), "identity")
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_chat_3xx_passthrough(self):
        self.upstream.status = 302
        self.upstream.response_payload = {"redirect": True}
        try:
            status, data = self.post(
                "/v1/responses",
                {"model": "deepseek-v4-flash", "input": "hi"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(status, 302)
            self.assertIn(b"redirect", data)
        finally:
            self.upstream.status = 200
            self.upstream.response_payload = {"ok": True}

    def test_chat_endpoint_with_query_converted(self):
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "id": "chatcmpl-q",
            "created": 1700000000,
            "model": "deepseek-v4-flash",
            "choices": [{"finish_reason": "stop",
                         "message": {"role": "assistant", "content": "ok"}}],
        }
        try:
            status, data = self.post(
                "/v1/responses?beta=1",
                {"model": "deepseek-v4-flash", "input": "hi"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(status, 200)
            self.assertEqual(self.upstream.last["path"], "/chat/completions")
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_failed_finish_reason(self):
        chat = {
            "id": "chatcmpl-f",
            "created": 1700000000,
            "model": "deepseek-v4-flash",
            "choices": [{"finish_reason": "insufficient_system_resource",
                         "message": {"role": "assistant", "content": ""}}],
        }
        resp = chat_to_responses(chat, request_model="deepseek-v4-flash")
        self.assertEqual(resp["status"], "failed")
        self.assertIsNotNone(resp["error"])

        t = ChatToResponsesStream("deepseek-v4-flash")
        t.feed_chunk({"choices": [{"delta": {}, "finish_reason": "insufficient_system_resource"}]})
        events = t.finish()
        last = json.loads(events[-1].decode("utf-8").split("data: ", 1)[1].strip())
        self.assertEqual(last["type"], "response.failed")
        self.assertEqual(last["response"]["error"]["code"], "insufficient_system_resource")

    def test_error_envelope_message_extracted(self):
        t = ChatToResponsesStream("deepseek-v4-flash")
        events = t.feed_chunk({"error": {"message": "boom", "type": "server_error"}})
        last = json.loads(events[-1].decode("utf-8").split("data: ", 1)[1].strip())
        self.assertEqual(last["response"]["error"]["message"], "boom")

    def test_chat_error_passthrough(self):
        self.upstream.status = 400
        self.upstream.response_payload = {
            "error": {"message": "bad model", "type": "invalid_request_error"}
        }
        try:
            status, data = self.post(
                "/v1/responses",
                {"model": "deepseek-v4-flash", "input": "hi"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(status, 400)
            self.assertIn(b"invalid_request_error", data)
            self.assertEqual(self.upstream.last["path"], "/chat/completions")
        finally:
            self.upstream.status = 200
            self.upstream.response_payload = {"ok": True}

    def test_upstream_error_message_in_request_log(self):
        request_logger = logging.getLogger("responses-proxy")
        self.upstream.status = 400
        self.upstream.response_payload = {
            "error": {"message": "boom detail", "type": "invalid_request_error"}
        }
        try:
            with self.assertLogs(request_logger, level="DEBUG") as captured:
                self.post(
                    "/v1/responses",
                    {"model": "deepseek-v4-flash", "input": "hi"},
                    proxy_port=self.chat_proxy_port,
                )
            lines = "\n".join(captured.output)
            self.assertIn("error=boom detail", lines)
        finally:
            self.upstream.status = 200
            self.upstream.response_payload = {"ok": True}

    def test_chat_stream_invalid_sse_logs_error(self):
        request_logger = logging.getLogger("responses-proxy")
        self.upstream.mode = "chat_sse_invalid"
        try:
            with self.assertLogs(request_logger, level="DEBUG") as captured:
                self.post(
                    "/v1/responses",
                    {"model": "deepseek-v4-flash", "input": "hi", "stream": True},
                    proxy_port=self.chat_proxy_port,
                )
            lines = "\n".join(captured.output)
            self.assertIn("error=upstream sent an invalid SSE data line", lines)
        finally:
            self.upstream.mode = "json"

    def test_content_length_preserved_passthrough(self):
        """上游带 Content-Length 的 2xx 响应不能被改成 chunked。"""
        self.upstream.response_payload = {"ok": True}
        conn = http.client.HTTPConnection("127.0.0.1", self.proxy_port, timeout=10)
        conn.request(
            "POST",
            "/v1/responses",
            body=json.dumps({"model": "m", "input": []}).encode(),
            headers={"Content-Type": "application/json"},
        )
        resp = conn.getresponse()
        data = resp.read()
        conn.close()
        self.assertEqual(resp.status, 200)
        self.assertEqual(resp.getheader("Content-Length"), str(len(data)))
        self.assertIsNone(resp.getheader("Transfer-Encoding"))

    def test_error_response_headers_preserved(self):
        """429/401 等错误响应必须保留 Retry-After/WWW-Authenticate 等关键头。"""
        self.upstream.status = 429
        self.upstream.response_payload = {
            "error": {"message": "rate limited", "type": "rate_limit"}
        }
        self.upstream.extra_headers = {
            "Retry-After": "30",
            "WWW-Authenticate": 'Bearer realm="proxy"',
        }
        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.proxy_port, timeout=10)
            conn.request(
                "POST",
                "/v1/responses",
                body=json.dumps({"model": "m", "input": []}).encode(),
                headers={"Content-Type": "application/json"},
            )
            resp = conn.getresponse()
            resp.read()
            conn.close()
            self.assertEqual(resp.status, 429)
            self.assertEqual(resp.getheader("Retry-After"), "30")
            self.assertEqual(resp.getheader("WWW-Authenticate"), 'Bearer realm="proxy"')
        finally:
            self.upstream.status = 200
            self.upstream.response_payload = {"ok": True}
            self.upstream.extra_headers = {}

    def test_error_body_truncation_marked(self):
        """超过 1MB 的错误体被截断时必须带 X-Upstream-Body-Truncated 标记。"""
        big = "x" * (1024 * 1024 + 100)
        self.upstream.status = 400
        self.upstream.response_payload = {
            "error": {"message": big, "type": "too_big"}
        }
        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.proxy_port, timeout=10)
            conn.request(
                "POST",
                "/v1/responses",
                body=json.dumps({"model": "m", "input": []}).encode(),
                headers={"Content-Type": "application/json"},
            )
            resp = conn.getresponse()
            data = resp.read()
            conn.close()
            self.assertEqual(resp.status, 400)
            self.assertEqual(resp.getheader("X-Upstream-Body-Truncated"), "true")
            self.assertLessEqual(len(data), 1024 * 1024)
        finally:
            self.upstream.status = 200
            self.upstream.response_payload = {"ok": True}

    def test_chat_json_body_too_large_returns_502(self):
        """非流式 Chat JSON 响应体超过上限时返回 502,不能无界读入。"""
        old_limit = proxy_module.MAX_CHAT_RESPONSE_BODY
        proxy_module.MAX_CHAT_RESPONSE_BODY = 1024
        try:
            self.upstream.mode = "chat_json"
            self.upstream.response_payload = {
                "id": "chatcmpl-big",
                "choices": [{
                    "finish_reason": "stop",
                    "message": {"role": "assistant", "content": "y" * 4096},
                }],
            }
            status, data = self.post(
                "/v1/responses",
                {"model": "deepseek-v4-flash", "input": "hi"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(status, 502)
            self.assertIn(b"upstream_response_too_large", data)
        finally:
            proxy_module.MAX_CHAT_RESPONSE_BODY = old_limit
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_chat_empty_choices_failed_response(self):
        """Chat 上游 200 但无 choices 时,客户端应收到 failed 而非空 completed。"""
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {"id": "chatcmpl-empty"}
        try:
            status, data = self.post(
                "/v1/responses",
                {"model": "deepseek-v4-flash", "input": "hi"},
                proxy_port=self.chat_proxy_port,
            )
            self.assertEqual(status, 200)
            body = json.loads(data)
            self.assertEqual(body["status"], "failed")
            self.assertEqual(
                body["error"]["code"], "invalid_upstream_response"
            )
            self.assertIn("no choices", body["error"]["message"])
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}

    def test_connection_named_header_stripped(self):
        """Connection 点名的自定义头属于逐跳头,不能转发给上游。"""
        self.post(
            "/v1/responses",
            {"model": "m", "input": []},
            headers={"Connection": "x-hop", "x-hop": "secret"},
        )
        headers = self.upstream.last["headers"]
        self.assertNotIn("x-hop", {k.lower() for k in headers})
        self.assertNotIn("secret", str(headers))

    def test_deep_nested_body_returns_400(self):
        """深嵌套 JSON 请求应显式 400,不能触发 RecursionError 变 500。"""
        nested = {"leaf": True}
        for _ in range(150):
            nested = {"child": nested}
        payload = {"model": "m", "input": [nested]}
        status, data = self.post(
            "/v1/responses", payload, proxy_port=self.fix_proxy_port
        )
        self.assertEqual(status, 400)
        self.assertIn(b"bad_request", data)

    def test_close_delimited_sse_relayed_promptly(self):
        """close-delimited 上游的 SSE 事件必须立即转发,不能等 64KB/EOF。"""
        self.upstream.mode = "close_sse"
        try:
            body = json.dumps({"model": "m", "input": [], "stream": True}).encode()
            request = (
                b"POST /v1/responses HTTP/1.1\r\n"
                b"Host: 127.0.0.1\r\n"
                b"Content-Type: application/json\r\n"
                b"Content-Length: "
                + str(len(body)).encode()
                + b"\r\nConnection: close\r\n\r\n"
                + body
            )
            sock = socket.create_connection(("127.0.0.1", self.proxy_port), timeout=5)
            sock.sendall(request)
            sock.settimeout(2.0)
            received = b""
            deadline = time.monotonic() + 2.0
            while b"hi" not in received and time.monotonic() < deadline:
                try:
                    chunk = sock.recv(4096)
                except socket.timeout:
                    break
                if not chunk:
                    break
                received += chunk
            sock.close()
            self.assertIn(b"hi", received)
        finally:
            self.upstream.mode = "json"

    def _start_extra_proxy(self, **overrides):
        kwargs = dict(
            upstream=self.upstream_url,
            chat_models=[],
            chat_upstream=None,
            fix_agent_message=False,
            timeout=5,
            client_timeout=10,
            max_connections=64,
            log_body=False,
        )
        kwargs.update(overrides)
        proxy = ProxyServer(("127.0.0.1", 0), argparse_namespace(**kwargs))
        threading.Thread(target=proxy.serve_forever, daemon=True).start()
        return proxy, proxy.server_address[1]

    def test_client_body_timeout_returns_408(self):
        """客户端声明大 body 但慢慢发,读超时必须回 408,不能永久占线程。"""
        proxy, port = self._start_extra_proxy(client_timeout=0.5)
        try:
            sock = socket.create_connection(("127.0.0.1", port), timeout=5)
            sock.sendall(
                b"POST /v1/responses HTTP/1.1\r\n"
                b"Host: 127.0.0.1\r\n"
                b"Content-Type: application/json\r\n"
                b"Content-Length: 100\r\n\r\n"
                b'{"model":'
            )
            sock.settimeout(5)
            data = b""
            while b"408" not in data:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                data += chunk
            sock.close()
            self.assertIn(b"408", data)
        finally:
            proxy.shutdown()
            proxy.server_close()

    def test_client_short_body_returns_400(self):
        """客户端提前断开导致短读,必须回 400,绝不能把残缺体转发给上游。"""
        proxy, port = self._start_extra_proxy(client_timeout=5)
        before = self.upstream.last
        try:
            sock = socket.create_connection(("127.0.0.1", port), timeout=5)
            sock.sendall(
                b"POST /v1/responses HTTP/1.1\r\n"
                b"Host: 127.0.0.1\r\n"
                b"Content-Type: application/json\r\n"
                b"Content-Length: 100\r\n\r\n"
                b'{"mod'
            )
            sock.shutdown(socket.SHUT_WR)
            sock.settimeout(5)
            data = b""
            while b"400" not in data:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                data += chunk
            sock.close()
            self.assertIn(b"400", data)
            self.assertIs(self.upstream.last, before)
        finally:
            proxy.shutdown()
            proxy.server_close()

    def test_cl_and_te_rejected_400(self):
        """Content-Length 与 Transfer-Encoding 并存时拒绝,避免读错字节。"""
        proxy, port = self._start_extra_proxy(client_timeout=5)
        try:
            sock = socket.create_connection(("127.0.0.1", port), timeout=5)
            sock.sendall(
                b"POST /v1/responses HTTP/1.1\r\n"
                b"Host: 127.0.0.1\r\n"
                b"Content-Type: application/json\r\n"
                b"Content-Length: 5\r\n"
                b"Transfer-Encoding: chunked\r\n\r\n"
                b"5\r\nhello\r\n0\r\n\r\n"
            )
            sock.settimeout(5)
            data = b""
            while b"400" not in data:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                data += chunk
            sock.close()
            self.assertIn(b"400", data)
        finally:
            proxy.shutdown()
            proxy.server_close()

    def test_max_connections_returns_503(self):
        """并发连接数达到上限时新请求回 503,而不是无限开线程。"""
        proxy, port = self._start_extra_proxy(max_connections=1, client_timeout=5)
        try:
            holder = socket.create_connection(("127.0.0.1", port), timeout=5)
            holder.sendall(
                b"POST /v1/responses HTTP/1.1\r\n"
                b"Host: 127.0.0.1\r\n"
                b"Content-Type: application/json\r\n"
                b"Content-Length: 100\r\n\r\n"
                b'{"model":'
            )
            time.sleep(0.5)

            body = json.dumps({"model": "m", "input": []}).encode()
            sock = socket.create_connection(("127.0.0.1", port), timeout=5)
            sock.sendall(
                b"POST /v1/responses HTTP/1.1\r\n"
                b"Host: 127.0.0.1\r\n"
                b"Content-Type: application/json\r\n"
                b"Content-Length: "
                + str(len(body)).encode()
                + b"\r\n\r\n"
                + body
            )
            sock.settimeout(5)
            data = b""
            while b"503" not in data:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                data += chunk
            sock.close()
            holder.close()
            self.assertIn(b"503", data)
        finally:
            proxy.shutdown()
            proxy.server_close()

    def test_server_listen_backlog_is_raised(self):
        """突发并发连接不能被内核 backlog 拒绝,默认应明显大于并发上限。"""
        self.assertGreaterEqual(ProxyServer.request_queue_size, 128)

    def test_chat_error_envelope_logged(self):
        request_logger = logging.getLogger("responses-proxy")
        self.upstream.mode = "chat_json"
        self.upstream.response_payload = {
            "error": {"message": "quota exceeded", "type": "insufficient_quota"}
        }
        try:
            with self.assertLogs(request_logger, level="DEBUG") as captured:
                status, data = self.post(
                    "/v1/responses",
                    {"model": "deepseek-v4-flash", "input": "hi"},
                    proxy_port=self.chat_proxy_port,
                )
            lines = "\n".join(captured.output)
            self.assertIn("error=quota exceeded", lines)
            self.assertEqual(status, 200)
            body = json.loads(data)
            self.assertEqual(body["status"], "failed")
            self.assertEqual(body["error"]["message"], "quota exceeded")
        finally:
            self.upstream.mode = "json"
            self.upstream.response_payload = {"ok": True}


def argparse_namespace(**kwargs):
    class Namespace:
        pass

    ns = Namespace()
    for key, value in kwargs.items():
        setattr(ns, key, value)
    return ns


if __name__ == "__main__":
    unittest.main(verbosity=2)