feat: keep RGB-D captures only on workstation

Data_Get/README.md

@@ -4,5 +4,6 @@
 `manifest.json`、`READY`。数据来自机器人 Nvidia 上独立的项目录制服务，不影响厂家
 `record_bag_node`。运行数据、临时同步目录及 `sync_status.json` 均不会提交到 Git。
 
-只有通过大小和 SHA-256 校验的目录才会从 `.incoming` 原子移动到这里。机器人端保留
+只有通过大小和 SHA-256 校验、fsync 和原子发布的目录才会从 `.incoming` 进入这里，
-ready 副本，传输中断后会自动续传，不要把 `.incoming` 当作完整数据集。
+并写入持久化 `VERIFIED` 凭据。之后才按 episode ID 与 manifest SHA 精确删除机器人端
+对应 `ready` 副本。传输或删除确认中断会自动重试，不要把 `.incoming` 当作完整数据集。

README.md

@@ -12,7 +12,7 @@
 - `tg3_local_teleop/`：部署到机器人 Nvidia；直接接收 OmniSocket 数据，发布双臂、
   双手及 `/hric/robot/cmd_vel`，提供限速回 Home，并独立录制控制数据 MCAP。
 - `tg3_data_collection/`：部署在本机；通过本地 SSH/rsync 拉取已完整收尾并校验的
-  episode，原子保存到 `Data_Get/`。不会删除机器人上的备份。
+  episode，原子保存到 `Data_Get/`；持久化验真凭据后自动回收机器人上的对应副本。
 - `Data_Get/`：最终数采目录。实际 MCAP、清单和同步状态默认不提交 Git。
 - `docs/`：可提交的跨机器人迁移步骤。含实测帧和现场拓扑的汇报/证据文档只保留在
   当前本地工作副本，不同步到匿名可读的远端仓库。
@@ -54,8 +54,9 @@ TOML 解析检查。若还要核对外置 OmniSocketGo 版本，可设置任务
 
 Nvidia 独立服务录制明确白名单内的机器人实测状态、双臂/双手命令与反馈、行走、
 IMU、电源状态、xTELE 应用帧以及头/腰 Orbbec 的 RGB-D。RGB 使用标准 JPEG 压缩流，
-16 位深度使用无损 compressedDepth PNG，并同时保存相机内参与静态 TF；腰部相机是
+16 位深度使用无损 compressedDepth PNG，并同时保存相机内参与静态 TF。头部和腰部
-必采，头部相机未启动时不阻塞，之后启动即可被同一录制自动发现。它不接管或停止厂家
+相机都是可选流：服务运行就自动采集，未运行不阻塞核心 episode；manifest 会分别标记
+`absent/healthy/partial/low_rate`。它不接管或停止厂家
 `/record_bag_node`，也不录点云。正常结束后先校验 MCAP、生成 SHA-256
 `manifest.json` 和 `READY`，再由
 本机服务拉取到：
@@ -64,8 +65,11 @@ IMU、电源状态、xTELE 应用帧以及头/腰 Orbbec 的 RGB-D。RGB 使用
 /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get/<episode_id>/
 ```
 
-传输中断只会留在 `Data_Get/.incoming/`，不会显示为完成 episode；机器人端 ready
+传输中断只会留在 `Data_Get/.incoming/`，不会显示为完成 episode。机器人在录制期间
-副本不会自动删除。部署、状态检查和恢复步骤见 `tg3_data_collection/README.md`。
+必须临时写 `active/ready`；只有本机完成全文件 SHA-256、fsync、原子发布并写入
+`VERIFIED` 后，才会自动删除机器人上同 ID、同 manifest 哈希的 `ready` 目录。EAI
+从不保存 MCAP 或图像 payload。部署、状态检查和恢复步骤见
+`tg3_data_collection/README.md`。
 
 ## 迁移前必须修改
 

docs/天工3.0本地同构臂遥操迁移部署指南.md

@@ -1031,6 +1031,7 @@ tg3_local_teleop.py
 data_collection.py
 data_recorder_protocol.py
 data_recorder_node.py
+delete_ready_episode.py
 config.toml
 run.sh
 run_data_recorder.sh
@@ -1061,11 +1062,13 @@ rosbag 加载 `RobotState/ArmCtrl` 时出现 undefined symbol。`run.sh` 仍需
 迁移时在 `config.toml [data_collection]` 核对 `base_directory`、100 GiB 余量、30 分钟
 上限、明确 topic 白名单和 required topics。不要改成 `ros2 bag record -a`，点云需另行
 估算带宽和磁盘后再加入。当前白名单已经包含头/腰 Orbbec 的 JPEG RGB、无损 PNG 深度、
-CameraInfo、Metadata 和 `/tf_static`；腰部 RGB-D/Metadata/静态 TF 为 required，两路
+CameraInfo、Metadata 和 `/tf_static`。头、腰相机都是 optional：服务已启动时 rosbag 自动
-图像收尾时还要求各自平均频率至少 `20 Hz`。头部未启动时是 optional，以后启动
+发现并采集；未启动时不影响核心 episode 开始或进入 READY。收尾后 manifest 会分别把
-`orbbec_head.service` 即可由正在录制的 rosbag 自动发现。
+两台相机标成 `absent`、`healthy`、`partial` 或 `low_rate`，后两种附带质量警告但不阻塞
+核心数据保存。
 
-新机器人先确认相机服务和发布者。相机必须在按 L3 开始 episode 之前启动：
+需要采集某台相机的完整 episode 时，先确认相机服务和发布者；不需要相机时可保持服务
+停止，L3 数采仍可正常启动：
 
 ```bash
 systemctl is-active orbbec_waist.service orbbec_head.service
@@ -1077,7 +1080,8 @@ ros2 topic info /ob_camera_waist/color/image_raw/compressed
 ros2 topic info /ob_camera_waist/depth/image_raw/compressedDepth
 ```
 
-两条腰部话题都必须显示 `Publisher count: 1`。原始 1280x720 RGB+深度约
+希望采集腰部图像时，两条腰部话题应显示 `Publisher count: 1`；头部同理。原始
+1280x720 RGB+深度约
 `8.3 GB/min/相机`，所以默认不重复录 raw，而是录标准 image_transport 压缩流；现场
 单相机约 `1.3–1.5 GB/min`，实际随画面变化。头部和腰部都运行时应按约两倍预留空间与
 同步时间。`minimum_free_gib=100` 是停止下限，不是整条 episode 的容量预算；双相机
@@ -1113,12 +1117,17 @@ systemctl --user restart tg3-data-get-sync.service
 2. 左摇杆按下（L3）连续 1 秒开始数采；该键是 `button_joystick.left`，不是 X/Y/Z；
 3. 松开 L3 至少 0.5 秒，再长按 1 秒结束；或正常 Z+C 结束遥操自动收尾；
 4. 查看 Nvidia `/tg3/data_collection/status`，必须先到 `ready`，不能从 `active/` 取数据；
-   recorder 会同时要求 required topics 有实时发布者且最终消息计数大于零；
+   recorder 会要求核心 required topics 有实时发布者且最终消息计数大于零；相机质量看
+   manifest 的 `optional_topic_groups`，相机未启动显示 `absent`，不会阻塞；
 5. 查看本机 `Data_Get/sync_status.json`，最终目录必须有 `READY`、`manifest.json`、
-   `bag/metadata.yaml` 和非空 `*.mcap`；
+   `VERIFIED`、`bag/metadata.yaml` 和非空 `*.mcap`；
 6. 执行 `ros2 bag info <episode>/bag` 并确认所需 topic 有消息；
-7. 断开 PS 网络再采一条，确认 Nvidia 保留数据；恢复网络后应续传且 SHA-256 通过。
+7. 断开 PS 网络再采一条，确认 Nvidia 暂存 `ready/`；恢复网络后应续传，完成 SHA-256、
+   fsync 和 `VERIFIED` 后，只删除机器人上 manifest 完全匹配的该 episode。
 
 每个新会话先要求 L3 稳定松开 0.5 秒，同一次持续按压只切换一次。数采故障不得解除
-遥操或延迟 STOP/Home；桥心跳中断超过 3 秒时 recorder 会自行 SIGINT 收尾。机器人
+遥操或延迟 STOP/Home；桥心跳中断超过 3 秒时 recorder 会自行 SIGINT 收尾。当前架构在
-`ready/` 不自动删除，确认本机和外部备份后才按具体 episode 清理。
+录制期间必须先写 Nvidia 的项目专用 `active/`，PS 离线时也会暂存在 `ready/`；它不是
+长期副本。本机完成全文件验真并持久化 `VERIFIED` 后，固定根目录删除助手才会原子移走并
+删除该 exact episode，不能触碰 `active/`、`failed/` 或根目录。`retain_failed_episodes=false`
+时失败录包也不在机器人保留。EAI 只转发控制数据，不保存 MCAP 或相机 payload。

tg3_data_collection/README.md

@@ -2,8 +2,10 @@
 
 机器人只在 `/home/nvidia/tg3_data_collection/ready` 暴露已经收到 SIGINT、写完
 `metadata.yaml`、通过 `ros2 bag info` 并生成 SHA-256 清单的 episode。本机服务用免密 SSH/rsync 复制到
-`Data_Get/.incoming`，逐文件校验后再原子改名为 `Data_Get/<episode_id>`。中断的复制不会
+`Data_Get/.incoming`，逐文件校验并 fsync 后再原子改名为 `Data_Get/<episode_id>`。本机
-显示成完成数据，也不会删除机器人上的备份。
+`VERIFIED` 收据持久化成功后，正式服务会再次完整哈希本机 payload，再让 Nvidia 侧固定根目录
+helper 比对原始 manifest SHA-256，原子移入 `deleting/` 后清理。因此完整数据最终只保留在 PS 本机，Nvidia 只是临时
+staging；中断、哈希错误或本机落盘失败都不会触发远端删除。
 
 数据只走本机到 Nvidia 的 SSH 链路，与公网或 EAI 本地 OmniSocket Hub 无关。当前默认：
 
@@ -12,6 +14,8 @@ nvidia@192.168.41.2:/home/nvidia/tg3_data_collection/ready/
   -> /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get/
 ```
 
+EAI 不运行 recorder/sync，也不保存 MCAP 或图像数据。
+
 先确认免密与依赖：
 
 ```bash
@@ -20,6 +24,18 @@ command -v rsync
 ssh nvidia@192.168.41.2 command -v rsync
 ```
 
+先把固定根目录删除 helper 部署到 Nvidia。它的 CLI 只接受安全 episode ID 和 64 位
+manifest SHA-256，不接受 root 参数，也不会枚举或触碰 `active/`、`failed/`。helper 在
+`deleting/` 树之外的 `delete_ledger/` 先持久化独立删除授权；即使递归删除中断且 tombstone
+里的 READY/manifest 已消失，仍能凭该授权安全续删，最后才删除授权文件：
+
+```bash
+scp tg3_local_teleop/delete_ready_episode.py \
+  nvidia@192.168.41.2:/home/nvidia/tg3_local_teleop/
+ssh nvidia@192.168.41.2 \
+  chmod 755 /home/nvidia/tg3_local_teleop/delete_ready_episode.py
+```
+
 安装本机用户服务：
 
 ```bash
@@ -39,10 +55,20 @@ python3 -m json.tool Data_Get/sync_status.json
 也可以只执行一次，便于首次部署验收：
 
 ```bash
-python3 tg3_data_collection/data_get_sync.py --once
+systemctl --user stop tg3-data-get-sync.service
+python3 tg3_data_collection/data_get_sync.py --once --delete-remote-after-sync
+systemctl --user start tg3-data-get-sync.service
 ```
 
+同步器使用 `Data_Get/.data_get_sync.lock` 做 flock；正式服务运行时，另一个 `--once` 会明确
+报“已有同步进程”，不会与服务同时复制或删除。不带 `--delete-remote-after-sync` 的手动命令
+只复制、验真并保留远端，正式 unit 已显式启用自动回收。
+
 日常轮询不会每 2 秒重新读取并哈希全部历史 MCAP；首次原子发布时已经完成深度校验。
+如果进程在远端删除 ACK 返回前退出，`VERIFIED` 会保持 `pending`。每次真正调用 helper
+之前都会重新完整哈希本机 payload；失败后从 30 秒开始指数退避，最大 15 分钟，进程重启也
+遵守持久化的 `next_retry_unix_s`，避免每 2 秒重读大型 MCAP。helper 使用独立 600 秒超时，
+能继续清理同摘要 tombstone/ledger，也会把已经不存在的同一目标视为成功。
 需要定期审计已有本地数据时单独执行（可能耗时较长）：
 
 ```bash
@@ -56,6 +82,7 @@ python3 tg3_data_collection/data_get_sync.py --once --verify-existing
 ```text
 <episode_id>/
   READY
+  VERIFIED
   manifest.json
   bag/metadata.yaml
   bag/*.mcap
@@ -64,10 +91,32 @@ python3 tg3_data_collection/data_get_sync.py --once --verify-existing
 ```
 
 同步端会再次核对 `READY`、episode ID、每个 MCAP/metadata 的长度和 SHA-256；远端
-manifest 在传输中发生变化也会拒绝发布最终目录。`.incoming`、`sync_status.json` 和全部
+manifest 在传输中发生变化也会拒绝发布最终目录。随后 fsync 所有复制文件和目录、以
-episode 已由 `Data_Get/.gitignore` 排除。服务绝不自动删除机器人端 `ready/` 数据；确认
+NOREPLACE 语义原子发布、fsync 父目录，最后写入并 fsync `VERIFIED`。收据绑定 episode、
-本机备份后如需清理，必须由操作者明确指定具体 episode，不能删除整个项目目录。
+远端身份、原始 manifest SHA-256、已校验文件数和字节数，并记录 `pending/deleted`。
+`.incoming`、`sync_status.json` 和全部 episode 已由 `Data_Get/.gitignore` 排除。
 
-迁移后需要同步修改 service 中的本机项目路径、机器人 SSH 地址和机器人 ready 路径。
+临时目录按 `<episode>.<manifest_sha256>.partial` 隔离，rsync 同时使用删除同步；发布前还会
-如果 Nvidia 的 `192.168.41.2` 改变，只改该 unit 的 `--remote`；它不在机器人
+拒绝 manifest 未列出的任何 `*.mcap` 或 `metadata.yaml`，避免旧中断文件混入新 episode。
-`config.toml` 或 OmniSocket Peer 设置中。
+
+`sync_status.json` 的 `state=delete_pending` 表示本机数据已完整发布，但远端回收尚未收到
+成功 ACK；`pending_remote_cleanup`、`last_delete_error`、`delete_count` 和
+`last_deleted_episode` 可用于排查。此状态下不要手工删除本机数据，服务会在下一轮重试。
+
+已有本机同名目录不会直接授权删除：没有可信 `VERIFIED` 时必须重新做完整哈希，并要求
+本机与当前远端 manifest 原始摘要一致，之后才补收据并回收远端。同名碰撞、损坏、符号
+链接、错误 root 或摘要不符都保留远端并报错。
+
+历史 `retained/deleted` 收据可以来自旧机器人 IP，不会阻塞新地址的同步；仍为 `pending` 的
+跨 IP 收据不会静默授权新机器人删除，只会逐项报错且不影响其他新 episode 继续同步。
+
+自动回收 helper 严格只覆盖 `ready/`，录制中的 `active/` 永远不能由 PS 同步器处理，
+也不能对 `failed/` 使用 READY 凭据。当前正式配置为
+`retain_failed_episodes=false`：录制器只把失败原因写入状态和 journal，随后精确删除自己
+刚创建的失败 payload，因此机器人不会长期保留失败 MCAP。若以后改回 `true` 以保留故障
+诊断，必须另建独立验真与回收协议，不能整目录清空。
+
+迁移后需要同步修改 service 中的本机项目路径和机器人 SSH 地址。如果 Nvidia 的
+`192.168.41.2` 改变，只改 unit 的 `--remote` 并重新部署 helper；它不在机器人
+`config.toml` 或 OmniSocket Peer 设置中。启用自动删除时远端路径必须保持固定的
+`/home/nvidia/tg3_data_collection/ready`，同步器会拒绝对其他 root 启用删除。

tg3_data_collection/test_data_get_sync.py

@@ -3,12 +3,20 @@ from __future__ import annotations
 
 import hashlib
 import json
+import os
+import shutil
+import subprocess
 import tempfile
 import unittest
 from pathlib import Path
+from unittest import mock
 
 from data_get_sync import (
     DataGetSync,
+    FIXED_REMOTE_READY,
+    ManifestDocument,
+    VERIFIED_RECEIPT_NAME,
+    fsync_episode_tree,
     parse_manifest,
     safe_episode_name,
     safe_relative_path,
@@ -16,6 +24,99 @@ from data_get_sync import (
 )
 
 
+class LocalRsyncRunner:
+    def __init__(self, remote_root: Path) -> None:
+        self.remote_root = remote_root
+        self.mode = "normal"
+        self.rsync_calls = 0
+
+    def run(
+        self, command: list[str], *, timeout: float
+    ) -> subprocess.CompletedProcess[str]:
+        del timeout
+        if not command or command[0] != "rsync":
+            return subprocess.CompletedProcess(command, 1, "", "unexpected command")
+        self.rsync_calls += 1
+        remote_source = command[-2].rstrip("/")
+        episode = remote_source.rsplit("/", 1)[-1]
+        destination = Path(command[-1])
+        shutil.copytree(
+            self.remote_root / episode,
+            destination,
+            dirs_exist_ok=True,
+            symlinks=True,
+        )
+        if self.mode == "truncate":
+            (destination / "bag" / "bag_0.mcap").write_bytes(b"truncated")
+        elif self.mode == "manifest_changed":
+            manifest = destination / "manifest.json"
+            manifest.write_bytes(manifest.read_bytes() + b"\n")
+        elif self.mode == "manifest_symlink":
+            manifest = destination / "manifest.json"
+            manifest.unlink()
+            manifest.symlink_to("bag/metadata.yaml")
+        return subprocess.CompletedProcess(command, 0, "", "")
+
+
+class LocalSync(DataGetSync):
+    def __init__(
+        self,
+        *,
+        remote_root: Path,
+        destination: Path,
+        runner: LocalRsyncRunner,
+        delete: bool = True,
+        remote: str = "nvidia@test",
+    ) -> None:
+        super().__init__(
+            remote=remote,
+            remote_ready=FIXED_REMOTE_READY,
+            destination=destination,
+            status_file=destination / "sync_status.json",
+            runner=runner,
+            reserve_bytes=0,
+            delete_remote_after_sync=delete,
+            remote_delete_helper="/fixed/delete_ready_episode.py",
+        )
+        self.remote_root = remote_root
+        self.delete_calls: list[tuple[str, str]] = []
+        self.delete_outcomes: list[str] = []
+
+    def list_remote_episodes(self) -> list[str]:
+        return sorted(
+            path.name
+            for path in self.remote_root.iterdir()
+            if path.is_dir() and not path.is_symlink()
+        )
+
+    def get_remote_manifest(self, episode: str) -> ManifestDocument:
+        raw = (self.remote_root / episode / "manifest.json").read_bytes()
+        return ManifestDocument.from_bytes(raw, episode)
+
+    def _invoke_delete_helper(
+        self, episode: str, manifest_sha256: str
+    ) -> dict[str, object]:
+        self.delete_calls.append((episode, manifest_sha256))
+        outcome = self.delete_outcomes.pop(0) if self.delete_outcomes else "success"
+        target = self.remote_root / episode
+        if outcome == "failure":
+            raise RuntimeError("injected delete failure")
+        if outcome == "ack_lost_after_delete":
+            if target.exists():
+                shutil.rmtree(target)
+            raise RuntimeError("injected lost acknowledgement")
+        if target.exists():
+            shutil.rmtree(target)
+            state = "deleted"
+        else:
+            state = "already_absent"
+        return {
+            "state": state,
+            "episode_id": episode,
+            "manifest_sha256": manifest_sha256,
+        }
+
+
 class DataGetSyncTests(unittest.TestCase):
     def test_safe_episode_name(self) -> None:
         self.assertTrue(safe_episode_name("episode_20260810T120000000_deadbeef"))
@@ -93,28 +194,377 @@ class DataGetSyncTests(unittest.TestCase):
             with self.assertRaises(ValueError):
                 validate_episode_dir(episode, manifest)
 
-    def test_existing_episode_skips_expensive_hash_unless_requested(self) -> None:
+    def test_fsync_episode_tree_rejects_intermediate_symlink(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:
             root = Path(temporary)
-            name = "episode_20260810T120000000_deadbeef"
+            episode = root / "episode"
-            episode, _manifest = self._episode(root, name)
+            episode.mkdir()
-            (episode / "bag" / "bag_0.mcap").write_bytes(b"changed")
+            outside = root / "outside"
-            fast = DataGetSync(
+            outside.mkdir()
-                remote="unused",
+            (outside / "file").write_bytes(b"sentinel")
-                remote_ready="/unused",
+            (episode / "linked").symlink_to(outside, target_is_directory=True)
-                destination=root,
-                status_file=root / "status.json",
-            )
-            self.assertFalse(fast.sync_episode(name))
-            deep = DataGetSync(
-                remote="unused",
-                remote_ready="/unused",
-                destination=root,
-                status_file=root / "status.json",
-                verify_existing=True,
-            )
             with self.assertRaises(ValueError):
-                deep.sync_episode(name)
+                fsync_episode_tree(episode)
+
+    def _local_sync(
+        self, root: Path, name: str = "episode_20260810T120000000_deadbeef"
+    ) -> tuple[LocalSync, LocalRsyncRunner, Path, Path]:
+        remote = root / "remote_ready"
+        destination = root / "Data_Get"
+        remote.mkdir()
+        destination.mkdir()
+        self._episode(remote, name)
+        runner = LocalRsyncRunner(remote)
+        syncer = LocalSync(
+            remote_root=remote,
+            destination=destination,
+            runner=runner,
+        )
+        return syncer, runner, remote, destination
+
+    def test_copy_is_verified_receipted_then_remote_is_deleted(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            syncer, _runner, remote, destination = self._local_sync(
+                Path(temporary), name
+            )
+            self.assertEqual(syncer.run_once(), 1)
+            final = destination / name
+            receipt = json.loads((final / VERIFIED_RECEIPT_NAME).read_text())
+            self.assertEqual(receipt["state"], "VERIFIED")
+            self.assertEqual(receipt["remote_identity"], syncer.remote_identity)
+            self.assertEqual(receipt["verified_files"], 2)
+            self.assertGreater(receipt["verified_bytes"], 0)
+            self.assertEqual(receipt["remote_delete"]["state"], "deleted")
+            self.assertFalse((remote / name).exists())
+            self.assertEqual(syncer.delete_count, 1)
+            status = json.loads((destination / "sync_status.json").read_text())
+            self.assertEqual(status["state"], "idle")
+            self.assertEqual(status["pending_remote_cleanup"], 0)
+
+    def test_truncated_copy_never_calls_delete(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            syncer, runner, remote, _destination = self._local_sync(Path(temporary))
+            runner.mode = "truncate"
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertEqual(len(list(remote.iterdir())), 1)
+
+    def test_manifest_change_during_copy_never_calls_delete(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            syncer, runner, remote, _destination = self._local_sync(Path(temporary))
+            runner.mode = "manifest_changed"
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertEqual(len(list(remote.iterdir())), 1)
+
+    def test_copied_manifest_symlink_never_calls_delete(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            syncer, runner, remote, _destination = self._local_sync(Path(temporary))
+            runner.mode = "manifest_symlink"
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertEqual(len(list(remote.iterdir())), 1)
+
+    def test_delete_failure_is_pending_and_retried(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            syncer, _runner, remote, destination = self._local_sync(
+                Path(temporary), name
+            )
+            syncer.delete_outcomes = ["failure", "success"]
+            self.assertEqual(syncer.run_once(), 1)
+            receipt_path = destination / name / VERIFIED_RECEIPT_NAME
+            receipt = json.loads(receipt_path.read_text())
+            self.assertEqual(receipt["remote_delete"]["state"], "pending")
+            self.assertTrue((remote / name).exists())
+            status = json.loads((destination / "sync_status.json").read_text())
+            self.assertEqual(status["state"], "delete_pending")
+            self.assertEqual(status["pending_remote_cleanup"], 1)
+
+            # Persistent backoff prevents a 2-second poll loop from repeatedly
+            # hashing a large MCAP after a network/helper failure.
+            self.assertEqual(syncer.run_once(), 0)
+            self.assertEqual(len(syncer.delete_calls), 1)
+            receipt = json.loads(receipt_path.read_text())
+            receipt["remote_delete"]["next_retry_unix_s"] = 0.0
+            receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
+
+            self.assertEqual(syncer.run_once(), 0)
+            receipt = json.loads(receipt_path.read_text())
+            self.assertEqual(receipt["remote_delete"]["state"], "deleted")
+            self.assertFalse((remote / name).exists())
+            self.assertEqual(len(syncer.delete_calls), 2)
+
+    def test_lost_delete_ack_is_idempotent_after_process_restart(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            root = Path(temporary)
+            first, runner, remote, destination = self._local_sync(root, name)
+            first.delete_outcomes = ["ack_lost_after_delete"]
+            self.assertEqual(first.run_once(), 1)
+            self.assertFalse((remote / name).exists())
+            receipt_path = destination / name / VERIFIED_RECEIPT_NAME
+            self.assertEqual(
+                json.loads(receipt_path.read_text())["remote_delete"]["state"],
+                "pending",
+            )
+
+            restarted = LocalSync(
+                remote_root=remote,
+                destination=destination,
+                runner=runner,
+            )
+            with mock.patch(
+                "data_get_sync.validate_episode_dir",
+                wraps=validate_episode_dir,
+            ) as deep_validator:
+                self.assertEqual(restarted.run_once(), 0)
+                self.assertEqual(deep_validator.call_count, 0)
+                self.assertEqual(restarted.delete_calls, [])
+
+                receipt = json.loads(receipt_path.read_text())
+                receipt["remote_delete"]["next_retry_unix_s"] = 0.0
+                receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
+                self.assertEqual(restarted.run_once(), 0)
+                self.assertGreaterEqual(deep_validator.call_count, 1)
+            self.assertEqual(
+                json.loads(receipt_path.read_text())["remote_delete"]["state"],
+                "deleted",
+            )
+            self.assertEqual(restarted.delete_calls[0][0], name)
+
+    def test_retry_rehashes_and_refuses_delete_after_local_tamper(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            syncer, _runner, remote, destination = self._local_sync(
+                Path(temporary), name
+            )
+            syncer.delete_outcomes = ["failure", "success"]
+            self.assertEqual(syncer.run_once(), 1)
+            (destination / name / "bag" / "bag_0.mcap").write_bytes(b"tampered")
+            receipt_path = destination / name / VERIFIED_RECEIPT_NAME
+            receipt = json.loads(receipt_path.read_text())
+            receipt["remote_delete"]["next_retry_unix_s"] = 0.0
+            receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
+
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertEqual(len(syncer.delete_calls), 1)
+            self.assertTrue((remote / name).exists())
+
+    def test_unlisted_stale_mcap_is_never_published_or_deleted(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            syncer, _runner, remote, destination = self._local_sync(
+                Path(temporary), name
+            )
+            (remote / name / "bag" / "stale_old_payload.mcap").write_bytes(
+                b"stale"
+            )
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertFalse((destination / name).exists())
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertTrue((remote / name).exists())
+
+    def test_reused_hash_staging_cannot_smuggle_old_mcap(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            syncer, _runner, remote, destination = self._local_sync(
+                Path(temporary), name
+            )
+            document = syncer.get_remote_manifest(name)
+            staging = (
+                destination
+                / ".incoming"
+                / f"{name}.{document.sha256}.partial"
+                / "bag"
+            )
+            staging.mkdir(parents=True)
+            (staging / "stale_old_payload.mcap").write_bytes(b"old interrupted bag")
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertFalse((destination / name).exists())
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertTrue((remote / name).exists())
+
+    def test_deleted_receipt_from_old_ip_does_not_block_new_ip(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            root = Path(temporary)
+            first, runner, remote, destination = self._local_sync(root, name)
+            self.assertEqual(first.run_once(), 1)
+            receipt_path = destination / name / VERIFIED_RECEIPT_NAME
+            receipt = json.loads(receipt_path.read_text())
+            receipt["remote"] = "nvidia@old-ip"
+            receipt["remote_identity"] = (
+                f"nvidia@old-ip:{FIXED_REMOTE_READY}"
+            )
+            receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
+
+            migrated = LocalSync(
+                remote_root=remote,
+                destination=destination,
+                runner=runner,
+                remote="nvidia@new-ip",
+            )
+            self.assertEqual(migrated.run_once(), 0)
+            status = json.loads((destination / "sync_status.json").read_text())
+            self.assertEqual(status["state"], "idle")
+            self.assertEqual(status["pending_remote_cleanup"], 0)
+
+    def test_bad_pending_old_ip_does_not_starve_new_episode(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            old_name = "episode_old"
+            first, _runner, _old_remote, destination = self._local_sync(
+                root, old_name
+            )
+            first.delete_outcomes = ["failure"]
+            self.assertEqual(first.run_once(), 1)
+
+            new_remote = root / "new_remote_ready"
+            new_remote.mkdir()
+            new_name = "episode_new"
+            self._episode(new_remote, new_name)
+            new_runner = LocalRsyncRunner(new_remote)
+            migrated = LocalSync(
+                remote_root=new_remote,
+                destination=destination,
+                runner=new_runner,
+                remote="nvidia@new-ip",
+            )
+            self.assertEqual(migrated.run_once(), 1)
+            self.assertFalse((new_remote / new_name).exists())
+            self.assertTrue((destination / new_name).is_dir())
+            status = json.loads((destination / "sync_status.json").read_text())
+            self.assertEqual(status["state"], "delete_pending")
+            self.assertEqual(status["pending_remote_cleanup"], 1)
+            self.assertIn("remote_identity", status["last_delete_error"])
+
+    def test_delete_helper_uses_independent_long_timeout(self) -> None:
+        class AckRunner:
+            def __init__(self) -> None:
+                self.timeouts: list[float] = []
+
+            def run(
+                self, command: list[str], *, timeout: float
+            ) -> subprocess.CompletedProcess[str]:
+                self.timeouts.append(timeout)
+                payload = {
+                    "state": "already_absent",
+                    "episode_id": "episode_timeout",
+                    "manifest_sha256": "0" * 64,
+                }
+                return subprocess.CompletedProcess(
+                    command, 0, json.dumps(payload) + "\n", ""
+                )
+
+        with tempfile.TemporaryDirectory() as temporary:
+            runner = AckRunner()
+            syncer = DataGetSync(
+                remote="nvidia@test",
+                remote_ready=FIXED_REMOTE_READY,
+                destination=Path(temporary),
+                status_file=Path(temporary) / "status.json",
+                runner=runner,
+                delete_remote_after_sync=True,
+                remote_delete_timeout_s=600.0,
+            )
+            syncer._invoke_delete_helper("episode_timeout", "0" * 64)
+            self.assertEqual(runner.timeouts, [600.0])
+
+    def test_existing_final_without_receipt_is_deep_verified_before_delete(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            root = Path(temporary)
+            syncer, _runner, remote, destination = self._local_sync(root, name)
+            shutil.copytree(remote / name, destination / name)
+            with mock.patch(
+                "data_get_sync.validate_episode_dir",
+                wraps=validate_episode_dir,
+            ) as deep_validator:
+                self.assertEqual(syncer.run_once(), 0)
+                self.assertGreaterEqual(deep_validator.call_count, 1)
+            self.assertTrue((destination / name / VERIFIED_RECEIPT_NAME).is_file())
+            self.assertFalse((remote / name).exists())
+
+    def test_existing_corrupt_final_without_receipt_never_deletes_remote(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            root = Path(temporary)
+            syncer, _runner, remote, destination = self._local_sync(root, name)
+            shutil.copytree(remote / name, destination / name)
+            (destination / name / "bag" / "bag_0.mcap").write_bytes(b"corrupt")
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertTrue((remote / name).exists())
+
+    def test_existing_manifest_collision_never_deletes_remote(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            name = "episode_20260810T120000000_deadbeef"
+            root = Path(temporary)
+            syncer, _runner, remote, destination = self._local_sync(root, name)
+            self._episode(destination, name)
+            local_manifest = destination / name / "manifest.json"
+            local_manifest.write_bytes(local_manifest.read_bytes() + b"\n")
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertTrue((remote / name).exists())
+
+    def test_receipt_write_failure_never_calls_delete(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            syncer, _runner, remote, _destination = self._local_sync(Path(temporary))
+            with mock.patch(
+                "data_get_sync.atomic_write_json",
+                side_effect=RuntimeError("injected receipt fsync failure"),
+            ):
+                with self.assertRaises(RuntimeError):
+                    syncer.sync_episode(next(remote.iterdir()).name)
+            self.assertEqual(syncer.delete_calls, [])
+            self.assertEqual(len(list(remote.iterdir())), 1)
+
+    def test_sync_flock_rejects_concurrent_once(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            first, runner, remote, destination = self._local_sync(root)
+            second = LocalSync(
+                remote_root=remote,
+                destination=destination,
+                runner=runner,
+            )
+            with first.process_lock():
+                with self.assertRaisesRegex(RuntimeError, "already holds"):
+                    second.run_once()
+
+    def test_symlink_lock_is_refused(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            syncer, _runner, _remote, destination = self._local_sync(root)
+            sentinel = root / "sentinel"
+            sentinel.write_text("do not touch", encoding="utf-8")
+            (destination / ".data_get_sync.lock").symlink_to(sentinel)
+            with self.assertRaises(RuntimeError):
+                syncer.run_once()
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "do not touch")
+
+    def test_delete_mode_rejects_nonfixed_remote_ready(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            with self.assertRaises(ValueError):
+                DataGetSync(
+                    remote="nvidia@test",
+                    remote_ready="/tmp/not-ready",
+                    destination=root,
+                    status_file=root / "status.json",
+                    delete_remote_after_sync=True,
+                )
 
     def test_bad_old_episode_does_not_starve_newer_episode(self) -> None:
         with tempfile.TemporaryDirectory() as temporary:

tg3_data_collection/tg3-data-get-sync.service

@@ -4,7 +4,7 @@ Description=TG3 completed data episode sync to Data_Get
 [Service]
 Type=simple
 WorkingDirectory=/home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop
-ExecStart=/usr/bin/python3 /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/tg3_data_collection/data_get_sync.py --remote nvidia@192.168.41.2 --remote-ready /home/nvidia/tg3_data_collection/ready --destination /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get --poll-seconds 2
+ExecStart=/usr/bin/python3 /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/tg3_data_collection/data_get_sync.py --remote nvidia@192.168.41.2 --remote-ready /home/nvidia/tg3_data_collection/ready --destination /home/ps/Desktop/TG3_TS1P_OmniSocket_Teleop/Data_Get --delete-remote-after-sync --poll-seconds 2
 Restart=always
 RestartSec=2
 KillSignal=SIGINT

tg3_local_teleop/README.md

@@ -167,21 +167,19 @@ OmniSocket 进程；在此之前桥会保持最后一帧目标。未 START 或
 为避免 1280x720 原始 RGB+深度每台相机约 `8.3 GB/min` 的持续写盘负载，默认录制
 `color/image_raw/compressed`（JPEG）和 `depth/image_raw/compressedDepth`（16UC1
 无损 PNG）。现场单台腰部相机实测合计约 `22–25 MB/s`（约 `1.3–1.5 GB/min`，随
-场景变化）。腰部图像、内参、Metadata 和 `/tf_static` 是 required topics；两路图像还
+场景变化）。头、腰两组相机都不是 required topics：服务在运行就由 rosbag 自动发现并
-必须在整条 bag 内分别达到至少 `20 Hz` 平均频率。相机服务未发布或图像严重掉流时，
+采集，未运行不阻塞 L3 或核心 READY。manifest 对每组记录 `absent`、`healthy`、
-L3 启动或最终校验会失败，不会生成缺图像却标记 READY 的 episode。头部话题在白名单
+`partial` 或 `low_rate`，并保存各话题消息数和图像平均频率；相机异常会产生明确质量
-中但不是 required：当前
+警告，但不会丢弃其余机器人控制数据。静态 TF 也尽力采集，不作为启动门槛。
-`orbbec_head.service` 未启动不影响腰部数采；将来在 episode 开始前启动后，rosbag 会
-自动发现并采集，无需再改配置。静态 TF 会被保存用于相机外参，但当前收尾只验证该
-Topic 非空，不解析每条 TF 的 frame 语义。
 
 机器人暂存目录：
 
 ```text
 /home/nvidia/tg3_data_collection/
-  active/   # 尚未完成，不能取走
+  active/    # 录制期间的临时数据，不能取走
-  ready/    # 已正常收尾、校验并生成 READY
+  ready/     # 等待 PS 本机完成持久化验真；随后自动回收
-  failed/   # 启动、磁盘、进程或校验失败，保留供诊断
+  failed/    # 当前策略不保留失败 payload
+  deleting/  # 固定根删除助手的短暂原子隔离目录
 ```
 
 MCAP 使用 `zstd_fast`、64 MiB cache、每 300 秒分片；每条 episode 最长 30 分钟，
@@ -191,6 +189,12 @@ MCAP 使用 `zstd_fast`、64 MiB cache、每 300 秒分片；每条 episode 最
 `manifest.json` 和 `READY`，最后才原子进入 `ready/`。Nvidia 需已安装 `python3-yaml`
 （现场已验证 PyYAML 6.0.1）。
 
+当前 `retain_failed_episodes=false`：失败原因仍进入 recorder 状态和 journal，但失败
+MCAP 不留在机器人磁盘。成功 episode 必须先在 PS 本机完成逐文件 SHA-256、fsync、
+原子发布并生成 `VERIFIED`，然后由固定根目录删除助手按 episode ID + manifest SHA
+精确回收；它不能访问 `active/`、`failed/` 或数据根目录。PS 断线时 `ready/` 会临时保留，
+避免尚未落到本机的数据丢失，恢复后自动续传和回收。
+
 相机由厂家系统服务管理，本项目只订阅、不自动启停。录制前检查或按需启动：
 
 ```bash
@@ -224,7 +228,8 @@ find /home/nvidia/tg3_data_collection/ready -mindepth 1 -maxdepth 1 -type d
 ```
 
 完成 episode 由 PS 本机服务校验后保存到项目 `Data_Get`，详见
-`../tg3_data_collection/README.md`。机器人 `ready/` 中的副本不会自动删除。
+`../tg3_data_collection/README.md`。本机 `VERIFIED` 持久化后，机器人对应 `ready/`
+副本会自动删除；EAI 不保存数采 payload。
 
 ## 完整重启顺序
 

tg3_local_teleop/config.toml

@@ -75,20 +75,17 @@ control_topic = "/tg3/data_collection/control"
 status_topic = "/tg3/data_collection/status"
 iarm_frame_topic = "/tg3/data_collection/iarm_frame"
 base_directory = "/home/nvidia/tg3_data_collection"
+# Failed/incomplete payloads are discarded after the error is reported through
+# recorder status and journal. Completed READY episodes remain only until the
+# PS sync has durably verified its local copy and requests exact-episode removal.
+retain_failed_episodes = false
 minimum_free_gib = 100.0
 max_duration_seconds = 1800.0
-# A camera endpoint merely existing is insufficient: completed episodes must
-# also sustain this conservative average rate across the rosbag data span.
-# The driver targets 30 Hz; 20 Hz tolerates startup/discovery jitter while
-# rejecting a stream that only delivered a few frames or stopped mid-episode.
-minimum_topic_rates_hz."/ob_camera_waist/color/image_raw/compressed" = 20.0
-minimum_topic_rates_hz."/ob_camera_waist/depth/image_raw/compressedDepth" = 20.0
 
 # Explicit data whitelist; never replace this with `-a` on the live robot.
 # RGB is recorded through image_transport JPEG and 16UC1 depth through its
-# lossless compressedDepth PNG transport. Waist RGB-D is required because that
+# lossless compressedDepth PNG transport. Both RGB-D cameras are optional:
-# camera is enabled on this robot. Head RGB-D stays optional: rosbag discovers
+# rosbag discovers and records either one whenever its service is running.
-# and records it automatically whenever orbbec_head.service is running.
 # /robot_state is the authoritative measured robot state.
 topics = [
   "/robot_state",
@@ -137,18 +134,43 @@ required_topics = [
   "/left_hand/motor_status",
   "/right_hand/motor_status",
   "/hric/robot/rl_state",
+  "/tg3/data_collection/control",
+  "/tg3/data_collection/status",
+  "/tg3/data_collection/iarm_frame",
+]
+
+# Optional sensor groups never block capture or core READY. Their message
+# counts and whole-episode average rates are recorded in manifest.json as
+# absent/healthy/partial/low_rate quality observations. The Orbbec driver
+# targets 30 Hz; 20 Hz leaves room for discovery and compression jitter.
+[data_collection.optional_topic_groups.head_rgbd]
+topics = [
+  "/ob_camera_head/color/image_raw/compressed",
+  "/ob_camera_head/color/camera_info",
+  "/ob_camera_head/color/metadata",
+  "/ob_camera_head/depth/image_raw/compressedDepth",
+  "/ob_camera_head/depth/camera_info",
+  "/ob_camera_head/depth/metadata",
+]
+
+[data_collection.optional_topic_groups.head_rgbd.minimum_topic_rates_hz]
+"/ob_camera_head/color/image_raw/compressed" = 20.0
+"/ob_camera_head/depth/image_raw/compressedDepth" = 20.0
+
+[data_collection.optional_topic_groups.waist_rgbd]
+topics = [
   "/ob_camera_waist/color/image_raw/compressed",
   "/ob_camera_waist/color/camera_info",
   "/ob_camera_waist/color/metadata",
   "/ob_camera_waist/depth/image_raw/compressedDepth",
   "/ob_camera_waist/depth/camera_info",
   "/ob_camera_waist/depth/metadata",
-  "/tf_static",
-  "/tg3/data_collection/control",
-  "/tg3/data_collection/status",
-  "/tg3/data_collection/iarm_frame",
 ]
 
+[data_collection.optional_topic_groups.waist_rgbd.minimum_topic_rates_hz]
+"/ob_camera_waist/color/image_raw/compressed" = 20.0
+"/ob_camera_waist/depth/image_raw/compressedDepth" = 20.0
+
 [locomotion]
 # The two original immediate bindings are independent: right C + left-stick
 # vertical controls translation; left Z + right-stick horizontal controls

tg3_local_teleop/data_collection.py

@@ -32,6 +32,15 @@ import yaml
 ToggleAction = Literal["start", "stop"]
 
 
+def _valid_ros_topic(topic: Any) -> bool:
+    return (
+        isinstance(topic, str)
+        and topic.startswith("/")
+        and topic.strip() == topic
+        and not any(character.isspace() for character in topic)
+    )
+
+
 def left_joystick_pressed(data: Mapping[str, Any]) -> bool | None:
     """Strictly parse the live-frame ``button_joystick.left`` value.
 
@@ -195,6 +204,55 @@ class RecordingToggleGate:
         return "active" if self.active else "idle"
 
 
+@dataclass(frozen=True)
+class OptionalTopicGroupConfig:
+    """Non-fatal completeness and rate checks for an optional sensor group."""
+
+    topics: Sequence[str]
+    minimum_topic_rates_hz: Mapping[str, float] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        topics = (
+            tuple(self.topics)
+            if isinstance(self.topics, Sequence)
+            and not isinstance(self.topics, str)
+            else ()
+        )
+        if not topics:
+            raise ValueError("optional topic group must contain at least one topic")
+        if any(not _valid_ros_topic(topic) for topic in topics):
+            raise ValueError(
+                "optional topic group topics must be absolute ROS topic names"
+            )
+        if len(set(topics)) != len(topics):
+            raise ValueError("optional topic group topics must not contain duplicates")
+        if not isinstance(self.minimum_topic_rates_hz, Mapping):
+            raise ValueError("optional minimum topic rates must be a mapping")
+        rates: dict[str, float] = {}
+        for topic, rate in self.minimum_topic_rates_hz.items():
+            if topic not in topics:
+                raise ValueError(
+                    "optional minimum-rate topics must be a subset of group topics"
+                )
+            if (
+                isinstance(rate, bool)
+                or not isinstance(rate, (int, float))
+                or not math.isfinite(rate)
+                or rate <= 0.0
+            ):
+                raise ValueError(
+                    f"optional minimum topic rate for {topic!r} must be "
+                    "positive and finite"
+                )
+            rates[topic] = float(rate)
+        object.__setattr__(self, "topics", topics)
+        object.__setattr__(
+            self,
+            "minimum_topic_rates_hz",
+            MappingProxyType(rates),
+        )
+
+
 @dataclass(frozen=True)
 class RecorderConfig:
     """Static configuration for :class:`DataRecorderManager`."""
@@ -203,6 +261,10 @@ class RecorderConfig:
     topics: Sequence[str] = ()
     required_topics: Sequence[str] = ()
     minimum_topic_rates_hz: Mapping[str, float] = field(default_factory=dict)
+    optional_topic_groups: Mapping[str, OptionalTopicGroupConfig] = field(
+        default_factory=dict
+    )
+    retain_failed_episodes: bool = True
     minimum_free_bytes: int = 5 * 1024**3
     max_duration_seconds: float = 30 * 60.0
     poll_interval_seconds: float = 0.1
@@ -219,25 +281,13 @@ class RecorderConfig:
         minimum_topic_rates: dict[str, float] = {}
         if not topics:
             raise ValueError("at least one recording topic is required")
-        if any(
+        if any(not _valid_ros_topic(topic) for topic in topics):
-            not isinstance(topic, str)
-            or not topic.startswith("/")
-            or topic.strip() != topic
-            or any(character.isspace() for character in topic)
-            for topic in topics
-        ):
             raise ValueError("recording topics must be absolute ROS topic names")
         if len(set(topics)) != len(topics):
             raise ValueError("recording topics must not contain duplicates")
         if len(set(required_topics)) != len(required_topics):
             raise ValueError("required topics must not contain duplicates")
-        if any(
+        if any(not _valid_ros_topic(topic) for topic in required_topics):
-            not isinstance(topic, str)
-            or not topic.startswith("/")
-            or topic.strip() != topic
-            or any(character.isspace() for character in topic)
-            for topic in required_topics
-        ):
             raise ValueError("required topics must be absolute ROS topic names")
         if not set(required_topics).issubset(topics):
             raise ValueError("required topics must be a subset of recording topics")
@@ -258,6 +308,45 @@ class RecorderConfig:
                     f"minimum topic rate for {topic!r} must be positive and finite"
                 )
             minimum_topic_rates[topic] = float(rate)
+        if not isinstance(self.optional_topic_groups, Mapping):
+            raise ValueError("optional topic groups must be a mapping")
+        optional_groups: dict[str, OptionalTopicGroupConfig] = {}
+        grouped_topics: set[str] = set()
+        for name, group in self.optional_topic_groups.items():
+            if (
+                not isinstance(name, str)
+                or not name
+                or len(name) > 64
+                or any(
+                    character
+                    not in "-_0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
+                    for character in name
+                )
+            ):
+                raise ValueError(
+                    "optional topic group names must contain 1-64 safe characters"
+                )
+            if not isinstance(group, OptionalTopicGroupConfig):
+                raise ValueError(
+                    f"optional topic group {name!r} has an invalid configuration"
+                )
+            group_topics = set(group.topics)
+            if not group_topics.issubset(topics):
+                raise ValueError(
+                    f"optional topic group {name!r} must be a subset of recording topics"
+                )
+            if group_topics.intersection(required_topics):
+                raise ValueError(
+                    f"optional topic group {name!r} must be disjoint from required topics"
+                )
+            overlap = group_topics.intersection(grouped_topics)
+            if overlap:
+                raise ValueError(
+                    "optional topic groups must be disjoint; repeated topics: "
+                    + ", ".join(sorted(overlap))
+                )
+            grouped_topics.update(group_topics)
+            optional_groups[name] = group
         if (
             isinstance(self.minimum_free_bytes, bool)
             or not isinstance(self.minimum_free_bytes, int)
@@ -277,6 +366,8 @@ class RecorderConfig:
             raise ValueError("ROS 2 executable must be a non-empty string")
         if type(self.validate_bag_info) is not bool:
             raise ValueError("validate bag info must be a bool")
+        if type(self.retain_failed_episodes) is not bool:
+            raise ValueError("retain failed episodes must be a bool")
         object.__setattr__(self, "base_directory", base)
         object.__setattr__(self, "topics", topics)
         object.__setattr__(self, "required_topics", required_topics)
@@ -285,6 +376,11 @@ class RecorderConfig:
             "minimum_topic_rates_hz",
             MappingProxyType(minimum_topic_rates),
         )
+        object.__setattr__(
+            self,
+            "optional_topic_groups",
+            MappingProxyType(optional_groups),
+        )
 
 
 class RecorderProcess(Protocol):
@@ -325,6 +421,8 @@ class _Episode:
     required_topic_message_counts: dict[str, int] | None = None
     minimum_topic_message_counts: dict[str, int] | None = None
     observed_topic_rates_hz: dict[str, float] | None = None
+    optional_topic_groups: dict[str, Any] | None = None
+    data_quality_warnings: list[str] | None = None
     metadata_duration_nanoseconds: int | None = None
     process: RecorderProcess | None = None
     stdout_stream: Any = None
@@ -767,6 +865,8 @@ class DataRecorderManager:
                 episode.minimum_topic_message_counts
             ),
             "observed_topic_rates_hz": episode.observed_topic_rates_hz,
+            "optional_topic_groups": episode.optional_topic_groups,
+            "data_quality_warnings": episode.data_quality_warnings,
             "metadata_duration_nanoseconds": (
                 episode.metadata_duration_nanoseconds
             ),
@@ -859,6 +959,7 @@ class DataRecorderManager:
             topic: all_counts.get(topic, 0) for topic in self.config.topics
         }
         episode.required_topic_message_counts = required_counts
+        self._observe_optional_topic_groups(episode, information, all_counts)
         missing = [
             topic for topic in self.config.required_topics if topic not in all_counts
         ]
@@ -877,6 +978,101 @@ class DataRecorderManager:
         self._validate_minimum_topic_rates(episode, information, all_counts)
         return required_counts
 
+    def _observe_optional_topic_groups(
+        self,
+        episode: _Episode,
+        information: Mapping[str, Any],
+        all_counts: Mapping[str, int],
+    ) -> None:
+        """Classify optional sensor quality without invalidating core data."""
+
+        duration_nanoseconds: int | None = None
+        duration = information.get("duration")
+        if isinstance(duration, Mapping):
+            candidate = duration.get("nanoseconds")
+            if (
+                type(candidate) is int
+                and candidate > 0
+            ):
+                duration_nanoseconds = candidate
+                episode.metadata_duration_nanoseconds = candidate
+        duration_seconds = (
+            None
+            if duration_nanoseconds is None
+            else duration_nanoseconds / 1_000_000_000.0
+        )
+
+        observations: dict[str, Any] = {}
+        warnings: list[str] = []
+        for name, group in self.config.optional_topic_groups.items():
+            counts = {
+                topic: int(all_counts.get(topic, 0)) for topic in group.topics
+            }
+            empty_topics = [
+                topic for topic, count in counts.items() if count <= 0
+            ]
+            all_absent = len(empty_topics) == len(group.topics)
+            minimum_counts: dict[str, int | None] = {}
+            observed_rates: dict[str, float | None] = {}
+            below_rate: list[str] = []
+            for topic, minimum_rate in group.minimum_topic_rates_hz.items():
+                if duration_seconds is None:
+                    minimum_counts[topic] = None
+                    observed_rates[topic] = None
+                    if not all_absent:
+                        below_rate.append(topic)
+                    continue
+                minimum_count = math.floor(duration_seconds * minimum_rate)
+                observed_rate = counts[topic] / duration_seconds
+                minimum_counts[topic] = minimum_count
+                observed_rates[topic] = observed_rate
+                if not all_absent and counts[topic] < minimum_count:
+                    below_rate.append(topic)
+
+            if all_absent:
+                state = "absent"
+            elif empty_topics:
+                state = "partial"
+                warnings.append(
+                    f"optional topic group {name!r} is partial; zero-message "
+                    "topics: " + ", ".join(empty_topics)
+                )
+            elif below_rate:
+                state = "low_rate"
+                if duration_seconds is None:
+                    warnings.append(
+                        f"optional topic group {name!r} rate could not be "
+                        "validated because bag duration is unavailable"
+                    )
+                else:
+                    details = [
+                        f"{topic}={observed_rates[topic]:.3f}Hz<"
+                        f"{group.minimum_topic_rates_hz[topic]:.3f}Hz"
+                        for topic in below_rate
+                    ]
+                    warnings.append(
+                        f"optional topic group {name!r} is below its observed "
+                        "minimum rate: " + "; ".join(details)
+                    )
+            else:
+                state = "healthy"
+
+            observations[name] = {
+                "state": state,
+                "topics": list(group.topics),
+                "topic_message_counts": counts,
+                "minimum_topic_rates_hz": dict(
+                    group.minimum_topic_rates_hz
+                ),
+                "minimum_topic_message_counts": minimum_counts,
+                "observed_topic_rates_hz": observed_rates,
+                "zero_message_topics": empty_topics,
+                "below_minimum_rate_topics": below_rate,
+            }
+
+        episode.optional_topic_groups = observations
+        episode.data_quality_warnings = warnings
+
     def _validate_minimum_topic_rates(
         self,
         episode: _Episode,
@@ -1036,6 +1232,31 @@ class DataRecorderManager:
     ) -> tuple[Path | None, str]:
         if not episode.active_directory.exists():
             return None, "active episode directory is missing"
+        if not self.config.retain_failed_episodes:
+            try:
+                active_root = Path(self.config.base_directory) / "active"
+                target = episode.active_directory
+                if (
+                    target.parent != active_root
+                    or target.name != episode.request.episode_id
+                    or target.is_symlink()
+                    or active_root.is_symlink()
+                ):
+                    raise RecordingError(
+                        "refusing to discard a failed episode outside its "
+                        "fixed active root"
+                    )
+                shutil.rmtree(target)
+                try:
+                    self._sync_directory(active_root)
+                except OSError:
+                    pass
+                return None, ""
+            except Exception as discard_error:
+                return None, (
+                    "failed to discard project-owned failed episode: "
+                    f"{type(discard_error).__name__}: {discard_error}"
+                )
         try:
             ready_marker = episode.active_directory / "READY"
             if ready_marker.exists():
@@ -1063,6 +1284,8 @@ class DataRecorderManager:
                     episode.minimum_topic_message_counts
                 ),
                 "observed_topic_rates_hz": episode.observed_topic_rates_hz,
+                "optional_topic_groups": episode.optional_topic_groups,
+                "data_quality_warnings": episode.data_quality_warnings,
                 "metadata_duration_nanoseconds": (
                     episode.metadata_duration_nanoseconds
                 ),
@@ -1242,6 +1465,7 @@ class DataRecorderManager:
 
 __all__ = [
     "DataRecorderManager",
+    "OptionalTopicGroupConfig",
     "RecorderConfig",
     "RecordingError",
     "RecordingToggleGate",

tg3_local_teleop/data_recorder_node.py

@@ -29,7 +29,11 @@ from rclpy.qos import (
 from rclpy.utilities import remove_ros_args
 from std_msgs.msg import String
 
-from data_collection import DataRecorderManager, RecorderConfig
+from data_collection import (
+    DataRecorderManager,
+    OptionalTopicGroupConfig,
+    RecorderConfig,
+)
 from data_recorder_protocol import (
     RecorderControlProtocol,
     topics_without_publishers,
@@ -102,6 +106,41 @@ def _minimum_topic_rates(value: Any) -> dict[str, float]:
     return result
 
 
+def _optional_topic_groups(
+    value: Any,
+) -> dict[str, OptionalTopicGroupConfig]:
+    name = "optional_topic_groups"
+    if not isinstance(value, dict):
+        raise ValueError(f"data_collection.{name} must be a TOML table")
+    result: dict[str, OptionalTopicGroupConfig] = {}
+    for group_name, raw_group in value.items():
+        if not isinstance(raw_group, dict):
+            raise ValueError(
+                f"data_collection.{name}.{group_name} must be a TOML table"
+            )
+        unknown = sorted(
+            set(raw_group) - {"topics", "minimum_topic_rates_hz"}
+        )
+        if unknown:
+            raise ValueError(
+                f"data_collection.{name}.{group_name} has unknown keys: "
+                + ", ".join(unknown)
+            )
+        topics = _topic_list(
+            raw_group.get("topics", []),
+            f"{name}.{group_name}.topics",
+            allow_empty=False,
+        )
+        rates = _minimum_topic_rates(
+            raw_group.get("minimum_topic_rates_hz", {})
+        )
+        result[group_name] = OptionalTopicGroupConfig(
+            topics=topics,
+            minimum_topic_rates_hz=rates,
+        )
+    return result
+
+
 @dataclass(frozen=True)
 class SupervisorSettings:
     enabled: bool
@@ -150,6 +189,9 @@ def load_settings(path: Path | str) -> SupervisorSettings:
     minimum_topic_rates = _minimum_topic_rates(
         section.get("minimum_topic_rates_hz", {})
     )
+    optional_topic_groups = _optional_topic_groups(
+        section.get("optional_topic_groups", {})
+    )
     unknown_rate_topics = sorted(
         set(minimum_topic_rates) - set(required_topics)
     )
@@ -187,6 +229,11 @@ def load_settings(path: Path | str) -> SupervisorSettings:
         topics=topics,
         required_topics=required_topics,
         minimum_topic_rates_hz=minimum_topic_rates,
+        optional_topic_groups=optional_topic_groups,
+        retain_failed_episodes=_boolean(
+            section.get("retain_failed_episodes", True),
+            "retain_failed_episodes",
+        ),
         minimum_free_bytes=int(minimum_free_gib * 1024**3),
         max_duration_seconds=max_duration,
         poll_interval_seconds=0.1,

tg3_local_teleop/delete_ready_episode.py

@@ -0,0 +1,537 @@
+#!/usr/bin/env python3
+"""Safely and idempotently remove one verified recorder ``ready`` episode.
+
+This helper intentionally has no command-line option for either root directory.
+It can only move a direct child of the fixed recorder ``ready`` directory into
+the fixed ``deleting`` staging directory and remove that tombstone.  The caller
+must provide both the episode id and the SHA-256 of the exact manifest that was
+verified and durably published on the PS workstation.
+
+Before recursive removal, an authorization ledger is fsynced in the fixed
+``delete_ledger`` sibling directory. It survives partial tombstone deletion and
+is removed only after the tree cleanup itself is durable.
+"""
+
+from __future__ import annotations
+
+import argparse
+import fcntl
+import hashlib
+import json
+import os
+import re
+import stat
+import sys
+import uuid
+from pathlib import Path
+from typing import Any
+
+
+BASE_ROOT = Path("/home/nvidia/tg3_data_collection")
+READY_ROOT = BASE_ROOT / "ready"
+DELETING_ROOT = BASE_ROOT / "deleting"
+LEDGER_ROOT = BASE_ROOT / "delete_ledger"
+SAFE_EPISODE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$")
+SAFE_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+MAX_MANIFEST_BYTES = 16 * 1024 * 1024
+MAX_LEDGER_BYTES = 64 * 1024
+LEDGER_SCHEMA_VERSION = 1
+
+
+class DeleteSafetyError(RuntimeError):
+    """The requested deletion did not satisfy the fixed safety contract."""
+
+
+class DeleteCollisionError(DeleteSafetyError):
+    """A conflicting ready/deleting episode exists and was left untouched."""
+
+
+def safe_episode_name(value: str) -> bool:
+    return bool(SAFE_EPISODE_NAME.fullmatch(value)) and value not in (".", "..")
+
+
+def _open_directory(path: Path) -> int:
+    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError as exc:
+        raise DeleteSafetyError(f"unsafe or missing directory: {path}: {exc}") from exc
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISDIR(metadata.st_mode):
+            raise DeleteSafetyError(f"not a directory: {path}")
+        return descriptor
+    except Exception:
+        os.close(descriptor)
+        raise
+
+
+def _open_child_directory(parent_fd: int, name: str) -> int:
+    flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(name, flags, dir_fd=parent_fd)
+    except OSError as exc:
+        raise DeleteSafetyError(f"unsafe or missing episode directory: {name}: {exc}") from exc
+    try:
+        if not stat.S_ISDIR(os.fstat(descriptor).st_mode):
+            raise DeleteSafetyError(f"episode is not a directory: {name}")
+        return descriptor
+    except Exception:
+        os.close(descriptor)
+        raise
+
+
+def _read_regular_file(parent_fd: int, name: str, *, max_bytes: int) -> bytes:
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(name, flags, dir_fd=parent_fd)
+    except OSError as exc:
+        raise DeleteSafetyError(f"missing or unsafe regular file {name!r}: {exc}") from exc
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode):
+            raise DeleteSafetyError(f"not a regular file: {name}")
+        if metadata.st_size > max_bytes:
+            raise DeleteSafetyError(f"file is too large: {name}")
+        chunks: list[bytes] = []
+        remaining = max_bytes + 1
+        while remaining:
+            block = os.read(descriptor, min(1024 * 1024, remaining))
+            if not block:
+                break
+            chunks.append(block)
+            remaining -= len(block)
+        payload = b"".join(chunks)
+        if len(payload) > max_bytes:
+            raise DeleteSafetyError(f"file is too large: {name}")
+        return payload
+    finally:
+        os.close(descriptor)
+
+
+def _validate_tree_entries(directory_fd: int) -> None:
+    """Require the quarantined recorder tree to contain only dirs/files."""
+
+    for child in os.listdir(directory_fd):
+        if child in (".", "..") or "/" in child or "\x00" in child:
+            raise DeleteSafetyError(f"unsafe directory entry: {child!r}")
+        metadata = os.stat(child, dir_fd=directory_fd, follow_symlinks=False)
+        if stat.S_ISDIR(metadata.st_mode):
+            child_fd = _open_child_directory(directory_fd, child)
+            try:
+                _validate_tree_entries(child_fd)
+            finally:
+                os.close(child_fd)
+        elif not stat.S_ISREG(metadata.st_mode):
+            raise DeleteSafetyError(f"episode contains a symlink/special entry: {child}")
+
+
+def _ledger_name(episode: str, manifest_sha256: str) -> str:
+    return f"{episode}.{manifest_sha256}.authorized.json"
+
+
+def _ledger_payload(
+    episode: str, manifest_sha256: str, tombstone: str
+) -> dict[str, Any]:
+    return {
+        "schema_version": LEDGER_SCHEMA_VERSION,
+        "state": "AUTHORIZED_DELETE",
+        "episode_id": episode,
+        "manifest_sha256": manifest_sha256,
+        "tombstone": tombstone,
+    }
+
+
+def _validate_ledger(
+    ledger_fd: int,
+    ledger_name: str,
+    *,
+    episode: str,
+    manifest_sha256: str,
+    tombstone: str,
+) -> None:
+    raw = _read_regular_file(ledger_fd, ledger_name, max_bytes=MAX_LEDGER_BYTES)
+    try:
+        payload: Any = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise DeleteSafetyError(f"invalid deletion authorization ledger: {exc}") from exc
+    if payload != _ledger_payload(episode, manifest_sha256, tombstone):
+        raise DeleteSafetyError("deletion authorization ledger identity mismatch")
+
+
+def _write_ledger(
+    ledger_fd: int,
+    ledger_name: str,
+    *,
+    episode: str,
+    manifest_sha256: str,
+    tombstone: str,
+) -> None:
+    payload = (
+        json.dumps(
+            _ledger_payload(episode, manifest_sha256, tombstone),
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+        + "\n"
+    ).encode("utf-8")
+    temporary = f".{ledger_name}.{uuid.uuid4().hex}.tmp"
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(temporary, flags, 0o600, dir_fd=ledger_fd)
+    try:
+        view = memoryview(payload)
+        while view:
+            written = os.write(descriptor, view)
+            if written <= 0:
+                raise OSError("short write while creating deletion ledger")
+            view = view[written:]
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+    try:
+        # A hard-link publish is atomic and refuses to replace any existing
+        # sidecar. The helper flock serializes legitimate writers as well.
+        os.link(
+            temporary,
+            ledger_name,
+            src_dir_fd=ledger_fd,
+            dst_dir_fd=ledger_fd,
+            follow_symlinks=False,
+        )
+        os.fsync(ledger_fd)
+    finally:
+        try:
+            os.unlink(temporary, dir_fd=ledger_fd)
+            os.fsync(ledger_fd)
+        except FileNotFoundError:
+            pass
+
+
+def _validate_episode(
+    parent_fd: int,
+    directory_name: str,
+    *,
+    episode: str,
+    expected_manifest_sha256: str,
+) -> None:
+    episode_fd = _open_child_directory(parent_fd, directory_name)
+    try:
+        # READY must be a real regular file.  Its contents are deliberately not
+        # trusted as a deletion credential; the exact manifest digest is.
+        _read_regular_file(episode_fd, "READY", max_bytes=4096)
+        raw_manifest = _read_regular_file(
+            episode_fd, "manifest.json", max_bytes=MAX_MANIFEST_BYTES
+        )
+        actual_digest = hashlib.sha256(raw_manifest).hexdigest()
+        if actual_digest != expected_manifest_sha256:
+            raise DeleteSafetyError(
+                "manifest SHA-256 mismatch: "
+                f"expected {expected_manifest_sha256}, got {actual_digest}"
+            )
+        try:
+            manifest: Any = json.loads(raw_manifest.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise DeleteSafetyError(f"invalid manifest JSON: {exc}") from exc
+        if not isinstance(manifest, dict):
+            raise DeleteSafetyError("manifest root is not an object")
+        if manifest.get("state") != "complete":
+            raise DeleteSafetyError("manifest state is not complete")
+        if manifest.get("episode_id") != episode:
+            raise DeleteSafetyError("manifest episode_id does not match target")
+        _validate_tree_entries(episode_fd)
+    finally:
+        os.close(episode_fd)
+
+
+def _remove_tree_at(parent_fd: int, name: str) -> None:
+    """Remove one child tree without following any symbolic link."""
+
+    directory_fd = _open_child_directory(parent_fd, name)
+    try:
+        for child in os.listdir(directory_fd):
+            if child in (".", "..") or "/" in child or "\x00" in child:
+                raise DeleteSafetyError(f"unsafe directory entry: {child!r}")
+            metadata = os.stat(child, dir_fd=directory_fd, follow_symlinks=False)
+            if stat.S_ISDIR(metadata.st_mode):
+                _remove_tree_at(directory_fd, child)
+            else:
+                # Symlinks and all non-directory entries are unlinked, never
+                # followed. Recorder output is expected to contain only files.
+                os.unlink(child, dir_fd=directory_fd)
+        os.fsync(directory_fd)
+    finally:
+        os.close(directory_fd)
+    os.rmdir(name, dir_fd=parent_fd)
+
+
+def _child_kind(parent_fd: int, name: str) -> str:
+    try:
+        metadata = os.stat(name, dir_fd=parent_fd, follow_symlinks=False)
+    except FileNotFoundError:
+        return "absent"
+    if stat.S_ISDIR(metadata.st_mode):
+        return "directory"
+    if stat.S_ISLNK(metadata.st_mode):
+        return "symlink"
+    if stat.S_ISREG(metadata.st_mode):
+        return "regular"
+    return "other"
+
+
+def delete_ready_episode(
+    episode: str,
+    manifest_sha256: str,
+    *,
+    ready_root: Path = READY_ROOT,
+    deleting_root: Path = DELETING_ROOT,
+    ledger_root: Path | None = None,
+) -> dict[str, str]:
+    """Compare, atomically quarantine, and remove exactly one ready episode.
+
+    Root arguments are injectable only for unit tests. The CLI below always
+    uses the immutable production constants, including a ledger directory that
+    is outside the recursively removed ``deleting`` tree.
+    """
+
+    if not safe_episode_name(episode):
+        raise DeleteSafetyError(f"unsafe episode id: {episode!r}")
+    if not SAFE_SHA256.fullmatch(manifest_sha256):
+        raise DeleteSafetyError("manifest SHA-256 must be 64 lowercase hex characters")
+
+    ready_root = Path(ready_root)
+    deleting_root = Path(deleting_root)
+    ledger_root = Path(ledger_root or (deleting_root.parent / LEDGER_ROOT.name))
+    if not (
+        ready_root.parent == deleting_root.parent == ledger_root.parent
+    ):
+        raise DeleteSafetyError("ready, deleting, and ledger roots must be siblings")
+    base_root = ready_root.parent
+
+    # Refuse a symlinked/wrong parent even in tests. resolve(strict=True) also
+    # proves that all existing parent components resolve to this exact path.
+    try:
+        if base_root.resolve(strict=True) != base_root.absolute():
+            raise DeleteSafetyError("base root must not contain symlinks")
+        if ready_root.resolve(strict=True) != ready_root.absolute():
+            raise DeleteSafetyError("ready root must not contain symlinks")
+    except FileNotFoundError as exc:
+        raise DeleteSafetyError(f"fixed recorder root is missing: {exc}") from exc
+
+    base_fd = _open_directory(base_root)
+    ready_fd = _open_directory(ready_root)
+    deleting_fd: int | None = None
+    ledger_fd: int | None = None
+    lock_fd: int | None = None
+    try:
+        base_stat = os.fstat(base_fd)
+        ready_stat = os.fstat(ready_fd)
+        if base_stat.st_dev != ready_stat.st_dev:
+            raise DeleteSafetyError("ready root is not on the recorder filesystem")
+
+        deleting_kind = _child_kind(base_fd, deleting_root.name)
+        if deleting_kind == "absent":
+            os.mkdir(deleting_root.name, mode=0o750, dir_fd=base_fd)
+            os.fsync(base_fd)
+        elif deleting_kind != "directory":
+            raise DeleteSafetyError("deleting root is not a real directory")
+        deleting_fd = _open_directory(deleting_root)
+        if os.fstat(deleting_fd).st_dev != ready_stat.st_dev:
+            raise DeleteSafetyError("deleting root is not on the ready filesystem")
+
+        ledger_kind = _child_kind(base_fd, ledger_root.name)
+        if ledger_kind == "absent":
+            os.mkdir(ledger_root.name, mode=0o750, dir_fd=base_fd)
+            os.fsync(base_fd)
+        elif ledger_kind != "directory":
+            raise DeleteSafetyError("delete ledger root is not a real directory")
+        ledger_fd = _open_directory(ledger_root)
+        if os.fstat(ledger_fd).st_dev != ready_stat.st_dev:
+            raise DeleteSafetyError("delete ledger is not on the ready filesystem")
+
+        lock_flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0)
+        lock_fd = os.open(".delete_ready.lock", lock_flags, 0o600, dir_fd=deleting_fd)
+        if not stat.S_ISREG(os.fstat(lock_fd).st_mode):
+            raise DeleteSafetyError("deletion lock is not a regular file")
+        fcntl.flock(lock_fd, fcntl.LOCK_EX)
+
+        tombstone = f"{episode}.{manifest_sha256}.deleting"
+        authorization = _ledger_name(episode, manifest_sha256)
+        ready_kind = _child_kind(ready_fd, episode)
+        tombstone_kind = _child_kind(deleting_fd, tombstone)
+        authorization_kind = _child_kind(ledger_fd, authorization)
+
+        # Any other tombstone for this episode is a manifest-identity collision.
+        prefix = f"{episode}."
+        collisions = [
+            name
+            for name in os.listdir(deleting_fd)
+            if name.startswith(prefix)
+            and name.endswith(".deleting")
+            and name != tombstone
+        ]
+        if collisions:
+            raise DeleteCollisionError(
+                f"conflicting deleting tombstone(s) exist: {sorted(collisions)!r}"
+            )
+        ledger_collisions = [
+            name
+            for name in os.listdir(ledger_fd)
+            if name.startswith(prefix)
+            and name.endswith(".authorized.json")
+            and name != authorization
+        ]
+        if ledger_collisions:
+            raise DeleteCollisionError(
+                "conflicting deletion authorization(s) exist: "
+                f"{sorted(ledger_collisions)!r}"
+            )
+        if ready_kind not in ("absent", "directory"):
+            raise DeleteSafetyError(f"ready target is unsafe: {ready_kind}")
+        if tombstone_kind not in ("absent", "directory"):
+            raise DeleteSafetyError(f"deleting target is unsafe: {tombstone_kind}")
+        if authorization_kind not in ("absent", "regular"):
+            raise DeleteSafetyError(
+                f"deletion authorization is unsafe: {authorization_kind}"
+            )
+        if ready_kind == "directory" and tombstone_kind == "directory":
+            raise DeleteCollisionError("ready target and deleting tombstone both exist")
+        if ready_kind == "directory" and authorization_kind == "regular":
+            raise DeleteCollisionError(
+                "ready target exists alongside a stale deletion authorization"
+            )
+
+        if ready_kind == "directory":
+            _validate_episode(
+                ready_fd,
+                episode,
+                episode=episode,
+                expected_manifest_sha256=manifest_sha256,
+            )
+            os.rename(
+                episode,
+                tombstone,
+                src_dir_fd=ready_fd,
+                dst_dir_fd=deleting_fd,
+            )
+            os.fsync(ready_fd)
+            os.fsync(deleting_fd)
+            # Re-validate after the rename.  If the source name was exchanged
+            # between the first comparison and rename, the unexpected tree is
+            # quarantined but is never unlinked.
+            _validate_episode(
+                deleting_fd,
+                tombstone,
+                episode=episode,
+                expected_manifest_sha256=manifest_sha256,
+            )
+            _write_ledger(
+                ledger_fd,
+                authorization,
+                episode=episode,
+                manifest_sha256=manifest_sha256,
+                tombstone=tombstone,
+            )
+            authorization_kind = "regular"
+            result_state = "deleted"
+        elif tombstone_kind == "directory":
+            if authorization_kind == "regular":
+                # The durable out-of-tree ledger remains usable even if an
+                # interrupted recursive removal already removed READY and the
+                # manifest from inside the tombstone.
+                _validate_ledger(
+                    ledger_fd,
+                    authorization,
+                    episode=episode,
+                    manifest_sha256=manifest_sha256,
+                    tombstone=tombstone,
+                )
+            else:
+                # A crash between rename and ledger publication is recoverable
+                # only while the tombstone still proves its original identity.
+                _validate_episode(
+                    deleting_fd,
+                    tombstone,
+                    episode=episode,
+                    expected_manifest_sha256=manifest_sha256,
+                )
+                _write_ledger(
+                    ledger_fd,
+                    authorization,
+                    episode=episode,
+                    manifest_sha256=manifest_sha256,
+                    tombstone=tombstone,
+                )
+                authorization_kind = "regular"
+            result_state = "resumed_delete"
+        else:
+            if authorization_kind == "regular":
+                # The tree was fully removed and only the final ledger cleanup
+                # (or its ACK) was interrupted.
+                _validate_ledger(
+                    ledger_fd,
+                    authorization,
+                    episode=episode,
+                    manifest_sha256=manifest_sha256,
+                    tombstone=tombstone,
+                )
+                os.unlink(authorization, dir_fd=ledger_fd)
+                os.fsync(ledger_fd)
+            return {
+                "state": "already_absent",
+                "episode_id": episode,
+                "manifest_sha256": manifest_sha256,
+            }
+
+        _remove_tree_at(deleting_fd, tombstone)
+        os.fsync(deleting_fd)
+        # Delete authorization is deliberately the final artifact removed.
+        # If this unlink/ACK is lost, the next invocation validates it and
+        # safely returns already_absent.
+        _validate_ledger(
+            ledger_fd,
+            authorization,
+            episode=episode,
+            manifest_sha256=manifest_sha256,
+            tombstone=tombstone,
+        )
+        os.unlink(authorization, dir_fd=ledger_fd)
+        os.fsync(ledger_fd)
+        return {
+            "state": result_state,
+            "episode_id": episode,
+            "manifest_sha256": manifest_sha256,
+        }
+    finally:
+        if lock_fd is not None:
+            os.close(lock_fd)
+        if deleting_fd is not None:
+            os.close(deleting_fd)
+        if ledger_fd is not None:
+            os.close(ledger_fd)
+        os.close(ready_fd)
+        os.close(base_fd)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("episode_id")
+    parser.add_argument("manifest_sha256")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        result = delete_ready_episode(args.episode_id, args.manifest_sha256)
+    except DeleteCollisionError as exc:
+        print(json.dumps({"state": "collision", "error": str(exc)}))
+        return 3
+    except (DeleteSafetyError, OSError) as exc:
+        print(json.dumps({"state": "refused", "error": str(exc)}))
+        return 2
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tg3_local_teleop/test_data_collection.py

@@ -15,12 +15,45 @@ from typing import Any
 
 from data_collection import (
     DataRecorderManager,
+    OptionalTopicGroupConfig,
     RecorderConfig,
     RecordingToggleGate,
     left_joystick_pressed,
 )
 
 
+HEAD_CAMERA_TOPICS = (
+    "/ob_camera_head/color/image_raw/compressed",
+    "/ob_camera_head/color/camera_info",
+    "/ob_camera_head/color/metadata",
+    "/ob_camera_head/depth/image_raw/compressedDepth",
+    "/ob_camera_head/depth/camera_info",
+    "/ob_camera_head/depth/metadata",
+)
+WAIST_CAMERA_TOPICS = tuple(
+    topic.replace("_head", "_waist") for topic in HEAD_CAMERA_TOPICS
+)
+
+
+def optional_camera_groups() -> dict[str, OptionalTopicGroupConfig]:
+    return {
+        "head_rgbd": OptionalTopicGroupConfig(
+            topics=HEAD_CAMERA_TOPICS,
+            minimum_topic_rates_hz={
+                HEAD_CAMERA_TOPICS[0]: 20.0,
+                HEAD_CAMERA_TOPICS[3]: 20.0,
+            },
+        ),
+        "waist_rgbd": OptionalTopicGroupConfig(
+            topics=WAIST_CAMERA_TOPICS,
+            minimum_topic_rates_hz={
+                WAIST_CAMERA_TOPICS[0]: 20.0,
+                WAIST_CAMERA_TOPICS[3]: 20.0,
+            },
+        ),
+    }
+
+
 class LeftJoystickParserTest(unittest.TestCase):
     def test_accepts_only_live_bool_or_binary_integer(self) -> None:
         self.assertIs(
@@ -291,6 +324,10 @@ class DataRecorderManagerTest(unittest.TestCase):
         topics: Sequence[str] = ("/joint_states", "/tf"),
         required_topics: Sequence[str] = ("/joint_states", "/tf"),
         minimum_topic_rates_hz: Mapping[str, float] | None = None,
+        optional_topic_groups: Mapping[
+            str, OptionalTopicGroupConfig
+        ] | None = None,
+        retain_failed_episodes: bool = True,
     ) -> tuple[DataRecorderManager, Any]:
         if factory is None:
             factory = FakeProcessFactory(**(process_options or {}))
@@ -305,6 +342,8 @@ class DataRecorderManagerTest(unittest.TestCase):
                 topics=topics,
                 required_topics=required_topics,
                 minimum_topic_rates_hz=minimum_topic_rates_hz or {},
+                optional_topic_groups=optional_topic_groups or {},
+                retain_failed_episodes=retain_failed_episodes,
                 minimum_free_bytes=minimum_free,
                 max_duration_seconds=max_duration,
                 poll_interval_seconds=0.005,
@@ -330,6 +369,36 @@ class DataRecorderManagerTest(unittest.TestCase):
                 self.fail("condition did not become true before timeout")
             time.sleep(0.002)
 
+    def record_with_optional_cameras(
+        self, topic_counts: Mapping[str, int]
+    ) -> tuple[dict[str, Any], dict[str, Any]]:
+        topics = (
+            "/joint_states",
+            "/tf",
+            *HEAD_CAMERA_TOPICS,
+            *WAIST_CAMERA_TOPICS,
+        )
+        manager, _ = self.make_manager(
+            topics=topics,
+            required_topics=("/joint_states", "/tf"),
+            optional_topic_groups=optional_camera_groups(),
+            process_options={
+                "topic_counts": topic_counts,
+                "duration_nanoseconds": 1_000_000_000,
+            },
+        )
+        self.assertTrue(manager.request_start("ca" * 16))
+        self.wait_for(lambda: manager.status()["recording"])
+        self.assertTrue(manager.request_stop())
+        self.assertTrue(manager.wait_until_idle(2.0))
+        status = manager.status()
+        manifest = json.loads(
+            (
+                Path(status["last_episode_directory"]) / "manifest.json"
+            ).read_text()
+        )
+        return status, manifest
+
     def test_manual_stop_uses_sigint_and_atomically_creates_ready_manifest(
         self,
     ) -> None:
@@ -627,6 +696,103 @@ class DataRecorderManagerTest(unittest.TestCase):
                     0 if head_count is None else head_count,
                 )
 
+    def test_optional_cameras_absent_do_not_block_core_ready(self) -> None:
+        status, manifest = self.record_with_optional_cameras(
+            {"/joint_states": 25, "/tf": 50}
+        )
+        self.assertEqual(status["last_result"], "ready")
+        self.assertEqual(manifest["data_quality_warnings"], [])
+        self.assertEqual(
+            {
+                name: observation["state"]
+                for name, observation in manifest[
+                    "optional_topic_groups"
+                ].items()
+            },
+            {"head_rgbd": "absent", "waist_rgbd": "absent"},
+        )
+        self.assertEqual(
+            manifest["optional_topic_groups"]["head_rgbd"][
+                "below_minimum_rate_topics"
+            ],
+            [],
+        )
+        self.assertTrue((Path(status["last_episode_directory"]) / "READY").exists())
+
+    def test_each_active_optional_camera_is_observed_as_healthy(self) -> None:
+        for active_name, active_topics in (
+            ("head_rgbd", HEAD_CAMERA_TOPICS),
+            ("waist_rgbd", WAIST_CAMERA_TOPICS),
+        ):
+            with self.subTest(active_name=active_name):
+                counts = {"/joint_states": 25, "/tf": 50}
+                counts.update({topic: 1 for topic in active_topics})
+                counts[active_topics[0]] = 25
+                counts[active_topics[3]] = 25
+                status, manifest = self.record_with_optional_cameras(counts)
+                self.assertEqual(status["last_result"], "ready")
+                observation = manifest["optional_topic_groups"][active_name]
+                self.assertEqual(observation["state"], "healthy")
+                self.assertEqual(
+                    observation["observed_topic_rates_hz"],
+                    {active_topics[0]: 25.0, active_topics[3]: 25.0},
+                )
+                inactive_name = (
+                    "waist_rgbd"
+                    if active_name == "head_rgbd"
+                    else "head_rgbd"
+                )
+                self.assertEqual(
+                    manifest["optional_topic_groups"][inactive_name]["state"],
+                    "absent",
+                )
+                self.assertEqual(manifest["data_quality_warnings"], [])
+
+    def test_partial_optional_camera_warns_but_remains_ready(self) -> None:
+        counts = {"/joint_states": 25, "/tf": 50}
+        counts.update(
+            {
+                HEAD_CAMERA_TOPICS[0]: 25,
+                HEAD_CAMERA_TOPICS[1]: 1,
+                HEAD_CAMERA_TOPICS[2]: 1,
+            }
+        )
+        status, manifest = self.record_with_optional_cameras(counts)
+        self.assertEqual(status["last_result"], "ready")
+        observation = manifest["optional_topic_groups"]["head_rgbd"]
+        self.assertEqual(observation["state"], "partial")
+        self.assertEqual(
+            observation["zero_message_topics"],
+            list(HEAD_CAMERA_TOPICS[3:]),
+        )
+        self.assertTrue(manifest["data_quality_warnings"])
+        self.assertIn("is partial", manifest["data_quality_warnings"][0])
+        self.assertTrue((Path(status["last_episode_directory"]) / "READY").exists())
+
+    def test_low_rate_optional_camera_warns_but_remains_ready(self) -> None:
+        counts = {"/joint_states": 25, "/tf": 50}
+        counts.update({topic: 1 for topic in WAIST_CAMERA_TOPICS})
+        counts[WAIST_CAMERA_TOPICS[0]] = 5
+        counts[WAIST_CAMERA_TOPICS[3]] = 10
+        status, manifest = self.record_with_optional_cameras(counts)
+        self.assertEqual(status["last_result"], "ready")
+        observation = manifest["optional_topic_groups"]["waist_rgbd"]
+        self.assertEqual(observation["state"], "low_rate")
+        self.assertEqual(
+            observation["observed_topic_rates_hz"],
+            {
+                WAIST_CAMERA_TOPICS[0]: 5.0,
+                WAIST_CAMERA_TOPICS[3]: 10.0,
+            },
+        )
+        self.assertEqual(
+            observation["below_minimum_rate_topics"],
+            [WAIST_CAMERA_TOPICS[0], WAIST_CAMERA_TOPICS[3]],
+        )
+        self.assertTrue(manifest["data_quality_warnings"])
+        self.assertIn("below", manifest["data_quality_warnings"][0])
+        self.assertTrue((Path(status["last_episode_directory"]) / "READY").exists())
+
     def test_topic_that_stops_mid_episode_fails_average_rate(self) -> None:
         bag_info = FakeBagInfoRunner()
         manager, _ = self.make_manager(
@@ -779,6 +945,20 @@ class DataRecorderManagerTest(unittest.TestCase):
         )
         self.assertEqual(failure["status"], "failed")
 
+    def test_failed_payload_is_discarded_when_retention_is_disabled(self) -> None:
+        manager, factory = self.make_manager(
+            free_bytes=lambda _path: 0,
+            retain_failed_episodes=False,
+        )
+        self.assertTrue(manager.request_start("d0" * 16))
+        self.assertTrue(manager.wait_until_idle(2.0))
+        status = manager.status()
+        self.assertEqual(status["last_result"], "failed")
+        self.assertIsNone(status["last_episode_directory"])
+        self.assertEqual(factory.processes, [])
+        self.assertEqual(list((self.base / "active").iterdir()), [])
+        self.assertEqual(list((self.base / "failed").iterdir()), [])
+
     def test_configuration_rejects_unsafe_or_ambiguous_values(self) -> None:
         with self.assertRaises(ValueError):
             RecorderConfig(self.base, ())
@@ -790,6 +970,8 @@ class DataRecorderManagerTest(unittest.TestCase):
             RecorderConfig(self.base, ("/ok",), minimum_free_bytes=-1)
         with self.assertRaises(ValueError):
             RecorderConfig(self.base, ("/ok",), validate_bag_info=1)
+        with self.assertRaises(ValueError):
+            RecorderConfig(self.base, ("/ok",), retain_failed_episodes=1)
         with self.assertRaises(ValueError):
             RecorderConfig(self.base, ("/ok",), bag_info_timeout_seconds=0)
         with self.assertRaises(ValueError):
@@ -827,6 +1009,45 @@ class DataRecorderManagerTest(unittest.TestCase):
                 required_topics=("/recorded",),
                 minimum_topic_rates_hz=[],  # type: ignore[arg-type]
             )
+        with self.assertRaises(ValueError):
+            OptionalTopicGroupConfig(topics=())
+        with self.assertRaises(ValueError):
+            OptionalTopicGroupConfig(
+                topics=("/camera/image",),
+                minimum_topic_rates_hz={"/other": 20.0},
+            )
+        camera_group = OptionalTopicGroupConfig(
+            topics=("/camera/image",),
+            minimum_topic_rates_hz={"/camera/image": 20.0},
+        )
+        with self.assertRaises(ValueError):
+            RecorderConfig(
+                self.base,
+                ("/recorded",),
+                optional_topic_groups={"camera": camera_group},
+            )
+        with self.assertRaises(ValueError):
+            RecorderConfig(
+                self.base,
+                ("/recorded", "/camera/image"),
+                required_topics=("/camera/image",),
+                optional_topic_groups={"camera": camera_group},
+            )
+        with self.assertRaises(ValueError):
+            RecorderConfig(
+                self.base,
+                ("/recorded", "/camera/image"),
+                optional_topic_groups={
+                    "camera_a": camera_group,
+                    "camera_b": camera_group,
+                },
+            )
+        with self.assertRaises(ValueError):
+            RecorderConfig(
+                self.base,
+                ("/recorded", "/camera/image"),
+                optional_topic_groups={"bad name": camera_group},
+            )
 
 
 if __name__ == "__main__":

tg3_local_teleop/test_delete_ready_episode.py

@@ -0,0 +1,376 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import hashlib
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from delete_ready_episode import (
+    DeleteCollisionError,
+    DeleteSafetyError,
+    delete_ready_episode,
+)
+
+
+class DeleteReadyEpisodeTests(unittest.TestCase):
+    def _roots(self, root: Path) -> tuple[Path, Path, Path]:
+        base = root / "tg3_data_collection"
+        ready = base / "ready"
+        deleting = base / "deleting"
+        active = base / "active"
+        failed = base / "failed"
+        for path in (ready, active, failed):
+            path.mkdir(parents=True, exist_ok=True)
+        (active / "ACTIVE_SENTINEL").write_text("active", encoding="utf-8")
+        (failed / "FAILED_SENTINEL").write_text("failed", encoding="utf-8")
+        return ready, deleting, base
+
+    def _episode(
+        self,
+        ready: Path,
+        name: str = "episode_20260810T120000000_deadbeef",
+        *,
+        state: str = "complete",
+    ) -> tuple[Path, str]:
+        episode = ready / name
+        (episode / "bag").mkdir(parents=True)
+        (episode / "bag" / "bag_0.mcap").write_bytes(b"mcap")
+        manifest = {"state": state, "episode_id": name, "files": []}
+        raw = (json.dumps(manifest, sort_keys=True) + "\n").encode("utf-8")
+        (episode / "manifest.json").write_bytes(raw)
+        (episode / "READY").write_text("ready\n", encoding="ascii")
+        return episode, hashlib.sha256(raw).hexdigest()
+
+    def _authorization(
+        self,
+        base: Path,
+        episode: str,
+        digest: str,
+        *,
+        payload_digest: str | None = None,
+    ) -> Path:
+        ledger = base / "delete_ledger"
+        ledger.mkdir(exist_ok=True)
+        tombstone = f"{episode}.{digest}.deleting"
+        path = ledger / f"{episode}.{digest}.authorized.json"
+        path.write_text(
+            json.dumps(
+                {
+                    "schema_version": 1,
+                    "state": "AUTHORIZED_DELETE",
+                    "episode_id": episode,
+                    "manifest_sha256": payload_digest or digest,
+                    "tombstone": tombstone,
+                }
+            ),
+            encoding="utf-8",
+        )
+        return path
+
+    def test_compare_rename_delete_and_absent_retry_are_idempotent(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready)
+            name = episode.name
+            result = delete_ready_episode(
+                name,
+                digest,
+                ready_root=ready,
+                deleting_root=deleting,
+            )
+            self.assertEqual(result["state"], "deleted")
+            self.assertFalse(episode.exists())
+            self.assertEqual(
+                delete_ready_episode(
+                    name,
+                    digest,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )["state"],
+                "already_absent",
+            )
+            self.assertEqual(
+                (base / "active" / "ACTIVE_SENTINEL").read_text(), "active"
+            )
+            self.assertEqual(
+                (base / "failed" / "FAILED_SENTINEL").read_text(), "failed"
+            )
+
+    def test_resume_matching_tombstone_after_lost_ack(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, _base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready)
+            deleting.mkdir()
+            tombstone = deleting / f"{episode.name}.{digest}.deleting"
+            episode.rename(tombstone)
+            result = delete_ready_episode(
+                "episode_20260810T120000000_deadbeef",
+                digest,
+                ready_root=ready,
+                deleting_root=deleting,
+            )
+            self.assertEqual(result["state"], "resumed_delete")
+            self.assertFalse(tombstone.exists())
+
+    def test_partial_tree_without_manifest_resumes_from_durable_ledger(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready)
+            tombstone = deleting / f"{episode.name}.{digest}.deleting"
+            ledger = (
+                base
+                / "delete_ledger"
+                / f"{episode.name}.{digest}.authorized.json"
+            )
+
+            def interrupt_after_control_files_removed(
+                _parent_fd: int, _name: str
+            ) -> None:
+                (tombstone / "READY").unlink()
+                (tombstone / "manifest.json").unlink()
+                raise RuntimeError("injected power loss during recursive delete")
+
+            with mock.patch(
+                "delete_ready_episode._remove_tree_at",
+                side_effect=interrupt_after_control_files_removed,
+            ):
+                with self.assertRaises(RuntimeError):
+                    delete_ready_episode(
+                        episode.name,
+                        digest,
+                        ready_root=ready,
+                        deleting_root=deleting,
+                    )
+            self.assertTrue(tombstone.is_dir())
+            self.assertFalse((tombstone / "manifest.json").exists())
+            self.assertTrue(ledger.is_file())
+
+            result = delete_ready_episode(
+                episode.name,
+                digest,
+                ready_root=ready,
+                deleting_root=deleting,
+            )
+            self.assertEqual(result["state"], "resumed_delete")
+            self.assertFalse(tombstone.exists())
+            self.assertFalse(ledger.exists())
+
+    def test_partial_tombstone_without_ledger_is_refused(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, _base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready)
+            deleting.mkdir()
+            tombstone = deleting / f"{episode.name}.{digest}.deleting"
+            episode.rename(tombstone)
+            (tombstone / "manifest.json").unlink()
+            with self.assertRaises(DeleteSafetyError):
+                delete_ready_episode(
+                    episode.name,
+                    digest,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )
+            self.assertTrue(tombstone.exists())
+
+    def test_wrong_digest_or_symlink_ledger_is_refused(self) -> None:
+        for mode in ("wrong_digest", "symlink"):
+            with self.subTest(mode=mode), tempfile.TemporaryDirectory() as temporary:
+                root = Path(temporary)
+                ready, deleting, base = self._roots(root)
+                episode, digest = self._episode(ready)
+                deleting.mkdir()
+                tombstone = deleting / f"{episode.name}.{digest}.deleting"
+                episode.rename(tombstone)
+                (tombstone / "manifest.json").unlink()
+                if mode == "wrong_digest":
+                    ledger = self._authorization(
+                        base, episode.name, digest, payload_digest="f" * 64
+                    )
+                else:
+                    ledger_root = base / "delete_ledger"
+                    ledger_root.mkdir()
+                    ledger = (
+                        ledger_root
+                        / f"{episode.name}.{digest}.authorized.json"
+                    )
+                    outside = root / "outside"
+                    outside.write_text("keep", encoding="utf-8")
+                    ledger.symlink_to(outside)
+                with self.assertRaises(DeleteSafetyError):
+                    delete_ready_episode(
+                        episode.name,
+                        digest,
+                        ready_root=ready,
+                        deleting_root=deleting,
+                    )
+                self.assertTrue(tombstone.exists())
+                self.assertTrue(ledger.exists())
+
+    def test_ledger_only_after_tree_removal_is_finalized_idempotently(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, base = self._roots(Path(temporary))
+            deleting.mkdir()
+            episode = "episode_20260810T120000000_deadbeef"
+            digest = "a" * 64
+            ledger = self._authorization(base, episode, digest)
+            result = delete_ready_episode(
+                episode,
+                digest,
+                ready_root=ready,
+                deleting_root=deleting,
+            )
+            self.assertEqual(result["state"], "already_absent")
+            self.assertFalse(ledger.exists())
+
+    def test_conflicting_ledger_digest_refuses_ready_target(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready)
+            collision = self._authorization(base, episode.name, "b" * 64)
+            with self.assertRaises(DeleteCollisionError):
+                delete_ready_episode(
+                    episode.name,
+                    digest,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )
+            self.assertTrue(episode.exists())
+            self.assertTrue(collision.exists())
+
+    def test_wrong_manifest_digest_refuses_and_preserves_target(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, _base = self._roots(Path(temporary))
+            episode, _digest = self._episode(ready)
+            with self.assertRaises(DeleteSafetyError):
+                delete_ready_episode(
+                    episode.name,
+                    "0" * 64,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )
+            self.assertTrue(episode.exists())
+
+    def test_noncomplete_manifest_refuses_and_preserves_target(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, _base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready, state="failed")
+            with self.assertRaises(DeleteSafetyError):
+                delete_ready_episode(
+                    episode.name,
+                    digest,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )
+            self.assertTrue(episode.exists())
+
+    def test_unsafe_episode_and_digest_are_rejected(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, _base = self._roots(Path(temporary))
+            for episode, digest in (("../active", "0" * 64), ("safe", "BAD")):
+                with self.subTest(episode=episode, digest=digest):
+                    with self.assertRaises(DeleteSafetyError):
+                        delete_ready_episode(
+                            episode,
+                            digest,
+                            ready_root=ready,
+                            deleting_root=deleting,
+                        )
+
+    def test_symlink_episode_is_refused_without_touching_outside(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            ready, deleting, _base = self._roots(root)
+            outside = root / "outside"
+            outside.mkdir()
+            sentinel = outside / "sentinel"
+            sentinel.write_text("keep", encoding="utf-8")
+            name = "episode_20260810T120000000_deadbeef"
+            (ready / name).symlink_to(outside, target_is_directory=True)
+            with self.assertRaises(DeleteSafetyError):
+                delete_ready_episode(
+                    name,
+                    "0" * 64,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )
+            self.assertEqual(sentinel.read_text(), "keep")
+
+    def test_symlink_manifest_ready_or_nested_entry_is_refused(self) -> None:
+        for target in ("manifest", "ready", "nested"):
+            with self.subTest(target=target), tempfile.TemporaryDirectory() as temporary:
+                root = Path(temporary)
+                ready, deleting, _base = self._roots(root)
+                episode, digest = self._episode(ready)
+                outside = root / "outside"
+                outside.write_text("keep", encoding="utf-8")
+                if target == "manifest":
+                    (episode / "manifest.json").unlink()
+                    (episode / "manifest.json").symlink_to(outside)
+                elif target == "ready":
+                    (episode / "READY").unlink()
+                    (episode / "READY").symlink_to(outside)
+                else:
+                    (episode / "nested_link").symlink_to(outside)
+                with self.assertRaises(DeleteSafetyError):
+                    delete_ready_episode(
+                        episode.name,
+                        digest,
+                        ready_root=ready,
+                        deleting_root=deleting,
+                    )
+                self.assertEqual(outside.read_text(), "keep")
+
+    def test_conflicting_tombstone_refuses_without_moving_ready(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, _base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready)
+            deleting.mkdir()
+            collision = deleting / f"{episode.name}.{'1' * 64}.deleting"
+            collision.mkdir()
+            with self.assertRaises(DeleteCollisionError):
+                delete_ready_episode(
+                    episode.name,
+                    digest,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )
+            self.assertTrue(episode.exists())
+            self.assertTrue(collision.exists())
+
+    def test_ready_and_same_tombstone_collision_refuses_both(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            ready, deleting, _base = self._roots(Path(temporary))
+            episode, digest = self._episode(ready)
+            deleting.mkdir()
+            tombstone = deleting / f"{episode.name}.{digest}.deleting"
+            tombstone.mkdir()
+            with self.assertRaises(DeleteCollisionError):
+                delete_ready_episode(
+                    episode.name,
+                    digest,
+                    ready_root=ready,
+                    deleting_root=deleting,
+                )
+            self.assertTrue(episode.exists())
+            self.assertTrue(tombstone.exists())
+
+    def test_wrong_parent_relationship_is_refused(self) -> None:
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            ready, _deleting, _base = self._roots(root)
+            elsewhere = root / "elsewhere" / "deleting"
+            elsewhere.mkdir(parents=True)
+            with self.assertRaises(DeleteSafetyError):
+                delete_ready_episode(
+                    "episode_safe",
+                    "0" * 64,
+                    ready_root=ready,
+                    deleting_root=elsewhere,
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

verify.sh

@@ -8,6 +8,7 @@ for executable in \
   "$repo_dir/tg3_omnisocket_transport/omnisocket_xtele_sender.py" \
   "$repo_dir/tg3_local_teleop/tg3_local_teleop.py" \
   "$repo_dir/tg3_local_teleop/data_recorder_node.py" \
+  "$repo_dir/tg3_local_teleop/delete_ready_episode.py" \
   "$repo_dir/tg3_local_teleop/run.sh" \
   "$repo_dir/tg3_local_teleop/run_data_recorder.sh" \
   "$repo_dir/tg3_local_teleop/wait_ros_ready.sh" \
@@ -27,6 +28,7 @@ python3 -m py_compile \
   "$repo_dir/tg3_local_teleop/data_collection.py" \
   "$repo_dir/tg3_local_teleop/data_recorder_protocol.py" \
   "$repo_dir/tg3_local_teleop/data_recorder_node.py" \
+  "$repo_dir/tg3_local_teleop/delete_ready_episode.py" \
   "$repo_dir/tg3_data_collection/data_get_sync.py"
 python3 "$repo_dir/tg3_omnisocket_transport/test_session_gate.py"
 python3 "$repo_dir/tg3_local_teleop/test_session_gate.py"
@@ -34,6 +36,7 @@ python3 "$repo_dir/tg3_local_teleop/test_gesture_toggle.py"
 python3 "$repo_dir/tg3_local_teleop/test_idle_session_refresh.py"
 python3 "$repo_dir/tg3_local_teleop/test_data_collection.py"
 python3 "$repo_dir/tg3_local_teleop/test_data_recorder_protocol.py"
+python3 "$repo_dir/tg3_local_teleop/test_delete_ready_episode.py"
 python3 "$repo_dir/tg3_data_collection/test_data_get_sync.py"
 python3 - "$repo_dir/tg3_local_teleop/config.toml" <<'PY'
 import sys
@@ -47,29 +50,39 @@ required = set(section["required_topics"])
 head_rgbd = {
     "/ob_camera_head/color/image_raw/compressed",
     "/ob_camera_head/color/camera_info",
+    "/ob_camera_head/color/metadata",
     "/ob_camera_head/depth/image_raw/compressedDepth",
     "/ob_camera_head/depth/camera_info",
+    "/ob_camera_head/depth/metadata",
 }
 waist_rgbd = {
     "/ob_camera_waist/color/image_raw/compressed",
     "/ob_camera_waist/color/camera_info",
+    "/ob_camera_waist/color/metadata",
     "/ob_camera_waist/depth/image_raw/compressedDepth",
     "/ob_camera_waist/depth/camera_info",
+    "/ob_camera_waist/depth/metadata",
 }
 assert head_rgbd <= topics, "head RGB-D topics must stay in the recording whitelist"
 assert waist_rgbd <= topics, "waist RGB-D topics must stay in the recording whitelist"
-assert waist_rgbd <= required, "live waist RGB-D topics must stay required"
+assert head_rgbd.isdisjoint(required), "head RGB-D must not block recording"
-assert head_rgbd.isdisjoint(required), "inactive head RGB-D must not block recording"
+assert waist_rgbd.isdisjoint(required), "waist RGB-D must not block recording"
-assert {
+assert "/tf_static" in topics and "/tf_static" not in required
-    "/ob_camera_waist/color/metadata",
+assert section.get("minimum_topic_rates_hz", {}) == {}
-    "/ob_camera_waist/depth/metadata",
+groups = section["optional_topic_groups"]
-    "/tf_static",
+assert set(groups) == {"head_rgbd", "waist_rgbd"}
-} <= required
+assert set(groups["head_rgbd"]["topics"]) == head_rgbd
-assert section["minimum_topic_rates_hz"] == {
+assert set(groups["waist_rgbd"]["topics"]) == waist_rgbd
+assert groups["head_rgbd"]["minimum_topic_rates_hz"] == {
+    "/ob_camera_head/color/image_raw/compressed": 20.0,
+    "/ob_camera_head/depth/image_raw/compressedDepth": 20.0,
+}
+assert groups["waist_rgbd"]["minimum_topic_rates_hz"] == {
     "/ob_camera_waist/color/image_raw/compressed": 20.0,
     "/ob_camera_waist/depth/image_raw/compressedDepth": 20.0,
 }
 assert section["minimum_free_gib"] >= 100.0
+assert section["retain_failed_episodes"] is False
 print("config.toml parse passed")
 PY