feat(harness): checkpoint 存 unit_id 序列，断点续跑孪生对不拆

核心算法保真 #3（断点续跑）：checkpoint 从逐题 question_id 改为存 unit_id
序列（孪生对折叠为单个 unit_id），恢复时 build_units + 按完整 unit 展开，
续跑后 pair 两成员同进同出、绝不被劈开。

- _batch_unit_ids/_batch_from_ids 对称折叠/展开，保序去重，纯非 AR 下
  unit_id==question_id、与旧逐题序列逐字节一致。
- momentum 采样抽取为 _sample_momentum_candidates 纯函数，docstring 显式
  记录 Phase 1 设计偏差：仅保证纯非 AR byte-identical，混格 momentum 不保证。
- 新增 test_checkpoint_pair（unit_id 落盘往返、pair 不拆）与
  test_non_ar_byte_identical（pools→batching→checkpoint→momentum 端到端黄金）。

app/harness/checkpoint.py

@@ -233,7 +233,8 @@ def write_checkpoint(
         global_step: 全局 step 序号。
         total_steps: 全局总 step 数。
         version_snapshot: skills/prompts 版本快照。
-        epoch_batches: 本 epoch 的 batch 划分（question_id 列表的列表）。
+        epoch_batches: 本 epoch 的 batch 划分（unit_id 列表的列表，孪生对折叠为
+            单个 unit_id；纯非 AR 下 unit_id==question_id）。
         config: 训练配置对象，用于计算 config_fingerprint。
 
     关键实现细节:

app/harness/runner.py

@@ -199,18 +199,78 @@ def _accumulate_slow_packs(diagnosis: DiagnosisResult, state: _TrainState) -> No
     state.tool_packs.extend(diagnosis.tool_case_packs.values())
 
 
-def _batch_from_ids(pools: Pools, ids: list[str]) -> list[GeneratedQuestion]:
-    """按 question_id 从诊断池重建一个 batch（保持原 epoch 划分）。
+def _batch_unit_ids(batch: list[GeneratedQuestion]) -> list[str]:
+    """把一个 batch 的扁平题目折叠为 unit_id 序列（孪生对成员去重为单个 unit_id）。
+
+    checkpoint 存 unit_id 序列而非逐题 question_id：断点续跑恢复时按完整 unit 展开，
+    保证孪生对整体重建、绝不被劈开（核心算法保真 #3 断点续跑）。
+
+    参数:
+        batch: 一个 mini-batch 的扁平题目列表（pair 两成员相邻）。
+
+    返回:
+        unit_id 列表，按题目在 batch 中的首次出现顺序去重；single 的 unit_id 即
+        question_id，故纯非 AR 输入下与旧逐题 question_id 序列逐字节一致。
+
+    关键实现:
+        用 dict 保序去重（pair 两成员共享 unit_id，仅记一次），无需额外集合。
+    """
+    ordered: dict[str, None] = {}
+    for q in batch:
+        ordered[q.unit_id] = None
+    return list(ordered)
+
+
+def _batch_from_ids(pools: Pools, unit_ids: list[str]) -> list[GeneratedQuestion]:
+    """按 unit_id 序列从诊断池重建一个 batch，按完整 unit 展开成题目列表。
+
+    与 _batch_unit_ids 对称：恢复时以完整 unit 为单位展开（pair 两成员同进同出），
+    断点续跑后孪生对绝不被拆开（核心算法保真 #3）。
 
     参数:
         pools: 三池容器。
-        ids: 一个 batch 的 question_id 列表。
+        unit_ids: 一个 batch 的 unit_id 序列（checkpoint 存的粒度）。
 
     返回:
-        按 ids 顺序取出的 GeneratedQuestion 列表。
+        按 unit_ids 顺序展开的 GeneratedQuestion 列表；每个 unit_id 展开为其全部
+        成员题（single 1 题、pair 2 题），顺序与原 batch 一致。
+
+    关键实现:
+        直接以 units_by_id[uid] 取值，unit_id 缺失触发 KeyError（P5 防静默兜底），
+        强制 checkpoint 与当前诊断池一致；纯非 AR 下 unit_id==question_id、单元即
+        单题，与旧逐题重建逐字节一致。
     """
-    by_id = {q.question_id: q for q in pools.diagnosis}
-    return [by_id[i] for i in ids]
+    units_by_id = {u.unit_id: u for u in build_units(pools.diagnosis)}
+    return [q for uid in unit_ids for q in units_by_id[uid].questions]
+
+
+def _sample_momentum_candidates(
+    pool: list[GeneratedQuestion],
+    allowed_task_types: set[str],
+    momentum_samples: int,
+    epoch: int,
+) -> list[GeneratedQuestion]:
+    """为 momentum 从诊断池按题型过滤后做确定性抽样（逐题粒度）。
+
+    设计偏差（Phase 1 显式记录）：momentum 采样在逐题粒度进行、不折叠 QuestionUnit，
+    故仅保证「纯非 AR 题库」的抽样序列与引入 QuestionUnit 前逐字节一致；混格题库下
+    孪生对可能被半采样、且候选集长度/顺序随 AR 成员增减而漂移，**Phase 1 不保证混格
+    momentum 的 byte-identical**（属可接受偏差，纯非 AR 必须不漂）。
+
+    参数:
+        pool: 诊断池扁平题目列表。
+        allowed_task_types: 允许参与的题型集合。
+        momentum_samples: 目标采样数上限。
+        epoch: 采样种子（同 epoch 可复现）。
+
+    返回:
+        采样到的题目列表；候选不足则全取，候选为空返回空列表。
+    """
+    candidates = [q for q in pool if q.task_type in allowed_task_types]
+    n = min(momentum_samples, len(candidates))
+    if n <= 0:
+        return []
+    return random.Random(epoch).sample(candidates, n)
 
 
 def _snapshot_current_skills(skills_dir: Path) -> dict[str, str]:
@@ -753,7 +813,8 @@ class Runner:
                     correct_ratio=self._config.batch_correct_ratio,
                 )
                 step_from = 0
-            batch_ids = [[q.question_id for q in b] for b in batches]
+            # checkpoint 存 unit_id 序列：断点续跑按完整 unit 展开，孪生对不拆
+            batch_unit_ids = [_batch_unit_ids(b) for b in batches]
             for step in range(step_from, len(batches)):
                 await self._run_step(epoch, step, total_steps, batches[step], pools, state)
                 state.global_step += 1
@@ -766,7 +827,7 @@ class Runner:
                     global_step=state.global_step,
                     total_steps=total_steps,
                     version_snapshot=self._current_version_snapshot(),
-                    epoch_batches=batch_ids,
+                    epoch_batches=batch_unit_ids,
                     config=self._config,
                 )
             await self._slow_update_cycle(epoch, pools, state)
@@ -782,7 +843,7 @@ class Runner:
                 global_step=state.global_step,
                 total_steps=total_steps,
                 version_snapshot=self._current_version_snapshot(),
-                epoch_batches=batch_ids,
+                epoch_batches=batch_unit_ids,
                 config=self._config,
             )
             if _should_early_stop(
@@ -1610,12 +1671,10 @@ class Runner:
         prev_skill = state.epoch_start_skills.get(target_file, skill_content)
         prev_guidance = momentum_inner(skill_content)
 
-        # 采样
-        allowed = set(task_types)
-        candidates = [q for q in pools.diagnosis if q.task_type in allowed]
-        rng = random.Random(epoch)
-        n = min(self._config.momentum_samples, len(candidates))
-        sampled = rng.sample(candidates, n) if n > 0 else []
+        # 采样（逐题粒度，不折叠 unit；混格偏差见 _sample_momentum_candidates docstring）
+        sampled = _sample_momentum_candidates(
+            pools.diagnosis, set(task_types), self._config.momentum_samples, epoch
+        )
 
         if not sampled:
             skill_path.write_text(