完善 O30 标定时钟同步与准备契约校验

2026-09-20 18:26:41 +08:00
parent 5ee4a3bb7c
commit db4c7af3e7
15 changed files with 543 additions and 13 deletions
@@ -861,8 +861,9 @@ DIP 只完成模拟后续访问的正式采集/父参考回读测试，使用真
 观测后，两关节各得到 248 帧。离线反事实回放还使用新 PIP 模型重新投影原 DIP 图像，
 验证整条依赖链；不修改原零位或历史日志，不把旧断点直接重新标记为通过。

-这项优化需要新采集的上游模型。验证整场稳定性使用 `--no-resume`；继续旧断点仍保留
-原冻结模型及不确定度，不能承诺仅重试 DIP 就会通过。正式运行命令：
+这项优化需要新采集的上游模型。新版会自动排除缺少当前准备契约的旧断点，重新采集
+完整准备链；新版本同契约断点仍支持恢复。验证独立整场运行时使用 `--no-resume`。
+正式运行命令：

 ```bash
 ros2 run linkerhand_calibration calibrate_hand \
@@ -883,3 +884,27 @@ ros2 run linkerhand_calibration calibrate_hand \
 报告和按角色哈希绑定的证据。辅助运动、独立准备、断点与最终验收的重新拟合均读取
 这项策略；无字段的旧日志继续使用原预算。删除策略、改写策略或使训练/留出帧数超过
 对应预算会拒绝回读，不能让在线与离线分别选择不同的子集。
+
+## 准备契约与时钟刷新修复（2026-09-20）
+
+`session_start.preparation_contract` 记录实际准备图：单/多关节采样预算、求解参数、
+父子来源、零位路径和辅助动作。自动发现断点先比较这份契约，缺失或不同则排除该断点，
+不读入大日志后再进入原停点。节点的显式恢复入口同样校验；产物回读还检查声明的新
+契约下不存在使用旧采样预算的已接受模型。旧原始记录保持可读、不会被重写。
+`resume_selection.json` 保存候选排除原因；显式重新开始的会话阻止回溯更早的旧谱系。
+
+这里采用整场重新准备，避免对冻结训练、共享 Tag 交接和派生角度进行局部替换。
+后续同契约恢复仍需基准及逐关节复核。此变更修复新旧模型混用，不代表已实现整手
+准备统一联合求解，也不保证所有现场视角都具备可观测性。
+
+相机在通过原连续性检查后，根据实测相位漂移缩短下一次校时的间隔，并扣除当前锚点
+已经经历的时间。刷新目标使用原 1 ms 漂移预算的四分之一；周期限制在 20–500 ms。
+真正的时钟跳变、计数器复位、过期映射及未来/乱序图像仍按原规则拒绝。
+这解决了固定 500 ms 刷新无法适应平滑漂移的机制；三台实机的具体异常仍需新日志验收。
+
+辅助旁观姿态拟合在计算前检查继承误差是否已耗尽原 1°/1 mm 预算。不可行来源记录为
+`observer_parent_precision_budget_exhausted`，不会通过重复拟合或降低门限试图挽救。
+
+针对回归、真实像素回放、合成完整 JSON→URDF 链路与只读配置检查均已通过，详见
+`software_review_stability_repair/software_validation.json`。本次设备被独立的
+`linkerhand_range_calibration` 占用，未中断它，也未执行新的实机标定或发布标定产物。
@@ -0,0 +1,119 @@
+# O30 标定稳定性评估（2026-09-20）
+
+当前结论：已有精度拒绝和产物发布边界应保留，但现有逐关节准备、立即冻结、
+再向下游传递的观测组织方式，尚不足以支撑稳定的一次整手标定。
+不能将局部软件回放通过或某个候选精度达标等同于整手稳定。
+
+后续修复进展：已实现准备契约隔离、发布入口预算核验、按漂移调整相机刷新间隔、
+启动锚点年龄扣除及辅助观测继承预算前置检查。详见 `O30_RIGHT_CALIBRATION.md`
+末节。下文保留评估时的事实和整体目标；不能将这些局部结构修复视为所有重构或
+整手实机稳定验收已完成。当前独立范围标定占用设备，尚无新版实机验收结果。
+
+## 最新实机证据
+
+会话 `calibration_output/O30_RIGHT_001/20260920_175156`，复用 12 个扫描单元后，
+在小指 DIP 准备阶段以 `OBS-POSE-AMBIGUOUS-117` 暂停。
+完整证据为该目录的 `assistant_resume_review.json`、`node_status.json`、
+`raw_samples.jsonl` 第 117841 行及三个 `camera_timing_*.jsonl`。
+
+| 项目 | 实测结果 | 含义 |
+| --- | --- | --- |
+| DIP 准备图像 | 228 帧，114 训练 / 114 留出 | 增加采样已实际生效 |
+| 候选 0 轴向不确定度界 | 2.915°，门限 3° | 合格但裕量很小 |
+| 候选 1 轴向不确定度界 | 3.035° | 不能发布，但仍能解释图像 |
+| 两候选留出 RMS | 0.350 / 0.395 px | 像素拟合接近，不足以直接选三维解 |
+| 校正后的比较 p 值 | 0.09375，要求 ≤ 0.01 | 歧义拒绝符合现有契约 |
+| 继承 PIP 轴向不确定度界 | 2.274° | 占用下游大部分精度预算 |
+| 旁观辅助观测 | 两次 `observer_image_validation_failed` | 本次没有提供可用的额外约束 |
+| SDK | 约 29.8 Hz，无报告故障 | 没有证据把直接停因归为 SDK 故障 |
+
+`axis_uncertainty_95_rad` 是现有字段名；实现使用局部线性化协方差的三倍标准差界，
+并保守加入继承项。它不是包含内参、标签系统偏差在内的实测绝对精度保证。
+
+## 为什么此前修复后仍会停
+
+1. **改动没有更新整条依赖链。** 密集采样用于新的 DIP 准备；恢复则保留已冻结的
+   旧 PIP 模型。参考复核仅确认当前安装/姿态可以复用，不会把旧模型重新训练成新模型。
+   因而恢复成功和下游可观测并不是同一个条件。
+2. **单视角平面标签仍有不同三维解释。** 增加同一运动弧上的帧数可以改善随机误差，
+   不保证增加区分两个候选所需的信息。本次两候选都收敛，不能靠增加求解次数解决。
+   也不能删掉略超精度门限、但仍可解释图像的备选候选。
+3. **辅助观测路径没有预先核算可行性。** `fit_observer_pose` 将父模型不确定度加入
+   辅助相对姿态的界，再要求旋转不超过 1°。旧 PIP 已有约 2.27°，沿用这个来源时，
+   即使像素验收过关也无法满足该界。本次实际首先失败于像素验收，不能把两者混为一项。
+4. **准备策略的组合复杂度超过已完成的实机覆盖。** 当前存在共享 Tag 交接、父几何
+   传递、旁观辅助、侧摆辅助及进入姿态联合准备等路径，部分组合显式不支持。
+   各路径局部回放通过，不构成它们在整手遮挡、回访和断点恢复中的组合保证。
+
+现有基准锁定、证据哈希、训练/留出隔离、禁止失败发布和有界恢复是合理的。
+需要调整的是观测规划、冻结范围、恢复依赖和验收顺序，而不是放宽拒绝条件。
+
+### 最新像素的父模型敏感性实验
+
+重新估计历史 PIP 准备模型，并仅在隔离的深拷贝中重算最新一轮 228 帧 DIP 的父姿态，
+按相同门限求解：PIP 继承轴向界由约 2.274° 降为 1.574°，DIP 候选 0 的轴向界
+为 2.213°，校正 p 值为 0.002227，离线解析成功。相较原实机的 2.915° / 0.09375，
+这支持“旧父模型是本次主要限制，必须连同依赖链更新”的判断。
+
+该实验使用历史 PIP 图像和最新 DIP 图像，没有实机重采，也没有重新完成当前父姿态
+准入。脚本显式使用离线准入标记，未改变原始记录，不允许发布。
+不能据此声称实机下次必过，更不能把重新计算的父模型注入旧正式产物。
+结果和复现脚本存于 `software_review_stability_architecture/latest_parent_counterfactual.json`
+及同目录 `parent_counterfactual.py`。脚本建立来源注册表时曾两次缺少依赖而被正式入口
+拒绝，补齐历史 MCP/PIP 来源后才完成实验；拒绝记录保存在该目录的验证摘要中。
+
+## 独立存在的时钟问题
+
+约 374 秒内，front / side / top 分别出现 33 / 31 / 27 次时钟失效并重同步；
+曝光时间序列的最大图像间隔分别约 372 / 338 / 371 ms。
+这不是此次歧义判定的直接错误码，但会削弱采集连续性，必须单独消除。
+
+三相机的主机时间与设备计时差在该轮累计同向变化约 201–203 ms，提示应优先检查
+共用的主机时间基础。另做 30 秒只读探测，墙钟相对 `CLOCK_MONOTONIC_RAW` 变化
+约 -13.9 ms；主机启用了时间同步。这些现象不足以证明历史失效就是 NTP 导致，
+也不能据此直接关时间同步或扩大同步容差。
+
+旧日志仅保存成功的锁存和失效原因，没有保存触发拒绝的锁存事务，缺少定位证据。
+本次补充了两个独立时钟的事务边界、被拒绝的原始锁存和旧锚点，拒绝时立即落盘。
+曝光时间映射、2 ms 锁存界、时钟连续性界和图像新鲜度界均未修改。
+对应相机时序测试 12 项通过，耗时 0.33 秒。这是诊断能力修复，不是时钟故障已解决。
+
+## 应采用的流程和实施顺序
+
+1. **先验收基础设施。** 固定本次控制进程的 ROS 发现范围；确保同一设备只有一个
+   控制者。三相机进行覆盖预期标定时长的连续采集检查，以原始双时钟证据定位并修复
+   同步异常。ROS 图隔离不能替代物理设备独占检查。
+2. **正式训练前完成整条手指的观测准备。** 每条依赖链先检查可见性、运动覆盖、
+   候选可区分性和完整精度预算，再整体接受准备模型。父模型应有经过数据论证的
+   下游预算，不能只按自身 3° 门限提前冻结。
+3. **独立信息不足时改变预先声明的观测设计。** 优先评估保持其他关节约束的不同
+   上游姿态，必要时采用经过验证的跨视角共同观测。额外动作需纳入运动计划、
+   原始证据和独立留出；不得观察留出结果后反复挑方案直到通过。
+   三台相机存在并不代表当前求解已使用多视角。本轮预检中光学内参和相对外参的
+   实测状态均为 `unverified`，跨视角方案必须先补足对应验证。
+4. **按依赖链管理准备版本和断点。** 准备策略升级或父模型需要更新时，在新会话中
+   明确重建受影响的父子准备模型及其派生数据；保留无依赖的数据和历史原件。
+   不得将新父模型与旧角度、旧零位或旧发布证据直接拼接。完全冻结后的训练版本
+   不能在正式留出失败后局部重训并继续沿用同一留出验收。
+5. **统一有限恢复。** 对短暂缺帧执行有界、可回放的局部恢复；基准变化、设备故障、
+   真正不可观测或证据损坏仍应终止/暂停。恢复必须由具体证据触发，同一信息不足的
+   运动不能无限重走。启动索引和重求解保持在实时控制回调之外。
+6. **最后验证完整交付。** 连续完成整手训练与独立验收，验证 JSON 的证据、数值、
+   单位和关节覆盖，从该 JSON 重建修正 URDF 并回读核验，最后才更新发布指针。
+   末节 CAD 零位保留假设必须继续在产物中明确，不能称所有零位均已独立实测。
+
+以上是结构调整的实施要求，并非这些重构已经完成。禁止用新增重试、降低门限或
+跳过候选比较来替代观测设计。
+
+## 稳定性的验收定义
+
+- 故障回放覆盖断帧、时钟跳变、SDK 失联、参考移动、候选歧义和恢复版本不兼容；
+  暂态在规定预算内恢复，其余明确拒绝，均无错误发布。
+- 固定代码、配置和现场布置后，至少连续三次完整实机运行无需人工介入，均完成独立
+  验收并生成可回读的 JSON/URDF，同时报告关节结果重复性。三次是工程验收建议，
+  不是数学上的成功率证明，也未改写当前产品 `required_independent_passes: 1`。
+- 单独验证正常中断后的断点恢复与一次完整运行结果一致；不能仅验证复用单元数量。
+- 保留每次失败和软件版本；不能只汇总成功运行或把反事实重放计为实机通过。
+
+当前尚未达到上述验收。本轮没有发布标定 JSON 或修正 URDF，不能宣称流程稳定或
+保证下一次一次通过。
@@ -1,5 +1,23 @@
 # 标定测试的执行范围

+准备契约升级用 `test_preparation_contract.py`，覆盖新旧断点分离、父子/预算/动作
+变化、历史记录保留、原会话边界、节点写入和发布入口禁止旧模型冒充新预算。
+时钟自适应刷新用 `test_camera_timing.py`，正负平滑漂移在模拟 30 Hz 的 60 秒内
+曝光时间误差保持小于原 1 ms 连续性预算，同时验证真跳变仍拒绝及启动锚点年龄。
+辅助精度预算前置用 `test_observer_pose.py` 和 `test_observer_reference.py`。
+
+此次增量验证分组记录在 `software_review_stability_repair`：初始针对组 79 项、8.84 秒；
+证据/真实 PIP-DIP 回放/runner/合成完整产物组 91 项、58.99 秒；修正启动锚点刷新
+时机后的相机组 15 项、0.47 秒。组间有重复用例，不将累计次数称为一次全量回归。
+实际目录的断点选择只读验证约 17 ms，排除了旧准备契约；配置 validate-only 通过。
+这些结果不替代新版本的连续实机验收。
+
+相机时钟诊断变更使用 `test_camera_timing.py`：同时覆盖原始计时单位、跳变、陈旧帧、
+重同步顺序及拒绝事务可重放。新增 `CLOCK_MONOTONIC_RAW` 仅记录诊断边界，不能
+替换 ROS 时间或放宽同步门限。2026-09-20 针对组 12 项通过，0.33 秒，记录于
+`calibration_output/O30_RIGHT_001/software_review_stability_architecture/camera_timing_tests.json`。
+这不表示实机重复失同步已经修复；整体评估见 `O30_STABILITY_ASSESSMENT_20260920.md`。
+
 日常修改运行受影响用例和必要契约用例，目标为 60 秒内完成。不要因为修改了一行代码就运行
 整手拟合或全量 `colcon test`。公共产物契约变更需要覆盖相关型号，发布前再执行全量回归。

@@ -13,6 +13,7 @@ import time


 CAMERA_TIMING_POLICY = "mvs_latched_exposure_time_v1"
+CLOCK_DRIFT_TOLERANCE_NS = 1_000_000


@dataclass(frozen=True)
@@ -20,6 +21,8 @@ class ClockLatch:
    device_tick: int
    before_ns: int
    after_ns: int
+    raw_before_ns: int | None = None
+    raw_after_ns: int | None = None

    @property
    def midpoint_ns(self):
@@ -45,6 +48,7 @@ class DeviceFrameClock:
        self.anchor = None
        self.last_frame_tick = None
        self.last_stamp_ns = None
+        self.refresh_interval_seconds = .5

    def elapsed_ns(self, ticks):
        return round(ticks*1_000_000_000/self.ticks_per_second)
@@ -59,9 +63,19 @@ class DeviceFrameClock:
        for current in latches:
            elapsed = self.elapsed_ns(current.device_tick-previous.device_tick)
            host_elapsed = current.midpoint_ns-previous.midpoint_ns
-            tolerance = current.half_width_ns+previous.half_width_ns+1_000_000
+            tolerance = current.half_width_ns+previous.half_width_ns+CLOCK_DRIFT_TOLERANCE_NS
            if elapsed < 0 or abs(elapsed-host_elapsed) > tolerance:
                raise ValueError("camera/host clock discontinuity or counter unit mismatch")
+        # Refresh before measured phase drift spends the existing 1 ms
+        # continuity allowance. A fixed 0.5 s period repeatedly invalidates a
+        # smooth slewing host clock. Scheduling changes; admission never does.
+        current = max(latches, key=lambda x: x.midpoint_ns)
+        host_elapsed = current.midpoint_ns-previous.midpoint_ns
+        if host_elapsed >= 50_000_000:
+            drift = abs(host_elapsed-self.elapsed_ns(current.device_tick-previous.device_tick))
+            rate = drift/host_elapsed
+            self.refresh_interval_seconds = max(.02, min(.5,
+                .25*CLOCK_DRIFT_TOLERANCE_NS/1e9/max(rate, 1e-12)))
        self.anchor = min(latches, key=lambda x: x.half_width_ns)

    def timestamp(self, tick, *, received_ns, exposure_us):
@@ -80,8 +94,11 @@ class DeviceFrameClock:
 class MvsCameraTiming:
    """The camera-specific clock latch and its optional append-only journal."""

-    def __init__(self, camera, mvs, clock_ns, *, exposure_us, journal_path=""):
+    def __init__(self, camera, mvs, clock_ns, *, exposure_us, journal_path="", raw_clock_ns=None):
        self.camera, self.mvs, self.clock_ns = camera, mvs, clock_ns
+        # CLOCK_MONOTONIC_RAW is independent of wall-clock synchronization.
+        # It is diagnostic evidence only, never a replacement ROS timestamp.
+        self.raw_clock_ns = raw_clock_ns or (lambda: time.clock_gettime_ns(time.CLOCK_MONOTONIC_RAW))
        self.exposure_us = exposure_us
        self.frame_clock = None
        self.last_stamp_ns = None
@@ -100,12 +117,14 @@ class MvsCameraTiming:
        return int(value.nCurValue)

    def _latch(self):
+        raw_before = self.raw_clock_ns()
        before = self.clock_ns()
        status = self.camera.MV_CC_SetCommandValue("DeviceTimestampLatch")
        if status != 0:
            raise ValueError(f"camera clock latch failed:0x{status:08x}")
        tick = self._integer("DeviceTimestamp")
-        return ClockLatch(tick, before, self.clock_ns())
+        after = self.clock_ns()
+        return ClockLatch(tick, before, after, raw_before, self.raw_clock_ns())

    def _write(self, row):
        if self.journal is not None:
@@ -135,13 +154,29 @@ class MvsCameraTiming:
            self._write(dict(kind="camera_clock_latch_rejected", latches=[asdict(x) for x in latches]))
            self.next_refresh = time.monotonic()+.02
            return
-        self.frame_clock.update(valid)
+        try:
+            self.frame_clock.update(valid)
+        except ValueError as error:
+            # Invalidation alone loses the transaction that failed. Preserve
+            # both clocks and the old anchor so drift, jumps and bad device
+            # counters can be distinguished without weakening the check.
+            self._write(dict(kind="camera_clock_sync_rejected", policy=CAMERA_TIMING_POLICY,
+                reason=str(error), ticks_per_second=self.frame_clock.ticks_per_second,
+                previous_anchor=asdict(self.frame_clock.anchor) if self.frame_clock.anchor else None,
+                latches=[asdict(x) for x in latches]))
+            if self.journal is not None:
+                self.journal.flush()
+            raise
        self._write(dict(kind="camera_clock_sync", policy=CAMERA_TIMING_POLICY,
            ticks_per_second=self.frame_clock.ticks_per_second,
+            refresh_interval_seconds=self.frame_clock.refresh_interval_seconds,
            latches=[asdict(x) for x in latches], anchor=asdict(self.frame_clock.anchor)))
        if self.journal is not None:
            self.journal.flush()
-        self.next_refresh = time.monotonic()+.5
+        # Startup may select a precise latch near the beginning of its window.
+        # That anchor has already spent part of the refresh budget.
+        anchor_age = (max(x.after_ns for x in latches)-self.frame_clock.anchor.midpoint_ns)/1e9
+        self.next_refresh = time.monotonic()+max(0., self.frame_clock.refresh_interval_seconds-anchor_age)

    def refresh_if_due(self):
        if self.frame_clock is None:
@@ -16,6 +16,17 @@ from .ippe import square_object_points
 from .relative import _relative_pose


+OBSERVER_POSE_LIMITS = (math.radians(1), .001)
+
+
+def require_observer_precision_budget(inherited):
+    """A parent alone must not exhaust the observer's final precision budget."""
+    if (len(inherited) != 2 or any(not math.isfinite(v) or v < 0 for v in inherited)):
+        raise ValueError('observer_parent_uncertainty_invalid')
+    if any(value >= limit for value, limit in zip(inherited, OBSERVER_POSE_LIMITS)):
+        raise ValueError('observer_parent_precision_budget_exhausted')
+
+
@dataclass(frozen=True)
 class ObserverPose:
    quaternion_xyzw: tuple
@@ -79,6 +90,7 @@ def _relative_covariance(fit, indices, parents, parent_corners, parent_size, chi

 def fit_observer_pose(parents, candidates, corners, matrices, size, stamps, *, inherited=(0., 0.),
        parent_corners=None, parent_size=None):
+    require_observer_precision_budget(inherited)
    count = len(stamps)
    if count < 24 or tuple(stamps) != tuple(sorted(set(stamps))) or any(not c for c in candidates):
        raise ValueError("observer_independent_images_missing")
@@ -142,7 +154,7 @@ def fit_observer_pose(parents, candidates, corners, matrices, size, stamps, *, i
            points, matrices, variance)
    bounds = [float(3*np.sqrt(max(0., np.linalg.eigvalsh(covariance[i:i+3, i:i+3])[-1])))
        + inherited[i//3] for i in (0, 3)]
-    if not all(math.isfinite(v) for v in bounds) or bounds[0] > math.radians(1) or bounds[1] > .001:
+    if not all(math.isfinite(v) for v in bounds) or any(v > limit for v, limit in zip(bounds, OBSERVER_POSE_LIMITS)):
        raise ValueError("observer_pose_uncertain")
    return ObserverPose(tuple(Rotation.from_rotvec(winner.x[:3]).as_quat()), tuple(winner.x[3:]),
        *bounds, tuple(stamps[i] for i in training), tuple(stamps[i] for i in validation), adjusted)
@@ -1,5 +1,6 @@
 """Explicit, bounded pose excitation at each task's existing clearance pose."""
-from dataclasses import replace
+from dataclasses import asdict, replace
+import json

 from ..core.domain.profile import ModelPreparation

@@ -21,6 +22,54 @@ def preparation_image_budget(profile, *, joint_count=1):
        else (128, image_parameters_for_sampling()))


+def preparation_contract(profile):
+    """Version the entire preparation graph independently of capture cycles.
+
+    A strategy upgrade changes upstream geometry as well as downstream image
+    budgets. Old captures remain readable, but cannot supply frozen models to
+    a new preparation graph. Legacy strategies keep their existing contract.
+    """
+    if not profile.motion.preparation_witnesses:
+        return None
+    tasks = {}
+    for task in profile.motion.tasks:
+        intervals, parameters = preparation_image_budget(profile, joint_count=len(task.joints))
+        tasks[task.key] = dict(joints=list(task.joints), command_intervals=intervals,
+            image_parameters=asdict(parameters),
+            parent_reference=asdict(task.parent_reference) if task.parent_reference else None)
+    value = dict(version='dependency_bound_preparation_v1', tasks=tasks,
+        zero_references={name: asdict(spec) for name, spec in profile.motion.joint_zero_references.items()},
+        witnesses={name: asdict(spec) for name, spec in profile.motion.preparation_witnesses.items()})
+    # The live header and decoded JSON must compare identically (including
+    # tuple-valued poses), without relying on incidental Python container types.
+    return json.loads(json.dumps(value, allow_nan=False))
+
+
+def validate_preparation_contract(profile, records):
+    """Audit a declared graph against both its profile and accepted models.
+
+    Older captures remain evaluable under their original per-model policies.
+    A new header must not advertise the new graph while importing old budgets.
+    """
+    from ..core.geometry.tag_pose.production_image_motion import image_parameters_for_sampling
+    declared = [row['preparation_contract'] for row in records
+        if row.get('kind') == 'session_start' and 'preparation_contract' in row]
+    if not declared:
+        return
+    if len(declared) != 1 or declared[0] != preparation_contract(profile):
+        raise ValueError('preparation_contract_profile_changed')
+    tasks = {task.key: set(task.joints) for task in profile.motion.tasks}
+    for row in records:
+        if row.get('kind') != 'motion_branch_initialization' or row.get('status') != 'resolved':
+            continue
+        joints = row.get('zero_joints', ())
+        if not joints or not set(joints) <= tasks.get(row.get('task_name'), set()):
+            raise ValueError('preparation_contract_model_dependency_changed')
+        _, expected = preparation_image_budget(profile, joint_count=len(joints))
+        if image_parameters_for_sampling(row.get('image_sampling_policy')) != expected:
+            raise ValueError('preparation_contract_model_budget_changed')
+
+
 def configure_preparation(profile, policy):
    if policy != 'occlusion_aware_witness_v1':
        raise ValueError('unknown_preparation_policy')
@@ -182,6 +182,10 @@ class CalibrationCoordinator:
                "source": "sdk_target_position_register" if parameters.initial_command else "adapter_feedback"},
            **({"diagnostic_capture": self.diagnostic_capture.as_dict()} if self.diagnostic_capture else {}),
        }
+        from ..profiles.preparation import preparation_contract
+        preparation = preparation_contract(profile)
+        if preparation is not None:
+            self._session_header["preparation_contract"] = preparation
        if profile.acquisition.command_capture_mode == "separate" and self.diagnostic_capture is None:
            from ..core.fitting.command_sampling import SAMPLING_POLICY
            self._session_header["command_sampling_policy"] = SAMPLING_POLICY
@@ -299,7 +299,9 @@ class CalibrationEngine:
        )

    def resume_compatible(self, session_start: Mapping[str, Any]) -> bool:
+        from ..profiles.preparation import preparation_contract
        return (self.capture_compatible(session_start)
+            and session_start.get("preparation_contract") == preparation_contract(self.profile)
            and session_start.get("capture_schedule_version") == capture_schedule_version(self.profile))


@@ -343,6 +343,8 @@ def validate_motion_provenance(profile, records, *, required=False, require_comp
    The explicit ``required`` flag is set by the production owner. Independent
    record markers preserve this requirement after a header is removed.
    """
+    from ..profiles.preparation import validate_preparation_contract
+    validate_preparation_contract(profile, records)
    if not required and not uses_motion_evidence(records):
        return {"status": "legacy_image_confirmation", "motion_evidence_verified": False}
    from .visual_motion import validate_visual_motion_evidence
@@ -6,7 +6,7 @@ import numpy as np

 from ..core.geometry.tag_pose.motion_evidence import MotionRelation
 from ..core.geometry.tag_pose.pose_bridge import SharedTagPoseBridge
-from ..core.geometry.tag_pose.observer_pose import fit_observer_pose
+from ..core.geometry.tag_pose.observer_pose import fit_observer_pose, require_observer_precision_budget
 from ..core.geometry.tag_pose.image_motion_projection import select_image_motion_frame
 from .branch_initialization import _pose
 from .shared_tag_reference import bridge_frames
@@ -37,6 +37,9 @@ def build_observer_bridge(profile, source):
    rows, held = source.rows, source.held_channels
    if len(parent_model.geometry) != 1 or parent_model.geometry[0].relation.child_role != spec.parent_role:
        raise ValueError('observer_parent_model_invalid')
+    uncertainty = {name: (angle, point) for name, angle, point in source.parent_report['geometry_uncertainty']}
+    inherited = uncertainty[parent_model.geometry[0].relation.joint]
+    require_observer_precision_budget(inherited)
    target = profile.motion.joint_zero_references[source.joint].command
    minimum = max(10, profile.acquisition.fixed_reference_minimum_frames)
    endpoints = source.endpoint_rows[-minimum:]
@@ -77,8 +80,6 @@ def build_observer_bridge(profile, source):
        matrices.append(row['camera_matrix'])
    if len(sizes) != 1:
        raise ValueError('observer_tag_size_changed')
-    uncertainty = {name: (angle, point) for name, angle, point in source.parent_report['geometry_uncertainty']}
-    inherited = uncertainty[parent_model.geometry[0].relation.joint]
    pose = fit_observer_pose(parents, candidates, corners, matrices, sizes.pop(),
        tuple(r['image_stamp_ns'] for r in rows), inherited=inherited,
        parent_corners=[r['tags'][spec.parent_role]['corners_xy'] for r in rows],
@@ -45,6 +45,8 @@ def discover_resume_candidate(
    session_root: Path, *, profile_id: str, serial_number: str,
    protected_hashes: Mapping[str, str],
    training_policy: str = "fixed",
+    preparation_contract: Mapping[str, Any] | None = None,
+    rejected_candidates: list | None = None,
 ) -> Path | None:
    """Locate evidence, never authorize its reuse or publish from a checkpoint.

@@ -99,6 +101,8 @@ def discover_resume_candidate(
                            raise ValueError("checkpoint training policy differs")
                        if not _compatible_header(start, profile_id, serial_number, root.name, protected_hashes):
                            raise ValueError("checkpoint header incompatible")
+                        if start.get("preparation_contract") != preparation_contract:
+                            raise ValueError("checkpoint preparation contract changed")
                    if kind == "fixed_base_reference_locked":
                        reference_found = True
                    if kind == "scan_unit_complete" or kind.endswith("sweep_observation_quality"):
@@ -116,7 +120,13 @@ def discover_resume_candidate(
                        previous = completed.get(identity)
                        if previous is None or attempt >= previous[0]:
                            completed[identity] = (attempt, passed)
-        except (OSError, ValueError):
+        except (OSError, ValueError) as error:
+            if rejected_candidates is not None:
+                rejected_candidates.append(dict(session=candidate.name, reason=str(error)))
+            # An explicit fresh run is a lineage boundary even if its header
+            # cannot be reused. Do not resurrect evidence from before it.
+            if start is not None and start.get("resume_checkpoint_requested") is False:
+                break
            continue
        if start is None or not reference_found or not completed_found:
            continue
@@ -141,11 +141,21 @@ def run_online(
    atomic_write_json(session/"measurement_contract.json", contract_report)
    if profile.artifacts.output_schema_version >= 3 and commands_enabled and diagnostic is None:
        require_release_geometry(contract_report)
+    from ..profiles.preparation import preparation_contract
+    rejected_checkpoints = []
+    preparation = preparation_contract(profile)
    resume = discover_resume_candidate(
        config.session_root, profile_id=profile.key.profile_id,
        serial_number=config.serial_number, protected_hashes=protected_inputs(config),
        training_policy=profile.quality.training_policy,
+        preparation_contract=preparation, rejected_candidates=rejected_checkpoints,
    ) if allow_resume and commands_enabled and diagnostic is None else None
+    atomic_write_json(session/"resume_selection.json", dict(
+        selected_session=resume.name if resume else None,
+        preparation_contract=preparation, rejected_candidates=rejected_checkpoints))
+    if any(row['reason'] == 'checkpoint preparation contract changed' for row in rejected_checkpoints):
+        print("旧断点的准备策略或父子模型预算与当前版本不同，已排除复用；"
+              + ("将使用兼容断点。" if resume else "本轮重新采集整条准备链，历史记录保留。"), flush=True)
    if contract_report["cad_constraint_conflicts"]:
        print("原始 CAD 存在限位/mimic 冲突，未修改原文件；详见 measurement_contract.json。最终范围验证不通过时禁止发布。", flush=True)
    log_path = session / "calibration.log"
@@ -67,6 +67,85 @@ def test_uncertain_control_transaction_does_not_refresh_the_clock():
    assert timing.frame_clock.anchor == latch(600_000_000)


+def test_rejected_sync_preserves_failed_transaction_and_previous_anchor(tmp_path):
+    path = tmp_path/'clock.jsonl'
+    timing = MvsCameraTiming(None, None, lambda: 0, exposure_us=5000., journal_path=path)
+    timing.frame_clock = initialized()
+    anchor = timing.frame_clock.anchor
+    jumped = latch(500_000_000, host_jump=10_000_000)
+    with pytest.raises(ValueError, match='discontinuity'):
+        timing._update([jumped])
+    assert timing.frame_clock.anchor == anchor
+    row = json.loads(path.read_text())  # Evidence is flushed before invalidation.
+    assert row['kind'] == 'camera_clock_sync_rejected'
+    assert ClockLatch(**row['previous_anchor']) == anchor
+    assert [ClockLatch(**x) for x in row['latches']] == [jumped]
+    replay = DeviceFrameClock(row['ticks_per_second'])
+    replay.anchor = ClockLatch(**row['previous_anchor'])
+    with pytest.raises(ValueError, match='discontinuity'):
+        replay.update([ClockLatch(**x) for x in row['latches']])
+    timing.close()
+
+
+def test_raw_clock_brackets_wall_clock_without_replacing_it(monkeypatch):
+    wall = iter((T0, T0+100_000))
+    raw = iter((20_000, 130_000))
+    camera = SimpleNamespace(MV_CC_SetCommandValue=lambda name: 0)
+    timing = MvsCameraTiming(camera, None, lambda: next(wall), exposure_us=5000.,
+                            raw_clock_ns=lambda: next(raw))
+    monkeypatch.setattr(timing, '_integer', lambda name: 123456)
+    measured = timing._latch()
+    assert measured == ClockLatch(123456, T0, T0+100_000, 20_000, 130_000)
+    assert measured.midpoint_ns == T0+50_000
+    assert measured.half_width_ns == 50_000
+
+
+@pytest.mark.parametrize('slew', [-.005, .005])
+def test_measured_drift_refreshes_before_continuity_budget_is_spent(monkeypatch, slew):
+    from dataclasses import replace
+    def slewed(elapsed_ns):
+        return latch(elapsed_ns, width=100_000, host_jump=round(elapsed_ns*slew))
+    # At the former fixed interval smooth slew fails the unchanged gate.
+    fixed = DeviceFrameClock(100_000_000)
+    fixed.update([slewed(0), slewed(100_000_000)])
+    with pytest.raises(ValueError, match='discontinuity'):
+        fixed.update([slewed(600_000_000)])
+    now = [.1]
+    monkeypatch.setattr('linkerhand_calibration.camera_timing.time.monotonic', lambda: now[0])
+    timing = MvsCameraTiming(None, None, lambda: 0, exposure_us=5000.)
+    timing.frame_clock = DeviceFrameClock(100_000_000)
+    timing._update([slewed(0), slewed(100_000_000)])
+    monkeypatch.setattr(timing, '_latch', lambda: slewed(round(now[0]*1e9)))
+    maximum_error = 0
+    for index in range(1, 1801):  # 60 seconds at the real camera's 30 Hz.
+        now[0] = .1+index/30
+        timing.refresh_if_due()
+        elapsed = round((now[0]-.016)*1e9)
+        stamp = timing.frame_clock.timestamp(latch(elapsed).device_tick,
+            received_ns=slewed(round(now[0]*1e9)).midpoint_ns, exposure_us=5000.)
+        expected = T0+round((elapsed+2_500_000)*(1+slew))
+        maximum_error = max(maximum_error, abs(stamp-expected))
+    assert maximum_error < 1_000_000
+    assert .02 <= timing.frame_clock.refresh_interval_seconds < .1
+    # A discontinuity is still rejected, including after successful adaptation.
+    anchor = timing.frame_clock.anchor
+    bad = slewed(round((now[0]+.04)*1e9))
+    with pytest.raises(ValueError, match='discontinuity'):
+        timing._update([replace(bad, before_ns=bad.before_ns+10_000_000, after_ns=bad.after_ns+10_000_000)])
+    assert timing.frame_clock.anchor == anchor
+
+
+def test_startup_anchor_age_is_charged_to_refresh_budget(monkeypatch):
+    monkeypatch.setattr('linkerhand_calibration.camera_timing.time.monotonic', lambda: 10.)
+    timing = MvsCameraTiming(None, None, lambda: 0, exposure_us=5000.)
+    timing.frame_clock = DeviceFrameClock(100_000_000)
+    initial = [latch(i*20_000_000, width=100_000 if i == 0 else 200_000,
+                     host_jump=i*100_000) for i in range(8)]
+    timing._update(initial)
+    assert timing.frame_clock.anchor == initial[0]
+    assert timing.next_refresh == 10.  # Renew before acquiring the next frame.
+
+
@pytest.mark.parametrize('host_jump', [0, 10_000_000, -1_000_000_000])
 def test_recovery_revalidates_clock_and_preserves_frame_time_order(monkeypatch, host_jump):
    timing = MvsCameraTiming(None, None, lambda: 0, exposure_us=5000.)
@@ -41,6 +41,14 @@ def test_all_legal_candidates_use_fixed_split_and_propagate_parent_uncertainty(o
    assert result.training_stamps == tuple(range(1, 49, 2))


+@pytest.mark.parametrize('inherited', [(math.radians(2.274), .0005), (0., .001)])
+def test_exhausted_parent_budget_is_rejected_before_fitting(observer_images, monkeypatch, inherited):
+    monkeypatch.setattr('linkerhand_calibration.core.geometry.tag_pose.observer_pose.least_squares',
+                        lambda *a, **kw: pytest.fail('fitting cannot recover spent parent precision'))
+    with pytest.raises(ValueError, match='parent_precision_budget_exhausted'):
+        fit_observer_pose(*observer_images, inherited=inherited)
+
+
@pytest.mark.parametrize('fault', ['missing', 'moving', 'installation', 'unobservable', 'uncertain_parent'])
 def test_invalid_observer_evidence_cannot_authorize_a_pose(observer_images, fault):
    parents, candidates, corners, matrices, size, stamps = observer_images
@@ -0,0 +1,156 @@
+"""Preparation upgrades cannot silently reuse frozen upstream geometry."""
+
+from copy import deepcopy
+from dataclasses import replace
+import json
+
+import pytest
+
+from linkerhand_calibration.core.domain.capture_plan import STAGED, configure_training
+from linkerhand_calibration.core.geometry.pnp import POSE_TRACKING_POLICY_VERSION
+from linkerhand_calibration.profiles import load_bundled_hand_profile
+from linkerhand_calibration.profiles.preparation import configure_preparation, preparation_contract
+from linkerhand_calibration.runtime.engine import ACQUISITION_POLICY_VERSION, CalibrationEngine
+from linkerhand_calibration.runtime.joint_resume import PreparedResume
+from linkerhand_calibration.runtime.resume import discover_resume_candidate
+
+
+def dense_profile():
+    return configure_preparation(configure_training(load_bundled_hand_profile('o30_right_18'), STAGED),
+                                 'occlusion_aware_witness_v1')
+
+
+def header(profile):
+    return dict(kind='session_start', serial_number='O30_TEST', profile_id=profile.key.profile_id,
+        acquisition_policy_version=ACQUISITION_POLICY_VERSION,
+        pose_tracking_policy_version=POSE_TRACKING_POLICY_VERSION,
+        capture_schedule_version=CalibrationEngine(profile).capture_schedule_version,
+        capture_plan={'policy': STAGED}, protected_hashes={'source': 'a'*64},
+        preparation_contract=preparation_contract(profile), resume_checkpoint_requested=True)
+
+
+def checkpoint(root, name, start, *, bad_tail=False):
+    path = root/name
+    path.mkdir()
+    rows = [start, dict(kind='fixed_base_reference_locked'),
+        dict(kind='scan_unit_complete', task_name='thumb_cmc_roll_front',
+             cycle=0, direction='increasing', passed=True)]
+    (path/'raw_samples.jsonl').write_text('\n'.join(json.dumps(row) for row in rows)
+        + ('\nnot-json' if bad_tail else '\n'))
+    return path
+
+
+def discover(root, profile, rejected):
+    return discover_resume_candidate(root, profile_id=profile.key.profile_id, serial_number='O30_TEST',
+        protected_hashes={'source': 'a'*64}, training_policy=STAGED,
+        preparation_contract=preparation_contract(profile), rejected_candidates=rejected)
+
+
+def test_old_budget_rejected_at_header_before_loading_large_history(tmp_path):
+    profile = dense_profile()
+    old = header(profile)
+    old.pop('preparation_contract')
+    checkpoint(tmp_path, '001', old, bad_tail=True)
+    rejected = []
+    assert discover(tmp_path, profile, rejected) is None
+    assert rejected == [dict(session='001', reason='checkpoint preparation contract changed')]
+    assert not CalibrationEngine(profile).resume_compatible(old)
+
+
+def test_matching_contract_resumes_and_json_roundtrip_is_identical(tmp_path):
+    profile = dense_profile()
+    start = json.loads(json.dumps(header(profile)))
+    source = checkpoint(tmp_path, '001', start)
+    assert CalibrationEngine(profile).resume_compatible(start)
+    assert discover(tmp_path, profile, []) == source
+    # Capture validity is separate: readable historical artifacts are not
+    # relabelled corrupt simply because a new live plan uses a different budget.
+    old = {key: value for key, value in start.items() if key != 'preparation_contract'}
+    assert CalibrationEngine(profile).capture_compatible(old)
+
+
+@pytest.mark.parametrize('field', ['budget', 'witness', 'parent', 'version'])
+def test_changed_dependency_graph_invalidates_whole_preparation_before_motion(tmp_path, field):
+    profile = dense_profile()
+    changed = header(profile)
+    contract = changed['preparation_contract']
+    if field == 'budget':
+        contract['tasks']['pinky_pip_side']['command_intervals'] = 128
+    elif field == 'witness':
+        contract['witnesses'].pop('pinky_mcp_pitch')
+    elif field == 'parent':
+        contract['tasks']['pinky_dip_side']['parent_reference']['geometry_joint'] = 'ring_pip'
+    else:
+        contract['version'] = 'future_strategy'
+    checkpoint(tmp_path, '001', changed)
+    assert discover(tmp_path, profile, []) is None
+    assert not CalibrationEngine(profile).resume_compatible(changed)
+    with pytest.raises(ValueError, match='identity/policy changed'):
+        PreparedResume.load(profile, CalibrationEngine(profile), tmp_path/'001/raw_samples.jsonl', 'O30_TEST')
+
+
+def test_incompatible_fresh_run_cannot_resurrect_older_lineage(tmp_path):
+    profile = dense_profile()
+    checkpoint(tmp_path, '001', header(profile))
+    fresh = header(profile)
+    fresh.pop('preparation_contract')
+    fresh['resume_checkpoint_requested'] = False
+    checkpoint(tmp_path, '002', fresh)
+    assert discover(tmp_path, profile, []) is None
+
+
+def test_resume_upgrade_does_not_rewrite_historical_capture(tmp_path):
+    profile = dense_profile()
+    old = header(profile)
+    old.pop('preparation_contract')
+    path = checkpoint(tmp_path, '001', old)/'raw_samples.jsonl'
+    before = path.read_bytes()
+    assert discover(tmp_path, profile, []) is None
+    assert path.read_bytes() == before
+
+
+def test_capture_cycle_supplement_keeps_preparation_but_changed_pose_does_not():
+    from linkerhand_calibration.core.domain.capture_plan import select_task_training
+    profile = dense_profile()
+    supplemented = select_task_training(profile, profile.motion.tasks[0].key, (0, 1, 2))
+    assert preparation_contract(supplemented) == preparation_contract(profile)
+    zeros = dict(profile.motion.joint_zero_references)
+    spec = zeros['pinky_pip']
+    command = list(spec.command)
+    command[0] -= 1
+    zeros['pinky_pip'] = replace(spec, command=tuple(command))
+    changed = replace(profile, motion=replace(profile.motion, joint_zero_references=zeros))
+    assert preparation_contract(changed) != preparation_contract(profile)
+
+
+def test_live_coordinator_writes_effective_preparation_graph(tmp_path):
+    from runtime_host_fixture import coordinator_fixture
+    host, _ = coordinator_fixture(tmp_path, model='o30', profile_transform=lambda _: dense_profile())
+    try:
+        row = json.loads(host.raw_path.read_text().splitlines()[0])
+        assert row['preparation_contract'] == preparation_contract(host.profile)
+        assert CalibrationEngine(host.profile).resume_compatible(row)
+    finally:
+        host.close()
+
+
+def test_legacy_profiles_keep_existing_resume_contract():
+    profile = load_bundled_hand_profile('o30_right_18')
+    assert preparation_contract(profile) is None
+
+
+def test_new_header_cannot_relabel_old_parent_models():
+    from linkerhand_calibration.profiles.preparation import validate_preparation_contract
+    from linkerhand_calibration.runtime.motion_provenance import validate_motion_provenance
+    from linkerhand_calibration.core.geometry.tag_pose.production_image_motion import DENSE_PREPARATION_SAMPLING
+    profile = dense_profile()
+    old = dict(kind='motion_branch_initialization', task_name='pinky_pip_side',
+               zero_joints=['pinky_pip'], status='resolved')
+    with pytest.raises(ValueError, match='model_budget_changed'):
+        validate_motion_provenance(profile, [header(profile), old])
+    dense = {**old, 'image_sampling_policy': DENSE_PREPARATION_SAMPLING}
+    validate_preparation_contract(profile, [header(profile), dense])
+    changed = deepcopy(header(profile))
+    changed['preparation_contract']['version'] = 'another_graph'
+    with pytest.raises(ValueError, match='profile_changed'):
+        validate_preparation_contract(profile, [changed, dense])