PolyGateway/tests/unit/test_completion_verdict.py

"""统一完成判定(#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__