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

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"