From decd7202b8ebd9256c4f94582df99213b5151829 Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sat, 10 Oct 2026 21:13:44 +0800
Subject: [PATCH 1/7] fix(dual-franka): calibration-samples review slice

---
 calibration_tools/base_handeye_collect.py     | 194 ++++++-
 .../test_collector_session.py                 | 478 ++++++++++++++++++
 2 files changed, 654 insertions(+), 18 deletions(-)
 create mode 100644 tests/unit_tests/calibration_tools/test_collector_session.py

diff --git a/calibration_tools/base_handeye_collect.py b/calibration_tools/base_handeye_collect.py
index 49ed0f9aa..8135022aa 100644
--- a/calibration_tools/base_handeye_collect.py
+++ b/calibration_tools/base_handeye_collect.py
@@ -19,6 +19,7 @@
 import base64
 import json
 import logging
+import re
 import shlex
 import shutil
 import subprocess
@@ -39,6 +40,7 @@
     check_opencv,
     check_state,
     delta,
+    rigid_transform,
     transform,
     write_json,
 )
@@ -94,12 +96,16 @@ def board_metadata(self) -> Record:
 class Collector:
     """Own acquisition state without mutating other imported modules."""
 
-    def __init__(self, config: CaptureConfig, root: Path) -> None:
+    def __init__(
+        self, config: CaptureConfig, root: Path, *, resume: bool = False
+    ) -> None:
         """Initialize an isolated session; hardware is read only on request."""
         self.config = config
         self.root = root
         self.pose_key = f"T_{config.arm}_base_ee"
         self.samples: list[Record] = []
+        self.reference: Record | None = None
+        self.next_index = 1
         self.last: Record = {}
         self.lock = threading.Lock()
         self.board = cv2.aruco.CharucoBoard(
@@ -109,6 +115,93 @@ def __init__(self, config: CaptureConfig, root: Path) -> None:
             cv2.aruco.getPredefinedDictionary(getattr(cv2.aruco, config.dictionary)),
         )
         self.board.setLegacyPattern(False)
+        if resume:
+            self._resume()
+        elif root.exists():
+            raise FileExistsError("Output directory already exists; use --resume")
+
+    def _resume(self) -> None:
+        """Load saved samples without changing their identities or files."""
+        if not self.root.is_dir():
+            raise ValueError("--resume requires an existing session directory")
+        reference: Record | None = None
+        indices: set[int] = set()
+        for parent in (self.root, self.root / "excluded"):
+            for folder in sorted(parent.glob("sample_*")):
+                match = re.fullmatch(r"sample_([0-9]+)", folder.name)
+                if not match or not folder.is_dir() or folder.is_symlink():
+                    raise ValueError(f"Invalid sample directory: {folder}")
+                index = int(match[1])
+                if (
+                    index < 1
+                    or index in indices
+                    or folder.name != f"sample_{index:03d}"
+                ):
+                    raise ValueError(f"Invalid or duplicate sample index: {folder}")
+                indices.add(index)
+                self.next_index = max(self.next_index, index + 1)
+                record = json.loads(
+                    (folder / "sample.json").read_text(encoding="utf-8")
+                )
+                if (
+                    type(record["index"]) is not int
+                    or record["index"] != index
+                    or record["arm"] != self.config.arm
+                    or record["calibration_mode"] != "eye_to_hand"
+                    or record["camera"]["serial"] != self.config.camera_serial
+                    or record["board"] != self.config.board_metadata
+                ):
+                    raise ValueError(f"Sample configuration does not match: {folder}")
+                if reference is None:
+                    reference = record
+                check_camera(record["camera"], reference["camera"])
+                for key in ("robot_before", "robot_after"):
+                    check_state(record[key])
+                    if not np.allclose(
+                        record[key]["F_T_EE"],
+                        reference["robot_before"]["F_T_EE"],
+                        rtol=0,
+                        atol=1e-8,
+                    ):
+                        raise ValueError(f"End-effector frame changed: {folder}")
+                pose = rigid_transform(record[self.pose_key], self.pose_key)
+                if not np.allclose(
+                    pose, transform(record["robot_before"]), rtol=0, atol=1e-8
+                ):
+                    raise ValueError(
+                        f"Stored robot pose disagrees with state: {folder}"
+                    )
+                rigid_transform(record["T_camera_board"], "T_camera_board")
+                for name in ("color.png", "annotated.png"):
+                    if not (folder / name).is_file():
+                        raise ValueError(f"Missing sample image: {folder / name}")
+                if parent == self.root:
+                    self.samples.append(record)
+        self.samples.sort(key=lambda sample: sample["index"])
+        self.reference = reference
+        self.last = {"resumed": True, "count": len(self.samples)}
+
+    def delete_sample(self, index: int) -> Record:
+        """Exclude a selected sample, preserving its files for manual recovery."""
+        if type(index) is not int or index < 1:
+            raise ValueError("Sample index must be a positive integer")
+        sample = next((s for s in self.samples if s["index"] == index), None)
+        if sample is None:
+            raise ValueError(f"No active sample with index {index}")
+        source = self.root / f"sample_{index:03d}"
+        destination = self.root / "excluded" / source.name
+        if destination.exists():
+            raise FileExistsError(f"Excluded sample already exists: {destination}")
+        destination.parent.mkdir(exist_ok=True)
+        source.rename(destination)
+        self.samples.remove(sample)
+        LOGGER.info("Excluded sample %d; files retained at %s", index, destination)
+        return {
+            "deleted": True,
+            "index": index,
+            "count": len(self.samples),
+            "archived": str(destination),
+        }
 
     def read_state(self) -> Record:
         """Run the configured readOnce probe with a bounded local timeout."""
@@ -150,9 +243,9 @@ def sample(self) -> Record:
             raise ValueError("采样窗口超过 3 秒，请检查网络后重试")
         if before["F_T_EE"] != after["F_T_EE"]:
             raise ValueError("末端坐标定义在采样期间发生变化")
-        if self.samples and not np.allclose(
+        if self.reference is not None and not np.allclose(
             before["F_T_EE"],
-            self.samples[0]["robot_before"]["F_T_EE"],
+            self.reference["robot_before"]["F_T_EE"],
             atol=1e-8,
             rtol=0,
         ):
@@ -161,7 +254,7 @@ def sample(self) -> Record:
             dm, dr = delta(a, np.array(old[self.pose_key]))
             if dm < 0.005 and dr < np.deg2rad(5):
                 raise ValueError("与已有姿态过于接近，请改变姿态后采集；不要重复点击")
-        check_camera(camera, self.samples[0]["camera"] if self.samples else camera)
+        check_camera(camera, self.reference["camera"] if self.reference else camera)
         png = base64.b64decode(camera.pop("png_b64"), validate=True)
         im = cv2.imdecode(np.frombuffer(png, np.uint8), cv2.IMREAD_COLOR)
         if im is None or im.shape[:2] != (camera["height"], camera["width"]):
@@ -203,7 +296,7 @@ def sample(self) -> Record:
         t[:3, :3] = rot
         t[:3, 3] = tv.ravel()
         record = {
-            "index": len(self.samples) + 1,
+            "index": self.next_index,
             "host_start_s": start,
             "host_end_s": time.time(),
             "robot_before": before,
@@ -241,6 +334,9 @@ def sample(self) -> Record:
             if temporary.exists():
                 shutil.rmtree(temporary)
         self.samples.append(record)
+        if self.reference is None:
+            self.reference = record
+        self.next_index += 1
         return {
             "accepted": True,
             "count": len(self.samples),
@@ -255,6 +351,14 @@ def status(self) -> Record:
             "count": len(self.samples),
             "session": str(self.root),
             "last": self.last,
+            "samples": [
+                {
+                    "index": sample["index"],
+                    "corners": sample["corners"],
+                    "reprojection_rms_px": round(sample["reprojection_rms_px"], 3),
+                }
+                for sample in self.samples
+            ],
         }
 
     def preview(self) -> bytes:
@@ -284,19 +388,47 @@ def preview(self) -> bytes:
 
 
 PAGE = b"""<!doctype html><meta charset="utf-8"><title>Hand-eye calibration</title>
-<style>body{font:18px sans-serif;max-width:900px;margin:30px auto}img{max-width:100%}button{padding:12px}</style>
+<style>body{font:18px sans-serif;max-width:900px;margin:30px auto}img{max-width:100%}button{padding:12px}
+table{border-collapse:collapse;width:100%}td,th{text-align:left;padding:8px;border-bottom:1px solid #ddd}</style>
 <h1>Stationary hand-eye calibration</h1>
 <p id="instruction"></p><p>Move the arm manually, release guidance, and wait two seconds.
 This collector only reads robot state. Keep the board visible and collect varied rotations.</p>
 <img id="view"><p><button id="button" onclick="capture()">Capture pose</button></p><pre id="status"></pre>
+<h2>Saved samples</h2>
+<p>Preview a sample before excluding it. Excluded samples are kept in the session's
+excluded/ directory and are not used by the solver. Re-solve after changing samples.</p>
+<table><thead><tr><th>Sample</th><th>Corners</th><th>RMS (px)</th><th>Actions</th></tr></thead><tbody id="samples"></tbody></table>
 <p>A recorded sample is not a successful calibration. Solve and validate on new poses.</p>
 <script>
+let busy=false, generation=0, samplesSignature='';
+function showStatus(state){
+ document.getElementById('status').textContent=JSON.stringify({count:state.count,session:state.session,last:state.last},null,2);
+ const signature=JSON.stringify(state.samples);if(signature===samplesSignature)return;samplesSignature=signature;
+ const rows=document.getElementById('samples');rows.replaceChildren();
+ for(const sample of state.samples){const row=rows.insertRow();
+  for(const value of [sample.index,sample.corners,sample.reprojection_rms_px])row.insertCell().textContent=value;
+  const actions=row.insertCell(), link=document.createElement('a');link.textContent='Preview';
+  link.href='/samples/'+sample.index+'/annotated.png';link.target='_blank';link.rel='noopener';actions.append(link,' ');
+  const button=document.createElement('button');button.textContent='Exclude';button.disabled=busy;
+  button.onclick=()=>excludeSample(sample.index);actions.append(button);
+ }
+}
+async function loadStatus(){const version=generation,state=await(await fetch('/status')).json();
+ if(version===generation&&!busy)showStatus(state);}
 async function refresh(){document.getElementById('view').src='/frame.jpg?t='+Date.now();
-try{document.getElementById('status').textContent=JSON.stringify(await(await fetch('/status')).json(),null,2);}catch(e){}
+try{if(!busy)await loadStatus();}catch(e){document.getElementById('status').textContent='Connection lost: '+e;}
 setTimeout(refresh,1000);}refresh();
-async function capture(){let b=document.getElementById('button');b.disabled=true;
-try{document.getElementById('status').textContent=JSON.stringify(await(await fetch('/sample',{method:'POST'})).json(),null,2);}
-catch(e){document.getElementById('status').textContent=String(e);}finally{b.disabled=false;}}
+async function mutate(path){if(busy)return;busy=true;generation++;
+ document.querySelectorAll('button').forEach(b=>b.disabled=true);
+ document.getElementById('status').textContent='Working...';
+ try{const response=await fetch(path,{method:'POST'}),result=await response.json();
+  document.getElementById('status').textContent=JSON.stringify(result,null,2);
+ }catch(e){document.getElementById('status').textContent=String(e);}
+ finally{busy=false;document.querySelectorAll('button').forEach(b=>b.disabled=false);
+  try{await loadStatus();}catch(e){}}
+}
+function capture(){return mutate('/sample');}
+function excludeSample(index){if(confirm('Exclude sample '+index+'? Files will be kept in excluded/.'))return mutate('/samples/'+index+'/delete');}
 </script>"""
 
 
@@ -330,6 +462,20 @@ def do_GET(self) -> None:
                 with collector.lock:
                     status = collector.status()
                 self.reply(200, json.dumps(status, allow_nan=False).encode())
+            elif match := re.fullmatch(r"/samples/([0-9]+)/annotated\.png", self.path):
+                index = int(match[1])
+                with collector.lock:
+                    if not any(s["index"] == index for s in collector.samples):
+                        self.send_error(404)
+                        return
+                    try:
+                        data = (
+                            collector.root / f"sample_{index:03d}" / "annotated.png"
+                        ).read_bytes()
+                    except OSError as error:
+                        self.reply(503, json.dumps({"error": str(error)}).encode())
+                        return
+                self.reply(200, data, "image/png")
             elif self.path.startswith("/frame.jpg"):
                 try:
                     self.reply(200, collector.preview(), "image/jpeg")
@@ -339,16 +485,21 @@ def do_GET(self) -> None:
                 self.send_error(404)
 
         def do_POST(self) -> None:
-            """Record one sample while rejecting concurrent acquisition."""
-            if self.path != "/sample":
+            """Serialize sample capture and recoverable exclusion."""
+            deletion = re.fullmatch(r"/samples/([0-9]+)/delete", self.path)
+            if self.path != "/sample" and deletion is None:
                 self.send_error(404)
                 return
             if not collector.lock.acquire(blocking=False):
-                self.reply(409, b'{"error":"sample in progress"}')
+                self.reply(409, b'{"error":"sample operation in progress"}')
                 return
             try:
                 try:
-                    collector.last = collector.sample()
+                    collector.last = (
+                        collector.delete_sample(int(deletion[1]))
+                        if deletion
+                        else collector.sample()
+                    )
                     code = 200
                 except (
                     ValueError,
@@ -357,7 +508,7 @@ def do_POST(self) -> None:
                     cv2.error,
                     subprocess.SubprocessError,
                 ) as error:
-                    LOGGER.warning("Sample rejected: %s", error)
+                    LOGGER.warning("Sample operation rejected: %s", error)
                     collector.last = {"accepted": False, "error": str(error)}
                     code = 422
                 self.reply(code, json.dumps(collector.last, allow_nan=False).encode())
@@ -386,6 +537,9 @@ def add_arguments(parser: argparse.ArgumentParser) -> None:
         help="Optional libfranka shared-library directory on the reader host",
     )
     parser.add_argument("--output", type=Path)
+    parser.add_argument(
+        "--resume", action="store_true", help="Resume --output with unchanged setup"
+    )
     parser.add_argument("--port", type=int, default=8767)
     parser.add_argument("--squares-x", type=int, default=6)
     parser.add_argument("--squares-y", type=int, default=8)
@@ -441,14 +595,18 @@ def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     add_arguments(parser)
     args = parser.parse_args()
+    if args.resume and args.output is None:
+        parser.error("--resume requires --output")
     config = build_config(args)
     root = args.output or Path(__file__).resolve().parent / "sessions" / (
         datetime.now().strftime("%Y%m%d_%H%M%S_%f") + f"_{config.arm}_eye_to_hand"
     )
-    if root.exists():
-        parser.error("Output directory already exists; use a new session directory")
+    try:
+        collector = Collector(config, root, resume=args.resume)
+    except (ValueError, OSError, KeyError) as error:
+        parser.error(str(error))
     logging.basicConfig(level=logging.INFO)
-    serve(Collector(config, root), args.port)
+    serve(collector, args.port)
 
 
 if __name__ == "__main__":
diff --git a/tests/unit_tests/calibration_tools/test_collector_session.py b/tests/unit_tests/calibration_tools/test_collector_session.py
new file mode 100644
index 000000000..f2d53cbb8
--- /dev/null
+++ b/tests/unit_tests/calibration_tools/test_collector_session.py
@@ -0,0 +1,478 @@
+# Copyright 2026 The RPent Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Offline regression coverage for interrupted and reviewed calibration sessions."""
+
+from __future__ import annotations
+
+import base64
+import copy
+import hashlib
+import importlib
+import io
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def collector_module(monkeypatch):
+    cv2 = pytest.importorskip("cv2")
+    if not hasattr(cv2, "aruco") or not hasattr(cv2.aruco, "CharucoDetector"):
+        pytest.skip("Calibration requires OpenCV with the ChArUco detector")
+    pytest.importorskip("scipy")
+    root = Path(__file__).resolve().parents[3]
+    monkeypatch.syspath_prepend(str(root / "calibration_tools"))
+    return importlib.import_module("base_handeye_collect")
+
+
+@pytest.fixture
+def config(collector_module):
+    return collector_module.CaptureConfig(
+        arm="left",
+        camera_serial="offline-camera",
+        camera_url="http://offline.invalid/frame",
+        reader_command=("offline-reader",),
+    )
+
+
+@pytest.fixture
+def record(collector_module, config):
+    identity = collector_module.np.eye(4)
+    pose = identity.copy()
+    pose[0, 3] = 0.1
+    target = identity.copy()
+    target[2, 3] = 0.8
+    state = {
+        "O_T_EE": pose.flatten(order="F").tolist(),
+        "F_T_EE": identity.flatten(order="F").tolist(),
+        "dq": [0.0] * 7,
+        "robot_mode": 1,
+        "has_errors": False,
+    }
+    return {
+        "index": 1,
+        "arm": config.arm,
+        "calibration_mode": "eye_to_hand",
+        "board": config.board_metadata,
+        "robot_before": state,
+        "robot_after": copy.deepcopy(state),
+        "T_left_base_ee": pose.tolist(),
+        "T_camera_board": target.tolist(),
+        "camera": {
+            "serial": config.camera_serial,
+            "width": 680,
+            "height": 880,
+            "K": [[800.0, 0.0, 340.0], [0.0, 800.0, 440.0], [0.0, 0.0, 1.0]],
+            "distortion": [0.0] * 5,
+            "distortion_model": "distortion.brown_conrady",
+            "host_received_s": 1.0,
+        },
+        "corners": 20,
+        "markers": 12,
+        "charuco_ids": list(range(20)),
+        "charuco_corners_px": [[float(i % 5), float(i // 5)] for i in range(20)],
+        "reprojection_rms_px": 0.2,
+        "drift_m": 0.0,
+        "drift_rad": 0.0,
+    }
+
+
+def save_record(root, record, index):
+    """Create a complete saved sample, preserving its serialized bytes for checks."""
+    folder = root / f"sample_{index:03d}"
+    folder.mkdir(parents=True)
+    value = copy.deepcopy(record)
+    value["index"] = index
+    (folder / "sample.json").write_text(json.dumps(value), encoding="utf-8")
+    (folder / "color.png").write_bytes(b"original-color-png")
+    (folder / "annotated.png").write_bytes(b"original-annotated-png")
+    return folder
+
+
+def snapshot(folder):
+    return {
+        str(path.relative_to(folder)): path.read_bytes()
+        for path in folder.rglob("*")
+        if path.is_file()
+    }
+
+
+@pytest.fixture
+def session_root(tmp_path, record):
+    root = tmp_path / "session"
+    save_record(root, record, 1)
+    return root
+
+
+@pytest.fixture
+def collector(collector_module, config, session_root):
+    return collector_module.Collector(config, session_root, resume=True)
+
+
+def test_existing_session_requires_explicit_resume(
+    collector_module, config, session_root
+):
+    original = snapshot(session_root)
+    with pytest.raises((ValueError, FileExistsError)):
+        collector_module.Collector(config, session_root)
+    assert snapshot(session_root) == original
+
+
+def test_resume_preserves_originals_and_continues_after_gaps(
+    collector_module, config, record, tmp_path
+):
+    root = tmp_path / "session"
+    save_record(root, record, 7)
+    save_record(root, record, 2)
+    original = snapshot(root)
+
+    collector = collector_module.Collector(config, root, resume=True)
+
+    assert [sample["index"] for sample in collector.samples] == [2, 7]
+    assert collector.next_index == 8
+    assert collector.status()["count"] == 2
+    assert collector.status()["samples"] == [
+        {"index": index, "corners": 20, "reprojection_rms_px": 0.2} for index in [2, 7]
+    ]
+    assert snapshot(root) == original
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("arm", "right"),
+        ("calibration_mode", "eye_in_hand"),
+        ("board", {"squares": [7, 8]}),
+        ("index", 99),
+    ],
+)
+def test_resume_rejects_incompatible_sample_metadata(
+    collector_module, config, session_root, field, value
+):
+    folder = session_root / "sample_001"
+    changed = json.loads((folder / "sample.json").read_text())
+    changed[field] = value
+    (folder / "sample.json").write_text(json.dumps(changed))
+    original = snapshot(session_root)
+
+    with pytest.raises(ValueError):
+        collector_module.Collector(config, session_root, resume=True)
+
+    assert snapshot(session_root) == original
+
+
+@pytest.mark.parametrize("field", ["serial", "K", "F_T_EE"])
+def test_resume_rejects_changed_camera_or_end_effector(
+    collector_module, config, record, session_root, field
+):
+    changed = copy.deepcopy(record)
+    if field == "serial":
+        changed["camera"]["serial"] = "another-camera"
+    elif field == "K":
+        changed["camera"]["K"][0][0] += 10.0
+    else:
+        for key in ("robot_before", "robot_after"):
+            changed[key]["F_T_EE"][12] = 0.05
+    save_record(session_root, changed, 2)
+
+    with pytest.raises(ValueError):
+        collector_module.Collector(config, session_root, resume=True)
+
+
+@pytest.mark.parametrize("missing", ["sample.json", "color.png", "annotated.png"])
+def test_resume_rejects_incomplete_sample(
+    collector_module, config, session_root, missing
+):
+    (session_root / "sample_001" / missing).unlink()
+
+    with pytest.raises((ValueError, OSError)):
+        collector_module.Collector(config, session_root, resume=True)
+
+
+def test_delete_archives_original_files_and_retains_monotonic_ids(
+    collector_module, config, record, tmp_path
+):
+    root = tmp_path / "session"
+    folder = save_record(root, record, 7)
+    original = snapshot(folder)
+    collector = collector_module.Collector(config, root, resume=True)
+
+    result = collector.delete_sample(7)
+
+    archived = root / "excluded" / "sample_007"
+    assert result == {
+        "deleted": True,
+        "index": 7,
+        "count": 0,
+        "archived": str(archived),
+    }
+    assert not folder.exists()
+    assert snapshot(archived) == original
+    assert collector.samples == []
+    assert collector.next_index == 8
+    resumed = collector_module.Collector(config, root, resume=True)
+    assert resumed.samples == []
+    assert resumed.next_index == 8
+
+
+@pytest.mark.parametrize("index", [0, -1, True, "1", None, 99])
+def test_delete_rejects_invalid_or_missing_ids_without_mutation(collector, index):
+    original = snapshot(collector.root)
+
+    with pytest.raises(ValueError):
+        collector.delete_sample(index)
+
+    assert snapshot(collector.root) == original
+    assert [sample["index"] for sample in collector.samples] == [1]
+
+
+def test_archive_conflict_never_overwrites_either_copy(collector, record):
+    archived = save_record(collector.root / "excluded", record, 1)
+    (archived / "color.png").write_bytes(b"previously-excluded-image")
+    original = snapshot(collector.root)
+
+    with pytest.raises((ValueError, FileExistsError)):
+        collector.delete_sample(1)
+
+    assert snapshot(collector.root) == original
+    assert collector.status()["count"] == 1
+
+
+def test_failed_archive_rename_preserves_active_sample(collector, monkeypatch):
+    original = snapshot(collector.root)
+
+    def fail_rename(path, target):
+        raise OSError("simulated archive rename failure")
+
+    monkeypatch.setattr(Path, "rename", fail_rename)
+    with pytest.raises(OSError, match="simulated"):
+        collector.delete_sample(1)
+
+    assert snapshot(collector.root) == original
+    assert collector.status()["count"] == 1
+    assert collector.next_index == 2
+
+
+def test_capture_after_resume_and_delete_never_reuses_saved_ids(
+    collector_module, config, record, tmp_path, monkeypatch
+):
+    root = tmp_path / "session"
+    save_record(root, record, 2)
+    save_record(root / "excluded", record, 9)
+    collector = collector_module.Collector(config, root, resume=True)
+    collector.delete_sample(2)
+    original = snapshot(root)
+
+    state = copy.deepcopy(record["robot_before"])
+    state["O_T_EE"][12] = 0.3
+    monkeypatch.setattr(collector, "read_state", lambda: copy.deepcopy(state))
+    monkeypatch.setattr(collector_module.time, "sleep", lambda _: None)
+    board_image = collector.board.generateImage((680, 880), marginSize=40)
+    success, encoded = collector_module.cv2.imencode(".png", board_image)
+    assert success
+
+    def camera_response(*args, **kwargs):
+        camera = copy.deepcopy(record["camera"])
+        camera["host_received_s"] = collector_module.time.time()
+        camera["png_b64"] = base64.b64encode(encoded.tobytes()).decode()
+        return io.BytesIO(json.dumps(camera).encode())
+
+    monkeypatch.setattr(collector_module.urllib.request, "urlopen", camera_response)
+
+    result = collector.sample()
+
+    assert result["accepted"] is True
+    assert result["count"] == 1
+    assert Path(result["saved"]).name == "sample_010"
+    assert collector.samples[0]["index"] == 10
+    assert collector.next_index == 11
+    assert all(
+        (root / name).read_bytes() == content for name, content in original.items()
+    )
+    resumed = collector_module.Collector(config, root, resume=True)
+    assert [sample["index"] for sample in resumed.samples] == [10]
+    assert resumed.next_index == 11
+
+
+def request_handler(collector_module, collector, path, method):
+    """Exercise the real routing methods without binding a network socket."""
+    handler_type = collector_module.make_handler(collector)
+    handler = handler_type.__new__(handler_type)
+    handler.path = path
+    responses = []
+    handler.reply = lambda code, body, kind="application/json": responses.append(
+        (code, body, kind)
+    )
+    handler.send_error = lambda code, *args: responses.append((code, b"", ""))
+    getattr(handler, f"do_{method}")()
+    assert len(responses) == 1
+    return responses[0]
+
+
+def test_http_review_delete_and_capture_share_mutation_lock(
+    collector_module, collector
+):
+    code, body, kind = request_handler(
+        collector_module, collector, "/samples/1/annotated.png", "GET"
+    )
+    assert (code, body, kind) == (200, b"original-annotated-png", "image/png")
+
+    with collector.lock:
+        for path in ("/sample", "/samples/1/delete"):
+            code, _, _ = request_handler(collector_module, collector, path, "POST")
+            assert code == 409
+    assert collector.status()["count"] == 1
+
+    code, body, _ = request_handler(
+        collector_module, collector, "/samples/1/delete", "POST"
+    )
+    assert code == 200
+    assert json.loads(body)["deleted"] is True
+    assert collector.status()["count"] == 0
+    assert not collector.lock.locked()
+    code, _, _ = request_handler(
+        collector_module, collector, "/samples/1/annotated.png", "GET"
+    )
+    assert code == 404
+
+
+@pytest.mark.parametrize("index", ["0", "-1", "wrong", "99"])
+def test_http_delete_invalid_id_retains_session(collector_module, collector, index):
+    code, _, _ = request_handler(
+        collector_module, collector, f"/samples/{index}/delete", "POST"
+    )
+
+    assert code in (404, 422)
+    assert collector.status()["count"] == 1
+    assert not collector.lock.locked()
+
+
+@pytest.mark.parametrize("restart", [False, True])
+@pytest.mark.parametrize("changed_field", ["K", "F_T_EE"])
+def test_deleting_all_samples_retains_camera_and_frame_baseline(
+    collector_module, collector, config, record, monkeypatch, restart, changed_field
+):
+    root = collector.root
+    collector.delete_sample(1)
+    if restart:
+        collector = collector_module.Collector(config, root, resume=True)
+    original = snapshot(root)
+    state = copy.deepcopy(record["robot_before"])
+    camera = copy.deepcopy(record["camera"])
+    if changed_field == "K":
+        camera["K"][0][0] += 20.0
+        expected_error = "Camera K"
+    else:
+        state["F_T_EE"][12] = 0.05
+        expected_error = "末端坐标定义与先前样本不同"
+    monkeypatch.setattr(collector, "read_state", lambda: copy.deepcopy(state))
+    monkeypatch.setattr(collector_module.time, "sleep", lambda _: None)
+
+    def camera_response(*args, **kwargs):
+        camera["host_received_s"] = collector_module.time.time()
+        return io.BytesIO(json.dumps(camera).encode())
+
+    monkeypatch.setattr(collector_module.urllib.request, "urlopen", camera_response)
+
+    with pytest.raises(ValueError, match=expected_error):
+        collector.sample()
+
+    assert collector.samples == []
+    assert collector.next_index == 2
+    assert snapshot(root) == original
+
+
+def test_deleting_sample_invalidates_previously_solved_candidate(
+    collector_module, config, record, tmp_path
+):
+    root = tmp_path / "session"
+    for index in range(1, 11):
+        save_record(root, record, index)
+    pose = collector_module.np.eye(4).tolist()
+    candidate = {
+        "status": "candidate_requires_independent_validation",
+        "arm": "left",
+        "calibration_mode": "eye_to_hand",
+        "sample_count": 10,
+        "selected_method": "PARK",
+        "camera_serial": config.camera_serial,
+        "T_left_base_camera": pose,
+        "T_ee_board": pose,
+    }
+    report = {
+        **candidate,
+        "serial": config.camera_serial,
+        "source_hashes": {
+            str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
+            for path in root.glob("sample_*/sample.json")
+        },
+        "methods": {"PARK": {"T_left_base_camera": pose, "T_ee_board": pose}},
+    }
+    candidate_path = root / "candidate.json"
+    candidate_path.write_text(json.dumps(candidate))
+    (root / "quality_report.json").write_text(json.dumps(report))
+    common = importlib.import_module("common")
+    assert common.load_candidate(candidate_path) == candidate
+
+    collector_module.Collector(config, root, resume=True).delete_sample(5)
+
+    with pytest.raises(ValueError, match="Training samples changed"):
+        common.load_candidate(candidate_path)
+    assert json.loads(candidate_path.read_text()) == candidate
+
+
+@pytest.mark.parametrize("resume_without_output", [False, True])
+def test_cli_requires_explicit_existing_session_selection(
+    collector_module,
+    config,
+    session_root,
+    monkeypatch,
+    capsys,
+    resume_without_output,
+):
+    original = snapshot(session_root)
+    arguments = [
+        "base_handeye_collect.py",
+        "--arm",
+        config.arm,
+        "--camera-serial",
+        config.camera_serial,
+        "--camera-url",
+        config.camera_url,
+        "--reader",
+        "offline-reader",
+        "--robot-ip",
+        "192.0.2.1",
+    ]
+    if resume_without_output:
+        arguments.append("--resume")
+        message = "--resume requires --output"
+    else:
+        arguments.extend(["--output", str(session_root)])
+        message = "already exists"
+    monkeypatch.setattr(sys, "argv", arguments)
+    monkeypatch.setattr(
+        collector_module,
+        "serve",
+        lambda *args: pytest.fail("Must reject before serving"),
+    )
+
+    with pytest.raises(SystemExit) as error:
+        collector_module.main()
+
+    assert error.value.code == 2
+    assert message in capsys.readouterr().err
+    assert snapshot(session_root) == original

From 4bf3f5cbc092247e619f1d67edeb1b7dcc07d91d Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sat, 10 Oct 2026 22:06:26 +0800
Subject: [PATCH 2/7] fix(ci): pin OpenCV for calibration regressions

---
 .github/workflows/unittest.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/unittest.yml b/.github/workflows/unittest.yml
index db2c31d0e..462d60d91 100644
--- a/.github/workflows/unittest.yml
+++ b/.github/workflows/unittest.yml
@@ -33,7 +33,7 @@ jobs:
       - name: Install test dependencies
         run: uv pip install --system -e ".[test]"
       - name: Install Flywheel export dependencies
-        run: uv pip install --system -e ".[flywheel]" --torch-backend cpu
+        run: uv pip install --system -e ".[flywheel]" "opencv-python-headless==4.10.0.84" --torch-backend cpu
       - name: Run unit tests with coverage
         run: coverage run --source=rpent,robots -m pytest tests/unit_tests -v
       - name: Generate coverage report

From 3dac862ea7755d63893a7bfc14931691427f15c3 Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sat, 10 Oct 2026 22:56:57 +0800
Subject: [PATCH 3/7] test(calibration): cover review and resume as one
 sampling workflow

---
 .../test_collector_session.py                 | 165 +++---------------
 1 file changed, 26 insertions(+), 139 deletions(-)

diff --git a/tests/unit_tests/calibration_tools/test_collector_session.py b/tests/unit_tests/calibration_tools/test_collector_session.py
index f2d53cbb8..79ec6d662 100644
--- a/tests/unit_tests/calibration_tools/test_collector_session.py
+++ b/tests/unit_tests/calibration_tools/test_collector_session.py
@@ -122,41 +122,10 @@ def collector(collector_module, config, session_root):
     return collector_module.Collector(config, session_root, resume=True)
 
 
-def test_existing_session_requires_explicit_resume(
-    collector_module, config, session_root
-):
-    original = snapshot(session_root)
-    with pytest.raises((ValueError, FileExistsError)):
-        collector_module.Collector(config, session_root)
-    assert snapshot(session_root) == original
-
-
-def test_resume_preserves_originals_and_continues_after_gaps(
-    collector_module, config, record, tmp_path
-):
-    root = tmp_path / "session"
-    save_record(root, record, 7)
-    save_record(root, record, 2)
-    original = snapshot(root)
-
-    collector = collector_module.Collector(config, root, resume=True)
-
-    assert [sample["index"] for sample in collector.samples] == [2, 7]
-    assert collector.next_index == 8
-    assert collector.status()["count"] == 2
-    assert collector.status()["samples"] == [
-        {"index": index, "corners": 20, "reprojection_rms_px": 0.2} for index in [2, 7]
-    ]
-    assert snapshot(root) == original
-
-
 @pytest.mark.parametrize(
     ("field", "value"),
     [
-        ("arm", "right"),
-        ("calibration_mode", "eye_in_hand"),
         ("board", {"squares": [7, 8]}),
-        ("index", 99),
     ],
 )
 def test_resume_rejects_incompatible_sample_metadata(
@@ -174,71 +143,6 @@ def test_resume_rejects_incompatible_sample_metadata(
     assert snapshot(session_root) == original
 
 
-@pytest.mark.parametrize("field", ["serial", "K", "F_T_EE"])
-def test_resume_rejects_changed_camera_or_end_effector(
-    collector_module, config, record, session_root, field
-):
-    changed = copy.deepcopy(record)
-    if field == "serial":
-        changed["camera"]["serial"] = "another-camera"
-    elif field == "K":
-        changed["camera"]["K"][0][0] += 10.0
-    else:
-        for key in ("robot_before", "robot_after"):
-            changed[key]["F_T_EE"][12] = 0.05
-    save_record(session_root, changed, 2)
-
-    with pytest.raises(ValueError):
-        collector_module.Collector(config, session_root, resume=True)
-
-
-@pytest.mark.parametrize("missing", ["sample.json", "color.png", "annotated.png"])
-def test_resume_rejects_incomplete_sample(
-    collector_module, config, session_root, missing
-):
-    (session_root / "sample_001" / missing).unlink()
-
-    with pytest.raises((ValueError, OSError)):
-        collector_module.Collector(config, session_root, resume=True)
-
-
-def test_delete_archives_original_files_and_retains_monotonic_ids(
-    collector_module, config, record, tmp_path
-):
-    root = tmp_path / "session"
-    folder = save_record(root, record, 7)
-    original = snapshot(folder)
-    collector = collector_module.Collector(config, root, resume=True)
-
-    result = collector.delete_sample(7)
-
-    archived = root / "excluded" / "sample_007"
-    assert result == {
-        "deleted": True,
-        "index": 7,
-        "count": 0,
-        "archived": str(archived),
-    }
-    assert not folder.exists()
-    assert snapshot(archived) == original
-    assert collector.samples == []
-    assert collector.next_index == 8
-    resumed = collector_module.Collector(config, root, resume=True)
-    assert resumed.samples == []
-    assert resumed.next_index == 8
-
-
-@pytest.mark.parametrize("index", [0, -1, True, "1", None, 99])
-def test_delete_rejects_invalid_or_missing_ids_without_mutation(collector, index):
-    original = snapshot(collector.root)
-
-    with pytest.raises(ValueError):
-        collector.delete_sample(index)
-
-    assert snapshot(collector.root) == original
-    assert [sample["index"] for sample in collector.samples] == [1]
-
-
 def test_archive_conflict_never_overwrites_either_copy(collector, record):
     archived = save_record(collector.root / "excluded", record, 1)
     (archived / "color.png").write_bytes(b"previously-excluded-image")
@@ -273,7 +177,30 @@ def test_capture_after_resume_and_delete_never_reuses_saved_ids(
     save_record(root, record, 2)
     save_record(root / "excluded", record, 9)
     collector = collector_module.Collector(config, root, resume=True)
-    collector.delete_sample(2)
+    active_original = snapshot(root / "sample_002")
+    code, body, kind = request_handler(
+        collector_module, collector, "/samples/2/annotated.png", "GET"
+    )
+    assert (code, body, kind) == (200, b"original-annotated-png", "image/png")
+    with collector.lock:
+        for path in ("/sample", "/samples/2/delete"):
+            assert request_handler(collector_module, collector, path, "POST")[0] == 409
+    code, body, _ = request_handler(
+        collector_module, collector, "/samples/2/delete", "POST"
+    )
+    assert code == 200
+    assert json.loads(body)["deleted"] is True
+    assert not collector.lock.locked()
+    assert (
+        request_handler(collector_module, collector, "/samples/2/annotated.png", "GET")[
+            0
+        ]
+        == 404
+    )
+    assert snapshot(root / "excluded" / "sample_002") == active_original
+    collector = collector_module.Collector(config, root, resume=True)
+    assert collector.samples == []
+    assert collector.next_index == 10
     original = snapshot(root)
 
     state = copy.deepcopy(record["robot_before"])
@@ -322,53 +249,13 @@ def request_handler(collector_module, collector, path, method):
     return responses[0]
 
 
-def test_http_review_delete_and_capture_share_mutation_lock(
-    collector_module, collector
-):
-    code, body, kind = request_handler(
-        collector_module, collector, "/samples/1/annotated.png", "GET"
-    )
-    assert (code, body, kind) == (200, b"original-annotated-png", "image/png")
-
-    with collector.lock:
-        for path in ("/sample", "/samples/1/delete"):
-            code, _, _ = request_handler(collector_module, collector, path, "POST")
-            assert code == 409
-    assert collector.status()["count"] == 1
-
-    code, body, _ = request_handler(
-        collector_module, collector, "/samples/1/delete", "POST"
-    )
-    assert code == 200
-    assert json.loads(body)["deleted"] is True
-    assert collector.status()["count"] == 0
-    assert not collector.lock.locked()
-    code, _, _ = request_handler(
-        collector_module, collector, "/samples/1/annotated.png", "GET"
-    )
-    assert code == 404
-
-
-@pytest.mark.parametrize("index", ["0", "-1", "wrong", "99"])
-def test_http_delete_invalid_id_retains_session(collector_module, collector, index):
-    code, _, _ = request_handler(
-        collector_module, collector, f"/samples/{index}/delete", "POST"
-    )
-
-    assert code in (404, 422)
-    assert collector.status()["count"] == 1
-    assert not collector.lock.locked()
-
-
-@pytest.mark.parametrize("restart", [False, True])
 @pytest.mark.parametrize("changed_field", ["K", "F_T_EE"])
 def test_deleting_all_samples_retains_camera_and_frame_baseline(
-    collector_module, collector, config, record, monkeypatch, restart, changed_field
+    collector_module, collector, config, record, monkeypatch, changed_field
 ):
     root = collector.root
     collector.delete_sample(1)
-    if restart:
-        collector = collector_module.Collector(config, root, resume=True)
+    collector = collector_module.Collector(config, root, resume=True)
     original = snapshot(root)
     state = copy.deepcopy(record["robot_before"])
     camera = copy.deepcopy(record["camera"])

From 3b9fc140484669998c36844001b9e7b09f33ae9c Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sat, 10 Oct 2026 23:07:07 +0800
Subject: [PATCH 4/7] test(calibration): verify complete sample management
 workflows

---
 .../test_collector_session.py                 | 365 +++++-------------
 1 file changed, 104 insertions(+), 261 deletions(-)

diff --git a/tests/unit_tests/calibration_tools/test_collector_session.py b/tests/unit_tests/calibration_tools/test_collector_session.py
index 79ec6d662..98f1802c1 100644
--- a/tests/unit_tests/calibration_tools/test_collector_session.py
+++ b/tests/unit_tests/calibration_tools/test_collector_session.py
@@ -11,7 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-"""Offline regression coverage for interrupted and reviewed calibration sessions."""
+"""Offline functional checks for reviewed and resumed calibration sessions."""
 
 from __future__ import annotations
 
@@ -21,7 +21,8 @@
 import importlib
 import io
 import json
-import sys
+import shutil
+from dataclasses import replace
 from pathlib import Path
 
 import pytest
@@ -33,73 +34,54 @@ def collector_module(monkeypatch):
     if not hasattr(cv2, "aruco") or not hasattr(cv2.aruco, "CharucoDetector"):
         pytest.skip("Calibration requires OpenCV with the ChArUco detector")
     pytest.importorskip("scipy")
-    root = Path(__file__).resolve().parents[3]
-    monkeypatch.syspath_prepend(str(root / "calibration_tools"))
+    monkeypatch.syspath_prepend(
+        str(Path(__file__).resolve().parents[3] / "calibration_tools")
+    )
     return importlib.import_module("base_handeye_collect")
 
 
 @pytest.fixture
-def config(collector_module):
-    return collector_module.CaptureConfig(
+def session(collector_module, tmp_path, monkeypatch):
+    config = collector_module.CaptureConfig(
         arm="left",
         camera_serial="offline-camera",
         camera_url="http://offline.invalid/frame",
         reader_command=("offline-reader",),
     )
-
-
-@pytest.fixture
-def record(collector_module, config):
-    identity = collector_module.np.eye(4)
-    pose = identity.copy()
-    pose[0, 3] = 0.1
-    target = identity.copy()
-    target[2, 3] = 0.8
+    collector = collector_module.Collector(config, tmp_path / "session")
     state = {
-        "O_T_EE": pose.flatten(order="F").tolist(),
-        "F_T_EE": identity.flatten(order="F").tolist(),
+        "O_T_EE": collector_module.np.eye(4).flatten(order="F").tolist(),
+        "F_T_EE": collector_module.np.eye(4).flatten(order="F").tolist(),
         "dq": [0.0] * 7,
         "robot_mode": 1,
         "has_errors": False,
     }
-    return {
-        "index": 1,
-        "arm": config.arm,
-        "calibration_mode": "eye_to_hand",
-        "board": config.board_metadata,
-        "robot_before": state,
-        "robot_after": copy.deepcopy(state),
-        "T_left_base_ee": pose.tolist(),
-        "T_camera_board": target.tolist(),
-        "camera": {
-            "serial": config.camera_serial,
-            "width": 680,
-            "height": 880,
-            "K": [[800.0, 0.0, 340.0], [0.0, 800.0, 440.0], [0.0, 0.0, 1.0]],
-            "distortion": [0.0] * 5,
-            "distortion_model": "distortion.brown_conrady",
-            "host_received_s": 1.0,
-        },
-        "corners": 20,
-        "markers": 12,
-        "charuco_ids": list(range(20)),
-        "charuco_corners_px": [[float(i % 5), float(i // 5)] for i in range(20)],
-        "reprojection_rms_px": 0.2,
-        "drift_m": 0.0,
-        "drift_rad": 0.0,
+    image = collector.board.generateImage((680, 880), marginSize=40)
+    success, encoded = collector_module.cv2.imencode(".png", image)
+    assert success
+    camera = {
+        "serial": config.camera_serial,
+        "width": 680,
+        "height": 880,
+        "K": [[800.0, 0.0, 340.0], [0.0, 800.0, 440.0], [0.0, 0.0, 1.0]],
+        "distortion": [0.0] * 5,
+        "distortion_model": "distortion.brown_conrady",
+        "png_b64": base64.b64encode(encoded.tobytes()).decode(),
     }
 
+    def camera_response(*args, **kwargs):
+        return io.BytesIO(
+            json.dumps(
+                {**camera, "host_received_s": collector_module.time.time()}
+            ).encode()
+        )
 
-def save_record(root, record, index):
-    """Create a complete saved sample, preserving its serialized bytes for checks."""
-    folder = root / f"sample_{index:03d}"
-    folder.mkdir(parents=True)
-    value = copy.deepcopy(record)
-    value["index"] = index
-    (folder / "sample.json").write_text(json.dumps(value), encoding="utf-8")
-    (folder / "color.png").write_bytes(b"original-color-png")
-    (folder / "annotated.png").write_bytes(b"original-annotated-png")
-    return folder
+    monkeypatch.setattr(
+        collector_module.Collector, "read_state", lambda self: copy.deepcopy(state)
+    )
+    monkeypatch.setattr(collector_module.time, "sleep", lambda _: None)
+    monkeypatch.setattr(collector_module.urllib.request, "urlopen", camera_response)
+    return collector, state, camera
 
 
 def snapshot(folder):
@@ -110,133 +92,9 @@ def snapshot(folder):
     }
 
 
-@pytest.fixture
-def session_root(tmp_path, record):
-    root = tmp_path / "session"
-    save_record(root, record, 1)
-    return root
-
-
-@pytest.fixture
-def collector(collector_module, config, session_root):
-    return collector_module.Collector(config, session_root, resume=True)
-
-
-@pytest.mark.parametrize(
-    ("field", "value"),
-    [
-        ("board", {"squares": [7, 8]}),
-    ],
-)
-def test_resume_rejects_incompatible_sample_metadata(
-    collector_module, config, session_root, field, value
-):
-    folder = session_root / "sample_001"
-    changed = json.loads((folder / "sample.json").read_text())
-    changed[field] = value
-    (folder / "sample.json").write_text(json.dumps(changed))
-    original = snapshot(session_root)
-
-    with pytest.raises(ValueError):
-        collector_module.Collector(config, session_root, resume=True)
-
-    assert snapshot(session_root) == original
-
-
-def test_archive_conflict_never_overwrites_either_copy(collector, record):
-    archived = save_record(collector.root / "excluded", record, 1)
-    (archived / "color.png").write_bytes(b"previously-excluded-image")
-    original = snapshot(collector.root)
-
-    with pytest.raises((ValueError, FileExistsError)):
-        collector.delete_sample(1)
-
-    assert snapshot(collector.root) == original
-    assert collector.status()["count"] == 1
-
-
-def test_failed_archive_rename_preserves_active_sample(collector, monkeypatch):
-    original = snapshot(collector.root)
-
-    def fail_rename(path, target):
-        raise OSError("simulated archive rename failure")
-
-    monkeypatch.setattr(Path, "rename", fail_rename)
-    with pytest.raises(OSError, match="simulated"):
-        collector.delete_sample(1)
-
-    assert snapshot(collector.root) == original
-    assert collector.status()["count"] == 1
-    assert collector.next_index == 2
-
-
-def test_capture_after_resume_and_delete_never_reuses_saved_ids(
-    collector_module, config, record, tmp_path, monkeypatch
-):
-    root = tmp_path / "session"
-    save_record(root, record, 2)
-    save_record(root / "excluded", record, 9)
-    collector = collector_module.Collector(config, root, resume=True)
-    active_original = snapshot(root / "sample_002")
-    code, body, kind = request_handler(
-        collector_module, collector, "/samples/2/annotated.png", "GET"
-    )
-    assert (code, body, kind) == (200, b"original-annotated-png", "image/png")
-    with collector.lock:
-        for path in ("/sample", "/samples/2/delete"):
-            assert request_handler(collector_module, collector, path, "POST")[0] == 409
-    code, body, _ = request_handler(
-        collector_module, collector, "/samples/2/delete", "POST"
-    )
-    assert code == 200
-    assert json.loads(body)["deleted"] is True
-    assert not collector.lock.locked()
-    assert (
-        request_handler(collector_module, collector, "/samples/2/annotated.png", "GET")[
-            0
-        ]
-        == 404
-    )
-    assert snapshot(root / "excluded" / "sample_002") == active_original
-    collector = collector_module.Collector(config, root, resume=True)
-    assert collector.samples == []
-    assert collector.next_index == 10
-    original = snapshot(root)
-
-    state = copy.deepcopy(record["robot_before"])
-    state["O_T_EE"][12] = 0.3
-    monkeypatch.setattr(collector, "read_state", lambda: copy.deepcopy(state))
-    monkeypatch.setattr(collector_module.time, "sleep", lambda _: None)
-    board_image = collector.board.generateImage((680, 880), marginSize=40)
-    success, encoded = collector_module.cv2.imencode(".png", board_image)
-    assert success
-
-    def camera_response(*args, **kwargs):
-        camera = copy.deepcopy(record["camera"])
-        camera["host_received_s"] = collector_module.time.time()
-        camera["png_b64"] = base64.b64encode(encoded.tobytes()).decode()
-        return io.BytesIO(json.dumps(camera).encode())
-
-    monkeypatch.setattr(collector_module.urllib.request, "urlopen", camera_response)
-
-    result = collector.sample()
-
-    assert result["accepted"] is True
-    assert result["count"] == 1
-    assert Path(result["saved"]).name == "sample_010"
-    assert collector.samples[0]["index"] == 10
-    assert collector.next_index == 11
-    assert all(
-        (root / name).read_bytes() == content for name, content in original.items()
-    )
-    resumed = collector_module.Collector(config, root, resume=True)
-    assert [sample["index"] for sample in resumed.samples] == [10]
-    assert resumed.next_index == 11
-
-
-def request_handler(collector_module, collector, path, method):
-    """Exercise the real routing methods without binding a network socket."""
-    handler_type = collector_module.make_handler(collector)
+def request_handler(module, collector, path, method="POST"):
+    """Exercise actual HTTP routing without opening a listening socket."""
+    handler_type = module.make_handler(collector)
     handler = handler_type.__new__(handler_type)
     handler.path = path
     responses = []
@@ -244,50 +102,79 @@ def request_handler(collector_module, collector, path, method):
         (code, body, kind)
     )
     handler.send_error = lambda code, *args: responses.append((code, b"", ""))
-    getattr(handler, f"do_{method}")()
+    if method == "GET":
+        handler.do_GET()
+    else:
+        handler.do_POST()
     assert len(responses) == 1
     return responses[0]
 
 
-@pytest.mark.parametrize("changed_field", ["K", "F_T_EE"])
-def test_deleting_all_samples_retains_camera_and_frame_baseline(
-    collector_module, collector, config, record, monkeypatch, changed_field
+def test_review_exclude_resume_and_continue_capture(collector_module, session):
+    collector, state, _ = session
+    code, body, _ = request_handler(collector_module, collector, "/sample")
+    assert code == 200 and json.loads(body)["accepted"]
+    original = snapshot(collector.root / "sample_001")
+    code, image, kind = request_handler(
+        collector_module, collector, "/samples/1/annotated.png", "GET"
+    )
+    assert (code, image, kind) == (200, original["annotated.png"], "image/png")
+    with collector.lock:
+        for path in ("/sample", "/samples/1/delete"):
+            assert request_handler(collector_module, collector, path)[0] == 409
+    code, body, _ = request_handler(collector_module, collector, "/samples/1/delete")
+    assert code == 200 and json.loads(body)["deleted"]
+    archived = collector.root / "excluded" / "sample_001"
+    assert snapshot(archived) == original
+
+    resumed = collector_module.Collector(collector.config, collector.root, resume=True)
+    assert resumed.status()["count"] == 0
+    state["O_T_EE"][12] = 0.1
+    code, body, _ = request_handler(collector_module, resumed, "/sample")
+    result = json.loads(body)
+    assert code == 200 and result["accepted"]
+    assert Path(result["saved"]).name == "sample_002"
+    final = collector_module.Collector(collector.config, collector.root, resume=True)
+    assert [sample["index"] for sample in final.status()["samples"]] == [2]
+    assert snapshot(archived) == original
+
+
+def test_existing_samples_are_protected_during_session_review(
+    collector_module, session
 ):
-    root = collector.root
-    collector.delete_sample(1)
-    collector = collector_module.Collector(config, root, resume=True)
-    original = snapshot(root)
-    state = copy.deepcopy(record["robot_before"])
-    camera = copy.deepcopy(record["camera"])
-    if changed_field == "K":
-        camera["K"][0][0] += 20.0
-        expected_error = "Camera K"
-    else:
-        state["F_T_EE"][12] = 0.05
-        expected_error = "末端坐标定义与先前样本不同"
-    monkeypatch.setattr(collector, "read_state", lambda: copy.deepcopy(state))
-    monkeypatch.setattr(collector_module.time, "sleep", lambda _: None)
-
-    def camera_response(*args, **kwargs):
-        camera["host_received_s"] = collector_module.time.time()
-        return io.BytesIO(json.dumps(camera).encode())
-
-    monkeypatch.setattr(collector_module.urllib.request, "urlopen", camera_response)
+    collector, _, camera = session
+    collector.sample()
+    archived = collector.root / "excluded" / "sample_001"
+    archived.parent.mkdir()
+    shutil.copytree(collector.root / "sample_001", archived)
+    original = snapshot(collector.root)
 
-    with pytest.raises(ValueError, match=expected_error):
-        collector.sample()
+    with pytest.raises(FileExistsError):
+        collector_module.Collector(collector.config, collector.root)
+    with pytest.raises(FileExistsError):
+        collector.delete_sample(1)
+    assert snapshot(collector.root) == original
 
-    assert collector.samples == []
-    assert collector.next_index == 2
-    assert snapshot(root) == original
+    archived.rename(collector.root / "saved_copy")
+    collector.delete_sample(1)
+    original = snapshot(collector.root)
+    changed = replace(collector.config, camera_serial="another-camera")
+    with pytest.raises(ValueError, match="configuration"):
+        collector_module.Collector(changed, collector.root, resume=True)
+    resumed = collector_module.Collector(collector.config, collector.root, resume=True)
+    camera["K"][0][0] += 20
+    with pytest.raises(ValueError, match="Camera K"):
+        resumed.sample()
+    assert resumed.status()["count"] == 0
+    assert snapshot(collector.root) == original
 
 
-def test_deleting_sample_invalidates_previously_solved_candidate(
-    collector_module, config, record, tmp_path
-):
-    root = tmp_path / "session"
-    for index in range(1, 11):
-        save_record(root, record, index)
+def test_reviewed_samples_require_a_new_calibration_result(collector_module, session):
+    collector, state, _ = session
+    for index in range(10):
+        state["O_T_EE"][12] = index * 0.02
+        collector.sample()
+    root = collector.root
     pose = collector_module.np.eye(4).tolist()
     candidate = {
         "status": "candidate_requires_independent_validation",
@@ -295,13 +182,13 @@ def test_deleting_sample_invalidates_previously_solved_candidate(
         "calibration_mode": "eye_to_hand",
         "sample_count": 10,
         "selected_method": "PARK",
-        "camera_serial": config.camera_serial,
+        "camera_serial": collector.config.camera_serial,
         "T_left_base_camera": pose,
         "T_ee_board": pose,
     }
     report = {
         **candidate,
-        "serial": config.camera_serial,
+        "serial": collector.config.camera_serial,
         "source_hashes": {
             str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
             for path in root.glob("sample_*/sample.json")
@@ -314,52 +201,8 @@ def test_deleting_sample_invalidates_previously_solved_candidate(
     common = importlib.import_module("common")
     assert common.load_candidate(candidate_path) == candidate
 
-    collector_module.Collector(config, root, resume=True).delete_sample(5)
+    collector.delete_sample(5)
 
     with pytest.raises(ValueError, match="Training samples changed"):
         common.load_candidate(candidate_path)
     assert json.loads(candidate_path.read_text()) == candidate
-
-
-@pytest.mark.parametrize("resume_without_output", [False, True])
-def test_cli_requires_explicit_existing_session_selection(
-    collector_module,
-    config,
-    session_root,
-    monkeypatch,
-    capsys,
-    resume_without_output,
-):
-    original = snapshot(session_root)
-    arguments = [
-        "base_handeye_collect.py",
-        "--arm",
-        config.arm,
-        "--camera-serial",
-        config.camera_serial,
-        "--camera-url",
-        config.camera_url,
-        "--reader",
-        "offline-reader",
-        "--robot-ip",
-        "192.0.2.1",
-    ]
-    if resume_without_output:
-        arguments.append("--resume")
-        message = "--resume requires --output"
-    else:
-        arguments.extend(["--output", str(session_root)])
-        message = "already exists"
-    monkeypatch.setattr(sys, "argv", arguments)
-    monkeypatch.setattr(
-        collector_module,
-        "serve",
-        lambda *args: pytest.fail("Must reject before serving"),
-    )
-
-    with pytest.raises(SystemExit) as error:
-        collector_module.main()
-
-    assert error.value.code == 2
-    assert message in capsys.readouterr().err
-    assert snapshot(session_root) == original

From 0cdf1ebc6eed2c14858d7e4016d5d9f9caf49158 Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sat, 10 Oct 2026 23:23:24 +0800
Subject: [PATCH 5/7] test(calibration): keep capture review and result
 invalidation flows

---
 .../test_collector_session.py                 | 32 -------------------
 1 file changed, 32 deletions(-)

diff --git a/tests/unit_tests/calibration_tools/test_collector_session.py b/tests/unit_tests/calibration_tools/test_collector_session.py
index 98f1802c1..9ea5ba867 100644
--- a/tests/unit_tests/calibration_tools/test_collector_session.py
+++ b/tests/unit_tests/calibration_tools/test_collector_session.py
@@ -21,8 +21,6 @@
 import importlib
 import io
 import json
-import shutil
-from dataclasses import replace
 from pathlib import Path
 
 import pytest
@@ -139,36 +137,6 @@ def test_review_exclude_resume_and_continue_capture(collector_module, session):
     assert snapshot(archived) == original
 
 
-def test_existing_samples_are_protected_during_session_review(
-    collector_module, session
-):
-    collector, _, camera = session
-    collector.sample()
-    archived = collector.root / "excluded" / "sample_001"
-    archived.parent.mkdir()
-    shutil.copytree(collector.root / "sample_001", archived)
-    original = snapshot(collector.root)
-
-    with pytest.raises(FileExistsError):
-        collector_module.Collector(collector.config, collector.root)
-    with pytest.raises(FileExistsError):
-        collector.delete_sample(1)
-    assert snapshot(collector.root) == original
-
-    archived.rename(collector.root / "saved_copy")
-    collector.delete_sample(1)
-    original = snapshot(collector.root)
-    changed = replace(collector.config, camera_serial="another-camera")
-    with pytest.raises(ValueError, match="configuration"):
-        collector_module.Collector(changed, collector.root, resume=True)
-    resumed = collector_module.Collector(collector.config, collector.root, resume=True)
-    camera["K"][0][0] += 20
-    with pytest.raises(ValueError, match="Camera K"):
-        resumed.sample()
-    assert resumed.status()["count"] == 0
-    assert snapshot(collector.root) == original
-
-
 def test_reviewed_samples_require_a_new_calibration_result(collector_module, session):
     collector, state, _ = session
     for index in range(10):

From 0266af721c23d3b5f5c79f465ecf957a5258df27 Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sun, 11 Oct 2026 16:10:46 +0800
Subject: [PATCH 6/7] refactor(calibration): share saved sample consistency
 validation

---
 calibration_tools/base_handeye_collect.py     | 39 ++++++-------------
 calibration_tools/common.py                   | 37 ++++++++++++++++++
 calibration_tools/solve_base_handeye.py       | 35 +++++------------
 .../test_collector_session.py                 | 12 ++++++
 4 files changed, 70 insertions(+), 53 deletions(-)

diff --git a/calibration_tools/base_handeye_collect.py b/calibration_tools/base_handeye_collect.py
index 8135022aa..57b97349c 100644
--- a/calibration_tools/base_handeye_collect.py
+++ b/calibration_tools/base_handeye_collect.py
@@ -40,7 +40,7 @@
     check_opencv,
     check_state,
     delta,
-    rigid_transform,
+    sample_poses,
     transform,
     write_json,
 )
@@ -143,35 +143,18 @@ def _resume(self) -> None:
                 record = json.loads(
                     (folder / "sample.json").read_text(encoding="utf-8")
                 )
-                if (
-                    type(record["index"]) is not int
-                    or record["index"] != index
-                    or record["arm"] != self.config.arm
-                    or record["calibration_mode"] != "eye_to_hand"
-                    or record["camera"]["serial"] != self.config.camera_serial
-                    or record["board"] != self.config.board_metadata
-                ):
-                    raise ValueError(f"Sample configuration does not match: {folder}")
+                if type(record["index"]) is not int or record["index"] != index:
+                    raise ValueError(f"Sample index does not match: {folder}")
                 if reference is None:
                     reference = record
-                check_camera(record["camera"], reference["camera"])
-                for key in ("robot_before", "robot_after"):
-                    check_state(record[key])
-                    if not np.allclose(
-                        record[key]["F_T_EE"],
-                        reference["robot_before"]["F_T_EE"],
-                        rtol=0,
-                        atol=1e-8,
-                    ):
-                        raise ValueError(f"End-effector frame changed: {folder}")
-                pose = rigid_transform(record[self.pose_key], self.pose_key)
-                if not np.allclose(
-                    pose, transform(record["robot_before"]), rtol=0, atol=1e-8
-                ):
-                    raise ValueError(
-                        f"Stored robot pose disagrees with state: {folder}"
-                    )
-                rigid_transform(record["T_camera_board"], "T_camera_board")
+                sample_poses(
+                    record,
+                    reference,
+                    arm=self.config.arm,
+                    serial=self.config.camera_serial,
+                    board=self.config.board_metadata,
+                    source=folder,
+                )
                 for name in ("color.png", "annotated.png"):
                     if not (folder / name).is_file():
                         raise ValueError(f"Missing sample image: {folder / name}")
diff --git a/calibration_tools/common.py b/calibration_tools/common.py
index dc8f2a8da..1efe5b093 100644
--- a/calibration_tools/common.py
+++ b/calibration_tools/common.py
@@ -161,6 +161,43 @@ def check_camera(camera: Record, reference: Record) -> None:
         raise ValueError("Camera K must be 3 x 3 with positive focal lengths")
 
 
+def sample_poses(
+    sample: Record,
+    reference: Record,
+    *,
+    arm: str,
+    serial: str,
+    board: Record,
+    source: Path,
+) -> tuple[Array, Array]:
+    """Validate saved sample identity and frames, returning robot and board poses.
+
+    Collection review and solving apply their own sample-count, image-quality,
+    and acquisition-drift requirements after this shared consistency check.
+    """
+    if (
+        sample.get("calibration_mode") != "eye_to_hand"
+        or sample.get("arm") != arm
+        or sample["camera"]["serial"] != serial
+        or sample["board"] != board
+    ):
+        raise ValueError(f"{source}: sample configuration does not match")
+    check_camera(sample["camera"], reference["camera"])
+    for key in ("robot_before", "robot_after"):
+        check_state(sample[key])
+        if not np.allclose(
+            sample[key]["F_T_EE"],
+            reference["robot_before"]["F_T_EE"],
+            rtol=0,
+            atol=1e-8,
+        ):
+            raise ValueError(f"{source}: end-effector frame changed")
+    pose = rigid_transform(sample[f"T_{arm}_base_ee"], str(source))
+    if not np.allclose(pose, transform(sample["robot_before"]), rtol=0, atol=1e-8):
+        raise ValueError(f"{source}: stored pose disagrees with robot_before")
+    return pose, rigid_transform(sample["T_camera_board"], str(source))
+
+
 def load_candidate(path: Path) -> Record:
     """Load a candidate only when its report and original samples still agree."""
     candidate = json.loads(path.read_text(encoding="utf-8"))
diff --git a/calibration_tools/solve_base_handeye.py b/calibration_tools/solve_base_handeye.py
index 9619ede42..b4e092bbe 100644
--- a/calibration_tools/solve_base_handeye.py
+++ b/calibration_tools/solve_base_handeye.py
@@ -27,13 +27,12 @@
     METHODS,
     Array,
     Record,
-    check_camera,
     check_opencv,
-    check_state,
     delta,
     errors,
     mean_pose,
     rigid_transform,
+    sample_poses,
     stats,
     transform,
     write_json,
@@ -101,28 +100,14 @@ def load_samples(
     expected_serial = serial or reference["camera"]["serial"]
     robot_poses, board_poses = [], []
     for path, sample in zip(files, samples):
-        if sample.get("calibration_mode") != "eye_to_hand" or sample.get("arm") != arm:
-            raise ValueError(
-                f"{path}: arm or calibration mode does not match this solver"
-            )
-        if (
-            sample["camera"]["serial"] != expected_serial
-            or sample["board"] != reference["board"]
-        ):
-            raise ValueError(f"{path}: camera serial or board definition changed")
-        check_camera(sample["camera"], reference["camera"])
-        for key in ("robot_before", "robot_after"):
-            check_state(sample[key])
-            if not np.allclose(
-                sample[key]["F_T_EE"],
-                reference["robot_before"]["F_T_EE"],
-                rtol=0,
-                atol=1e-8,
-            ):
-                raise ValueError(f"{path}: end-effector frame changed")
-        pose = rigid_transform(sample[f"T_{arm}_base_ee"], str(path))
-        if not np.allclose(pose, transform(sample["robot_before"]), rtol=0, atol=1e-8):
-            raise ValueError(f"{path}: stored pose disagrees with robot_before")
+        pose, board_pose = sample_poses(
+            sample,
+            reference,
+            arm=arm,
+            serial=expected_serial,
+            board=reference["board"],
+            source=path,
+        )
         drift_m, drift_rad = delta(pose, transform(sample["robot_after"]))
         if drift_m > 0.001 or drift_rad > 0.005:
             raise ValueError(f"{path}: robot moved during image acquisition")
@@ -141,7 +126,7 @@ def load_samples(
                 f"{path}: insufficient corners or excessive reprojection error"
             )
         robot_poses.append(pose)
-        board_poses.append(rigid_transform(sample["T_camera_board"], str(path)))
+        board_poses.append(board_pose)
     return files, samples, np.array(robot_poses), np.array(board_poses)
 
 
diff --git a/tests/unit_tests/calibration_tools/test_collector_session.py b/tests/unit_tests/calibration_tools/test_collector_session.py
index 9ea5ba867..4218e46d2 100644
--- a/tests/unit_tests/calibration_tools/test_collector_session.py
+++ b/tests/unit_tests/calibration_tools/test_collector_session.py
@@ -143,6 +143,18 @@ def test_reviewed_samples_require_a_new_calibration_result(collector_module, ses
         state["O_T_EE"][12] = index * 0.02
         collector.sample()
     root = collector.root
+    solver = importlib.import_module("solve_base_handeye")
+    assert len(solver.load_samples(root, "left")[1]) == 10
+    sample_path = root / "sample_005" / "sample.json"
+    original = sample_path.read_text()
+    changed = json.loads(original)
+    changed["T_left_base_ee"][0][3] += 0.1
+    sample_path.write_text(json.dumps(changed))
+    with pytest.raises(ValueError, match="stored pose disagrees"):
+        solver.load_samples(root, "left")
+    with pytest.raises(ValueError, match="stored pose disagrees"):
+        collector_module.Collector(collector.config, root, resume=True)
+    sample_path.write_text(original)
     pose = collector_module.np.eye(4).tolist()
     candidate = {
         "status": "candidate_requires_independent_validation",

From 0fc3dd992a217671a661ad63c1f086ad4856b13e Mon Sep 17 00:00:00 2001
From: Codex <codex@openai.com>
Date: Sun, 11 Oct 2026 17:34:07 +0800
Subject: [PATCH 7/7] test: narrow offline regressions to essential workflows

---
 .../test_collector_session.py                 | 71 ++++---------------
 1 file changed, 12 insertions(+), 59 deletions(-)

diff --git a/tests/unit_tests/calibration_tools/test_collector_session.py b/tests/unit_tests/calibration_tools/test_collector_session.py
index 4218e46d2..73ef61e8b 100644
--- a/tests/unit_tests/calibration_tools/test_collector_session.py
+++ b/tests/unit_tests/calibration_tools/test_collector_session.py
@@ -17,7 +17,6 @@
 
 import base64
 import copy
-import hashlib
 import importlib
 import io
 import json
@@ -27,7 +26,7 @@
 
 
 @pytest.fixture
-def collector_module(monkeypatch):
+def session(monkeypatch, tmp_path):
     cv2 = pytest.importorskip("cv2")
     if not hasattr(cv2, "aruco") or not hasattr(cv2.aruco, "CharucoDetector"):
         pytest.skip("Calibration requires OpenCV with the ChArUco detector")
@@ -35,11 +34,7 @@ def collector_module(monkeypatch):
     monkeypatch.syspath_prepend(
         str(Path(__file__).resolve().parents[3] / "calibration_tools")
     )
-    return importlib.import_module("base_handeye_collect")
-
-
-@pytest.fixture
-def session(collector_module, tmp_path, monkeypatch):
+    collector_module = importlib.import_module("base_handeye_collect")
     config = collector_module.CaptureConfig(
         arm="left",
         camera_serial="offline-camera",
@@ -49,11 +44,11 @@ def session(collector_module, tmp_path, monkeypatch):
     collector = collector_module.Collector(config, tmp_path / "session")
     state = {
         "O_T_EE": collector_module.np.eye(4).flatten(order="F").tolist(),
-        "F_T_EE": collector_module.np.eye(4).flatten(order="F").tolist(),
         "dq": [0.0] * 7,
         "robot_mode": 1,
         "has_errors": False,
     }
+    state["F_T_EE"] = state["O_T_EE"].copy()
     image = collector.board.generateImage((680, 880), marginSize=40)
     success, encoded = collector_module.cv2.imencode(".png", image)
     assert success
@@ -79,7 +74,7 @@ def camera_response(*args, **kwargs):
     )
     monkeypatch.setattr(collector_module.time, "sleep", lambda _: None)
     monkeypatch.setattr(collector_module.urllib.request, "urlopen", camera_response)
-    return collector, state, camera
+    return collector_module, collector, state
 
 
 def snapshot(folder):
@@ -92,24 +87,19 @@ def snapshot(folder):
 
 def request_handler(module, collector, path, method="POST"):
     """Exercise actual HTTP routing without opening a listening socket."""
-    handler_type = module.make_handler(collector)
-    handler = handler_type.__new__(handler_type)
+    handler = object.__new__(module.make_handler(collector))
     handler.path = path
     responses = []
     handler.reply = lambda code, body, kind="application/json": responses.append(
         (code, body, kind)
     )
-    handler.send_error = lambda code, *args: responses.append((code, b"", ""))
-    if method == "GET":
-        handler.do_GET()
-    else:
-        handler.do_POST()
+    getattr(handler, f"do_{method}")()
     assert len(responses) == 1
     return responses[0]
 
 
-def test_review_exclude_resume_and_continue_capture(collector_module, session):
-    collector, state, _ = session
+def test_review_exclude_resume_and_continue_capture(session):
+    collector_module, collector, state = session
     code, body, _ = request_handler(collector_module, collector, "/sample")
     assert code == 200 and json.loads(body)["accepted"]
     original = snapshot(collector.root / "sample_001")
@@ -134,55 +124,18 @@ def test_review_exclude_resume_and_continue_capture(collector_module, session):
     assert Path(result["saved"]).name == "sample_002"
     final = collector_module.Collector(collector.config, collector.root, resume=True)
     assert [sample["index"] for sample in final.status()["samples"]] == [2]
-    assert snapshot(archived) == original
-
-
-def test_reviewed_samples_require_a_new_calibration_result(collector_module, session):
-    collector, state, _ = session
-    for index in range(10):
-        state["O_T_EE"][12] = index * 0.02
+    collector = final
+    for index in range(1, 10):
+        state["O_T_EE"][12] = 0.1 + index * 0.02
         collector.sample()
     root = collector.root
     solver = importlib.import_module("solve_base_handeye")
     assert len(solver.load_samples(root, "left")[1]) == 10
     sample_path = root / "sample_005" / "sample.json"
-    original = sample_path.read_text()
-    changed = json.loads(original)
+    changed = json.loads(sample_path.read_text())
     changed["T_left_base_ee"][0][3] += 0.1
     sample_path.write_text(json.dumps(changed))
     with pytest.raises(ValueError, match="stored pose disagrees"):
         solver.load_samples(root, "left")
     with pytest.raises(ValueError, match="stored pose disagrees"):
         collector_module.Collector(collector.config, root, resume=True)
-    sample_path.write_text(original)
-    pose = collector_module.np.eye(4).tolist()
-    candidate = {
-        "status": "candidate_requires_independent_validation",
-        "arm": "left",
-        "calibration_mode": "eye_to_hand",
-        "sample_count": 10,
-        "selected_method": "PARK",
-        "camera_serial": collector.config.camera_serial,
-        "T_left_base_camera": pose,
-        "T_ee_board": pose,
-    }
-    report = {
-        **candidate,
-        "serial": collector.config.camera_serial,
-        "source_hashes": {
-            str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
-            for path in root.glob("sample_*/sample.json")
-        },
-        "methods": {"PARK": {"T_left_base_camera": pose, "T_ee_board": pose}},
-    }
-    candidate_path = root / "candidate.json"
-    candidate_path.write_text(json.dumps(candidate))
-    (root / "quality_report.json").write_text(json.dumps(report))
-    common = importlib.import_module("common")
-    assert common.load_candidate(candidate_path) == candidate
-
-    collector.delete_sample(5)
-
-    with pytest.raises(ValueError, match="Training samples changed"):
-        common.load_candidate(candidate_path)
-    assert json.loads(candidate_path.read_text()) == candidate