fix: normalize non-scalar prediction; harden predictions insert

app/harness/inference.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+import sqlite3
 from collections import defaultdict
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
@@ -162,6 +163,24 @@ def _to_text_field(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False)
 
 
+def _normalize_prediction(answer: object) -> str | None:
+    """归一化 prediction 落库值。
+
+    LLM 提交的 answer 有时是 list/dict（如 {'answer': ['B']}），sqlite 无法绑定
+    非标量类型直接入库会抛 ProgrammingError 击穿整轮 gather。None 保留（INFRA 空
+    预测语义，供正确率判定天然计错）；str 原样；其余 JSON 序列化为文本。
+
+    参数:
+        answer: LoopResult.result 中的 answer 原始值（可能是 None/str/list/dict）。
+
+    返回:
+        None（保留空预测语义）或可直接入库的字符串。
+    """
+    if answer is None or isinstance(answer, str):
+        return answer
+    return _to_text_field(answer)
+
+
 def _zero_result(run_id: str) -> InferenceResult:
     """空记录时的零值 InferenceResult。
 
@@ -419,7 +438,7 @@ async def _run_single_question(
         reasoning = _to_text_field(result_dict.get("reasoning", ""))
         record.update(
             {
-                "prediction": result_dict.get("answer"),
+                "prediction": _normalize_prediction(result_dict.get("answer")),
                 "evidence": evidence,
                 "reasoning": reasoning,
                 "steps_used": loop_result.steps_used,
@@ -442,8 +461,18 @@ async def _run_single_question(
     except Exception:
         logger.exception("[{}] QA {} 执行异常", qa.video_id, qa.question_id)
 
-    # prediction 必落库（try 外，无论成败）
+    # prediction 必落库（try 外，无论成败）；绑定异常降级为最小 error 行，不击穿 gather
-    await asyncio.to_thread(log.insert, "predictions", record)
+    try:
+        await asyncio.to_thread(log.insert, "predictions", record)
+    except (sqlite3.InterfaceError, sqlite3.ProgrammingError):
+        logger.exception("[{}] QA {} 落库绑定异常，降级为 error 行", qa.video_id, qa.question_id)
+        record["prediction"] = None
+        record["stop_reason"] = "error"
+        await asyncio.to_thread(
+            log.insert,
+            "predictions",
+            {k: v for k, v in record.items() if isinstance(v, (str, int, float, type(None)))},
+        )
     return record
 
 

tests/unit/test_harness_inference.py

@@ -559,6 +559,65 @@ class TestPredictionAlwaysWritten:
         assert rows[0]["prediction"] is None
 
 
+def _make_nonscalar_llm_response() -> LLMResponse:
+    """构造 submit_answer 提交非标量 answer（list）的 LLMResponse。"""
+    content = json.dumps(
+        {
+            "reflect": {"observation": "找到答案"},
+            "plan": {"next_step": "提交"},
+            "action": {
+                "tool": "submit_answer",
+                "args": {
+                    "answer": ["B"],
+                    "evidence": "证据文本",
+                    "reasoning": "推理过程",
+                },
+            },
+        }
+    )
+    return LLMResponse(
+        content=content,
+        thinking="思考过程",
+        model="test-model",
+        provider="test",
+        prompt_tokens=100,
+        completion_tokens=50,
+        latency_ms=200,
+        ttft_ms=30.0,
+        max_inter_token_ms=5.0,
+        cache_hit=False,
+        call_id="test-call-nonscalar",
+    )
+
+
+class TestNonScalarPrediction:
+    """非标量 prediction 归一化 + 落库加固测试。"""
+
+    @pytest.mark.asyncio
+    async def test_nonscalar_prediction_does_not_crash(self, harness_log: HarnessLog) -> None:
+        """submit_answer 返回 {'answer': ['B']} 时归一化落库，不抛 sqlite 绑定异常。"""
+        llm = AsyncMock()
+        llm.chat.return_value = _make_nonscalar_llm_response()
+
+        result = await run_inference(
+            [_make_question(answer="B")],
+            llm=llm,
+            tool_dispatch_fn=_stub_tool_dispatch,
+            prompt_builder=_stub_prompt_builder,
+            log=harness_log,
+            run_id="run-nonscalar",
+            concurrency=1,
+            max_steps=10,
+            skill_mode="auto",
+        )
+
+        assert result.total == 1
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        assert len(rows) == 1
+        # prediction 被 JSON 序列化为字符串，不再是 Python list
+        assert rows[0]["prediction"] == '["B"]'
+
+
 class TestPluginsFactory:
     """plugins_factory 调用测试。"""