test: apply evidence-based live checks without hiding regressions

tests/e2e/test_compat_projects.py

@@ -1,98 +1,119 @@
-"""GovDoc 与 Video-Tree 最小接入冒烟(2026-07-20 拍板: 两个项目都做)。
-
-复刻两项目的真实调用点形态,对真实网关跑一次治理调用,证明"调用点零改动
-迁移"成立;并验证 VT 现有平铺键名(LLM_TIMEOUT 等)可直接装配。
-reference/ 只读——本文件只 import 其 Protocol,绝不修改。
-"""
+"""历史接入调用形态的真实冒烟；不能替代缺失下游的现行配置验收。"""
 
 import os
 import sys
 from pathlib import Path
+from uuid import uuid4
 
 import pytest
 from dotenv import dotenv_values
 
-from polygateway import GatewayClient
+from polygateway import Effort, GatewayClient, GatewaySettings
+from tests.e2e.conftest import (
+    LiveCapture,
+    captured_chat_round,
+    chat_expectations,
+    enforce_verdict,
+    observed_client,
+    source_controls,
+)
+from tests.live_evidence import write_live_round
 
 _REPO = Path(__file__).resolve().parents[2]
 _ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None}
-_HAS_SOURCE = any(k.split("__")[0] == "LLM" and k.endswith("__API_KEY") for k in _ENV)
-
-# 真实网关调用: 与 test_thinking_live.py 同待遇标 slow(pytest addopts 默认排除,
-# 显式 `pytest -m slow` 运行)。理由是这些用例的成败取决于网关此刻快不快,而
-# pre-commit 关卡跑全套件——网关一抖就挡住与之无关的提交,久了会把"测试红了
-# 先怀疑网关"变成惯性,真 bug 也会被当成抖动重试掉。发版清单负责让它们真跑。
+_HAS_SOURCE = any(k.startswith("LLM__") and k.endswith("__API_KEY") for k in _ENV)
 pytestmark = [
     pytest.mark.slow,
-    pytest.mark.skipif(
-        not _HAS_SOURCE, reason="需真实网关凭据: 在 .env 配置 LLM__{PROVIDER}__1__*"
-    ),
+    pytest.mark.skipif(not _HAS_SOURCE, reason="缺少矩阵必需凭据，未覆盖"),
 ]
+_OUT = Path("tests/outputs/134/live")
 
 
-@pytest.fixture
-async def client():
-    c = GatewayClient.from_env("LLM", env=_ENV)
-    yield c
-    await c.aclose()
+async def _call_shape(matrix, **kwargs):
+    """session/parent 总由逐轮 UUID 传入；保留 cache_salt 调用形态。"""
+    settings = GatewaySettings.from_env("LLM", env=_ENV)
+    messages = [{"role": "user", "content": "Reply with exactly: compatibility-ok"}]
+    capture = LiveCapture(
+        expectations=chat_expectations(
+            settings, messages=messages, stream=True, controls=source_controls(settings)
+        )
+    )
+
+    def validate(response):
+        assert response.content.strip() and response.call_id
+
+    async with observed_client(settings, capture) as client:
+        _, verdict = await captured_chat_round(
+            client,
+            capture,
+            run_id=uuid4().hex,
+            matrix_id=matrix,
+            round_index=1,
+            output_dir=_OUT,
+            messages=messages,
+            models={s.name: s.model for s in settings.sources},
+            providers={s.name: s.provider for s in settings.sources},
+            source_efforts={
+                s.name: s.reasoning_effort
+                if s.reasoning_effort is not None
+                else (Effort.AUTO if s.enable_thinking else Effort.NONE)
+                if s.enable_thinking is not None
+                else None
+                for s in settings.sources
+            },
+            aliases={},
+            validate=validate,
+            **kwargs,
+        )
+        enforce_verdict(verdict)
 
 
 class TestGovDocOnboarding:
-    """GovDoc agent/loop.py:377 调用形态: session_id + parent_call_id。"""
+    """历史 session_id＋parent_call_id 调用点契约。"""
 
-    async def test_call_site_shape_runs_governed(self, client):
-        response = await client.chat(
-            [{"role": "user", "content": "Reply with exactly: govdoc-ok"}],
-            session_id="govdoc-e2e",
-            parent_call_id="step-1",
-        )
-        assert response.content.strip()
-        assert response.call_id  # GovernedLLMClient 契约字段全在
+    async def test_call_site_shape_runs_governed(self):
+        await _call_shape("compat-parent")
 
-    async def test_structural_protocol_match(self, client):
+    async def test_structural_protocol_match(self):
+        """外部 Protocol 缺包单列未覆盖；合成契约另在 unit 跑。"""
+        run_id = uuid4().hex
         sys.path.insert(0, str(_REPO / "reference/GovDoc-SaaS/packages/docagent-core/src"))
         try:
             from docagent_core.protocols import LLMProvider
         except ImportError:
-            pytest.skip("GovDoc protocols 依赖不可导入(结构断言已由单测兜底覆盖)")
+            write_live_round(
+                _OUT,
+                run_id=run_id,
+                matrix_id="external-protocol",
+                round_index=0,
+                safe_fields={
+                    "status": "UNCOVERED",
+                    "reason": "外部 Protocol 包缺失；未验证真实下游",
+                },
+            )
+            pytest.skip("外部 Protocol 包缺失，未覆盖")
         finally:
             sys.path.pop(0)
-        assert isinstance(client, LLMProvider)
+        status = "FAIL"
+        client = None
+        try:
+            client = GatewayClient.from_env("LLM", env=_ENV)
+            assert isinstance(client, LLMProvider)
+            status = "PASS"
+        finally:
+            if client is not None:
+                await client.aclose()
+            write_live_round(
+                _OUT,
+                run_id=run_id,
+                matrix_id="external-protocol",
+                round_index=0,
+                safe_fields={"status": status, "reason": "外部 Protocol 结构契约，不是模型能力"},
+            )
 
 
 class TestVideoTreeOnboarding:
-    """VT loop.py:336 调用形态: session_id + cache_salt(跨 epoch 重采样)。"""
+    """历史 cache_salt 调用点契约，平铺键装配已移至 unit。"""
 
-    async def test_call_site_shape_with_cache_salt(self, client):
-        response = await client.chat(
-            [{"role": "user", "content": "Reply with exactly: vt-ok"}],
-            session_id="vt-e2e",
-            cache_salt="epoch-1",
-        )
-        assert response.content.strip()
-
-    async def test_flat_legacy_keys_assemble(self):
-        """VT 现有键名(LLM_TIMEOUT/LLM_MAX_RETRIES 等)零改名装配成功。"""
-        source_keys = {k: v for k, v in _ENV.items() if k.split("__")[0] == "LLM" and "__" in k}
-        flat_env = {
-            **source_keys,
-            # 与 .env 的 LLM__MINIMAX__1__TIMEOUT_S 同值。取 120(VT 旧值)会让本用例的
-            # 超时比生产配置还紧一半,在慢网关上必然间歇红——而本用例断言的是平铺
-            # 键名能否解析成 SourceConfig.timeout_s,超时取值本身不是被测对象
-            "LLM_TIMEOUT": "300",
-            "LLM_MAX_RETRIES": "3",
-            "LLM_RETRY_BASE_DELAY": "2.0",
-            "LLM_RETRY_MAX_DELAY": "30.0",
-            "LLM_CIRCUIT_BREAKER_THRESHOLD": "5",
-            "LLM_CIRCUIT_BREAKER_COOLDOWN": "60",
-            "LLM_TTFT_TIMEOUT": "30",
-            "LLM_INTER_TOKEN_TIMEOUT": "15",
-            "PGW_CACHE_BACKEND": "none",
-            "PGW_TELEMETRY_BACKEND": "none",
-        }
-        client = GatewayClient.from_env("LLM", env=flat_env)
-        try:
-            resp = await client.chat([{"role": "user", "content": "Reply: flat-ok"}])
-            assert resp.content.strip()
-        finally:
-            await client.aclose()
+    async def test_call_site_shape_with_cache_salt(self):
+        await _call_shape("compat-salt", cache_salt="epoch-1")