diff --git a/tests/unit/test_embedding.py b/tests/unit/test_embedding.py index 1880f4d..94b2c6c 100644 --- a/tests/unit/test_embedding.py +++ b/tests/unit/test_embedding.py @@ -533,3 +533,70 @@ class TestEmbeddingSettings: s = EmbeddingSettings.from_env("EMBED", env=self._ENV) client = EmbeddingClient.from_settings(s) assert isinstance(client, EmbeddingClient) + + +class TestReasonlessTelemetryContract: + """从真实客户端到落库，误配推理配置也不能产生推理档。""" + + @pytest.mark.parametrize("config", [{"enable_thinking": True}, {"reasoning_effort": "high"}]) + @pytest.mark.parametrize("backend", ["memory", "sqlite"]) + async def test_failed_then_successful_attempts_have_null_effort( + self, config, backend, tmp_path + ): + import sqlite3 + + from polygateway.telemetry.sqlite import SQLiteRecorder + + path = tmp_path / "embed.sqlite" + recorder = ( + _MemoryRecorder() if backend == "memory" else SQLiteRecorder(path, auto_migrate=True) + ) + client, _ = _embed_client( + [_src(**config)], [TransientError("retry"), "ok"], telemetry=recorder + ) + try: + await client.embed(["text"], session_id="run", parent_call_id="embed") + if backend == "memory": + rows = [(r["error"], r["reasoning_effort"]) for r in recorder.rows] + else: + with sqlite3.connect(path) as db: + rows = db.execute("SELECT error, reasoning_effort FROM llm_calls").fetchall() + assert len(rows) == 2 + assert sum(bool(error) for error, _ in rows) == 1 + assert [tier for _, tier in rows] == [None, None] + finally: + await client.aclose() + if backend == "sqlite": + recorder.close() + + @pytest.mark.parametrize("exhausted", [False, True]) + async def test_failed_attempts_still_have_null_effort(self, exhausted): + from polygateway.errors import AllSourcesExhausted + + script = ( + [TransientError("retry")] * 3 if exhausted else [RequestRejectedError("bad request")] + ) + recorder = _MemoryRecorder() + client, _ = _embed_client([_src(enable_thinking=True)], script, telemetry=recorder) + with pytest.raises(AllSourcesExhausted if exhausted else RequestRejectedError): + await client.embed(["text"]) + assert len(recorder.rows) == len(script) + assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows) + + async def test_embedding_wire_ignores_reasoning_configuration(self): + seen = [] + + def handler(request): + seen.append(json.loads(request.content)) + return httpx.Response(200, json=_ok_body([[1.0]], usage={"prompt_tokens": 1})) + + transport = _transport_with(handler) + try: + await transport.embed( + texts=["text"], + source=_src(enable_thinking=True, reasoning_effort="high"), + call_id="wire", + ) + assert seen == [{"model": "embed-1", "input": ["text"]}] + finally: + await transport.aclose() diff --git a/tests/unit/test_monkey_ocr.py b/tests/unit/test_monkey_ocr.py index 77c4461..f0c100e 100644 --- a/tests/unit/test_monkey_ocr.py +++ b/tests/unit/test_monkey_ocr.py @@ -405,3 +405,31 @@ class TestLifecycle: await t.check_health(source=_source()) await t.aclose() await t.aclose() + + +@pytest.mark.parametrize("method", ["recognize_text", "parse_layout"]) +async def test_ocr_wire_does_not_send_reasoning_configuration(method): + """真实 multipart 与 ZIP 下载两段均不发送源级推理配置。""" + sent = [] + + def handler(request): + sent.append(request) + if request.method == "GET": + return httpx.Response(200, content=_zip_bytes()) + return httpx.Response( + 200, json=_text_body() if method == "recognize_text" else _parse_body() + ) + + transport = _transport_for(handler) + try: + await getattr(transport, method)( + image=b"image", + source=_source(enable_thinking=True, reasoning_effort="high"), + call_id="wire", + ) + assert len(sent) == (1 if method == "recognize_text" else 2) + for request in sent: + for key in (b"reasoning_effort", b"enable_thinking", b"thinking_budget"): + assert key not in request.content + finally: + await transport.aclose() diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index 85e875b..00e435b 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -564,3 +564,54 @@ class TestAssembly: client = OcrClient.from_env("OCR", env=dict(self._ENV)) await client.aclose() await client.aclose() + + +class TestReasonlessTelemetryContract: + """text/layout 两个入口分别验证错误行不受源级推理配置污染。""" + + @pytest.mark.parametrize( + "method,action", [("recognize_text", "text"), ("parse_layout", "layout")] + ) + @pytest.mark.parametrize("config", [{"enable_thinking": True}, {"reasoning_effort": "high"}]) + @pytest.mark.parametrize("backend", ["memory", "sqlite"]) + async def test_failed_then_successful_attempts_have_null_effort( + self, method, action, config, backend, tmp_path + ): + import sqlite3 + + from polygateway.telemetry.sqlite import SQLiteRecorder + + path = tmp_path / "ocr.sqlite" + recorder = ( + _MemoryRecorder() if backend == "memory" else SQLiteRecorder(path, auto_migrate=True) + ) + client, _, _ = _client( + [_src(**config)], [TransientError("retry"), action], telemetry=recorder + ) + try: + await getattr(client, method)(b"image", session_id="run", parent_call_id=method) + if backend == "memory": + rows = [(r["error"], r["reasoning_effort"]) for r in recorder.rows] + else: + with sqlite3.connect(path) as db: + rows = db.execute("SELECT error, reasoning_effort FROM llm_calls").fetchall() + assert len(rows) == 2 + assert sum(bool(error) for error, _ in rows) == 1 + assert [tier for _, tier in rows] == [None, None] + finally: + await client.aclose() + if backend == "sqlite": + recorder.close() + + @pytest.mark.parametrize("method", ["recognize_text", "parse_layout"]) + @pytest.mark.parametrize("exhausted", [False, True]) + async def test_failed_attempts_still_have_null_effort(self, method, exhausted): + script = ( + [TransientError("retry")] * 3 if exhausted else [RequestRejectedError("bad request")] + ) + recorder = _MemoryRecorder() + client, _, _ = _client([_src(enable_thinking=True)], script, telemetry=recorder) + with pytest.raises(AllSourcesExhausted if exhausted else RequestRejectedError): + await getattr(client, method)(b"image") + assert len(recorder.rows) == len(script) + assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows) diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 693db6f..49aacca 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -2839,3 +2839,90 @@ class TestPostgresFailureClassification: assert [s for s in conn.statements if s.startswith("INSERT INTO llm_calls")] assert recorder.telemetry_status.degraded is False + + +class TestConcurrentReasoningPathContracts: + """真实治理链路并发时，按逻辑调用归组且 attempts 不串档。""" + + async def test_chat_and_reasonless_clients_keep_distinct_rows(self): + from polygateway.errors import TransientError + from tests.unit.test_client import _client as chat_client + from tests.unit.test_client import _source, _sse + from tests.unit.test_embedding import _embed_client + from tests.unit.test_embedding import _src as embed_source + from tests.unit.test_ocr_client import _client as ocr_client + from tests.unit.test_ocr_client import _src as ocr_source + + recorder = _MemoryRecorder() + calls = 0 + + def handler(request): + nonlocal calls + calls += 1 + if calls == 1: + import httpx + + return httpx.Response(503, json={"error": {"message": "retry"}}) + return _sse() + + chat = chat_client( + sources=[_source(provider="zhipu", model="glm-5.3", effort_fallback="nearest")], + handler=handler, + telemetry=recorder, + retry=RetryPolicy(3, 0.001, 0.01), + ) + embed, _ = _embed_client( + [embed_source(enable_thinking=True)], + [TransientError("retry"), "ok"], + telemetry=recorder, + ) + ocr, _, _ = ocr_client( + [ocr_source(enable_thinking=True)], + [TransientError("retry"), "text"], + telemetry=recorder, + ) + try: + await asyncio.gather( + chat.chat([], reasoning_effort="medium", session_id="run", parent_call_id="chat"), + embed.embed(["text"], session_id="run", parent_call_id="embed"), + ocr.recognize_text(b"image", session_id="run", parent_call_id="ocr"), + ) + groups = { + name: [r for r in recorder.rows if r["parent_call_id"] == name] + for name in ("chat", "embed", "ocr") + } + assert len(recorder.rows) == 6 + assert len({r["call_id"] for r in recorder.rows}) == 6 + assert all(r["session_id"] == "run" for r in recorder.rows) + assert [r["reasoning_effort"] for r in groups["chat"]] == ["medium", "low"] + for name in ("embed", "ocr"): + assert len(groups[name]) == 2 + assert [r["reasoning_effort"] for r in groups[name]] == [None, None] + finally: + await chat._transport.aclose() + await chat.aclose() + await embed.aclose() + await ocr.aclose() + + async def test_chat_sugar_failure_records_auto_through_retry(self): + import httpx + + from polygateway.errors import AllSourcesExhausted + from tests.unit.test_client import _client, _source + + recorder = _MemoryRecorder() + client = _client( + sources=[_source(model="qwen3.7-plus", enable_thinking=True)], + handler=lambda request: httpx.Response(503), + telemetry=recorder, + retry=RetryPolicy(1, 0.001, 0.01), + ) + try: + with pytest.raises(AllSourcesExhausted): + await client.chat([]) + attempts = [r for r in recorder.rows if r["source_name"]] + assert len(attempts) == 1 + assert attempts[0]["reasoning_effort"] == "auto" + assert attempts[0]["error"] + finally: + await client._transport.aclose()