iomgaa
39a319e34e
Both transport paths now share one verdict: a stream is complete on [DONE] or on finish_reason followed by measured usage and a clean EOF (both modes; missing_done only governs real truncation). Outcomes follow the design's first-match table: upstream_error finishes stay transient, fidelity mode delivers reasoning-only/length results, text mode raises ResultInvalid for content-less length/filter endings, and non-stream fidelity judges before building the snapshot so truncated tools reach tool_truncated. Projections honour the reasoning alias, dialect warnings are bounded, and responses carry finish_reason and stream_end. Red: 39 of 46 tests in tests/unit/test_completion_transport.py failed before this change (e.g. eof_after_finish raised 保真响应缺 [DONE]; reasoning-only length raised empty_completion; reasoning-only stream hit the TTFT watchdog, re-confirmed by disabling alias liveness).
377 lines
13 KiB
Python
377 lines
13 KiB
Python
"""OpenAI 消息保真回归：按网关 Chat Completions 形态构造。"""
|
|
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from polygateway.errors import AllSourcesExhausted, TransientError
|
|
from tests.unit.test_client import _client
|
|
from tests.unit.test_openai_compat import _complete, _source, _sse_stream, _transport_for
|
|
from tests.unit.test_openai_completion import accumulate, chunk, tool_frames
|
|
|
|
|
|
async def test_error_before_done_is_failure():
|
|
"""文本模式也不能接受 error 后的 DONE。"""
|
|
transport = _transport_for(
|
|
lambda _: _sse_stream(chunk({"content": "partial"}), 'data: {"error": null}\n\n')
|
|
)
|
|
try:
|
|
with pytest.raises(TransientError):
|
|
await _complete(transport, _source())
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
async def test_mixed_content_reasoning_survive():
|
|
"""同一帧双文本不互斥。"""
|
|
transport = _transport_for(
|
|
lambda _: _sse_stream(chunk({"content": "answer", "reasoning_content": "reason"}))
|
|
)
|
|
try:
|
|
result = await _complete(transport, _source())
|
|
assert (result.content, result.thinking) == ("answer", "reason")
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
async def test_tools_roundtrip_and_retry():
|
|
"""真实 transport 经过治理，失败重试不污染工具，消息可直接往返。"""
|
|
requests = []
|
|
|
|
def handler(request):
|
|
requests.append(json.loads(request.content))
|
|
if len(requests) == 1:
|
|
return _sse_stream(
|
|
chunk({"content": "partial"}),
|
|
'data: {"error": {"code": "upstream_stream_read_error"}}\n\n',
|
|
)
|
|
return _sse_stream(*tool_frames())
|
|
|
|
client = _client(handler=handler)
|
|
try:
|
|
response = await client.chat(
|
|
[{"role": "user", "content": "hi"}],
|
|
response_mode="openai",
|
|
overlay={"reasoning_effort": "max"},
|
|
)
|
|
assert response.call_stats.attempts == 2
|
|
message = response.completion.message
|
|
assert message["role"] == "assistant" and message["content"] is None
|
|
assert [t["id"] for t in message["tool_calls"]] == ["a", "b"]
|
|
assert message["tool_calls"][0]["function"]["arguments"] == '{"v": "中" }'
|
|
assert response.completion.response_id == "upstream-1"
|
|
assert response.completion.finish_reason == "tool_calls"
|
|
messages = [message, {"role": "tool", "tool_call_id": "a", "content": "ok"}]
|
|
await client.chat(messages, response_mode="openai")
|
|
assert requests[-1]["messages"] == messages
|
|
assert requests[0]["reasoning_effort"] == "max"
|
|
finally:
|
|
await client._transport.aclose()
|
|
|
|
|
|
@pytest.mark.parametrize("role", [None, "user"])
|
|
async def test_final_role_required(role):
|
|
"""非流式缺失或错误 role 必须协议失败。"""
|
|
message = {"content": "ok"}
|
|
if role:
|
|
message["role"] = role
|
|
client = _client(
|
|
handler=lambda _: httpx.Response(200, json={"choices": [{"message": message}]})
|
|
)
|
|
try:
|
|
with pytest.raises(AllSourcesExhausted):
|
|
await client.chat(
|
|
[{"role": "user", "content": "hi"}], stream=False, response_mode="openai"
|
|
)
|
|
finally:
|
|
await client._transport.aclose()
|
|
|
|
|
|
@pytest.mark.parametrize("stream", [True, False])
|
|
async def test_completion_cache_roundtrip(stream):
|
|
"""完整结果回放，修改返回对象不毒化缓存；模式隔离。"""
|
|
from polygateway.backends.memory.cache import InMemoryCache
|
|
|
|
completion = accumulate().finish()
|
|
|
|
def handler(_):
|
|
if stream:
|
|
return _sse_stream(*tool_frames())
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"id": "upstream-1",
|
|
"choices": [{"message": completion.message, "finish_reason": "tool_calls"}],
|
|
},
|
|
)
|
|
|
|
client = _client(
|
|
handler=handler, cache=InMemoryCache(), cache_namespace="tools", cache_ttl_s=60
|
|
)
|
|
try:
|
|
first = await client.chat([], stream=stream, response_mode="openai")
|
|
second = await client.chat([], stream=stream, response_mode="openai")
|
|
assert not first.cache_hit and second.cache_hit
|
|
assert second.completion == first.completion
|
|
assert second.call_id != first.call_id and second.call_stats.attempts == 0
|
|
second.completion.message["tool_calls"].clear()
|
|
third = await client.chat([], stream=stream, response_mode="openai")
|
|
assert len(third.completion.message["tool_calls"]) == 2
|
|
finally:
|
|
await client._transport.aclose()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"kwargs",
|
|
[
|
|
{"response_mode": "bad"},
|
|
{"response_mode": "openai", "structured": "json"},
|
|
{"response_mode": "openai", "overlay": {"n": 2}},
|
|
{"response_mode": "openai", "overlay": {"n": True}},
|
|
],
|
|
)
|
|
async def test_invalid_mode_combinations_fail_before_http(kwargs):
|
|
"""组合错误入口拒绝。"""
|
|
|
|
def unexpected(_):
|
|
pytest.fail("不应发送 HTTP")
|
|
|
|
client = _client(handler=unexpected)
|
|
try:
|
|
with pytest.raises(ValueError):
|
|
await client.chat([], **kwargs)
|
|
finally:
|
|
await client._transport.aclose()
|
|
|
|
|
|
async def test_openai_never_salvages():
|
|
"""无 finish 后 usage 的截断流即使收全工具也不打捞(1.3.10: 判据由 DONE 改为完成判定)。"""
|
|
transport = _transport_for(lambda _: _sse_stream(*tool_frames(), done=False))
|
|
try:
|
|
with pytest.raises(TransientError, match="missing_done"):
|
|
await transport.complete(
|
|
messages=[],
|
|
source=_source(missing_done="salvage"),
|
|
stream=True,
|
|
overlay={},
|
|
call_id="c",
|
|
reasoning_effort=None,
|
|
first_token_event=None,
|
|
content_event=None,
|
|
response_mode="openai",
|
|
)
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
async def test_openai_finish_with_trailing_usage_is_complete_without_done():
|
|
"""对照: 同一工具流补上 finish 之后的 usage 帧,即使无 DONE 也完整交回(#31)。"""
|
|
usage = 'data: {"choices": [], "usage": {"prompt_tokens": 3, "completion_tokens": 4}}\n\n'
|
|
transport = _transport_for(lambda _: _sse_stream(*tool_frames(), usage, done=False))
|
|
try:
|
|
result = await transport.complete(
|
|
messages=[],
|
|
source=_source(),
|
|
stream=True,
|
|
overlay={},
|
|
call_id="c",
|
|
reasoning_effort=None,
|
|
first_token_event=None,
|
|
content_event=None,
|
|
response_mode="openai",
|
|
)
|
|
finally:
|
|
await transport.aclose()
|
|
assert result.stream_end == "eof_after_finish"
|
|
assert [t["id"] for t in result.completion.message["tool_calls"]] == ["a", "b"]
|
|
|
|
|
|
async def test_tool_progress_sets_events():
|
|
"""工具首进展驱动两个事件与时间观测。"""
|
|
import asyncio
|
|
|
|
first, progress = asyncio.Event(), asyncio.Event()
|
|
transport = _transport_for(lambda _: _sse_stream(*tool_frames()))
|
|
try:
|
|
result = await transport.complete(
|
|
messages=[],
|
|
source=_source(),
|
|
stream=True,
|
|
overlay={},
|
|
call_id="c",
|
|
reasoning_effort=None,
|
|
first_token_event=first,
|
|
content_event=progress,
|
|
response_mode="openai",
|
|
)
|
|
assert first.is_set() and progress.is_set()
|
|
assert result.ttft_ms is not None and result.stall_count == 0
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
async def test_real_transport_stall_hedge_cancels_loser():
|
|
"""真实工具流沉默触发异源对冲，连接关闭且赢家不混入半成品。"""
|
|
import asyncio
|
|
import time
|
|
|
|
from polygateway.sources import RoundRobinSelector
|
|
from polygateway.types import ChatRequest, _CallContext
|
|
from tests.unit.test_retry import _src
|
|
from tests.unit.test_stall_hedge import _harness
|
|
|
|
entered, closed = asyncio.Event(), asyncio.Event()
|
|
|
|
class Hanging(httpx.AsyncByteStream):
|
|
"""工具首帧后阻塞，取消由真实 httpx 流处理。"""
|
|
|
|
async def __aiter__(self):
|
|
yield tool_frames()[0].encode()
|
|
yield tool_frames()[1].encode()
|
|
entered.set()
|
|
await asyncio.Event().wait()
|
|
|
|
async def aclose(self):
|
|
closed.set()
|
|
|
|
count = 0
|
|
|
|
def handler(_):
|
|
nonlocal count
|
|
count += 1
|
|
if count == 1:
|
|
return httpx.Response(200, stream=Hanging())
|
|
return _sse_stream(*tool_frames())
|
|
|
|
transport = _transport_for(handler)
|
|
sources = [_src("s1"), _src("s2")]
|
|
mw, limiter, *_ = _harness(
|
|
sources, transport, stall_after_s=0.05, selector=RoundRobinSelector()
|
|
)
|
|
context = _CallContext(now=time.monotonic)
|
|
try:
|
|
response = await asyncio.wait_for(
|
|
mw(ChatRequest(messages=[], response_mode="openai", call_context=context)), 3
|
|
)
|
|
assert entered.is_set() and closed.is_set()
|
|
assert response.source_name == "s2" and len(response.completion.message["tool_calls"]) == 2
|
|
assert context.snapshot().hedges == 1 and context.snapshot().attempts == 2
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
@pytest.mark.parametrize("stream", [True, False])
|
|
async def test_bad_usage_is_domain_failure(stream):
|
|
"""坏 usage 不能穿透成裸 AttributeError，包括默认文本模式。"""
|
|
transport = _transport_for(
|
|
lambda _: (
|
|
_sse_stream(chunk({"content": "ok"}), 'data: {"choices": [], "usage": [1]}\n\n')
|
|
if stream
|
|
else httpx.Response(
|
|
200, json={"choices": [{"message": {"content": "ok"}}], "usage": [1]}
|
|
)
|
|
)
|
|
)
|
|
try:
|
|
with pytest.raises(TransientError):
|
|
await _complete(transport, _source(), stream=stream)
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
async def test_nonstream_tool_result_has_no_liveness_events():
|
|
"""非流式工具不制造流式观测。"""
|
|
import asyncio
|
|
|
|
completion = accumulate().finish()
|
|
first, progress = asyncio.Event(), asyncio.Event()
|
|
transport = _transport_for(
|
|
lambda _: httpx.Response(
|
|
200, json={"choices": [{"message": completion.message, "finish_reason": "tool_calls"}]}
|
|
)
|
|
)
|
|
try:
|
|
result = await transport.complete(
|
|
messages=[],
|
|
source=_source(),
|
|
stream=False,
|
|
overlay={},
|
|
call_id="c",
|
|
reasoning_effort=None,
|
|
first_token_event=first,
|
|
content_event=progress,
|
|
response_mode="openai",
|
|
)
|
|
assert result.completion.message == completion.message
|
|
assert not first.is_set() and not progress.is_set()
|
|
assert result.ttft_ms is None and result.stall_ms is None
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
async def test_openai_does_not_bypass_managed_conflict():
|
|
"""保真模式不静默撤销源级受管意图。"""
|
|
from polygateway.errors import RequestRejectedError
|
|
|
|
def unexpected(_):
|
|
pytest.fail("冲突请求不应发送 HTTP")
|
|
|
|
transport = _transport_for(unexpected)
|
|
try:
|
|
with pytest.raises(RequestRejectedError):
|
|
await transport.complete(
|
|
messages=[],
|
|
source=_source(enable_thinking=True),
|
|
stream=True,
|
|
overlay={"reasoning_effort": "max"},
|
|
call_id="c",
|
|
reasoning_effort=None,
|
|
first_token_event=None,
|
|
content_event=None,
|
|
response_mode="openai",
|
|
)
|
|
finally:
|
|
await transport.aclose()
|
|
|
|
|
|
async def test_cancel_tool_stream_releases_resources():
|
|
"""显式取消真实工具流，连接关闭且配额归还。"""
|
|
import asyncio
|
|
|
|
from polygateway.backends.memory.limiter import InMemoryLimiter
|
|
from polygateway.types import GlobalLimits
|
|
|
|
entered, closed = asyncio.Event(), asyncio.Event()
|
|
|
|
class Hanging(httpx.AsyncByteStream):
|
|
"""可确定取消窗口的 HTTP 流。"""
|
|
|
|
async def __aiter__(self):
|
|
for frame in tool_frames()[:2]:
|
|
yield frame.encode()
|
|
entered.set()
|
|
await asyncio.Event().wait()
|
|
|
|
async def aclose(self):
|
|
closed.set()
|
|
|
|
source = _source(max_concurrency=1)
|
|
limiter = InMemoryLimiter(
|
|
scope="llm", sources={source.name: source}, global_limits=GlobalLimits(1, 0, 0)
|
|
)
|
|
client = _client(
|
|
sources=[source], limiter=limiter, handler=lambda _: httpx.Response(200, stream=Hanging())
|
|
)
|
|
task = asyncio.create_task(client.chat([], response_mode="openai"))
|
|
try:
|
|
await asyncio.wait_for(entered.wait(), 2)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert closed.is_set()
|
|
assert (await limiter.source_stats(source.name)).inflight == 0
|
|
finally:
|
|
task.cancel()
|
|
await client._transport.aclose()
|