test(harness): 补 _ladder_units 直测 + 澄清 gate 观测表 unit_id 口径注释

M1：quadrant_pair / gate_evidence 的 question_id 列注释与 write_* docstring
更正为承载 unit_id（single=question_id、pair=pair_id），提示逐题明细在
predictions 表溯源、按 pair_id join 真实 question 表会 join 不上。

M2：给 _ladder_units 补直接单测——纯非 AR 恒等（unit 序==原题序、
unit_id==question_id）、混格交错保持信息阶梯序（按单元最早出现下标重排、
pair 整锁）、且与 build_units 的 single-first 默认序显式区分（防阶梯序被污染）。

app/harness/observation.py

@@ -87,6 +87,8 @@ _HOLDOUT_COLS: dict[str, str] = {
 _QUADRANT_COLS: dict[str, str] = {
     "epoch": "INTEGER",
     "step": "INTEGER",
+    # question_id 列承载 unit_id（single=question_id，pair=pair_id）；
+    # 逐题明细在 predictions 表溯源，按 pair_id join 真实 question 表会 join 不上。
     "question_id": "TEXT",
     "task_type": "TEXT",
     "prev_correct": "INTEGER",
@@ -98,6 +100,8 @@ _GATE_EVIDENCE_COLS: dict[str, str] = {
     "epoch": "INTEGER",
     "step": "INTEGER",
     "task_type": "TEXT",
+    # question_id 列承载 unit_id（single=question_id，pair=pair_id）；
+    # 逐题明细在 predictions 表溯源，按 pair_id join 真实 question 表会 join 不上。
     "question_id": "TEXT",
     "block_idx": "INTEGER",
     "baseline_correct": "INTEGER",
@@ -275,7 +279,7 @@ def write_quadrant_pairs(
     step: int,
     pairs: list[dict[str, Any]],
 ) -> None:
-    """落 quadrant_pair 多行：fast gate 后逐题四象限（prev/curr 翻转 + category）落库。
+    """落 quadrant_pair 多行：fast gate 后按 **unit** 四象限（prev/curr 翻转 + category）落库。
 
     参数:
         db_path: SQLite 路径。
@@ -283,7 +287,10 @@ def write_quadrant_pairs(
         epoch: 轮次（1-based）。
         step: epoch 内 step 序号（0-based）。
         pairs: 每条含 question_id/task_type/prev_correct/curr_correct/category；
-            prev_correct/curr_correct 为 bool，写库前转 0/1。
+            question_id 字段承载 **unit_id**（single=question_id，pair=pair_id，
+            与 gate e-process 同粒度）——逐题明细在 predictions 表溯源，按 pair_id
+            join 真实 question 表会 join 不上；prev_correct/curr_correct 为 bool，
+            写库前转 0/1。
 
     关键实现:
         用 insert_many 批量落库；pairs 为空时只建表不插入（fast gate 无翻转的极端情况）。
@@ -326,16 +333,18 @@ def write_gate_evidence(
     step: int,
     rows: list[dict[str, Any]],
 ) -> None:
-    """落 gate_evidence 逐题行：CE-Gate 每次决策的可回放审计记录。
+    """落 gate_evidence 单元行：CE-Gate 每次决策的可回放审计记录（unit 口径）。
 
     参数:
         db_path: SQLite 路径。
         run_id: 训练 run ID。
         epoch: 该 gate 所属的轮次（1-based）。
         step: epoch 内 step 序号（0-based）。
-        rows: 每题一行，含 question_id/task_type/block_idx/baseline_correct/
-            candidate_correct/e_value（该题所在块判定后的累计 e 值）/
-            stop_reason（仅最后一题携带最终 stop_reason，其余空串）。
+        rows: 每 **单元** 一行，含 question_id/task_type/block_idx/baseline_correct/
+            candidate_correct/e_value（该单元所在块判定后的累计 e 值）/
+            stop_reason（仅最后一单元携带最终 stop_reason，其余空串）。
+            question_id 字段承载 **unit_id**（single=question_id，pair=pair_id）——
+            逐题明细在 predictions 表溯源，按 pair_id join 真实 question 表会 join 不上。
 
     关键实现:
         逐行 insert（非 insert_many），保证每行独立事务。

tests/unit/test_gate_block_unit.py

@@ -15,7 +15,7 @@ import pytest
 from app.harness.gate_ladder import BaselineCache, skill_hash
 from app.harness.inference import PREDICTIONS_SCHEMA, InferenceResult
 from app.harness.log import HarnessLog
-from app.harness.validate import validate_skill_local
+from app.harness.validate import _ladder_units, validate_skill_local
 from core.evolution import GateParams
 from core.types import GeneratedQuestion
 
@@ -136,6 +136,45 @@ def _make_mock_run_inference(
     return mock_fn, call_log
 
 
+class TestLadderUnits:
+    """_ladder_units：阶梯题序聚合为单元并保持信息阶梯序。"""
+
+    def test_pure_non_ar_is_identity(self) -> None:
+        """纯非 AR 输入 → 单元序恒等（unit 序==原题序、unit_id==question_id）。"""
+        ladder = [_single("s0"), _single("s1"), _single("s2")]
+        units = _ladder_units(ladder)
+        assert [u.kind for u in units] == ["single", "single", "single"]
+        # unit_id 逐一等于原 question_id，且顺序与输入完全一致
+        assert [u.unit_id for u in units] == ["s0", "s1", "s2"]
+        assert [q.question_id for u in units for q in u.questions] == ["s0", "s1", "s2"]
+
+    def test_mixed_preserves_ladder_order(self) -> None:
+        """混格交错输入 → 按单元最早出现下标重排，pair 整锁、不被 single-first 污染。"""
+        p1o, p1m = _pair("p1")
+        # 交错布置：pair 两成员分居 idx 1、3；single 分居 idx 0、2、4
+        ladder = [_single("s0"), p1o, _single("s1"), p1m, _single("s2")]
+        units = _ladder_units(ladder)
+        # 最早出现下标：s0=0, p1=min(1,3)=1, s1=2, s2=4 → 阶梯序 [s0,p1,s1,s2]
+        assert [u.unit_id for u in units] == ["s0", "p1", "s1", "s2"]
+        # pair 整锁为一个单元（含两成员），不被拆
+        p1_unit = next(u for u in units if u.unit_id == "p1")
+        assert p1_unit.kind == "pair"
+        assert {q.question_id for q in p1_unit.questions} == {"p1_o", "p1_m"}
+
+    def test_mixed_differs_from_build_units_default(self) -> None:
+        """混格重排必须纠正 build_units 的 single-first 顺序（否则阶梯序被污染）。"""
+        from app.harness.question_units import build_units
+
+        p1o, p1m = _pair("p1")
+        ladder = [_single("s0"), p1o, _single("s1"), p1m, _single("s2")]
+        default_order = [u.unit_id for u in build_units(ladder)]
+        ladder_order = [u.unit_id for u in _ladder_units(ladder)]
+        # build_units 把 pair 排到 single 之后；_ladder_units 恢复信息阶梯序
+        assert default_order == ["s0", "s1", "s2", "p1"]
+        assert ladder_order == ["s0", "p1", "s1", "s2"]
+        assert ladder_order != default_order
+
+
 @pytest.mark.asyncio
 async def test_gate_n_used_counts_units_not_questions(tmp_path: Path) -> None:
     """混格阶梯（1 pair + 2 single）→ n_used=3 单元，非 4 题。"""