PolyGateway/tests/unit/test_live_evidence.py

"""live_evidence 与测试侧取证装配的日常离线反例。"""

import asyncio
import inspect
import json
from dataclasses import replace

import httpx
import pytest

from polygateway.errors import RequestRejectedError, SourceDeadError, TransientError
from polygateway.ports import EmbeddingTransport, Transport
from polygateway.transports.openai_compat import OpenAICompatTransport
from polygateway.types import SourceConfig
from polygateway.types import ThinkingObservation as O
from tests.e2e.conftest import LiveCapture, ObservedTransport
from tests.live_evidence import (
    AttemptEvidence,
    HttpEvidence,
    LiveVerdict,
    assess_expected_rejection,
    assess_model_identity,
    assess_thinking_coverage,
    classify_live_failure,
    messages_digest,
    request_is_valid,
    safe_attempts,
    summarize_verdicts,
    write_live_round,
)

_CHECKS = tuple(
    (key, True)
    for key in (
        "method",
        "origin",
        "path",
        "model",
        "stream",
        "authorization",
        "control",
        "messages_digest",
    )
)
_BODY = b'{"error":{"type":"model_not_found"}}'
_SECRET = "fake-unique-credential-sentinel"
_PROMPT = "fake-private-prompt-sentinel"
_MESSAGES = [{"role": "user", "content": _PROMPT}]


def _failure(*, body=_BODY, status=404, checks=_CHECKS):
    """完整的唯一错误证据，供逐维破坏。"""
    error = RequestRejectedError("safe", status_code=status)
    event = HttpEvidence("a", checks, status, body, (False, None))
    return error, (AttemptEvidence("a", (event,), error),)


def test_exact_model_not_found_is_uncovered():
    error, attempts = _failure()
    assert classify_live_failure(error, attempts).status == "UNCOVERED"


@pytest.mark.parametrize(
    "body",
    [
        None,
        b"",
        b"{",
        b"[]",
        b'{"error":[]}',
        b'{"error":{"type":7}}',
        b'{"error":{"message":"model_not_found"}}',
        b'{"error":{"type":"MODEL_NOT_FOUND"}}',
        b'{"error":{"type":"wrong","type":"model_not_found"}}',
        b'{"error":{},"error":{"type":"model_not_found"}}',
        b"\xff",
        _BODY + b" " * 65536,
    ],
)
def test_incomplete_or_ambiguous_error_body_fails(body):
    assert classify_live_failure(*_failure(body=body)).status == "FAIL"


@pytest.mark.parametrize("status", [400, 401, 403, 429, 500, 503])
def test_status_classes_never_automatically_skip(status):
    assert classify_live_failure(*_failure(status=status)).status == "FAIL"


@pytest.mark.parametrize(
    "error",
    [TransientError("safe"), SourceDeadError("safe"), ValueError("safe"), AssertionError("safe")],
)
def test_whole_exception_class_skip_is_forbidden(error):
    assert classify_live_failure(error, ()).status == "FAIL"


@pytest.mark.parametrize("key", [key for key, _ in _CHECKS])
@pytest.mark.parametrize("mode", ["missing", "false", "duplicate"])
def test_all_request_checks_are_required(key, mode):
    checks = tuple(
        (k, False if k == key and mode == "false" else v)
        for k, v in _CHECKS
        if mode != "missing" or k != key
    )
    if mode == "duplicate":
        checks += ((key, True),)
    assert classify_live_failure(*_failure(checks=checks)).status == "FAIL"


@pytest.mark.parametrize(
    "mode",
    ["empty", "second_attempt", "second_http", "wrong_id", "different_error", "status_mismatch"],
)
def test_attempt_pairing_cannot_be_guessed(mode):
    error, attempts = _failure()
    first = attempts[0]
    if mode == "empty":
        attempts = (replace(first, http=()),)
    elif mode == "second_attempt":
        attempts += (AttemptEvidence("b", (), TransientError("safe")),)
    elif mode == "second_http":
        attempts = (replace(first, http=first.http * 2),)
    elif mode == "wrong_id":
        attempts = (replace(first, call_id="other"),)
    elif mode == "different_error":
        attempts = (replace(first, error=RequestRejectedError("safe", status_code=404)),)
    else:
        attempts = (replace(first, http=(replace(first.http[0], status_code=400),)),)
    assert classify_live_failure(error, attempts).status == "FAIL"


@pytest.mark.parametrize(
    ("reported", "raw", "expected"),
    [
        ("m", (False, None), "PASS"),
        ("alias", (True, "alias"), "PASS"),
        (None, (False, None), "FAIL"),
        ("other", (False, None), "FAIL"),
        (None, (True, None), "UNCOVERED"),
        ("other", (True, "other"), "UNCOVERED"),
        (None, (True, "m"), "FAIL"),
        ("wrong", (True, "m"), "FAIL"),
        ("m", (True, None), "FAIL"),
    ],
)
def test_identity_requires_independent_raw_evidence(reported, raw, expected):
    assert (
        assess_model_identity(
            requested="m",
            aliases=frozenset({"alias"}),
            reported=reported,
            raw_identity=raw,
            request_valid=True,
        ).status
        == expected
    )
    assert (
        assess_model_identity(
            requested="m",
            aliases=frozenset(),
            reported=reported,
            raw_identity=raw,
            request_valid=False,
        ).status
        == "FAIL"
    )


@pytest.mark.parametrize(
    ("proposition", "observations", "expected"),
    [
        ("enabled", [O.OBSERVED, O.OBSERVED, O.UNKNOWN], "PASS"),
        ("enabled", [O.UNKNOWN] * 3, "UNCOVERED"),
        ("enabled", [O.ABSENT] * 3, "FAIL"),
        ("disabled", [O.ABSENT] * 3, "PASS"),
        ("disabled", [O.ABSENT, O.UNKNOWN, O.ABSENT], "UNCOVERED"),
        ("disabled", [O.OBSERVED, O.UNKNOWN, O.ABSENT], "FAIL"),
        ("cannot_disable", [O.OBSERVED, O.UNKNOWN, O.ABSENT], "PASS"),
        ("cannot_disable", [O.ABSENT] * 3, "FAIL"),
        ("cannot_disable", [O.UNKNOWN] * 3, "UNCOVERED"),
    ],
)
def test_coverage_is_proposition_specific(proposition, observations, expected):
    assert (
        assess_thinking_coverage(observations, planned_rounds=3, proposition=proposition).status
        == expected
    )


def test_missing_round_never_reduces_denominator():
    assert (
        assess_thinking_coverage([O.OBSERVED] * 2, planned_rounds=3, proposition="enabled").status
        == "FAIL"
    )
    assert summarize_verdicts(
        [LiveVerdict("PASS", "ok"), LiveVerdict("FAIL", "bad")], planned_rounds=3
    ) == {"PASS": 1, "FAIL": 1, "UNCOVERED": 0, "missing": 1}


def test_expected_400_is_a_separate_negative_proposition():
    error, attempts = _failure(status=400, body=b'{"error":{"type":"unsupported_value"}}')
    assert classify_live_failure(error, attempts).status == "FAIL"
    assert (
        assess_expected_rejection(
            error, attempts, status_code=400, machine_type="unsupported_value"
        ).status
        == "PASS"
    )
    assert (
        assess_expected_rejection(error, attempts, status_code=400, machine_type="other").status
        == "FAIL"
    )


def _source():
    """假凭据与非默认超时，不接真实网络。"""
    return SourceConfig(
        name="source",
        provider="openai",
        model="gpt-5.5",
        base_url="https://example.test/v1",
        api_key=_SECRET,
        timeout_s=137,
        trust_env=False,
    )


def _capture(*, stream=False, embedding=False, **changes):
    """预期独立于生产 payload。"""
    expected = {
        "model": "gpt-5.5",
        "origin": "https://example.test",
        "path": "/v1/embeddings" if embedding else "/v1/chat/completions",
        "control": {},
        "messages_digest": messages_digest([_PROMPT] if embedding else _MESSAGES),
    }
    expected.update({"input_shape": 1} if embedding else {"stream": stream})
    expected.update(changes)
    return LiveCapture(expectations={"source": expected})


def _response(model="gpt-5.5"):
    """OpenAI 非流式完整响应形态。"""
    result = {
        "id": "test",
        "choices": [
            {"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}
        ],
        "usage": {"prompt_tokens": 2, "completion_tokens": 1, "total_tokens": 3},
    }
    if model != "missing":
        result["model"] = model
    return result


def _real_transport(capture, handler, *, tamper=None):
    """仅替换 HTTP 边界，保留真实 hooks 与 production transport。"""

    def factory(source):
        """在网络边界换成 MockTransport。"""
        client = capture.client_factory(source)
        client._transport = httpx.MockTransport(handler)
        if tamper is not None:
            client.event_hooks["request"].insert(0, tamper)
        return client

    return OpenAICompatTransport(client_factory=factory)


async def _complete(observed, *, call_id="a", stream=False):
    """沿原参数调用薄委托器。"""
    return await observed.complete(
        messages=_MESSAGES,
        source=_source(),
        stream=stream,
        overlay={},
        call_id=call_id,
        reasoning_effort=None,
    )


@pytest.mark.parametrize("model", ["gpt-5.5", "missing", None])
async def test_raw_identity_snapshot_reaches_round_consumer(model):
    capture = _capture()
    real = _real_transport(capture, lambda request: httpx.Response(200, json=_response(model)))
    try:
        with capture.round_context(session_id="s", parent_call_id="p"):
            result = await _complete(ObservedTransport(real, capture))
        raw = capture.raw_identity(session_id="s", parent_call_id="p", call_id="a")
        assert raw == (True, None if model == "missing" else model)
        assert request_is_valid(capture.attempts(session_id="s", parent_call_id="p")[0].http[0])
        verdict = assess_model_identity(
            requested="gpt-5.5",
            aliases=frozenset(),
            reported=result.model_reported,
            raw_identity=raw,
            request_valid=True,
        )
        assert verdict.status == ("PASS" if model == "gpt-5.5" else "UNCOVERED")
        if model == "gpt-5.5":
            assert (
                assess_model_identity(
                    requested="gpt-5.5",
                    aliases=frozenset(),
                    reported=None,
                    raw_identity=raw,
                    request_valid=True,
                ).status
                == "FAIL"
            )
    finally:
        await real.aclose()


@pytest.mark.parametrize(
    "body", [b"[]", b"{", b'{"model":"a","model":"b"}', b'{"model":7}', b"\xff"]
)
async def test_invalid_raw_json_is_not_missing_identity(body):
    capture = _capture()
    real = _real_transport(capture, lambda request: httpx.Response(200, content=body))
    try:
        with (
            capture.round_context(session_id="s", parent_call_id="p"),
            pytest.raises((ValueError, AttributeError, KeyError, TransientError)),
        ):
            await _complete(ObservedTransport(real, capture))
        assert capture.raw_identity(session_id="s", parent_call_id="p", call_id="a") == (
            False,
            None,
        )
        assert capture.notes(session_id="s", parent_call_id="p")
    finally:
        await real.aclose()


@pytest.mark.parametrize(
    "field", ["model", "authorization", "path", "stream", "control", "messages_digest"]
)
async def test_hooks_detect_corrupted_request_without_repair(field):
    capture = _capture()

    async def tamper(request):
        """故意改错实发请求，而不是改预期。"""
        if field == "authorization":
            request.headers["Authorization"] = "Bearer wrong"
        elif field == "path":
            request.url = request.url.copy_with(path="/wrong")
        else:
            body = json.loads(request.content)
            key, value = {
                "model": ("model", "wrong"),
                "stream": ("stream", True),
                "control": ("reasoning_effort", "high"),
                "messages_digest": ("messages", []),
            }[field]
            body[key] = value
            request._content = json.dumps(body).encode()

    real = _real_transport(
        capture, lambda request: httpx.Response(404, content=_BODY), tamper=tamper
    )
    try:
        with (
            capture.round_context(session_id="s", parent_call_id="p"),
            pytest.raises(RequestRejectedError) as caught,
        ):
            await _complete(ObservedTransport(real, capture))
        attempts = capture.attempts(session_id="s", parent_call_id="p")
        assert dict(attempts[0].http[0].request_checks)[field] is False
        assert classify_live_failure(caught.value, attempts).status == "FAIL"
    finally:
        await real.aclose()


async def test_embed_uses_same_capture_and_source_factory_contract():
    capture = _capture(embedding=True)
    real = _real_transport(
        capture,
        lambda request: httpx.Response(
            200,
            json={"data": [{"index": 0, "embedding": [0.1, 0.2]}], "usage": {"prompt_tokens": 1}},
        ),
    )
    try:
        with capture.round_context(session_id="s", parent_call_id="p"):
            result = await ObservedTransport(real, capture).embed(
                texts=[_PROMPT], source=_source(), call_id="e"
            )
        assert result.dim == 2
        assert request_is_valid(capture.attempts(session_id="s", parent_call_id="p")[0].http[0])
        client = real._clients["source"]
        assert client.timeout.read == 137 and client.trust_env is False
        assert client.headers["Authorization"] == f"Bearer {_SECRET}"
    finally:
        await real.aclose()
        assert real._clients == {}


class _SSE(httpx.AsyncByteStream):
    """可检查是否被 hook 提前消费的 SSE。"""

    def __init__(self):
        self.reads = 0
        self.closed = False

    async def __aiter__(self):
        self.reads += 1
        yield b'data: {"model":"gpt-5.5","choices":[{"delta":{"content":"ok"},"finish_reason":null}]}\n\n'
        yield b'data: {"choices":[],"usage":{"prompt_tokens":2,"completion_tokens":1,"total_tokens":3}}\n\ndata: [DONE]\n\n'

    async def aclose(self):
        self.closed = True


async def test_success_sse_is_not_preconsumed_and_has_no_raw_snapshot():
    capture = _capture(stream=True)
    stream = _SSE()

    def handler(request):
        return httpx.Response(200, stream=stream)

    def factory(source):
        client = capture.client_factory(source)

        async def after_hook(response):
            assert stream.reads == 0

        client.event_hooks["response"].append(after_hook)
        client._transport = httpx.MockTransport(handler)
        return client

    real = OpenAICompatTransport(client_factory=factory)
    try:
        with capture.round_context(session_id="s", parent_call_id="p"):
            result = await _complete(ObservedTransport(real, capture), stream=True)
        assert result.content == "ok" and stream.reads == 1 and stream.closed
        raw = capture.raw_identity(session_id="s", parent_call_id="p", call_id="a")
        assert raw == (False, None)
        assert (
            assess_model_identity(
                requested="gpt-5.5",
                aliases=frozenset(),
                reported=None,
                raw_identity=raw,
                request_valid=True,
            ).status
            == "FAIL"
        )
    finally:
        await real.aclose()


async def test_concurrent_rounds_retry_and_cancellation_reset_context():
    capture = _capture()
    reached = asyncio.Event()

    async def handler(request):
        await asyncio.sleep(0)
        return httpx.Response(200, json=_response())

    real = _real_transport(capture, handler)
    observed = ObservedTransport(real, capture)

    async def one(parent):
        with capture.round_context(session_id="s", parent_call_id=parent):
            # 零 HTTP 的前次失败与成功 call_id 不得混淆。
            with pytest.raises(ValueError), capture.attempt_context(parent + "-failed"):
                raise ValueError("safe")
            await _complete(observed, call_id=parent + "-ok")
        return capture.raw_identity(session_id="s", parent_call_id=parent, call_id=parent + "-ok")

    async def cancelled():
        with (
            capture.round_context(session_id="s", parent_call_id="cancel"),
            capture.attempt_context("cancelled"),
        ):
            reached.set()
            await asyncio.Future()

    try:
        assert await asyncio.gather(one("p1"), one("p2")) == [(True, "gpt-5.5")] * 2
        for parent in ("p1", "p2"):
            attempts = capture.attempts(session_id="s", parent_call_id=parent)
            assert len(attempts) == 2 and len(attempts[0].http) == 0 and len(attempts[1].http) == 1
        with pytest.raises(ValueError, match="跨轮"):
            capture.raw_identity(session_id="s", parent_call_id="p1", call_id="p2-ok")
        task = asyncio.create_task(cancelled())
        await reached.wait()
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert len(capture.attempts(session_id="s", parent_call_id="cancel")) == 1
        with pytest.raises(LookupError):
            capture._round.get()
        with pytest.raises(LookupError):
            capture._attempt.get()
        with (
            capture.round_context(session_id="s", parent_call_id="new"),
            pytest.raises(ValueError, match="重复"),
            capture.attempt_context("p1-ok"),
        ):
            pytest.fail("不应进入")
    finally:
        await real.aclose()


def test_delegator_signatures_match_ports():
    assert inspect.signature(ObservedTransport.complete) == inspect.signature(Transport.complete)
    assert inspect.signature(ObservedTransport.embed) == inspect.signature(EmbeddingTransport.embed)


def test_missing_expectations_fail_before_network():
    with pytest.raises(ValueError):
        LiveCapture(expectations={"s": {"model": "m"}})


def test_round_reports_preserve_prior_round_and_redact_sentinels(tmp_path):
    error, attempts = _failure(
        body=(f'{{"error":{{"type":"model_not_found","message":"{_SECRET} {_PROMPT}"}}}}').encode()
    )
    error.args = (f"{_SECRET} {_PROMPT}",)
    first = write_live_round(
        tmp_path,
        run_id="run",
        matrix_id="matrix",
        round_index=1,
        safe_fields={"status": "PASS", "attempts": safe_attempts(attempts)},
    )
    second = write_live_round(
        tmp_path,
        run_id="run",
        matrix_id="matrix",
        round_index=2,
        safe_fields={"status": "FAIL", "attempts": safe_attempts(attempts)},
    )
    assert first != second and first.exists()
    for path in tmp_path.rglob("*.md"):
        text = path.read_text()
        assert _SECRET not in text and _PROMPT not in text
    assert '"status": "PASS"' in first.read_text()
    assert '"status": "FAIL"' in second.read_text()


@pytest.mark.parametrize(
    "fields",
    [
        {"raw": _BODY},
        {"error_type": ValueError("secret")},
        {"attempts": [AttemptEvidence("a", (), None)]},
        {"attempts": [{"call_id": "a", "error_type": None, "http": [{"Authorization": "secret"}]}]},
    ],
)
def test_report_refuses_raw_objects_and_unknown_fields(tmp_path, fields):
    with pytest.raises((ValueError, TypeError)):
        write_live_round(
            tmp_path, run_id="run", matrix_id="matrix", round_index=1, safe_fields=fields
        )


def test_report_write_failure_is_not_skip(tmp_path):
    blocked = tmp_path / "file"
    blocked.write_text("not a directory")
    with pytest.raises(OSError):
        write_live_round(
            blocked,
            run_id="run",
            matrix_id="matrix",
            round_index=1,
            safe_fields={"status": "UNCOVERED"},
        )


@pytest.mark.parametrize(
    "body", [b"", _BODY, _BODY + b" " * (65536 - len(_BODY)), _BODY + b" " * 65536]
)
async def test_error_snapshot_keeps_only_complete_bounded_body(body):
    """真实 hook 的边界，不把生产摘要当完整错误 JSON。"""
    capture = _capture()
    real = _real_transport(capture, lambda request: httpx.Response(404, content=body))
    try:
        with (
            capture.round_context(session_id="s", parent_call_id="p"),
            pytest.raises(RequestRejectedError) as caught,
        ):
            await _complete(ObservedTransport(real, capture))
        attempts = capture.attempts(session_id="s", parent_call_id="p")
        assert attempts[0].http[0].error_body == (body if len(body) <= 65536 else None)
        expected = "UNCOVERED" if 0 < len(body) <= 65536 else "FAIL"
        assert classify_live_failure(caught.value, attempts).status == expected
    finally:
        await real.aclose()


async def test_multiple_success_http_candidates_fail_identity_accessor():
    """一个 attempt 多 HTTP 不能猜最后一条。"""
    capture = _capture()
    client = capture.client_factory(_source())
    client._transport = httpx.MockTransport(lambda request: httpx.Response(200, json=_response()))
    try:
        with (
            capture.round_context(session_id="s", parent_call_id="p"),
            capture.attempt_context("a"),
        ):
            for _ in range(2):
                await client.post(
                    "https://example.test/v1/chat/completions",
                    json={"model": "gpt-5.5", "stream": False, "messages": _MESSAGES},
                )
        with pytest.raises(ValueError, match="多个成功"):
            capture.raw_identity(session_id="s", parent_call_id="p", call_id="a")
    finally:
        await client.aclose()


@pytest.mark.parametrize(
    "body", [b"data: not-json\n\n", b'data: {"choices":[]}\n\ndata: [DONE]\n\n']
)
async def test_sse_parser_failure_is_not_external_uncovered(body):
    capture = _capture(stream=True)
    real = _real_transport(capture, lambda request: httpx.Response(200, content=body))
    try:
        with (
            capture.round_context(session_id="s", parent_call_id="p"),
            pytest.raises(TransientError) as caught,
        ):
            await _complete(ObservedTransport(real, capture), stream=True)
        assert (
            classify_live_failure(
                caught.value, capture.attempts(session_id="s", parent_call_id="p")
            ).status
            == "FAIL"
        )
    finally:
        await real.aclose()


async def test_round_consumer_keeps_first_success_when_second_assertion_fails(tmp_path):
    """真实 GatewayClient 与报告出口连通；断言异常不得漏报告。"""
    from polygateway import GatewaySettings
    from tests.e2e.conftest import captured_chat_round, observed_client
    from tests.unit.test_config import _BASE_ENV

    settings = GatewaySettings.from_env("LLM", env=_BASE_ENV)
    settings = replace(settings, sources=(_source(),))
    capture = _capture()
    original_factory = capture.client_factory
    clients = []

    def factory(source):
        client = original_factory(source)
        client._transport = httpx.MockTransport(
            lambda request: httpx.Response(200, json=_response())
        )
        clients.append(client)
        return client

    capture.client_factory = factory

    def fail(response):
        raise AssertionError(_SECRET + _PROMPT)

    async with observed_client(settings, capture) as client:
        for index, validate in ((1, None), (2, fail)):
            _, verdict = await captured_chat_round(
                client,
                capture,
                run_id="run",
                matrix_id="matrix",
                round_index=index,
                output_dir=tmp_path,
                messages=_MESSAGES,
                models={"source": "gpt-5.5"},
                aliases={},
                stream=False,
                validate=validate,
            )
            assert verdict.status == ("PASS" if index == 1 else "FAIL")
    paths = list(tmp_path.rglob("*.md"))
    assert len(paths) == 2
    text = "".join(path.read_text() for path in paths)
    assert '"status": "PASS"' in text and '"status": "FAIL"' in text
    assert _SECRET not in text and _PROMPT not in text
    assert clients and all(client.is_closed for client in clients)


def test_partial_uncovered_and_failed_rounds_never_become_model_pass():
    from tests.live_evidence import combine_live_verdicts, qualify_live_rounds

    passed = LiveVerdict("PASS", "safe")
    uncovered = LiveVerdict("UNCOVERED", "safe")
    failed = LiveVerdict("FAIL", "safe")
    assert combine_live_verdicts([passed, uncovered]).status == "UNCOVERED"
    assert combine_live_verdicts([passed, uncovered, failed]).status == "FAIL"
    assert qualify_live_rounds([passed, passed], planned_rounds=3).status == "FAIL"
    assert qualify_live_rounds([passed, failed, passed], planned_rounds=3).status == "FAIL"


async def test_real_retry_success_call_id_selects_exact_raw_identity(tmp_path):
    """真实 RetryMW 先503再成功，两个并发 parent 的身份不串线。"""
    from polygateway import GatewaySettings
    from tests.e2e.conftest import captured_chat_round, observed_client
    from tests.unit.test_config import _BASE_ENV

    settings = GatewaySettings.from_env("LLM", env=_BASE_ENV)
    settings = replace(
        settings,
        sources=(_source(),),
        retry=replace(settings.retry, backoff_base_s=0.001, backoff_max_s=0.002),
    )
    capture = _capture()
    original_factory = capture.client_factory
    counts = {}

    async def handler(request):
        key = capture._round.get()
        counts[key] = counts.get(key, 0) + 1
        await asyncio.sleep(0)
        if counts[key] == 1:
            return httpx.Response(503, json={"error": {"type": "temporary"}, "model": "wrong"})
        return httpx.Response(200, json=_response())

    def factory(source):
        client = original_factory(source)
        client._transport = httpx.MockTransport(handler)
        return client

    capture.client_factory = factory
    async with observed_client(settings, capture) as client:

        async def one(index):
            return await captured_chat_round(
                client,
                capture,
                run_id="retry",
                matrix_id="retry",
                round_index=index,
                output_dir=tmp_path,
                messages=_MESSAGES,
                models={"source": "gpt-5.5"},
                aliases={},
                stream=False,
            )

        results = await asyncio.gather(one(1), one(2))
    assert all(verdict.status == "PASS" for _, verdict in results)
    assert len(counts) == 2 and set(counts.values()) == {2}
    for key in counts:
        attempts = capture.attempts(session_id=key[0], parent_call_id=key[1])
        assert len(attempts) == 2
        assert attempts[0].http[0].raw_identity == (False, None)
        assert capture.raw_identity(
            session_id=key[0], parent_call_id=key[1], call_id=attempts[1].call_id
        ) == (True, "gpt-5.5")


def test_live_model_provider_mapping_covers_default_capabilities_without_env_import():
    """L8 映射装配守卫离线化，只读取显式字面矩阵，不执行 .env 模块。"""
    import ast
    from pathlib import Path

    from polygateway.thinking import DEFAULT_CAPABILITIES

    tree = ast.parse((Path(__file__).parents[1] / "e2e/test_thinking_live.py").read_text())
    mapping = next(
        ast.literal_eval(node.value)
        for node in tree.body
        if isinstance(node, ast.Assign)
        and any(
            isinstance(target, ast.Name) and target.id == "_MODEL_PROVIDER"
            for target in node.targets
        )
    )
    assert not set(DEFAULT_CAPABILITIES) - mapping.keys()


async def test_unbuffered_error_is_insufficient_evidence():
    """response hook 不预读错误流；未完成缓冲不能借片段归因。"""
    capture = _capture()
    stream = _SSE()
    client = capture.client_factory(_source())
    client._transport = httpx.MockTransport(lambda request: httpx.Response(404, stream=stream))
    error = RequestRejectedError("safe", status_code=404)
    try:
        with (
            pytest.raises(RequestRejectedError),
            capture.round_context(session_id="s", parent_call_id="p"),
            capture.attempt_context("a"),
        ):
            async with client.stream(
                "POST",
                "https://example.test/v1/chat/completions",
                json={"model": "gpt-5.5", "messages": _MESSAGES, "stream": False},
            ):
                raise error
        attempts = capture.attempts(session_id="s", parent_call_id="p")
        assert attempts[0].http[0].error_body is None and stream.reads == 0
        assert classify_live_failure(error, attempts).status == "FAIL"
    finally:
        await client.aclose()


def test_pytest_report_hook_preserves_report_and_uses_safe_fallback(tmp_path, monkeypatch):
    """pytest wrapper generator 的 return 值是协议必需，不可按错误 LSP 建议删除。"""
    from types import SimpleNamespace

    from tests.e2e.conftest import pytest_runtest_makereport

    monkeypatch.chdir(tmp_path)
    report = SimpleNamespace(skipped=False, failed=True, when="call")
    hook = pytest_runtest_makereport(SimpleNamespace(nodeid="safe-node"), None)
    assert next(hook) is None
    with pytest.raises(StopIteration) as finished:
        hook.send(report)
    assert finished.value.value is report
    paths = list(tmp_path.rglob("*.md"))
    assert len(paths) == 1 and '"status": "FAIL"' in paths[0].read_text()