test: keep cancelled embedding probe rounds incomplete

tests/e2e/test_embed_probe.py

@@ -53,6 +53,7 @@ async def test_probe_real_gateway_embeddings():
     transport = ObservedTransport(real, capture)
     run_id, parent, call_id = uuid4().hex, uuid4().hex, uuid4().hex
     verdict = LiveVerdict("FAIL", "轮次未完成")
+    completed_rounds = 0
     try:
         with capture.round_context(session_id=run_id, parent_call_id=parent):
             try:
@@ -65,10 +66,12 @@ async def test_probe_real_gateway_embeddings():
                 assert len(events) == 1 and request_is_valid(events[0])
                 assert result.dim > 0 and len(result.vectors) == 1
                 verdict = LiveVerdict("PASS", "向量形状与实发请求合格")
+                completed_rounds = 1
             except Exception as error:
                 verdict = classify_live_failure(
                     error, capture.attempts(session_id=run_id, parent_call_id=parent)
                 )
+                completed_rounds = 1
             finally:
                 write_live_round(
                     Path("tests/outputs/134/live"),
@@ -79,7 +82,7 @@ async def test_probe_real_gateway_embeddings():
                         "requested_model": source.model,
                         "provider": source.provider,
                         "planned_rounds": 1,
-                        "completed_rounds": 1,
+                        "completed_rounds": completed_rounds,
                         "status": verdict.status,
                         "reason": verdict.reason,
                         "session_id": run_id,

tests/unit/test_live_evidence.py

@@ -692,6 +692,7 @@ async def test_round_consumer_keeps_first_success_when_second_assertion_fails(tm
         ("503", "FAIL", "TransientError"),
         ("404", "UNCOVERED", "RequestRejectedError"),
         ("request_error", "FAIL", "TransientError"),
+        ("cancelled", "FAIL", None),
     ],
 )
 async def test_embed_probe_report_keeps_actual_source_and_round_identity(
@@ -731,18 +732,22 @@ async def test_embed_probe_report_keeps_actual_source_and_round_identity(
     original_factory = LiveCapture.client_factory
     clients = []
     calls = []
+    reached = asyncio.Event()
 
     def factory(capture, source):
         """保留真实取证 hooks、源与逻辑 ID，只隔离网络出口。"""
         client = original_factory(capture, source)
 
-        def handler(request):
+        async def handler(request):
             """提供完整成功／错误样本，敏感回显不得进入报告。"""
             payload = json.loads(request.content)
             assert payload["model"] == model == source.model
             assert source.model != env["LLM__MINIMAX__1__MODEL"]
             assert request.headers["Authorization"] == f"Bearer {_SECRET}"
             calls.append((source, capture._round.get(), capture._attempt.get().call_id))
+            if outcome == "cancelled":
+                reached.set()
+                await asyncio.Future()
             if outcome == "request_error":
                 raise httpx.ConnectError(_SECRET + _PROMPT, request=request)
             if outcome == "success":
@@ -765,7 +770,19 @@ async def test_embed_probe_report_keeps_actual_source_and_round_identity(
 
     monkeypatch.setattr(LiveCapture, "client_factory", factory)
     probe = namespace["test_probe_real_gateway_embeddings"]
-    if status == "PASS":
+    if outcome == "cancelled":
+        task = asyncio.create_task(probe())
+        try:
+            async with asyncio.timeout(5):
+                await reached.wait()
+                task.cancel()
+                with pytest.raises(asyncio.CancelledError):
+                    await task
+            assert task.cancelled()
+        finally:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+    elif status == "PASS":
         await probe()
     elif status == "UNCOVERED":
         with pytest.raises(pytest.skip.Exception):
@@ -785,7 +802,10 @@ async def test_embed_probe_report_keeps_actual_source_and_round_identity(
     assert row["status"] == status
     assert row["requested_model"] == source.model == model
     assert row["provider"] == source.provider == "minimax"
-    assert row["planned_rounds"] == row["completed_rounds"] == 1
+    assert row["planned_rounds"] == 1
+    assert row["completed_rounds"] == (0 if outcome == "cancelled" else 1)
+    if outcome == "cancelled":
+        assert row["reason"] == "轮次未完成"
     assert row["session_id"] == paths[0].parent.name == session_id
     assert row["parent_call_id"] == parent_call_id
     assert len({session_id, parent_call_id, call_id}) == 3
@@ -804,12 +824,13 @@ async def test_embed_probe_report_keeps_actual_source_and_round_identity(
             "503": 503,
             "404": 404,
             "request_error": 0,
+            "cancelled": 0,
         }[outcome]
     )
     if outcome in {"503", "404"}:
         assert event["error_body_complete"] is True
         assert event["machine_type"] == "model_not_found"
-    elif outcome == "request_error":
+    elif outcome in {"request_error", "cancelled"}:
         assert "无可配对响应" in row["evidence_notes"]