iomgaa
abcdf6930e
- fidelity cacheability also requires a non-empty projected body, so think-tag-only answers are no longer cached - a no-argument tool sent with arguments null only is a deterministic dialect: ResultInvalid tool_arguments_null instead of retried - text mode treats choice index null as 0 and no longer lets False pose as 0 - null-only tool_calls elements no longer produce an empty tool_calls list; non-stream fidelity checks role before the verdict; warning paths are length-bounded - docs: ARCHITECTURE 7.8 field counts, README cache/watchdog wording, CHANGELOG notes, review record in the design, hook note in the plan Red: 8 new tests failed before this change (e.g. test_openai_cacheable_requires_projected_body, test_null_arguments_tool_is_result_invalid_stream, test_text_mode_choice_index_null_is_zero[None-True]).
266 lines
10 KiB
Python
266 lines
10 KiB
Python
"""统一完成判定(#30/#31): 内核纯函数真值表。"""
|
|
|
|
import pytest
|
|
|
|
from polygateway.types import (
|
|
LLMResponse,
|
|
OpenAIChatCompletion,
|
|
TransportResult,
|
|
_classify_finish,
|
|
_effective_finish,
|
|
_is_cacheable,
|
|
_judge_outcome,
|
|
_OutputSummary,
|
|
_stream_is_complete,
|
|
_summarize_message,
|
|
_summarize_text,
|
|
_tool_snapshot_ok,
|
|
)
|
|
|
|
_TOOL = {"id": "t1", "type": "function", "function": {"name": "f", "arguments": "{}"}}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("value", "expected"),
|
|
[
|
|
("stop", "natural"),
|
|
("tool_calls", "natural"),
|
|
("function_call", "natural"),
|
|
("length", "truncated"),
|
|
("content_filter", "filtered"),
|
|
("sensitive", "filtered"),
|
|
("network_error", "upstream_error"),
|
|
("error", "upstream_error"),
|
|
("end_turn", "unknown"),
|
|
("STOP", "unknown"),
|
|
(None, "absent"),
|
|
],
|
|
)
|
|
def test_classify_finish(value, expected):
|
|
assert _classify_finish(value) == expected
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("done", "finish", "after", "measured", "expected"),
|
|
[
|
|
(True, None, False, False, "done"),
|
|
(True, "stop", True, True, "done"),
|
|
(False, "stop", True, True, "eof_after_finish"),
|
|
(False, "length", True, True, "eof_after_finish"),
|
|
(False, None, True, True, None), # 无 finish 即截断
|
|
(False, "stop", False, True, None), # usage 早于 finish(累计 usage)
|
|
(False, "stop", True, False, None), # finish 后 usage 不可用
|
|
],
|
|
)
|
|
def test_stream_is_complete(done, finish, after, measured, expected):
|
|
got = _stream_is_complete(
|
|
done=done, finish_reason=finish, usage_after_finish=after, usage_measured=measured
|
|
)
|
|
assert got == expected
|
|
|
|
|
|
def test_tool_snapshot_ok():
|
|
assert _tool_snapshot_ok({"role": "assistant"})
|
|
assert _tool_snapshot_ok({"tool_calls": None})
|
|
assert _tool_snapshot_ok({"tool_calls": [None, _TOOL]})
|
|
assert not _tool_snapshot_ok({"tool_calls": {}})
|
|
assert not _tool_snapshot_ok({"tool_calls": [{"id": "t1", "type": "function"}]})
|
|
assert not _tool_snapshot_ok({"tool_calls": [_TOOL, _TOOL]}) # 重复 id
|
|
|
|
|
|
def test_summarize_text():
|
|
assert _summarize_text("", "") == _OutputSummary(False, False, False, True)
|
|
assert _summarize_text(" ", "") == _OutputSummary(True, False, False, True)
|
|
assert _summarize_text("", "think") == _OutputSummary(True, False, False, True)
|
|
assert _summarize_text("hi", "") == _OutputSummary(True, True, False, True)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("message", "projected", "expected"),
|
|
[
|
|
({"role": "assistant"}, "", _OutputSummary(False, False, False, True)),
|
|
({"role": "assistant", "content": None}, "", _OutputSummary(False, False, False, True)),
|
|
({"role": "assistant", "refusal": "no"}, "", _OutputSummary(True, False, False, True)),
|
|
(
|
|
{"role": "assistant", "reasoning_content": "r"},
|
|
"",
|
|
_OutputSummary(True, False, False, True),
|
|
),
|
|
({"role": "assistant", "content": " "}, " ", _OutputSummary(True, False, False, True)),
|
|
(
|
|
{"role": "assistant", "x": {"y": ["", "z"]}},
|
|
"",
|
|
_OutputSummary(True, False, False, True),
|
|
),
|
|
(
|
|
{"role": "assistant", "tool_calls": [None]},
|
|
"",
|
|
_OutputSummary(False, False, False, True),
|
|
),
|
|
(
|
|
{"role": "assistant", "tool_calls": [_TOOL]},
|
|
"",
|
|
_OutputSummary(True, True, True, True),
|
|
),
|
|
(
|
|
{"role": "assistant", "tool_calls": [{"id": "t"}]},
|
|
"",
|
|
_OutputSummary(True, False, True, False),
|
|
),
|
|
({"role": "assistant", "content": "a"}, "a", _OutputSummary(True, True, False, True)),
|
|
],
|
|
)
|
|
def test_summarize_message(message, projected, expected):
|
|
assert _summarize_message(message, projected) == expected
|
|
|
|
|
|
_NONE = _OutputSummary(False, False, False, True)
|
|
_THINK = _OutputSummary(True, False, False, True)
|
|
_BODY = _OutputSummary(True, True, False, True)
|
|
_BAD_TOOL = _OutputSummary(True, True, True, False)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("mode", "summary", "finish", "expected"),
|
|
[
|
|
# 行 1: upstream_error 优先于一切
|
|
("openai", _BODY, "upstream_error", ("transient", "upstream_finish_error")),
|
|
("text", _BODY, "upstream_error", ("transient", "upstream_finish_error")),
|
|
# 行 2: 坏工具(仅保真),正文不得掩盖
|
|
("openai", _BAD_TOOL, "truncated", ("result_invalid", "tool_truncated")),
|
|
("openai", _BAD_TOOL, "natural", ("transient", "tool_incomplete")),
|
|
("text", _BAD_TOOL, "natural", None),
|
|
# 行 3: 无产物 + truncated/filtered
|
|
("openai", _NONE, "truncated", None),
|
|
("openai", _NONE, "filtered", None),
|
|
("text", _NONE, "truncated", ("result_invalid", "length_without_content")),
|
|
("text", _NONE, "filtered", ("result_invalid", "filtered_without_content")),
|
|
# 行 4: 零增量抖动(2026-07-20 裁决)
|
|
("openai", _NONE, "natural", ("transient", "empty_completion")),
|
|
("openai", _NONE, "absent", ("transient", "empty_completion")),
|
|
("openai", _NONE, "unknown", ("transient", "empty_completion")),
|
|
("text", _NONE, "absent", ("transient", "empty_completion")),
|
|
# 行 5: 有正文
|
|
("openai", _BODY, "truncated", None),
|
|
("text", _BODY, "filtered", None),
|
|
("text", _BODY, "absent", None),
|
|
# 行 6-9: 有产物无正文
|
|
("openai", _THINK, "truncated", None),
|
|
("text", _THINK, "truncated", ("result_invalid", "length_without_content")),
|
|
("openai", _THINK, "filtered", None),
|
|
("text", _THINK, "filtered", ("result_invalid", "filtered_without_content")),
|
|
("openai", _THINK, "natural", None),
|
|
("openai", _THINK, "unknown", None),
|
|
("text", _THINK, "natural", ("transient", "empty_completion")),
|
|
("openai", _THINK, "absent", ("transient", "empty_completion")),
|
|
("text", _THINK, "absent", ("transient", "empty_completion")),
|
|
],
|
|
)
|
|
def test_judge_outcome_rows(mode, summary, finish, expected):
|
|
verdict = _judge_outcome(mode=mode, summary=summary, finish=finish)
|
|
if expected is None:
|
|
assert verdict.deliver and verdict.error_kind is None and verdict.code is None
|
|
else:
|
|
assert not verdict.deliver
|
|
assert (verdict.error_kind, verdict.code) == expected
|
|
|
|
|
|
def _completion(content="a", finish="stop", tools=None):
|
|
message = {"role": "assistant", "content": content}
|
|
if tools is not None:
|
|
message["tool_calls"] = tools
|
|
return OpenAIChatCompletion(message, finish, None, None)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kwargs", "expected"),
|
|
[
|
|
({"mode": "text", "content": "a", "finish_reason": "stop", "stream_end": "done"}, True),
|
|
({"mode": "text", "content": "a", "finish_reason": None, "stream_end": None}, True),
|
|
(
|
|
{
|
|
"mode": "text",
|
|
"content": "a",
|
|
"finish_reason": "stop",
|
|
"stream_end": "eof_after_finish",
|
|
},
|
|
True,
|
|
),
|
|
({"mode": "text", "content": "a", "finish_reason": None, "stream_end": "salvaged"}, False),
|
|
({"mode": "text", "content": " ", "finish_reason": "stop", "stream_end": "done"}, False),
|
|
({"mode": "text", "content": "a", "finish_reason": "length", "stream_end": None}, False),
|
|
({"mode": "text", "content": "a", "finish_reason": "sensitive", "stream_end": None}, False),
|
|
({"mode": "text", "content": "a", "finish_reason": "end_turn", "stream_end": None}, False),
|
|
({"mode": "text", "content": "a", "finish_reason": "stop", "stream_end": "bogus"}, False),
|
|
],
|
|
)
|
|
def test_is_cacheable_text(kwargs, expected):
|
|
assert _is_cacheable(completion=None, **kwargs) is expected
|
|
|
|
|
|
def test_is_cacheable_openai():
|
|
ok = _completion()
|
|
assert _is_cacheable(
|
|
mode="openai", content="a", finish_reason="stop", stream_end="done", completion=ok
|
|
)
|
|
assert not _is_cacheable(
|
|
mode="openai", content="a", finish_reason="stop", stream_end="done", completion=None
|
|
)
|
|
# 原文为空且无工具: 不信任旧投影
|
|
assert not _is_cacheable(
|
|
mode="openai",
|
|
content="a",
|
|
finish_reason="stop",
|
|
stream_end="done",
|
|
completion=_completion(content=""),
|
|
)
|
|
tool_only = _completion(content=None, finish="tool_calls", tools=[_TOOL])
|
|
assert _is_cacheable(
|
|
mode="openai", content="", finish_reason="tool_calls", stream_end=None, completion=tool_only
|
|
)
|
|
|
|
|
|
def test_effective_finish():
|
|
assert _effective_finish("stop", _completion(finish="length")) == "stop"
|
|
assert _effective_finish(None, _completion(finish="length")) == "length"
|
|
assert _effective_finish(None, None) is None
|
|
|
|
|
|
def test_completion_accepts_null_tool_calls():
|
|
snap = OpenAIChatCompletion({"role": "assistant", "tool_calls": None}, "stop", None, None)
|
|
assert "tool_calls" in snap.message and snap.message["tool_calls"] is None
|
|
with pytest.raises(ValueError):
|
|
OpenAIChatCompletion({"role": "assistant", "tool_calls": {}}, "stop", None, None)
|
|
|
|
|
|
def test_response_fields_default_none():
|
|
result = TransportResult("a", "", 0, 0, "unavailable", None, None, {})
|
|
assert result.finish_reason is None and result.stream_end is None
|
|
response = LLMResponse("a", "", "m", "p", 0, 0, 0, None, None, False, "c")
|
|
assert response.finish_reason is None and response.stream_end is None
|
|
|
|
|
|
def test_stream_end_exported():
|
|
import polygateway
|
|
|
|
assert "StreamEnd" in polygateway.__all__
|
|
|
|
|
|
def test_openai_cacheable_requires_projected_body():
|
|
"""审查 I1: think 标签剥离后投影为空,即使原文非空也不可缓存。"""
|
|
think_only = _completion(content="<think>x</think>")
|
|
assert not _is_cacheable(
|
|
mode="openai", content="", finish_reason="stop", stream_end="done", completion=think_only
|
|
)
|
|
|
|
|
|
def test_null_arguments_only_tool_is_result_invalid():
|
|
"""审查 I2: 无参函数 arguments 全程 null(其余齐全)→ 确定性方言,ResultInvalid。"""
|
|
tool = {"id": "t1", "type": "function", "function": {"name": "f", "arguments": None}}
|
|
summary = _summarize_message({"role": "assistant", "tool_calls": [tool]}, "")
|
|
assert summary.null_arguments and not summary.tools_ok
|
|
verdict = _judge_outcome(mode="openai", summary=summary, finish="natural")
|
|
assert (verdict.error_kind, verdict.code) == ("result_invalid", "tool_arguments_null")
|
|
broken = {"id": "t1", "type": "function", "function": {"arguments": None}}
|
|
assert not _summarize_message({"role": "assistant", "tool_calls": [broken]}, "").null_arguments
|