fix: predictions row carries arm run_id under shared gate_log; drain evolve gather on failure (algo #6)

app/harness/inference.py

@@ -409,7 +409,11 @@ async def _run_single_question(
     返回:
         预测结果字典（含 video_id, question_id, prediction, answer 等）。
     """
+    # run_id 必须显式入 record:HarnessLog.insert 缺省用**实例** run_id 填充,
+    # 连续并发 gate 共享单一 gate_log(实例 run_id 为 step 级)时,各臂行必须
+    # 落自己的臂 run_id,否则 validate 回读 _load_run_rows(臂 run_id) 为空。
     record: dict[str, Any] = {
+        "run_id": run_id,
         "video_id": qa.video_id,
         "question_id": qa.question_id,
         "task_type": qa.task_type,

app/harness/runner.py

@@ -1370,13 +1370,17 @@ class Runner:
                 rejected=state.rejected_buffer.get(task_type, []),
             )
 
-        records = dict(
-            zip(
-                active_types,
-                await asyncio.gather(*[_evolve_one(t) for t in active_types]),
-                strict=True,
-            )
-        )
+        # 首异常先取消其余进化任务并排水再向上传播(与 validate_skills_concurrent
+        # 同款语义):避免失败后残留 in-flight LLM 任务与 pending task 警告。
+        tasks = [asyncio.ensure_future(_evolve_one(t)) for t in active_types]
+        try:
+            evolved = await asyncio.gather(*tasks)
+        except BaseException:
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            raise
+        records = dict(zip(active_types, evolved, strict=True))
 
         # 无真实改动的题型照旧写 skipped 后出队
         gated: dict[str, EvolutionRecord] = {}

tests/unit/test_gate_batch_parallel.py

@@ -236,3 +236,65 @@ def test_gate_batch_parallel_evolve_and_alphabetical_settle(
         assert "_gate_" in spec.gate_run_prefix
     assert isinstance(captured["log"], _FakeHarnessLog)
     assert callable(captured["run_inference"])
+
+
+# ---------------------------------------------------------------------------
+# 共享 gate_log 的 run_id 契约(真 SQLite,Codex 质量审 C1):
+# HarnessLog.insert 缺省用实例 run_id 填充;record 自带 run_id 必须覆盖它,
+# 否则连续并发 gate 下所有臂的 predictions 会落成 step 级 run_id,
+# validate 按臂 run_id 回读为空 → gate 静默废掉。
+# ---------------------------------------------------------------------------
+
+
+def test_harness_log_insert_record_run_id_overrides_instance(tmp_path: Path) -> None:
+    """record 自带 run_id 覆盖实例 run_id;缺省时回落实例 run_id(锁死 enriched.update 语义)。"""
+    from app.harness.inference import PREDICTIONS_SCHEMA
+    from app.harness.log import HarnessLog
+
+    with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as log:
+        log.create_table("predictions", PREDICTIONS_SCHEMA)
+        log.insert(
+            "predictions",
+            {"run_id": "run_e1_s0_gate_a_base_u0", "question_id": "q1", "prediction": "A"},
+        )
+        log.insert("predictions", {"question_id": "q2", "prediction": "B"})
+        rows = log.query("SELECT question_id, run_id FROM predictions ORDER BY question_id")
+    assert [(r["question_id"], r["run_id"]) for r in rows] == [
+        ("q1", "run_e1_s0_gate_a_base_u0"),
+        ("q2", "gate_e1_s0"),
+    ]
+
+
+def test_inference_prediction_row_carries_arm_run_id(tmp_path: Path) -> None:
+    """经共享 gate_log 落库的 prediction 行 run_id 必须是臂 run_id 而非实例 run_id。
+
+    prompt_builder 抛错走异常路径即落库,无需真实 LLM;
+    该路径与成功路径共用同一 record 初始 dict,契约一致。
+    """
+    from app.harness.inference import PREDICTIONS_SCHEMA, _run_single_question
+    from app.harness.log import HarnessLog
+
+    def _broken_prompt_builder(qa: GeneratedQuestion) -> tuple[str, str]:
+        raise RuntimeError("测试注入:跳过真实推理")
+
+    async def _noop_dispatch(tool_name: str, args: dict, *, context: dict) -> str:
+        raise NotImplementedError
+
+    with HarnessLog(str(tmp_path / "harness.db"), "gate_e1_s0") as gate_log:
+        gate_log.create_table("predictions", PREDICTIONS_SCHEMA)
+        asyncio.run(
+            _run_single_question(
+                _question("q-arm", _TYPE_A),
+                llm=object(),  # prompt_builder 先抛错,不会触达
+                tool_dispatch_fn=_noop_dispatch,
+                prompt_builder=_broken_prompt_builder,
+                log=gate_log,
+                max_steps=3,
+                plugins=[],
+                run_id="run_e1_s0_gate_action-reasoning_cand_u0",
+            )
+        )
+        rows = gate_log.query("SELECT run_id, stop_reason FROM predictions")
+    assert len(rows) == 1
+    assert rows[0]["run_id"] == "run_e1_s0_gate_action-reasoning_cand_u0"
+    assert rows[0]["stop_reason"] == "error"

tests/unit/test_harness_inference.py

@@ -117,8 +117,9 @@ def harness_log(tmp_path: Any, request: Any) -> HarnessLog:
     """创建临时 HarnessLog 实例。
 
     使用 test 节点名称的 hash 作为 db 文件名，避免冲突。
-    run_id 固定为 "test-run"，实际 run_inference 中传入的 run_id
-    由 HarnessLog.insert 自动覆盖为 HarnessLog 构造时的值。
+    实例 run_id 固定为 "test-run"；predictions 行的 run_id 由 inference
+    record 显式携带（run_inference 传入值），不回落实例 run_id——
+    连续并发 gate 共享单一 HarnessLog 的契约。
     """
     db_name = f"harness_{id(request)}.db"
     db_path = str(tmp_path / db_name)
@@ -528,8 +529,9 @@ class TestPredictionAlwaysWritten:
         assert result.correct == 0
         assert result.stop_reason_counts.get("error") == 1
 
-        # 验证 DB 中的记录（HarnessLog.insert 使用构造时的 run_id）
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # 验证 DB 中的记录（record 显式携带 run_inference 的 run_id，
+        # 不再回落 HarnessLog 实例 run_id——连续并发 gate 共享 log 的契约）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-error",))
         assert len(rows) == 1
         assert rows[0]["stop_reason"] == "error"
         assert rows[0]["prediction"] is None
@@ -553,8 +555,8 @@ class TestPredictionAlwaysWritten:
         )
 
         assert result.total == 1
-        # HarnessLog.insert 使用构造时的 run_id
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # record 显式携带 run_inference 的 run_id（共享 log 契约）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-parse-err",))
         assert len(rows) == 1
         assert rows[0]["prediction"] is None
 
@@ -612,7 +614,7 @@ class TestNonScalarPrediction:
         )
 
         assert result.total == 1
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-nonscalar",))
         assert len(rows) == 1
         # prediction 被 JSON 序列化为字符串，不再是 Python list
         assert rows[0]["prediction"] == '["B"]'

tests/unit/test_inference_pair_aggregate.py

@@ -332,8 +332,8 @@ class TestRunInferencePairEndToEnd:
         assert result.total == 1
         assert result.correct == 1
 
-        # 逐题溯源：predictions 表两条 record 都在
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # 逐题溯源：predictions 表两条 record 都在（record 显式携带传入的 run_id）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-pair-e2e",))
         qids = {r["question_id"] for r in rows}
         assert qids == {"po", "pm"}
 
@@ -360,6 +360,6 @@ class TestRunInferencePairEndToEnd:
         )
 
         assert result.total == 1  # single 存活，孤儿剔除
-        # 逐题溯源：孤儿题仍逐题落库（推理不变）
-        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("test-run",))
+        # 逐题溯源：孤儿题仍逐题落库（推理不变；record 显式携带传入的 run_id）
+        rows = harness_log.query("SELECT * FROM predictions WHERE run_id = ?", ("run-orphan-e2e",))
         assert {r["question_id"] for r in rows} == {"s1", "po"}