fix(request-body): 使用 deepcopy 防止请求体在处理流程中被意外修改

handler 基类和格式转换 registry 中,原始请求体通过浅拷贝或直接引用传递,
导致下游处理(模型映射、格式转换、重试整流)可能修改原始数据,
影响后续重试或并发请求的正确性。统一改用 copy.deepcopy 隔离副本。
This commit is contained in:
fawney19
2026-03-17 13:23:02 +08:00
parent d63d5eff85
commit d480aa11f3
7 changed files with 313 additions and 33 deletions

View File

@@ -0,0 +1,132 @@
from __future__ import annotations
import copy
from typing import Any
from src.core.api_format.conversion.internal import (
FormatCapabilities,
InternalMessage,
InternalRequest,
InternalResponse,
Role,
TextBlock,
)
from src.core.api_format.conversion.normalizer import FormatNormalizer
from src.core.api_format.conversion.registry import FormatConversionRegistry
class _BaseTestNormalizer(FormatNormalizer):
capabilities = FormatCapabilities()
def response_to_internal(self, response: dict[str, Any]) -> InternalResponse:
return InternalResponse(id=str(response.get("id") or ""), model="", content=[])
def response_from_internal(
self,
internal: InternalResponse,
*,
requested_model: str | None = None,
) -> dict[str, Any]:
return {
"id": internal.id,
"model": requested_model or internal.model,
}
class _SameFormatNormalizer(_BaseTestNormalizer):
FORMAT_ID = "TEST:SAME"
def request_to_internal(self, request: dict[str, Any]) -> InternalRequest:
return InternalRequest(model=str(request.get("model") or ""), messages=[])
def request_from_internal(
self,
internal: InternalRequest,
*,
target_variant: str | None = None,
) -> dict[str, Any]:
return {"model": internal.model, "variant": target_variant}
class _MutatingSourceNormalizer(_BaseTestNormalizer):
FORMAT_ID = "TEST:MUTSRC"
def request_to_internal(self, request: dict[str, Any]) -> InternalRequest:
request.pop("ephemeral", None)
request["messages"][0]["content"][0]["text"] = "mutated"
return InternalRequest(
model=str(request.get("model") or ""),
messages=[
InternalMessage(
role=Role.USER,
content=[TextBlock(text=str(request["messages"][0]["content"][0]["text"]))],
)
],
)
def request_from_internal(
self,
internal: InternalRequest,
*,
target_variant: str | None = None,
) -> dict[str, Any]:
return {"model": internal.model, "variant": target_variant}
class _TargetNormalizer(_BaseTestNormalizer):
FORMAT_ID = "TEST:MUTTGT"
def request_to_internal(self, request: dict[str, Any]) -> InternalRequest:
return InternalRequest(model=str(request.get("model") or ""), messages=[])
def request_from_internal(
self,
internal: InternalRequest,
*,
target_variant: str | None = None,
) -> dict[str, Any]:
text = ""
if internal.messages and internal.messages[0].content:
first = internal.messages[0].content[0]
if isinstance(first, TextBlock):
text = first.text
return {"model": internal.model, "text": text, "variant": target_variant}
def test_convert_request_same_format_returns_detached_copy() -> None:
registry = FormatConversionRegistry()
registry.register(_SameFormatNormalizer())
original = {
"model": "gpt-test",
"messages": [{"role": "user", "content": [{"type": "text", "text": "hello"}]}],
}
out = registry.convert_request(original, "test:same", "test:same")
assert out == original
assert out is not original
assert out["messages"] is not original["messages"]
out["messages"][0]["content"][0]["text"] = "changed"
assert original["messages"][0]["content"][0]["text"] == "hello"
def test_convert_request_cross_format_does_not_mutate_original_input() -> None:
registry = FormatConversionRegistry()
registry.register(_MutatingSourceNormalizer())
registry.register(_TargetNormalizer())
original = {
"model": "gpt-test",
"ephemeral": "keep-me",
"messages": [{"role": "user", "content": [{"type": "text", "text": "hello"}]}],
}
snapshot = copy.deepcopy(original)
out = registry.convert_request(original, "test:mutsrc", "test:muttgt")
assert out["model"] == "gpt-test"
assert out["text"] == "mutated"
assert original == snapshot