From 39a319e34ee40598011c4576362aede8f9707dd8 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 14 Oct 2026 23:31:54 -0400 Subject: [PATCH] fix: judge stream completion by finish reason and trailing usage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- src/polygateway/middleware/retry.py | 2 + src/polygateway/transports/openai_compat.py | 299 ++++++++++---- tests/unit/test_completion_transport.py | 418 ++++++++++++++++++++ tests/unit/test_openai_fidelity.py | 26 +- 4 files changed, 661 insertions(+), 84 deletions(-) create mode 100644 tests/unit/test_completion_transport.py diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 03c4f96..b4867ef 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -741,6 +741,8 @@ class RetryMW: return LLMResponse( content=result.content, completion=result.completion, + finish_reason=result.finish_reason, + stream_end=result.stream_end, thinking=result.thinking, model=source.model, provider=source.provider, diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index 3956bcc..d19f951 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -12,7 +12,7 @@ import json import math import re import time -from dataclasses import replace +from dataclasses import dataclass, field, replace from typing import TYPE_CHECKING, Any import httpx @@ -41,16 +41,21 @@ from polygateway.transports._http_errors import compose_message, summarize_body from polygateway.transports._openai_completion import ( ChatCompletionAccumulator, ChatDelta, - parse_chat_completion, + build_completion, + check_completion_body, ) from polygateway.types import ( Effort, EmbeddingTransportResult, - OpenAIChatCompletion, ResponseMode, SourceConfig, + StreamEnd, TransportResult, - _has_completion_output, + _classify_finish, + _judge_outcome, + _stream_is_complete, + _summarize_message, + _summarize_text, ) if TYPE_CHECKING: @@ -58,6 +63,9 @@ if TYPE_CHECKING: from collections.abc import AsyncIterator, Callable, Mapping _THINK_PATTERN = re.compile(r"(.*?)", re.DOTALL) +# 方言告警节流上限与 finish 值截断(设计 §5.4): 键由上游决定,必须有界 +_WARN_CAP = 1024 +_FINISH_WARN_CHARS = 64 # —— SSE 纯函数(VT llm.py 同款)—— @@ -135,6 +143,125 @@ async def _iter_sse_deltas( yield delta +@dataclass +class _StreamTiming: + """一次流式尝试的计时与文本片段(每尝试新建,不跨重试/对冲共享)。""" + + content_parts: list[str] = field(default_factory=list) + reasoning_parts: list[str] = field(default_factory=list) + alias_parts: list[str] = field(default_factory=list) + chunk_times: list[float] = field(default_factory=list) + ttft_ms: float | None = None + max_gap: float = 0.0 + + @property + def has_reasoning(self) -> bool: + """出现过非空 reasoning_content: 别名 `reasoning` 让位(设计 §4.4)。""" + return bool(self.reasoning_parts) + + +async def _consume_deltas( + guarded: AsyncIterator[ChatDelta], + started: float, + first_token_event: asyncio.Event | None, + content_event: asyncio.Event | None, +) -> _StreamTiming: + """消费进展事件: 计时、置位对冲事件、收集文本片段。""" + timing = _StreamTiming() + last = started + async for delta in guarded: + now = time.monotonic() + # 增量 chunk 时间戳(1.3.8 沉默计量): 循环体只会被真实增量触发—— + # 空 chunk/usage 帧/SSE keepalive 已在聚合器被过滤 + timing.chunk_times.append(now) + if content_event is not None: + # 每个增量 chunk 置位一次: 中段沉默对冲的活性复位信号 + content_event.set() + if timing.ttft_ms is None: + timing.ttft_ms = (now - started) * 1000 + # 首个增量即对冲语义上的"首 token"(思考增量同样是存活证据, + # 与看门狗活性口径一致);None = 调用方未启用对冲,零分支成本 + if first_token_event is not None: + first_token_event.set() + else: + timing.max_gap = max(timing.max_gap, (now - last) * 1000) + last = now + for parts, text in ( + (timing.content_parts, delta.content), + (timing.reasoning_parts, delta.reasoning_content), + (timing.alias_parts, delta.reasoning), + ): + if text: + parts.append(text) + return timing + + +def _raise_unless_deliverable( + mode: ResponseMode, + message: dict[str, Any] | None, + content: str, + thinking: str, + finish_reason: str | None, + source: SourceConfig, +) -> None: + """结局判定(设计 §4.3)的唯一翻译点: 判定值 → 领域错误。 + + 原因码以固定格式进 message(遥测 error 列可检索);不回显推理/工具内容。 + """ + if mode == "openai" and message is not None: + summary = _summarize_message(message, content) + else: + summary = _summarize_text(content, thinking) + verdict = _judge_outcome(mode=mode, summary=summary, finish=_classify_finish(finish_reason)) + if verdict.deliver: + return + text = ( + f"{source.name} {verdict.code}: finish_reason={finish_reason!r}, " + f"thinking_len={len(thinking)}" + ) + if verdict.error_kind == "result_invalid": + raise ResultInvalidError(text, raw_text="", source_name=source.name, operation="chat") + raise TransientError(text, source_name=source.name, operation="chat") + + +def _text_choice(body: dict[str, Any], source: SourceConfig) -> tuple[Any, str | None]: + """文本模式取首个 choice 的 message 与 finish(非字符串 finish 记 None)。""" + choices = body.get("choices") or [] + if not choices: + raise TransientError( + f"{source.name} 响应缺 choices", source_name=source.name, operation="chat" + ) + if not isinstance(choices, list) or not isinstance(choices[0], dict): + raise TransientError("chat choices 形态非法", operation="chat") + finish = choices[0].get("finish_reason") + return choices[0].get("message", {}), finish if isinstance(finish, str) else None + + +def _nonstream_message( + body: dict[str, Any], mode: ResponseMode, source: SourceConfig +) -> tuple[dict[str, Any], str | None]: + """非流式取单 choice 的原始 message 与 finish;只校验形态,不构造快照。""" + if mode == "openai": + message, finish = check_completion_body(body) + else: + message, finish = _text_choice(body, source) + if not isinstance(message, dict) or any( + message.get(key) is not None and not isinstance(message[key], str) + for key in ("content", "reasoning_content") + ): + raise TransientError("chat message 形态非法", operation="chat") + return message, finish + + +def _reasoning_of(message: dict[str, Any]) -> str: + """非流式思考投影: 非空 reasoning_content 优先,否则 `reasoning` 别名。""" + primary = message.get("reasoning_content") + if isinstance(primary, str) and primary: + return primary + alias = message.get("reasoning") + return alias if isinstance(alias, str) else "" + + # —— 错误翻译(CHS invokers.py 同款)—— @@ -392,6 +519,9 @@ class OpenAICompatTransport: # 都会连带改掉另一侧的行为。(键空间恰好不相交,故当下**不会**互相压制; # 分开维护的理由是语义,不是碰撞) self._warned_mismatches: set[tuple[str, str, Effort | None]] = set() + # 方言告警(1.3.10)另立集合: 冲突删键/未知 finish 与推理对账语义无关 + self._warned_fields: set[tuple[str, str]] = set() + self._warn_saturated = False self._client_factory = client_factory or _default_client_factory self._clients: dict[str, httpx.AsyncClient] = {} @@ -636,54 +766,48 @@ class OpenAICompatTransport: inter_token_s=source.inter_token_timeout_s or source.timeout_s, total_s=source.timeout_s, ) - content_parts: list[str] = [] - thinking_parts: list[str] = [] - ttft_ms: float | None = None - last = started - max_gap = 0.0 - chunk_times: list[float] = [] - async for delta in guarded: - now = time.monotonic() - # 增量 chunk 时间戳(1.3.8 沉默计量): 循环体只会被真实增量触发—— - # 空 chunk/usage 帧/SSE keepalive 已在 _sse_data_payload/_sse_delta - # 被过滤,这里逐字追加即可;阈值判定不在热路径,派生收敛在 - # _stall_from_chunk_times(设计 §3 派生原则) - chunk_times.append(now) - if content_event is not None: - # 每个增量 chunk 置位一次: 中段沉默对冲的活性复位信号 - content_event.set() - if ttft_ms is None: - ttft_ms = (now - started) * 1000 - # 首个增量即对冲语义上的"首 token"(思考增量同样是存活证据, - # 与看门狗活性口径一致);None = 调用方未启用对冲,零分支成本 - if first_token_event is not None: - first_token_event.set() - else: - max_gap = max(max_gap, (now - last) * 1000) - last = now - if delta.content: - content_parts.append(delta.content) - if delta.reasoning_content: - thinking_parts.append(delta.reasoning_content) - if response_mode == "openai" and not sink.get("done"): - raise TransientError( - "保真响应缺 [DONE]，禁止打捞", source_name=source.name, operation="chat" - ) + timing = await _consume_deltas(guarded, started, first_token_event, content_event) + return self._stream_result(sink, accumulator, timing, source, profile, response_mode) + + def _stream_result( + self, + sink: dict[str, Any], + accumulator: ChatCompletionAccumulator, + timing: _StreamTiming, + source: SourceConfig, + profile: ProviderProfile, + response_mode: ResponseMode, + ) -> TransportResult: + """流结束后的统一判定(设计 §4.2-§4.3): 完整性 → 结局 → 组装。""" + self._warn_dialect(source, accumulator.conflicts, accumulator.finish_reason) + thinking_parts = timing.reasoning_parts if timing.has_reasoning else timing.alias_parts + content, thinking = self._finalize_text(timing.content_parts, thinking_parts, profile) + stream_end = _stream_is_complete( + done=bool(sink.get("done")), + finish_reason=accumulator.finish_reason, + usage_after_finish=accumulator.usage_after_finish, + usage_measured=_resolve_usage(sink.get("usage") or {})[2] == "measured", + ) + if stream_end is None: + stream_end = self._truncated(timing, thinking_parts, source, response_mode) + message = accumulator.raw_message() if response_mode == "openai" else None + _raise_unless_deliverable( + response_mode, message, content, thinking, accumulator.finish_reason, source + ) completion_result = accumulator.finish() if response_mode == "openai" else None - salvaged = self._check_done(sink, content_parts, thinking_parts, source) - content, thinking = self._finalize_text(content_parts, thinking_parts, profile) - self._reject_empty_completion(content, completion_result, source) - prompt, completion, usage_source = _resolve_stream_usage(sink, salvaged) + prompt, completion, usage_source = _resolve_stream_usage(sink, stream_end == "salvaged") reasoning_tokens = _coerce_reasoning_tokens(sink.get("usage")) - stall_ms, stall_count = _stall_from_chunk_times(chunk_times, source.stall_threshold_ms) + stall_ms, stall_count = _stall_from_chunk_times( + timing.chunk_times, source.stall_threshold_ms + ) return TransportResult( content=content, thinking=thinking, prompt_tokens=prompt, completion_tokens=completion, usage_source=usage_source, - ttft_ms=ttft_ms, - max_inter_token_ms=(max_gap if ttft_ms is not None else None), + ttft_ms=timing.ttft_ms, + max_inter_token_ms=(timing.max_gap if timing.ttft_ms is not None else None), stall_ms=stall_ms, stall_count=stall_count, raw={"usage": sink.get("usage")}, @@ -696,40 +820,55 @@ class OpenAICompatTransport: thinking_observation=observe_thinking( thinking=thinking, reasoning_tokens=reasoning_tokens ), + finish_reason=accumulator.finish_reason, + stream_end=stream_end, ) - def _check_done( + def _truncated( self, - sink: dict[str, Any], - content_parts: list[str], + timing: _StreamTiming, thinking_parts: list[str], source: SourceConfig, - ) -> bool: - """缺 [DONE] 语义(设计 §6): 零内容恒 retry;有内容按 missing_done 策略。""" - if sink.get("done"): - return False + response_mode: ResponseMode, + ) -> StreamEnd: + """截断流(设计 §4.2): 保真不打捞;文本按 missing_done,零产物恒 retry。""" ctx: dict[str, Any] = {"source_name": source.name, "operation": "chat"} - if not content_parts and not thinking_parts: + if response_mode == "openai": + raise TransientError(f"{source.name} missing_done: 保真响应截断,禁止打捞", **ctx) + if not any(timing.content_parts) and not any(thinking_parts): raise TransientError(f"{source.name} SSE early_eof: 零内容断流", **ctx) if source.missing_done == "retry": raise TransientError(f"{source.name} SSE missing_done: 截断且无 [DONE]", **ctx) - return True + return "salvaged" - def _reject_empty_completion( - self, content: str, completion: OpenAIChatCompletion | None, source: SourceConfig + def _warn_dialect( + self, source: SourceConfig, conflicts: frozenset[str], finish_reason: str | None ) -> None: - """空补全 → 瞬时错误(2026-07-20 人类裁决,M1 验证发现)。 - - 服务 200 且流程完整([DONE]/usage 正常)但 content 为空——MiniMax 等 - 网关的间歇异常形态。视为服务抖动: 退避重试/换源,**绝不缓存空响应**; - 承 CHS "VLM 零 content"归瞬时的先例(invokers.py:309)。 - """ - if not _has_completion_output(content, completion): - raise TransientError( - f"{source.name} 空补全(empty_completion): 流程完整但零内容", - source_name=source.name, - operation="chat", + """方言告警: 冲突删键与未登记 finish 各按 (源, 路径/值) 只喊一次。""" + for path in sorted(conflicts): + self._warn_once( + source.name, f"field:{path}", f"字段 {path} 合并冲突已删除,保真结果缺该字段" ) + if finish_reason is not None and _classify_finish(finish_reason) == "unknown": + value = finish_reason[:_FINISH_WARN_CHARS] + self._warn_once( + source.name, + f"finish:{value}", + f"未登记的 finish_reason {value!r},按正常结束交付,该值的结果不缓存", + ) + + def _warn_once(self, source_name: str, key: str, text: str) -> None: + """有界节流: 键由上游决定,长跑 worker 不能让集合无限增长(设计 §5.4)。""" + if (source_name, key) in self._warned_fields: + return + if len(self._warned_fields) >= _WARN_CAP: + if not self._warn_saturated: + self._warn_saturated = True + logger.warning("方言告警节流集合已饱和({}),后续同类告警省略", _WARN_CAP) + return + self._warned_fields.add((source_name, key)) + # 单参数传入 loguru: 文案可能含上游给的花括号(同 `_warn_on_thinking_mismatch`) + logger.warning("源 {} —— {}", source_name, text) def _finalize_text( self, content_parts: list[str], thinking_parts: list[str], profile: ProviderProfile @@ -759,6 +898,9 @@ class OpenAICompatTransport: 为纯时间阈值(对冲只能靠 `hedge_after_s` 触发)。`content_event`(1.3.8) 同款: 接收但**永不置位**——非流式无"中段"概念,stall 两值走 TransportResult 默认 None(排除条款)。 + + 无流完整性可言,直接以 body 的 finish_reason 走结局判定;保真模式**先判定 + 后构造快照**,否则半截工具在构造时就抛 Transient,到不了 `tool_truncated`。 """ resp = await client.post(url, json=payload) if resp.status_code != 200: @@ -772,24 +914,16 @@ class OpenAICompatTransport: f"{source.name} 非流式响应非法 JSON", source_name=source.name, operation="chat" ) from exc _check_chat_error(body) - completion_result = parse_chat_completion(body) if response_mode == "openai" else None - choices = body.get("choices") or [] - if not choices: - raise TransientError( - f"{source.name} 响应缺 choices", source_name=source.name, operation="chat" - ) - if not isinstance(choices, list) or not isinstance(choices[0], dict): - raise TransientError("chat choices 形态非法", operation="chat") - message = choices[0].get("message", {}) - if not isinstance(message, dict) or any( - message.get(key) is not None and not isinstance(message[key], str) - for key in ("content", "reasoning_content") - ): - raise TransientError("chat message 形态非法", operation="chat") + message, finish = _nonstream_message(body, response_mode, source) content, thinking = self._finalize_text( - [message.get("content") or ""], [message.get("reasoning_content") or ""], profile + [message.get("content") or ""], [_reasoning_of(message)], profile + ) + self._warn_dialect(source, frozenset(), finish) + raw_message = message if response_mode == "openai" else None + _raise_unless_deliverable(response_mode, raw_message, content, thinking, finish, source) + completion_result = ( + build_completion(message, body, finish) if response_mode == "openai" else None ) - self._reject_empty_completion(content, completion_result, source) prompt, completion, usage_source = _resolve_usage(body.get("usage") or {}) reasoning_tokens = _coerce_reasoning_tokens(body.get("usage")) return TransportResult( @@ -810,6 +944,7 @@ class OpenAICompatTransport: thinking_observation=observe_thinking( thinking=thinking, reasoning_tokens=reasoning_tokens ), + finish_reason=finish, ) async def aclose(self) -> None: diff --git a/tests/unit/test_completion_transport.py b/tests/unit/test_completion_transport.py new file mode 100644 index 0000000..a5efe93 --- /dev/null +++ b/tests/unit/test_completion_transport.py @@ -0,0 +1,418 @@ +"""统一完成判定在 transport 两条路径上的落点(#30/#31,设计 §4.2-§4.4、§5.4)。 + +假 SSE 按 magpie/glm 实测形态二次构造: 末块 finish_reason + usage 后干净关闭、 +不发 [DONE];以及只有思考的 length 截断、null/未知字段等方言。 +""" + +import asyncio +import json + +import httpx +import pytest +from loguru import logger + +from polygateway.errors import AllSourcesExhausted, ResultInvalidError, TransientError +from polygateway.transports.openai_compat import OpenAICompatTransport +from tests.unit.test_openai_compat import _complete, _source, _sse_stream, _transport_for + +_USAGE = {"prompt_tokens": 11, "completion_tokens": 7} +_TOOL = { + "index": 0, + "id": "call_1", + "type": "function", + "function": {"name": "lookup", "arguments": '{"q": 1}'}, +} + + +def _f(delta=None, finish=None, usage=None, choices=True): + """单 choice 帧;`choices=False` 构造 usage-only 帧。""" + body = {"id": "up-1"} + if choices: + choice = {"index": 0, "delta": delta or {}} + if finish is not None: + choice["finish_reason"] = finish + body["choices"] = [choice] + else: + body["choices"] = [] + if usage is not None: + body["usage"] = usage + return f"data: {json.dumps(body)}\n\n" + + +def _assistant(**delta): + return _f({"role": "assistant", **delta}) + + +async def _run(frames, *, done, mode="text", **source_kw): + transport = _transport_for(lambda _: _sse_stream(*frames, done=done)) + try: + return await _complete(transport, _source(**source_kw), response_mode=mode) + finally: + await transport.aclose() + + +async def _run_once(body, *, mode="text"): + transport = _transport_for(lambda _: httpx.Response(200, json=body)) + try: + return await _complete(transport, _source(), stream=False, response_mode=mode) + finally: + await transport.aclose() + + +# —— #31: 缺 [DONE] 但 finish + 其后实测 usage 完整 —— + + +@pytest.mark.parametrize("mode", ["text", "openai"]) +@pytest.mark.parametrize("finish", ["stop", "length"]) +async def test_eof_after_finish_with_usage_succeeds(mode, finish): + """验收 1: 末块 finish + usage 后干净关闭 → 成功、measured、eof_after_finish。""" + frames = [_assistant(content="hel"), _f({"content": "lo"}, finish=finish, usage=_USAGE)] + result = await _run(frames, done=False, mode=mode) + assert result.content == "hello" and result.stream_end == "eof_after_finish" + assert result.usage_source == "measured" and result.finish_reason == finish + assert (result.prompt_tokens, result.completion_tokens) == (11, 7) + if mode == "openai": + assert result.completion.finish_reason == finish + + +async def test_eof_after_finish_usage_only_frame_and_tools(): + """验收 2: tool_calls 结束、usage 在其后的 usage-only 帧 → 工具完整交回。""" + frames = [ + _assistant(content=None, tool_calls=[_TOOL]), + _f({}, finish="tool_calls"), + _f(choices=False, usage=_USAGE), + ] + result = await _run(frames, done=False, mode="openai") + tool = result.completion.message["tool_calls"][0] + assert tool["function"] == {"name": "lookup", "arguments": '{"q": 1}'} + assert result.stream_end == "eof_after_finish" + + +@pytest.mark.parametrize( + "frames", + [ + [_assistant(content="partial"), _f(choices=False, usage=_USAGE)], # 验收 3: 无 finish + [_assistant(content="partial"), _f({}, finish="stop")], # 验收 4: 无 usage + [_assistant(content="a", **{}), _f({}, usage=_USAGE), _f({}, finish="stop")], # usage 早 + [_assistant(content="a"), _f({}, finish="stop", usage={})], # usage 不可用 + ], +) +async def test_incomplete_eof_still_fails(frames): + """截断不放宽: 保真仍不打捞,文本 retry 档仍抛。""" + with pytest.raises(TransientError, match="missing_done"): + await _run(frames, done=False, mode="openai", missing_done="salvage") + with pytest.raises(TransientError, match="missing_done"): + await _run(frames, done=False) + + +async def test_text_salvage_marks_stream_end(): + """文本 salvage 打捞 → stream_end=salvaged,usage 降为 estimated。""" + frames = [_assistant(content="partial"), _f(choices=False, usage=_USAGE)] + result = await _run(frames, done=False, missing_done="salvage") + assert result.stream_end == "salvaged" and result.usage_source == "estimated" + + +async def test_read_error_mid_stream_fails(): + """验收 5: 读取中途 RemoteProtocolError → 失败,不当干净 EOF。""" + + class Broken(httpx.AsyncByteStream): + async def __aiter__(self): + yield _f({"role": "assistant", "content": "a"}, finish="stop", usage=_USAGE).encode() + raise httpx.RemoteProtocolError("peer closed") + + async def aclose(self): + return None + + transport = _transport_for(lambda _: httpx.Response(200, stream=Broken())) + try: + with pytest.raises(TransientError, match="网络错误"): + await _complete(transport, _source(), response_mode="openai") + finally: + await transport.aclose() + + +@pytest.mark.parametrize("mode", ["text", "openai"]) +async def test_done_stream_marks_done(mode): + """验收 6: 带 [DONE] 的正常流行为不变,stream_end=done。""" + result = await _run([_assistant(content="ok"), _f({}, "stop", _USAGE)], done=True, mode=mode) + assert result.stream_end == "done" and result.finish_reason == "stop" + + +@pytest.mark.parametrize("done", [True, False]) +@pytest.mark.parametrize("mode", ["text", "openai"]) +async def test_upstream_error_finish_is_transient(done, mode): + """GLM network_error: 上游自报失败,有正文也不交付。""" + frames = [_assistant(content="half"), _f({}, finish="network_error", usage=_USAGE)] + with pytest.raises(TransientError, match="upstream_finish_error"): + await _run(frames, done=done, mode=mode) + + +# —— #30 与结局表(设计 §4.3)—— + +_THINK = [_assistant(reasoning_content="long thought")] + + +@pytest.mark.parametrize("done", [True, False]) +async def test_openai_reasoning_only_length_delivered(done): + """#30 验收: 只有思考 + length → 交回,content 空,思考完整。""" + result = await _run([*_THINK, _f({}, "length", _USAGE)], done=done, mode="openai") + message = result.completion.message + assert message.get("content") in (None, "") and message["reasoning_content"] == "long thought" + assert result.finish_reason == "length" and result.thinking == "long thought" + + +@pytest.mark.parametrize( + ("frames", "finish", "openai_ok", "text_error"), + [ + (_THINK, "length", True, (ResultInvalidError, "length_without_content")), + ([_assistant()], "length", True, (ResultInvalidError, "length_without_content")), + (_THINK, "content_filter", True, (ResultInvalidError, "filtered_without_content")), + ([_assistant()], "sensitive", True, (ResultInvalidError, "filtered_without_content")), + (_THINK, "stop", True, (TransientError, "empty_completion")), + ([_assistant(refusal="no")], "stop", True, (TransientError, "empty_completion")), + ([_assistant()], "stop", False, (TransientError, "empty_completion")), + ], +) +async def test_outcome_rows_stream(frames, finish, openai_ok, text_error): + """行 3/4/6/7/8: 两模式的分叉。""" + full = [*frames, _f({}, finish, _USAGE)] + if openai_ok: + assert (await _run(full, done=True, mode="openai")).finish_reason == finish + else: + with pytest.raises(TransientError, match="empty_completion"): + await _run(full, done=True, mode="openai") + error, code = text_error + with pytest.raises(error, match=code) as caught: + await _run(full, done=True) + assert caught.type is error # ResultInvalid 不得被 Transient 子类掩盖 + + +@pytest.mark.parametrize("mode", ["text", "openai"]) +async def test_reasoning_only_without_finish_is_transient(mode): + """行 9: 只靠 DONE 收尾、无 finish → 无法区分抖动与结局,维持瞬时。""" + with pytest.raises(TransientError, match="empty_completion"): + await _run([*_THINK, _f(choices=False, usage=_USAGE)], done=True, mode=mode) + + +async def test_openai_whitespace_content_stop_delivered(): + """正文只有空白 + stop: 保真交回(#30 一-2)。""" + result = await _run( + [_assistant(content=" "), _f({}, "stop", _USAGE)], done=True, mode="openai" + ) + assert result.completion.message["content"] == " " + + +async def test_truncated_tool_is_result_invalid(): + """行 2: length 截断的半截工具(缺 name)→ ResultInvalid tool_truncated。""" + broken = {"index": 0, "id": "c", "type": "function", "function": {"arguments": '{"q'}} + frames = [_assistant(tool_calls=[broken]), _f({}, "length", _USAGE)] + with pytest.raises(ResultInvalidError, match="tool_truncated"): + await _run(frames, done=True, mode="openai") + + +async def test_content_cannot_mask_broken_tool(): + """行 2: 正文非空 + 坏工具 + stop → Transient tool_incomplete。""" + broken = {"index": 0, "id": "c", "type": "function", "function": {"arguments": "{}"}} + frames = [_assistant(content="text", tool_calls=[broken]), _f({}, "stop", _USAGE)] + with pytest.raises(TransientError, match="tool_incomplete"): + await _run(frames, done=True, mode="openai") + + +async def test_null_refusal_and_tool_calls_stream(): + """#30 验收二: delta 带 refusal:null 与 tool_calls:null → 正常交回。""" + frames = [_assistant(content="ok", refusal=None, tool_calls=None), _f({}, "stop", _USAGE)] + result = await _run(frames, done=True, mode="openai") + assert result.completion.message["content"] == "ok" + + +# —— 非流式镜像 —— + + +def _body(message, finish): + choice = {"index": 0, "message": {"role": "assistant", **message}} + if finish is not None: + choice["finish_reason"] = finish + return {"id": "up", "choices": [choice], "usage": _USAGE} + + +async def test_nonstream_null_tool_calls_and_refusal(): + """非流式 message 带 tool_calls:null + refusal:null → 成功。""" + body = _body({"content": "hi", "tool_calls": None, "refusal": None}, "stop") + result = await _run_once(body, mode="openai") + assert result.completion.message["tool_calls"] is None and result.stream_end is None + + +@pytest.mark.parametrize( + ("message", "finish", "openai_expect", "text_expect"), + [ + ( + {"reasoning_content": "t"}, + "length", + None, + (ResultInvalidError, "length_without_content"), + ), + ({"content": ""}, "content_filter", None, (ResultInvalidError, "filtered_without_content")), + ({"reasoning_content": "t"}, "stop", None, (TransientError, "empty_completion")), + ({"reasoning_content": "t"}, None, (TransientError, "empty_completion"), None), + ({"content": "a"}, "network_error", (TransientError, "upstream_finish_error"), None), + ({"content": "a"}, "length", None, None), + ], +) +async def test_outcome_rows_nonstream(message, finish, openai_expect, text_expect): + """非流式跳过流完整性,直接以 body finish 走结局表。""" + body = _body(message, finish) + for mode, expect in (("openai", openai_expect), ("text", text_expect or openai_expect)): + if expect is None: + result = await _run_once(body, mode=mode) + assert result.finish_reason == finish and result.stream_end is None + else: + with pytest.raises(expect[0], match=expect[1]) as caught: + await _run_once(body, mode=mode) + assert caught.type is expect[0] + + +async def test_nonstream_truncated_tool_is_result_invalid(): + """非流式先判定后构造快照: 缺 name + length → tool_truncated 而非构造期 Transient。""" + tool = {"id": "c", "type": "function", "function": {"arguments": "{"}} + with pytest.raises(ResultInvalidError, match="tool_truncated"): + await _run_once(_body({"tool_calls": [tool]}, "length"), mode="openai") + + +# —— reasoning 别名(设计 §4.4)—— + + +async def test_reasoning_alias_projects_thinking(): + """只有 reasoning 字段: 文本 thinking 取别名;非空 reasoning_content 优先。""" + alias = await _run( + [_assistant(reasoning="r1"), _f({"content": "a"}, "stop", _USAGE)], done=True + ) + assert alias.thinking == "r1" + both = [ + _assistant(reasoning="r1", reasoning_content="rc"), + _f({"content": "a"}, "stop", _USAGE), + ] + assert (await _run(both, done=True)).thinking == "rc" + empty_rc = [_assistant(reasoning="r1", reasoning_content=""), _f({"content": "a"}, "stop")] + assert (await _run(empty_rc, done=True)).thinking == "r1" + nonstream = await _run_once(_body({"content": "a", "reasoning": "nr"}, "stop")) + assert nonstream.thinking == "nr" + + +async def test_reasoning_alias_keeps_ttft_alive(): + """只有 reasoning 的思考期刷新活性,不被 TTFT 看门狗误杀。""" + + class Slow(httpx.AsyncByteStream): + async def __aiter__(self): + yield _assistant().encode() + for _ in range(5): # 0.5s 只有 reasoning,远超 ttft,但每帧间隔 < inter_token + yield _f({"reasoning": "t"}).encode() + await asyncio.sleep(0.1) + yield _f({"content": "answer"}, "stop", _USAGE).encode() + yield b"data: [DONE]\n\n" + + async def aclose(self): + return None + + transport = _transport_for(lambda _: httpx.Response(200, stream=Slow())) + try: + source = _source(ttft_timeout_s=0.3, inter_token_timeout_s=0.25) + result = await _complete(transport, source) + assert result.content == "answer" and result.thinking == "ttttt" + finally: + await transport.aclose() + + +# —— 有界告警节流(设计 §5.4)—— + + +def _sink(): + records = [] + handle = logger.add(lambda m: records.append(m.record["message"]), level="WARNING") + return records, handle + + +async def test_conflict_and_unknown_finish_warn_once(): + """同源同路径冲突、同一未知 finish 值各只告警一次,不回显字段值。""" + frames = [ + _assistant(content="a", ext=1), + _f({"ext": 2}), + _f({}, "end_turn", _USAGE), + ] + transport = _transport_for(lambda _: _sse_stream(*frames)) + records, handle = _sink() + try: + for _ in range(2): + result = await _complete(transport, _source(), response_mode="openai") + assert "ext" not in result.completion.message + finally: + logger.remove(handle) + await transport.aclose() + conflict = [r for r in records if "delta.ext" in r] + unknown = [r for r in records if "end_turn" in r] + assert len(conflict) == 1 and len(unknown) == 1 and "不缓存" in unknown[0] + + +async def test_warning_set_is_bounded(): + """节流集合饱和后只记一次饱和告警,不再增长。""" + transport = OpenAICompatTransport() + transport._warned_fields.update((f"s{i}", "k") for i in range(1024)) + records, handle = _sink() + try: + transport._warn_once("qwen_1", "field:x", "a") + transport._warn_once("qwen_1", "field:y", "b") + finally: + logger.remove(handle) + await transport.aclose() + assert len(transport._warned_fields) == 1024 + assert sum("饱和" in r for r in records) == 1 + + +# —— 治理链: 确定性结局不重试、不计熔断 —— + + +async def test_text_length_without_content_not_retried_nor_breaker_failure(): + """文本 length 无正文 → ResultInvalid 直抛,attempts=1,熔断不记失败。""" + from tests.unit.test_client import _client + + calls = [] + + def handler(_): + calls.append(1) + return _sse_stream(*_THINK, _f({}, "length", _USAGE)) + + client = _client(handler=handler) + failures = [] + original = client._breaker_backend.record_failure + + async def spy(*args, **kwargs): + failures.append(args) + return await original(*args, **kwargs) + + client._breaker_backend.record_failure = spy + try: + with pytest.raises(ResultInvalidError, match="length_without_content"): + await client.chat([{"role": "user", "content": "hi"}]) + finally: + await client._transport.aclose() + assert len(calls) == 1 and failures == [] + + +async def test_openai_reasoning_length_single_attempt_via_client(): + """#30 验收经治理链: attempts=1;#31 形态经治理链不再重试到耗尽。""" + from tests.unit.test_client import _client + + calls = [] + + def handler(_): + calls.append(1) + return _sse_stream(*_THINK, _f({}, "length", _USAGE), done=False) + + client = _client(handler=handler) + try: + response = await client.chat([{"role": "user", "content": "hi"}], response_mode="openai") + except AllSourcesExhausted: # pragma: no cover - 失败即回归 + pytest.fail("reasoning-only length 不应重试到耗尽") + finally: + await client._transport.aclose() + assert len(calls) == 1 and response.call_stats.attempts == 1 + assert response.stream_end == "eof_after_finish" and response.finish_reason == "length" diff --git a/tests/unit/test_openai_fidelity.py b/tests/unit/test_openai_fidelity.py index bc7e77f..87f892f 100644 --- a/tests/unit/test_openai_fidelity.py +++ b/tests/unit/test_openai_fidelity.py @@ -146,10 +146,10 @@ async def test_invalid_mode_combinations_fail_before_http(kwargs): async def test_openai_never_salvages(): - """已收完整工具也必须有 DONE。""" + """无 finish 后 usage 的截断流即使收全工具也不打捞(1.3.10: 判据由 DONE 改为完成判定)。""" transport = _transport_for(lambda _: _sse_stream(*tool_frames(), done=False)) try: - with pytest.raises(TransientError, match="DONE"): + with pytest.raises(TransientError, match="missing_done"): await transport.complete( messages=[], source=_source(missing_done="salvage"), @@ -165,6 +165,28 @@ async def test_openai_never_salvages(): 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