PolyGateway/tests/unit/test_live_evidence.py

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

import asyncio
import inspect
import json
from dataclasses import replace
from typing import Any

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()


@pytest.mark.parametrize(
    "damage", [None, "prefix", "role", "content", "unpaired", "budget", "wire"]
)
async def test_structured_reask_preserves_message_contract(tmp_path, damage, monkeypatch):
    """真实 StructuredMW 缺字段后重问成功；前缀、反馈结构及 wire 破坏均失败。"""
    from pydantic import BaseModel

    from polygateway import GatewaySettings
    from polygateway.middleware.structured import StructuredMW
    from tests.e2e.conftest import captured_chat_round, observed_client
    from tests.unit.test_config import _BASE_ENV

    class Answer(BaseModel):
        """离线最小结构化契约。"""

        answer: int
        reason: str

    settings = replace(
        GatewaySettings.from_env("LLM", env=_BASE_ENV),
        sources=(_source(),),
        structured_max_retries=1,
    )
    capture = _capture(messages_prefix_length=1, structured_max_retries=1)
    original_feedback = StructuredMW._with_feedback

    def feedback(self, *args):
        """只破坏重问产物，不替代生产阶梯或解析。"""
        request = original_feedback(self, *args)
        messages = [dict(message) for message in request.messages]
        if damage == "prefix":
            messages[0]["content"] = "changed"
        elif damage == "role":
            messages[-1]["role"] = "assistant"
        elif damage == "content":
            messages[-1]["content"] = ["wrong-type"]
        elif damage == "unpaired":
            messages.pop()
        elif damage == "budget":
            messages.extend(messages[-2:])
        return replace(request, messages=messages)

    monkeypatch.setattr(StructuredMW, "_with_feedback", feedback)
    requests = []

    def handler(request):
        requests.append(json.loads(request.content))
        body = _response()
        body["choices"][0]["message"]["content"] = (
            '{"answer":5}' if len(requests) == 1 else '{"answer":5,"reason":"sum"}'
        )
        return httpx.Response(200, json=body)

    original_factory = capture.client_factory

    def factory(source):
        client = original_factory(source)
        client._transport = httpx.MockTransport(handler)
        if damage == "wire":

            async def corrupt(request):
                payload = json.loads(request.content)
                if len(payload["messages"]) > 1:
                    payload["messages"][-1]["content"] = "well-shaped-but-corrupted"
                    request._content = json.dumps(payload).encode()

            client.event_hooks["request"].insert(0, corrupt)
        return client

    capture.client_factory = factory
    async with observed_client(settings, capture) as client:
        response, verdict = await captured_chat_round(
            client,
            capture,
            run_id="structured",
            matrix_id="reask",
            round_index=1,
            output_dir=tmp_path,
            messages=_MESSAGES,
            models={"source": "gpt-5.5"},
            aliases={},
            stream=False,
            structured=Answer,
        )
    assert response.structured_data.answer == 5
    assert len(requests) == 2
    assert verdict.status == ("PASS" if damage is None else "FAIL")
    text = "".join(path.read_text() for path in tmp_path.rglob("*.md"))
    assert _SECRET not in text and _PROMPT not in text


def _thinking_consumer():
    """仅加载 live 消费者定义与字面矩阵，跳过所有环境读取语句。"""
    import ast
    from pathlib import Path
    from types import SimpleNamespace

    path = Path(__file__).parents[1] / "e2e/test_thinking_live.py"
    tree = ast.parse(path.read_text())
    excluded = {
        "_ENV",
        "_HAS_SOURCE",
        "pytestmark",
        "_ROUNDS",
        "_TIER_ROUNDS",
        "_TIER_LONG_ROUNDS",
        "_TIER_CONCURRENCY",
    }
    tree.body = [
        node
        for node in tree.body
        if not (
            isinstance(node, ast.Assign)
            and any(
                isinstance(target, ast.Name) and target.id in excluded for target in node.targets
            )
        )
    ]
    namespace: dict[str, Any] = {
        "_ROUNDS": 3,
        "_TIER_ROUNDS": 2,
        "_TIER_LONG_ROUNDS": 1,
        "_TIER_CONCURRENCY": 1,
    }
    exec(compile(tree, str(path), "exec"), namespace)
    return SimpleNamespace(**namespace), namespace


@pytest.mark.parametrize(
    ("observation", "qualification", "expected"),
    [
        (O.ABSENT, "PASS", "UNCOVERED"),
        (O.OBSERVED, "PASS", "UNCOVERED"),
        (O.UNKNOWN, "PASS", "UNCOVERED"),
        (O.ABSENT, "FAIL", "FAIL"),
    ],
)
async def test_unregistered_candidate_has_no_disable_declaration(
    observation, qualification, expected
):
    """执行真实 T10 消费者，未登记不等于不可关闭，资格失败仍红。"""
    from types import SimpleNamespace

    live, namespace = _thinking_consumer()
    conclusions = []
    calls = []

    async def probe(model, effort, *, rounds, **kwargs):
        calls.append(rounds)
        return [
            {
                "verdict": LiveVerdict(qualification, "safe"),
                "response": SimpleNamespace(thinking_observation=observation),
            }
            for _ in range(rounds)
        ]

    namespace["_probe_effort"] = probe
    namespace["_conclude"] = lambda matrix, verdict, **kwargs: conclusions.append(verdict)
    await live.TestTierProbe().test_t10_none_direction_matches_declaration("claude-haiku-5")
    assert conclusions[0].status == expected
    assert calls == ([2, 1] if observation is not O.OBSERVED and qualification == "PASS" else [2])


@pytest.mark.parametrize("case", ["none", "tiers", "L8"])
async def test_capability_conclusions_link_models_and_all_subruns(tmp_path, case):
    """两个型号的 PASS／UNCOVERED 结论必须关联原件和完整短长／档位轮数。"""
    from contextlib import asynccontextmanager
    from types import SimpleNamespace

    from polygateway import GatewaySettings
    from tests.unit.test_config import _BASE_ENV

    live, namespace = _thinking_consumer()
    settings = GatewaySettings.from_env("LLM", env=_BASE_ENV)
    namespace["_OUT_DIR"] = tmp_path
    namespace["_tier_settings"] = lambda model: replace(
        settings, sources=(replace(_source(), model=model),)
    )
    namespace["_settings"] = lambda **kwargs: replace(
        settings, sources=(replace(_source(), **kwargs),)
    )
    namespace["enforce_verdict"] = lambda verdict: None

    @asynccontextmanager
    async def client(*args, **kwargs):
        yield None

    async def round_call(client, capture, **kwargs):
        model = kwargs["models"]["source"]
        observation = (
            O.UNKNOWN if model == "gpt-5.4" else O.OBSERVED if case == "tiers" else O.ABSENT
        )
        write_live_round(
            kwargs["output_dir"],
            run_id=kwargs["run_id"],
            matrix_id=kwargs["matrix_id"],
            round_index=kwargs["round_index"],
            safe_fields={
                "requested_model": model,
                "session_id": kwargs["run_id"],
                "status": "PASS",
                "thinking_observation": observation,
            },
        )
        return SimpleNamespace(thinking_observation=observation), LiveVerdict("PASS", "safe")

    namespace["observed_client"] = client
    namespace["captured_chat_round"] = round_call
    for model in ("gpt-5.5", "gpt-5.4"):
        if case == "none":
            await live.TestTierProbe().test_t10_none_direction_matches_declaration(model)
        elif case == "tiers":
            await live.TestTierProbe().test_t10_declared_tiers_actually_reason(model)
        else:
            await live.TestCapabilityDrift().test_declared_capability_matches_reality(model)
    matrix = {"none": "T10-none", "tiers": "T10-tiers", "L8": "L8"}[case]
    finals = list(tmp_path.rglob(f"{matrix}-0-*.md"))
    assert len(finals) == 2
    seen = set()
    for path in finals:
        row = json.loads(path.read_text().split("```json\n")[1].split("\n```")[0])
        assert "requested_model" in row, "结论缺型号，无法关联逐轮原件"
        model = row["requested_model"]
        seen.add(model)
        assert row["session_id"] == path.parent.name
        assert row["status"] == ("PASS" if model == "gpt-5.5" else "UNCOVERED")
        assert row["proposition"]
        subruns = row["subruns"]
        expected_groups = (
            2
            if case == "none"
            else len(
                [
                    effort
                    for effort in live.DEFAULT_CAPABILITIES[model].supported_efforts
                    if effort is not live.Effort.NONE
                ]
            )
            if case == "tiers"
            else 1
        )
        assert len(subruns) == expected_groups
        assert row["planned_rounds"] == sum(subrun["planned_rounds"] for subrun in subruns)
        assert row["completed_rounds"] == row["planned_rounds"]
        for subrun in subruns:
            originals = [
                p
                for p in path.parent.glob(f"{subrun['matrix_id']}-*.md")
                if p.name[len(subrun["matrix_id"]) + 1 :].split("-", 1)[0].isdigit()
                and not p.name.startswith(subrun["matrix_id"] + "-0-")
            ]
            assert len(originals) == subrun["planned_rounds"] == subrun["completed_rounds"]
            for original in originals:
                data = json.loads(original.read_text().split("```json\n")[1].split("\n```")[0])
                assert data["requested_model"] == model
                assert data["session_id"] == row["session_id"]
    assert seen == {"gpt-5.5", "gpt-5.4"}


@pytest.mark.parametrize("machine_type", ["model_not_found", _SECRET, "arbitrary-upstream-text"])
def test_safe_machine_type_retains_only_known_enum(tmp_path, machine_type):
    """认可机器枚举可复核；任意机器正文与 sentinel 不得落盘。"""
    _, attempts = _failure(body=json.dumps({"error": {"type": machine_type}}).encode())
    fields = {"attempts": safe_attempts(attempts)}
    event = fields["attempts"][0]["http"][0]
    assert event.get("machine_type") == (
        "model_not_found" if machine_type == "model_not_found" else "omitted"
    )
    path = write_live_round(
        tmp_path, run_id="safe", matrix_id="machine", round_index=1, safe_fields=fields
    )
    text = path.read_text()
    assert _SECRET not in text and "arbitrary-upstream-text" not in text
    if machine_type != "model_not_found":
        event["machine_type"] = machine_type
        with pytest.raises(ValueError, match="机器"):
            write_live_round(
                tmp_path, run_id="unsafe", matrix_id="machine", round_index=1, safe_fields=fields
            )


async def test_structured_first_attempt_requires_exact_initial_messages():
    """结构化窄规则不能允许首轮凭空带入一对反馈。"""
    capture = _capture(messages_prefix_length=1, structured_max_retries=1)
    real = _real_transport(capture, lambda request: httpx.Response(200, json=_response()))
    try:
        with capture.round_context(session_id="first", parent_call_id="parent"):
            await ObservedTransport(real, capture).complete(
                messages=[
                    *_MESSAGES,
                    {"role": "assistant", "content": "old"},
                    {"role": "user", "content": "retry"},
                ],
                source=_source(),
                stream=False,
                overlay={},
                call_id="first",
                reasoning_effort=None,
            )
        event = capture.attempts(session_id="first", parent_call_id="parent")[0].http[0]
        assert not request_is_valid(event)
    finally:
        await real.aclose()