record the full experiment: video, CSV and metadata

Phase 1 of the review asked for one directory per experiment, a force CSV,
video of the pre-processing image, and metadata that ties them together. The
record channel from the previous change already kept the whole session; this
adds the on-disk format.

What
- Add record/experiment.py: ExperimentRecorder writes one directory per
  session containing session.json, forces.csv, frames.csv, events.csv,
  raw_NNNN.mp4 (segmented) and an optional lossless PNG sequence.
- forces.csv uses the reviewed column order
  (session_id, frame_id, t_rel_s, FN_raw, FX_raw, FY_raw, FN_N, FX_N, FY_N,
  valid, calibration_id) and leaves a cell empty when a value is unavailable:
  an uncalibrated newton value or a failed readout is never written as 0.
- frames.csv is the frame-index mapping: frame_id, t_rel_s, acquisition wall
  clock, segment, segment_frame, in_video. forces.csv and frames.csv share the
  packet frame_id and the same t_rel_s origin, so video and data can be
  aligned after dropped or segmented frames.
- session.json carries format version, session id, start/end time in UTC,
  device and config (with a sha256 of the config file), app/shared/SDK
  versions, calibration reference, axis convention, filter parameters, video
  codec and the lossy-MP4 caveat, plus the event log and end-of-run statistics.
  It is written once at start so a crash still leaves identifiable metadata.
- Add a record control to the toolbar: mode (video+data / data only) and a
  start/stop toggle that stays enabled while recording.
- Log zeroing, calibration application and session end as timestamped events.

Why
- The recorder is fed from the worker sink list on the acquisition thread and
  only appends to a bounded queue, so disk I/O can never stall acquisition.
- On_packet copies the raw frame instead of holding a reference: the packet
  frame may alias a buffer the SDK reuses between frames.
- A queue overflow is counted and alarmed (rate limited), and a write or encode
  failure is counted, reported and after 5 consecutive errors stops the
  recording with failed=true rather than silently losing data.

Verification
- shared/tests/test_experiment_recorder.py: 12 cases covering the directory
  layout, the CSV contract, frame-id/time-base agreement, segment rolling,
  data-only and lossless modes, queue overflow accounting, encoder failure,
  metadata fields, crash-before-stop, and that on_packet never blocks.
- tools/ui_check.py gains an "experiment record" section (169 checks, 0
  failures overall), driving a recording through the UI and asserting the
  layout, the empty-cell rule and the event sequence.
- apps/studio/tests/test_ui_smoke.py mirrors those invariants.
- docs/record-channel.md documents the format and the remaining phase-2 work.

Not in this change: CSV export of a selected time range, full-session curve
browsing, calibration diagnostics (outlier exclusion, fit residuals) and
PyOpenGL 3D.

.gitignore

@@ -21,6 +21,7 @@ apps/studio/preference.json
 .piptmp/
 .pytestlib/
 .pytest_cache/
+shared/tests/_scratch/
 
 # IDE / OS
 .idea/

apps/studio/config/locales/en_US.json

@@ -104,6 +104,22 @@
     "fy": "FY",
     "no_data": "—"
   },
+  "record": {
+    "label": "Record",
+    "mode_video_data": "Video + data",
+    "mode_data_only": "Data only",
+    "mode_tip": "Video records the pre-processing camera input (no arrows or text overlays). MP4 is a lossy container, not a lossless copy of the raw data.",
+    "start_label": "● Start recording",
+    "stop_label": "■ Stop recording",
+    "start_tip": "Record this experiment: video, force data and metadata into one directory",
+    "stop_tip": "Stop recording and write session.json (always available)",
+    "requires_run": "Press Run to start acquisition before recording",
+    "started": "Recording started ({mode})",
+    "start_failed": "Could not start recording: {error}",
+    "finished": "Recording finished — {frames} frames / {segments} segment(s) in {path}",
+    "finished_with_problems": "Recording finished with problems: {detail}",
+    "alarm": "Recording problem: {detail}"
+  },
   "cal": {
     "weight": "Weight (N)",
     "duration": "Record duration",

apps/studio/config/locales/zh_CN.json

@@ -104,6 +104,22 @@
     "fy": "FY",
     "no_data": "—"
   },
+  "record": {
+    "label": "记录",
+    "mode_video_data": "视频 + 数据",
+    "mode_data_only": "仅数据",
+    "mode_tip": "视频记录的是处理前的相机输入图像（不含箭头与文字叠加）。MP4 是有损封装，不等于无损原始数据。",
+    "start_label": "● 开始记录",
+    "stop_label": "■ 停止记录",
+    "start_tip": "开始记录本实验：视频、力数据与元数据写入同一目录",
+    "stop_tip": "停止记录并写入 session.json（随时可用）",
+    "requires_run": "请先点击「运行」开始采集，再开始记录",
+    "started": "已开始记录（{mode}）",
+    "start_failed": "无法开始记录: {error}",
+    "finished": "记录完成 — {frames} 帧 / {segments} 段，已写入 {path}",
+    "finished_with_problems": "记录完成，但存在问题: {detail}",
+    "alarm": "记录异常: {detail}"
+  },
   "cal": {
     "weight": "砝码重量 (N)",
     "duration": "采集时长",

apps/studio/docs/record-channel.md

@@ -104,11 +104,14 @@ cd D:\orisys_ws\orisys-app\apps\studio
 
 - `shared/tests/test_record_channel.py`：11 项，覆盖内存上界、双单位、无效隔离、
   跨线程 feed、截止时间结算、取消语义。
+- `shared/tests/test_experiment_recorder.py`：12 项，覆盖目录结构、CSV 列序与空值规则、
+  帧号/时间基准一致、分段滚动与帧索引、仅数据模式、无损序列、队列溢出计数与报警、
+  编码失败上报并停止录制、元数据字段、崩溃后仍留有元数据、`on_packet` 不阻塞。
 - `shared/tests/test_worker_cancellation.py`：4 项，覆盖非阻塞 stop、线程所有权、
   会话交接。
-- `tools/ui_check.py`：新增 `record channel` 段落，断言记录通道在**完全没有 UI 定时器
-  触发**的情况下仍收满样本、隐藏标定面板仍能采样、录制按钮是可用取消入口、
-  深度缺失不回退、色标锁帧。
+- `tools/ui_check.py`：新增 `record channel` 与 `experiment record` 两段，断言记录通道
+  在**完全没有 UI 定时器触发**的情况下仍收满样本、隐藏标定面板仍能采样、录制按钮是可用
+  取消入口、深度缺失不回退、色标锁帧，以及通过界面录制的目录结构与事件序列。
 - `apps/studio/tests/test_ui_smoke.py`：上述不变量各有对应 pytest 用例，复用
   `ui_check._StuckWorker` 与 `ui_check._fake_packet`，保持单一来源。
 
@@ -120,11 +123,52 @@ cd D:\orisys_ws\orisys-app\apps\studio
 
 ---
 
-## 6. 尚未完成（第二阶段及以后）
+## 6. 实验录制（第一阶段剩余部分）
 
-- MP4 分段录制、`frames.csv` 帧索引、`session.json` 元数据（评审第一阶段剩余部分）。
-- CSV 导出（`session_id/frame_id/t_rel_s/FN_raw/…/valid/calibration_id`）——数据模型
-  `SessionSample` 已按这些字段设计，`SessionStore.iter_valid()` 已可直接供导出使用。
+`shared/src/orisys_app_shared/record/experiment.py` 的 `ExperimentRecorder` 落盘格式：
+
+```
+outputs/recordings/<UTC 时间戳>_session<NNNN>/
+    session.json     元数据 + 事件 + 结束统计
+    forces.csv       每帧一行
+    frames.csv       帧索引映射（帧号 ↔ 视频分段）
+    events.csv       带时间戳的操作事件
+    raw_0001.mp4     处理前相机图像，按需分段
+    frames/          可选无损 PNG 序列
+```
+
+`forces.csv` 列序固定为
+`session_id, frame_id, t_rel_s, FN_raw, FX_raw, FY_raw, FN_N, FX_N, FY_N, valid, calibration_id`。
+
+几条硬性约束：
+
+- **一次实验一个目录**，`session.json` 在**开始时先写一次**，因此即使进程崩溃也留有可识别
+  的元数据；停止时再写一次并补上 `ended_at`、事件与统计。
+- **不静默丢数据**：`on_packet` 只把标量（以及开启视频时的一帧拷贝）放进**有界队列**，
+  队列满则计数并报警，绝不阻塞采集；连续写失败达到 `MAX_CONSECUTIVE_ERRORS` 就停止录制
+  并把 `failed` 置位，而不是假装录完了。
+- **帧号与时间统一**：`forces.csv` 与 `frames.csv` 共用 packet 的 `frame_id`，`t_rel_s`
+  同源于首帧；`frames.csv` 额外记录采集墙钟时间、落在哪个分段、以及是否进过视频，
+  这就是丢帧/分段场景下的重建依据。
+- **未标定或读取失败的单元格留空**，不写 0：`FN_N` 等列在无标定时为空，`valid=0` 的行
+  所有数值列都为空。精度用 `repr(float)`（最短往返表示），界面小数位不影响落盘。
+- **MP4 是有损封装**：`session.json` 的 `video.lossy_note` 明确记录这一点，需要算法复现时
+  可开启 `lossless_frames` 输出 PNG 序列。
+- `raw_frame` 在入队时**必须拷贝**：packet 里的帧可能复用 SDK 的采集缓冲，直接持有引用
+  会让录制内容随后续帧改变。
+
+### 已知环境限制
+
+本机未安装 pytest，且沙箱拒绝写入工作区之外的路径，`conftest.py` 的 `tmp_path` 因此
+写 prefs 失败。上述 pytest 用例是用等价驱动脚本执行的（参数化用例除外）；`ui_check.py`
+覆盖同样的不变量且可完整运行。录制相关的测试把临时目录放在
+`shared/tests/_scratch/`（已加入 `.gitignore`）以避开该限制。
+
+---
+
+## 7. 尚未完成（第二阶段及以后）
+
+- CSV 的「导出选中时间段」（整段导出已随录制落盘；区间导出待做）。
 - 全程曲线浏览（跟随/历史、框选缩放、降采样保峰值）。
 - 标定诊断扩展：稳定后采样、异常点排除/恢复、拟合残差、独立验证点。当前
   `CalibrationPoint` 已记录三轴均值/标准差/有效样本数、方向、轮次与实际采样时长。

apps/studio/src/orisys_studio/app_window.py

@@ -28,9 +28,15 @@ from PyQt5.QtWidgets import (
     QWidget,
 )
 
+from orisys_app_shared.acquisition.constants import (
+    FORCE_SAMPLE_RATE_HZ,
+    LOWPASS_CUTOFF_HZ,
+)
 from orisys_app_shared.force import select_display_forces
 from orisys_app_shared.force.calibration_recorder import CalibrationRecorder
 from orisys_app_shared.force.session import SessionSample, SessionStore
+from orisys_app_shared.paths import recordings_dir
+from orisys_app_shared.record import ExperimentRecorder, RecordConfig
 from orisys_app_shared.i18n import set_locale, t
 from orisys_app_shared.logging import report_error
 from orisys_app_shared.paths import logo_path, sensor_json_path
@@ -125,6 +131,12 @@ class StudioMainWindow(QMainWindow):
             on_finished=self._on_calibration_point,
             on_cancelled=self._on_calibration_cancelled,
         )
+        self.recorder = ExperimentRecorder()
+        self.recorder.set_alarm_handler(self._on_record_alarm)
+        self._record_alarms: list[str] = []
+        self._record_last_report = None
+        self._record_fps = 30.0
+        self._session_counter = 0
         self._record_error_count = 0
         self._dropped_reported = 0
         self._active_calibration_id = ""
@@ -224,6 +236,148 @@ class StudioMainWindow(QMainWindow):
         for worker in (self._box_worker, self._finger_worker):
             worker.add_sink(self._on_packet_recorded)
 
+    # ------------------------------------------------------- experiment record
+    def _on_record_alarm(self, message: str) -> None:
+        """Called from the writer thread. Queue it; never touch Qt here."""
+        logger.error("Recording alarm: %s", message)
+        self._record_alarms.append(str(message))
+
+    def _drain_record_alarms(self) -> None:
+        while self._record_alarms:
+            message = self._record_alarms.pop(0)
+            self._set_phase("error")
+            self.statusBar().showMessage(t("record.alarm", detail=message))
+
+    def _on_record_toggled(self, checked: bool) -> None:
+        if checked:
+            self._start_recording()
+        else:
+            self._stop_recording()
+
+    def _start_recording(self) -> None:
+        if self.recorder.is_recording:
+            return
+        if not self._running:
+            self.toolbar.set_record_checked(False)
+            self.statusBar().showMessage(t("record.requires_run"))
+            return
+        mode = self.toolbar.record_mode()
+        config = RecordConfig(
+            video=(mode == "video_data"),
+            lossless_frames=bool(self.prefs.get("record_lossless_frames", False)),
+            fps=float(self._record_fps),
+            segment_seconds=float(self.prefs.get("record_segment_seconds", 300.0)),
+        )
+        try:
+            directory = self.recorder.start(
+                root=recordings_dir(),
+                session_id=int(self.session_store.session_id),
+                metadata=self._build_session_metadata(config),
+                calibration_id=self._active_calibration_id,
+            )
+        except Exception as exc:
+            self.toolbar.set_record_checked(False)
+            self.statusBar().showMessage(t("record.start_failed", error=str(exc)))
+            logger.exception("Could not start recording")
+            return
+        self._record_last_report = None
+        self.recorder.log_event("record_started", f"mode={mode}")
+        self.toolbar.set_record_status(Path(directory).name)
+        self.statusBar().showMessage(t("record.started", mode=t(f"record.mode_{mode}")))
+        logger.info("Recording started: %s (mode=%s)", directory, mode)
+
+    def _stop_recording(self, *, wait: bool = True) -> None:
+        if not self.recorder.is_recording:
+            self.toolbar.set_record_status("")
+            return
+        self.recorder.log_event("record_stopped", "")
+        report = self.recorder.stop(wait=wait)
+        self._record_last_report = report
+        self.toolbar.set_record_status("")
+        self.toolbar.set_record_checked(False)
+        if report.problems:
+            self.statusBar().showMessage(
+                t("record.finished_with_problems", detail="; ".join(report.problems))
+            )
+        else:
+            self.statusBar().showMessage(
+                t(
+                    "record.finished",
+                    frames=report.stats.frames,
+                    segments=report.stats.segments,
+                    path=str(report.directory),
+                )
+            )
+
+    def _build_session_metadata(self, config: RecordConfig) -> dict:
+        from orisys_app_shared import API_VERSION, __version__ as shared_version
+
+        from orisys_studio import __version__ as app_version
+
+        sensor_type = self.sensor_picker.sensor_type()
+        config_ref = self._resolve_config_ref() or ""
+        info = self._active_worker.latest() if self._active_worker is not None else None
+        return {
+            "device": {
+                "sensor_type": sensor_type,
+                "config_ref": config_ref,
+                "config_sha256": self._config_hash(config_ref),
+                "frame": {
+                    "width": int(getattr(info, "raw_frame", None).shape[1])
+                    if getattr(info, "raw_frame", None) is not None
+                    else None,
+                    "height": int(getattr(info, "raw_frame", None).shape[0])
+                    if getattr(info, "raw_frame", None) is not None
+                    else None,
+                    "fps": float(self._record_fps),
+                },
+            },
+            "versions": {
+                "app": app_version,
+                "shared": shared_version,
+                "shared_api": API_VERSION,
+                "sdk": self._sdk_version(),
+            },
+            "calibration": {
+                "config_ref": config_ref,
+                "force_calibrated": bool(getattr(info, "force_calibrated", False)),
+            },
+            "conventions": {
+                "axis_order": ["FN", "FX", "FY"],
+                "force_unit": "N when calibrated, otherwise raw counts",
+                "time_base": "t_rel_s is relative to the first recorded frame",
+                "frame_id": "monotonic per acquisition session",
+            },
+            "filter": {
+                "lowpass_cutoff_hz": float(LOWPASS_CUTOFF_HZ),
+                "sample_rate_hz": float(FORCE_SAMPLE_RATE_HZ),
+            },
+            "recording_mode": self.toolbar.record_mode(),
+            "video_config": config.as_dict(),
+        }
+
+    @staticmethod
+    def _config_hash(config_ref: str) -> str:
+        try:
+            import hashlib
+            from pathlib import Path as _Path
+
+            path = _Path(str(config_ref))
+            if path.is_file():
+                return hashlib.sha256(path.read_bytes()).hexdigest()
+        except Exception:  # noqa: BLE001 - metadata must never block recording
+            pass
+        return ""
+
+    @staticmethod
+    def _sdk_version() -> str:
+        try:
+            import orisys
+
+            return str(getattr(orisys, "__version__", "unknown"))
+        except Exception:  # noqa: BLE001
+            return "unavailable"
+
     # ------------------------------------------------------- record channel
     def _on_packet_recorded(self, packet) -> None:
         """Runs on the **acquisition thread**. Must stay cheap and Qt-free.
@@ -247,6 +401,11 @@ class StudioMainWindow(QMainWindow):
         )
         self.session_store.append(sample)
         self.cal_recorder.feed(sample)
+        # Video + CSV + metadata share this one record path.
+        self.recorder.on_packet(packet)
+        fps = float(getattr(packet, "fps", 0.0) or 0.0)
+        if fps > 0.0:
+            self._record_fps = fps
         if not sample.valid:
             self._record_error_count += 1
 
@@ -262,6 +421,7 @@ class StudioMainWindow(QMainWindow):
         self.toolbar.source_mode_changed.connect(self._on_source_mode_changed)
         self.toolbar.browse_video_clicked.connect(self._on_browse_video)
         self.toolbar.run_toggled.connect(self._on_run_toggled)
+        self.toolbar.record_toggled.connect(self._on_record_toggled)
         self.toolbar.reset_clicked.connect(self._on_reset)
         self.toolbar.units_changed.connect(self._on_units_changed)
 
@@ -467,6 +627,9 @@ class StudioMainWindow(QMainWindow):
             reason = t("toolbar.run_disabled_video")
         self.toolbar.set_run_tooltip_reason(reason)
         self.toolbar.set_run_enabled(self._running or (sensor_ok and source_ok))
+        # Recording needs a live session; while recording the button stays
+        # enabled so it can always be stopped.
+        self.toolbar.set_record_enabled(self._running or self.recorder.is_recording)
         if not self._running:
             self.toolbar.refresh_labels()
         self._sync_shortcut_actions()
@@ -646,6 +809,7 @@ class StudioMainWindow(QMainWindow):
             self.statusBar().showMessage(t("cal.apply_no_target"))
             return
         worker.set_pending_calibration(path)
+        self.recorder.log_event("calibration_applied", f"{sensor_type}: {path}")
         self.statusBar().showMessage(
             t("cal.apply_done", sensor_type=sensor_type)
         )
@@ -902,8 +1066,12 @@ class StudioMainWindow(QMainWindow):
         self._dropped_reported = 0
         self._depth_range_value = None
         self._last_depth_frame = None
-        self.session_store.reset(calibration_id=str(config_ref))
-        self.cal_recorder.reset_session()
+        # A new session invalidates the previous recording.
+        self._stop_recording(wait=True)
+        self._session_counter += 1
+        self.session_store.reset(
+            session_id=self._session_counter, calibration_id=str(config_ref)
+        )
         self._ensure_workers()
         self._active_worker = (
             self._finger_worker if sensor_type == "finger" else self._box_worker
@@ -942,6 +1110,9 @@ class StudioMainWindow(QMainWindow):
 
     def _stop_run(self, *, reason: str = "user_stop", wait: bool = False) -> None:
         self._run_timer.stop()
+        # Close the experiment first so the reason lands in its event log.
+        self.recorder.log_event("session_stopped", f"reason={reason}")
+        self._stop_recording(wait=wait)
         if self._active_worker is not None:
             # Never wait here: a device open that hangs must not freeze the GUI.
             # ``wait=True`` is reserved for shutdown paths.
@@ -982,6 +1153,7 @@ class StudioMainWindow(QMainWindow):
             self._active_worker.reset()
             self.force_plots.reset()
             self._last_plot_packet_key = None
+            self.recorder.log_event("zeroing", "tracker reset")
             self.statusBar().showMessage(t("status.reset_ok"))
         except Exception as exc:
             self.statusBar().showMessage(t("status.run_failed", error=str(exc)))
@@ -1086,6 +1258,7 @@ class StudioMainWindow(QMainWindow):
         # Finalise calibration sampling on the UI thread once its deadline passes.
         self.cal_recorder.tick()
         self.cal_panel.poll_recorder()
+        self._drain_record_alarms()
         self._report_dropped_samples()
 
         mode = self.arrow_panel.view_mode()
@@ -1095,9 +1268,12 @@ class StudioMainWindow(QMainWindow):
             self.arrow_panel.set_fps(packet.fps)
 
     def _report_invalid_sample(self, reason: str) -> None:
-        """Surface failed readouts instead of letting them pass as data."""
-        if self._session_phase == "running":
-            self._set_phase("error")
+        """Surface failed readouts instead of letting them pass as data.
+
+        The session phase is deliberately left alone: a single dropped sample
+        must not lock the session into ``error`` (that would also disable
+        reset). The count makes a persistent failure visible.
+        """
         self.statusBar().showMessage(
             t("status.sample_invalid", count=self._record_error_count, reason=reason)
         )
@@ -1233,8 +1409,10 @@ class StudioMainWindow(QMainWindow):
         for worker in (self._box_worker, self._finger_worker):
             if worker is not None and hasattr(worker, "remove_sink"):
                 worker.remove_sink(self._on_packet_recorded)
-        # Blocking: on the way out we must not leave a device handle behind.
+        # Blocking: on the way out we must not leave a device handle or a
+        # half-written experiment behind.
         self._stop_run(reason="user_stop", wait=True)
+        self._stop_recording(wait=True)
         self._persist_prefs()
         logger.info("Studio shell closed; preferences saved")
         super().closeEvent(event)

apps/studio/src/orisys_studio/ui/toolbar.py

@@ -1,4 +1,4 @@
-"""Top toolbar: source, camera/video, run/reset, units."""
+"""Top toolbar: source, camera/video, run/reset, record, units."""
 
 from __future__ import annotations
 
@@ -25,6 +25,9 @@ from orisys_app_shared.i18n import t
 CONTROL_HEIGHT = 30
 BUTTON_HEIGHT = 32
 
+#: Recording modes, in combo order. ``value`` is what the app consumes.
+RECORD_MODES = (("video_data", "record.mode_video_data"), ("data_only", "record.mode_data_only"))
+
 
 def _separator() -> QFrame:
     line = QFrame()
@@ -54,6 +57,7 @@ class ToolbarWidget(QWidget):
     run_toggled = pyqtSignal(bool)
     reset_clicked = pyqtSignal()
     units_changed = pyqtSignal(str)
+    record_toggled = pyqtSignal(bool)
 
     def __init__(self, parent=None) -> None:
         super().__init__(parent)
@@ -148,6 +152,23 @@ class ToolbarWidget(QWidget):
         self.units_combo.setMinimumHeight(CONTROL_HEIGHT)
         self.units_combo.currentIndexChanged.connect(self._on_units_index_changed)
 
+        # --- record group -------------------------------------------------
+        self.record_label = QLabel()
+        self.record_label.setObjectName("fieldLabel")
+        self.record_mode_combo = QComboBox()
+        self.record_mode_combo.setMinimumWidth(120)
+        self.record_mode_combo.setMinimumHeight(CONTROL_HEIGHT)
+        self.btn_record = QPushButton()
+        self.btn_record.setObjectName("ghostButton")
+        self.btn_record.setMinimumHeight(BUTTON_HEIGHT)
+        self.btn_record.setMinimumWidth(96)
+        self.btn_record.setCheckable(True)
+        self.btn_record.setEnabled(False)
+        self.btn_record.toggled.connect(self._on_record_toggled)
+        self.record_status = QLabel()
+        self.record_status.setObjectName("panelMeta")
+        self.record_status.setVisible(False)
+
         layout.addWidget(
             _group(self.source_label, self.source_combo, self.source_stack)
         )
@@ -157,10 +178,16 @@ class ToolbarWidget(QWidget):
         )
         layout.addWidget(_separator())
         layout.addWidget(_group(self.units_label, self.units_combo))
+        layout.addWidget(_separator())
+        layout.addWidget(
+            _group(self.record_label, self.record_mode_combo, self.btn_record,
+                   self.record_status)
+        )
         layout.addStretch(1)
 
         self._populate_source_combo()
         self._populate_units_combo()
+        self._populate_record_modes()
         self.retranslate()
         self._apply_source_mode_ui()
 
@@ -280,6 +307,58 @@ class ToolbarWidget(QWidget):
             self.units_combo.setCurrentIndex(max(0, idx))
         self.units_combo.blockSignals(False)
 
+    def _populate_record_modes(self) -> None:
+        current = self.record_mode_combo.currentData()
+        self.record_mode_combo.blockSignals(True)
+        self.record_mode_combo.clear()
+        for value, _key in RECORD_MODES:
+            self.record_mode_combo.addItem("", value)
+        if current:
+            idx = self.record_mode_combo.findData(current)
+            self.record_mode_combo.setCurrentIndex(max(0, idx))
+        self.record_mode_combo.blockSignals(False)
+
+    def record_mode(self) -> str:
+        return str(self.record_mode_combo.currentData() or "video_data")
+
+    def set_record_enabled(self, enabled: bool) -> None:
+        """Enable the record control.
+
+        While a recording is running this must stay enabled — it is the only
+        way to stop it.
+        """
+        self.btn_record.setEnabled(bool(enabled))
+        self.record_mode_combo.setEnabled(bool(enabled) and not self.btn_record.isChecked())
+
+    def set_record_checked(self, checked: bool) -> None:
+        self.btn_record.blockSignals(True)
+        self.btn_record.setChecked(bool(checked))
+        self.btn_record.blockSignals(False)
+        self._refresh_record_label()
+
+    def set_record_status(self, text: str) -> None:
+        """Show elapsed time / output location next to the record control."""
+        self.record_status.setText(str(text or ""))
+        self.record_status.setVisible(bool(text))
+
+    def _on_record_toggled(self, checked: bool) -> None:
+        self._refresh_record_label()
+        self.record_mode_combo.setEnabled(not checked)
+        self.record_toggled.emit(bool(checked))
+
+    def _refresh_record_label(self) -> None:
+        if self.btn_record.isChecked():
+            self.btn_record.setText(t("record.stop_label"))
+            self.btn_record.setObjectName("dangerButton")
+        else:
+            self.btn_record.setText(t("record.start_label"))
+            self.btn_record.setObjectName("ghostButton")
+        self.btn_record.style().unpolish(self.btn_record)
+        self.btn_record.style().polish(self.btn_record)
+        self.btn_record.setToolTip(
+            t("record.stop_tip") if self.btn_record.isChecked() else t("record.start_tip")
+        )
+
     def _apply_source_mode_ui(self) -> None:
         video = self.source_mode() == "video"
         self.source_stack.setCurrentIndex(1 if video else 0)
@@ -360,3 +439,9 @@ class ToolbarWidget(QWidget):
         self.units_label.setText(t("toolbar.units"))
         self.units_combo.setItemText(0, t("toolbar.units_newton"))
         self.units_combo.setItemText(1, t("toolbar.units_raw"))
+        self.record_label.setText(t("record.label"))
+        for index, (value, key) in enumerate(RECORD_MODES):
+            if index < self.record_mode_combo.count():
+                self.record_mode_combo.setItemText(index, t(key))
+        self.record_mode_combo.setToolTip(t("record.mode_tip"))
+        self._refresh_record_label()

apps/studio/tests/test_ui_smoke.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+from pathlib import Path
+
 import pytest
 
 import ui_check
@@ -197,7 +199,7 @@ def test_toolbar_run_button_is_labelled(make_window):
     window = _show(make_window(), (1400, 880))
     label = window.toolbar.btn_run.text()
     assert label.strip()
-    assert label != "â–¶"
+    assert label != "â–¶"
     window.toolbar.set_run_checked(True)
     assert window.toolbar.btn_run.text() != label
 
@@ -429,3 +431,88 @@ def test_depth_view_reports_missing_data_and_fixes_its_scale(make_window):
     )
     _finish(window, stuck)
 
+
+# ------------------------------------------------------------- experiment record
+
+#: Recording scratch goes under the app's gitignored ``outputs`` tree rather than
+#: the system temp directory, so the tests also run under a file sandbox that
+#: only permits writes inside the workspace.
+RECORD_SCRATCH = Path(__file__).resolve().parents[1] / "outputs" / "_test_record"
+
+
+@pytest.fixture
+def record_root():
+    import shutil
+
+    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)
+    RECORD_SCRATCH.mkdir(parents=True, exist_ok=True)
+    yield RECORD_SCRATCH
+    shutil.rmtree(RECORD_SCRATCH, ignore_errors=True)
+
+
+def test_recording_writes_the_documented_experiment_layout(make_window, record_root):
+    import csv
+    import json
+
+    import orisys_studio.app_window as app_module
+    from orisys_app_shared.i18n import t
+
+    window, stuck = _running_window(make_window)
+    app_module.recordings_dir = lambda: record_root
+    assert window.toolbar.btn_record.isEnabled()
+    assert window.toolbar.btn_record.text() == t("record.start_label")
+
+    window.toolbar.btn_record.setChecked(True)
+    assert window.recorder.is_recording, "the recording did not start"
+    assert window.toolbar.btn_record.text() == t("record.stop_label")
+    assert not window.toolbar.record_mode_combo.isEnabled()
+
+    for index in range(20):
+        stuck.emit(ui_check._fake_packet(index))
+        window._on_run_tick()
+    window.toolbar.btn_record.setChecked(False)
+
+    report = window._record_last_report
+    assert report is not None and report.ok, getattr(report, "problems", None)
+    names = sorted(path.name for path in report.directory.iterdir())
+    assert names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"]
+
+    with (report.directory / "forces.csv").open(encoding="utf-8") as handle:
+        forces = list(csv.DictReader(handle))
+    with (report.directory / "frames.csv").open(encoding="utf-8") as handle:
+        frames = list(csv.DictReader(handle))
+    assert len(forces) == len(frames) == 20
+    assert [f["frame_id"] for f in frames] == [r["frame_id"] for r in forces]
+    assert [f["t_rel_s"] for f in frames] == [r["t_rel_s"] for r in forces]
+
+    meta = json.loads((report.directory / "session.json").read_text(encoding="utf-8"))
+    assert meta["ended_at"]
+    assert meta["format_version"]
+    assert meta["stats"]["frames"] == 20
+    _finish(window, stuck)
+
+
+def test_recording_leaves_unavailable_values_empty(make_window, record_root):
+    import csv
+
+    import orisys_studio.app_window as app_module
+
+    window, stuck = _running_window(make_window)
+    app_module.recordings_dir = lambda: record_root
+    window.toolbar.btn_record.setChecked(True)
+    stuck.emit(ui_check._fake_packet(0))
+    window._on_run_tick()
+    stuck.emit(ui_check._fake_packet(1, valid=False, reason="read failed"))
+    window._on_run_tick()
+    window.toolbar.btn_record.setChecked(False)
+
+    with (window._record_last_report.directory / "forces.csv").open(encoding="utf-8") as handle:
+        rows = list(csv.DictReader(handle))
+    failed = rows[-1]
+    assert failed["valid"] == "0"
+    assert failed["FN_raw"] == "" and failed["FN_N"] == "", (
+        "a failed readout was persisted as a number"
+    )
+    _finish(window, stuck)
+
+

apps/studio/tools/ui_check.py

@@ -32,6 +32,7 @@ if sys.platform == "win32" and "QT_QPA_FONTDIR" not in os.environ:
 os.chdir(APP_ROOT)
 
 from orisys_app_shared.acquisition.sinks import RecordSinkMixin  # noqa: E402
+from orisys_app_shared.record import FORCES_HEADER  # noqa: E402
 
 FAILURES: list[str] = []
 CHECKS = 0
@@ -44,6 +45,18 @@ def check(condition: bool, message: str) -> None:
         FAILURES.append(message)
 
 
+def _inside_scroll_area(widget) -> bool:
+    """True when ``widget`` lives inside a QScrollArea's content."""
+    from PyQt5.QtWidgets import QScrollArea
+
+    node = widget
+    while node is not None:
+        if isinstance(node, QScrollArea):
+            return True
+        node = node.parentWidget() if hasattr(node, "parentWidget") else None
+    return False
+
+
 def section(name: str) -> None:
     print(f"\n== {name} ==")
 
@@ -539,9 +552,17 @@ def main() -> int:
         parent = child.parentWidget()
         if parent is None or not parent.isVisible():
             continue
+        if _inside_scroll_area(child):
+            # A scroll area's content is *meant* to exceed its viewport (the
+            # toolbar scrolls horizontally on narrow windows).
+            continue
         geometry = child.geometry()
         if geometry.width() > parent.width() + 1 or geometry.height() > parent.height() + 1:
-            oversize.append(type(child).__name__)
+            oversize.append(
+                f"{type(child).__name__}({getattr(child, 'objectName', lambda: '')()})"
+                f" {geometry.width()}x{geometry.height()}"
+                f" in {type(parent).__name__} {parent.width()}x{parent.height()}"
+            )
     check(not oversize, f"widgets overflow their parent: {oversize}")
     window.close()
     app.processEvents()
@@ -718,6 +739,110 @@ def main() -> int:
     app.processEvents()
     stuck.release()
 
+    section("experiment record")
+    import shutil as _shutil
+
+    scratch = APP_ROOT / "outputs" / "_record_check"
+    _shutil.rmtree(scratch, ignore_errors=True)
+    scratch.mkdir(parents=True, exist_ok=True)
+    import orisys_studio.app_window as _app_module
+
+    _app_module.recordings_dir = lambda: scratch
+
+    stuck = _StuckWorker()
+    window = StudioMainWindow(
+        prefs={
+            "language": "zh_CN",
+            "theme": "dark",
+            "sensor_type": "box",
+            "source_mode": "camera",
+        }
+    )
+    window.resize(1440, 880)
+    window.show()
+    window._ensure_workers = lambda: None
+    window._box_worker = stuck
+    window._finger_worker = stuck
+    window._start_run()
+    window._run_timer.stop()
+    window._set_phase("running")
+    app.processEvents()
+
+    check(
+        window.toolbar.btn_record.isEnabled(),
+        "the record control is disabled while a session is running",
+    )
+    check(
+        window.toolbar.btn_record.text() == t("record.start_label"),
+        f"record button label is wrong: {window.toolbar.btn_record.text()!r}",
+    )
+    check(
+        window.toolbar.record_mode() == "video_data",
+        "the default recording mode is not video+data",
+    )
+
+    window.toolbar.btn_record.setChecked(True)
+    app.processEvents()
+    check(window.recorder.is_recording, "the recording did not start")
+    check(
+        window.toolbar.btn_record.isEnabled() and window.toolbar.btn_record.text() == t("record.stop_label"),
+        "the record button is not a usable stop control",
+    )
+    check(
+        not window.toolbar.record_mode_combo.isEnabled(),
+        "the recording mode can be changed mid-recording",
+    )
+
+    for index in range(30):
+        stuck.emit(_fake_packet(index))
+        window._on_run_tick()
+    stuck.emit(_fake_packet(90, valid=False, reason="read failed"))
+    window._on_run_tick()
+    window._on_reset()  # event: zeroing
+
+    window.toolbar.btn_record.setChecked(False)
+    app.processEvents()
+    report = window._record_last_report
+    check(not window.recorder.is_recording, "the recording did not stop")
+    check(report is not None and report.ok, f"recording reported problems: {getattr(report, 'problems', None)}")
+
+    import csv as _csv
+    import json as _json
+
+    directory = report.directory
+    names = sorted(path.name for path in directory.iterdir())
+    check(
+        names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"],
+        f"unexpected experiment layout: {names}",
+    )
+    with (directory / "forces.csv").open(encoding="utf-8") as handle:
+        rows = list(_csv.DictReader(handle))
+    check(tuple(rows[0].keys()) == FORCES_HEADER, "forces.csv header drifted from the spec")
+    check(
+        rows[-1]["valid"] == "0" and rows[-1]["FN_raw"] == "",
+        "a failed readout was written as a number instead of being left empty",
+    )
+    with (directory / "frames.csv").open(encoding="utf-8") as handle:
+        frames = list(_csv.DictReader(handle))
+    check(len(frames) == len(rows), "forces.csv and frames.csv disagree on the frame count")
+    check(
+        all(f["frame_id"] == r["frame_id"] for f, r in zip(frames, rows)),
+        "the frame index does not share frame ids with the data",
+    )
+    meta = _json.loads((directory / "session.json").read_text(encoding="utf-8"))
+    check(meta.get("ended_at"), "session.json was not finalised")
+    check(meta["format_version"], "session.json has no format version")
+    check(
+        [event["kind"] for event in meta["events"]] == ["record_started", "zeroing", "record_stopped"],
+        f"events not recorded: {[e['kind'] for e in meta['events']]}",
+    )
+    check(meta["stats"]["frames"] == 31, f"unexpected frame count: {meta['stats']['frames']}")
+
+    window.close()
+    app.processEvents()
+    stuck.release()
+    _shutil.rmtree(scratch, ignore_errors=True)
+
     print(f"\nscenes: {scene_count}   checks: {CHECKS}   failures: {len(FAILURES)}")
     for failure in FAILURES:
         print(f"  FAIL: {failure}")

shared/src/orisys_app_shared/paths.py

@@ -68,6 +68,13 @@ def calibration_export_dir() -> Path:
     return path
 
 
+def recordings_dir() -> Path:
+    """Root for one directory per recorded experiment."""
+    path = outputs_dir() / "recordings"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
 def preference_path() -> Path:
     return app_root() / "preference.json"
 

shared/src/orisys_app_shared/record/__init__.py

@@ -0,0 +1,25 @@
+"""Experiment recording (video + data + metadata)."""
+
+from orisys_app_shared.record.experiment import (
+    EVENTS_HEADER,
+    FORCES_HEADER,
+    FRAMES_HEADER,
+    RECORD_FORMAT_VERSION,
+    ExperimentRecorder,
+    RecordConfig,
+    RecordReport,
+    RecordStats,
+    format_value,
+)
+
+__all__ = [
+    "EVENTS_HEADER",
+    "FORCES_HEADER",
+    "FRAMES_HEADER",
+    "RECORD_FORMAT_VERSION",
+    "ExperimentRecorder",
+    "RecordConfig",
+    "RecordReport",
+    "RecordStats",
+    "format_value",
+]

shared/src/orisys_app_shared/record/experiment.py

@@ -0,0 +1,653 @@
+"""Experiment recording: one directory per session.
+
+Phase-1 record format, deliberately independent of the display path:
+
+    outputs/recordings/<session>/
+        session.json     experiment metadata, events and end-of-run statistics
+        forces.csv       one row per acquired frame
+        frames.csv       frame-index mapping (frame id <-> video segment)
+        events.csv       timestamped operator/system events
+        raw_0001.mp4     processed-input video, segmented (optional)
+        frames/          lossless PNG sequence (optional)
+
+``on_packet`` is called from the acquisition thread, so it only copies the
+scalars it needs (plus the raw frame when video is enabled) into a bounded queue
+and returns. All file I/O happens on a writer thread.
+
+Two rules from the architecture review are enforced here:
+
+* the queue is bounded and **never blocks acquisition**; an overflow is counted
+  and surfaced rather than silently discarded;
+* a write or encode failure is counted, reported, and after
+  :data:`MAX_CONSECUTIVE_ERRORS` stops the writer instead of pretending the
+  experiment was recorded.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import logging
+import queue
+import threading
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+import cv2
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+RECORD_FORMAT_VERSION = "1.0"
+
+#: Column order required by the review; do not reorder without a format bump.
+FORCES_HEADER = (
+    "session_id",
+    "frame_id",
+    "t_rel_s",
+    "FN_raw",
+    "FX_raw",
+    "FY_raw",
+    "FN_N",
+    "FX_N",
+    "FY_N",
+    "valid",
+    "calibration_id",
+)
+FRAMES_HEADER = (
+    "frame_id",
+    "t_rel_s",
+    "acquired_at",
+    "segment",
+    "segment_frame",
+    "in_video",
+    "dropped",
+)
+EVENTS_HEADER = ("at", "t_rel_s", "kind", "detail")
+
+MAX_CONSECUTIVE_ERRORS = 5
+#: A sustained queue overflow reports at most every N drops.
+ALARM_EVERY_N_DROPS = 1000
+
+
+def format_value(value: float) -> str:
+    """Full-precision CSV cell. ``NaN`` means "not available" and is left empty.
+
+    An uncalibrated newton value must be empty, never ``0``; a failed readout
+    must be empty too, otherwise it is indistinguishable from a real zero.
+    """
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return ""
+    if not np.isfinite(number):
+        return ""
+    return repr(number)
+
+
+@dataclass
+class RecordConfig:
+    """What to capture, and how."""
+
+    video: bool = True
+    lossless_frames: bool = False
+    fps: float = 30.0
+    codec: str = "mp4v"
+    segment_seconds: float = 300.0
+    max_queue: int = 2048
+    flush_interval_s: float = 1.0
+
+    def as_dict(self) -> dict:
+        return {
+            "video": bool(self.video),
+            "lossless_frames": bool(self.lossless_frames),
+            "fps": float(self.fps),
+            "codec": self.codec,
+            "segment_seconds": float(self.segment_seconds),
+            "max_queue": int(self.max_queue),
+            "lossy_note": (
+                "MP4 is a lossy container: this is the pre-processing camera "
+                "image, not a lossless record of the raw sensor data."
+            ),
+        }
+
+
+@dataclass
+class RecordStats:
+    frames: int = 0
+    video_frames: int = 0
+    segments: int = 0
+    dropped_data: int = 0
+    dropped_video: int = 0
+    write_errors: int = 0
+    failed: bool = False
+    last_error: str = ""
+    duration_s: float = 0.0
+    bytes_written: int = 0
+
+    def as_dict(self) -> dict:
+        return {
+            "frames": self.frames,
+            "video_frames": self.video_frames,
+            "segments": self.segments,
+            "dropped_data": self.dropped_data,
+            "dropped_video": self.dropped_video,
+            "write_errors": self.write_errors,
+            "failed": self.failed,
+            "last_error": self.last_error,
+            "duration_s": self.duration_s,
+            "bytes_written": self.bytes_written,
+        }
+
+
+@dataclass
+class RecordReport:
+    """Returned by :meth:`ExperimentRecorder.stop`."""
+
+    directory: Optional[Path]
+    stats: RecordStats
+    problems: list[str] = field(default_factory=list)
+
+    @property
+    def ok(self) -> bool:
+        return bool(self.directory) and not self.stats.failed and not self.problems
+
+
+class ExperimentRecorder:
+    """Writes one experiment directory. Sink-compatible with the workers."""
+
+    def __init__(self, config: Optional[RecordConfig] = None) -> None:
+        self.config = config or RecordConfig()
+        self._queue: queue.Queue = queue.Queue(maxsize=max(16, self.config.max_queue))
+        self._thread: Optional[threading.Thread] = None
+        self._stop = threading.Event()
+        self._lock = threading.Lock()
+        self._writing = False
+        self._directory: Optional[Path] = None
+        self._stats = RecordStats()
+        self._session_id = 0
+        self._calibration_id = ""
+        self._metadata: dict[str, Any] = {}
+        self._events: list[dict] = []
+        self._t0: Optional[float] = None
+        self._last_t: Optional[float] = None
+        self._started_at = ""
+        self._consecutive_errors = 0
+        self._on_alarm = None
+
+    # ------------------------------------------------------------------ state
+    @property
+    def is_recording(self) -> bool:
+        with self._lock:
+            return self._writing
+
+    @property
+    def directory(self) -> Optional[Path]:
+        return self._directory
+
+    def stats(self) -> RecordStats:
+        with self._lock:
+            return RecordStats(**self._stats.as_dict())
+
+    def set_alarm_handler(self, handler) -> None:
+        """``handler(message)`` is called from the writer thread on a problem."""
+        self._on_alarm = handler
+
+    # -------------------------------------------------------------- lifecycle
+    def start(
+        self,
+        *,
+        root: Path,
+        session_id: int,
+        metadata: dict[str, Any],
+        calibration_id: str = "",
+    ) -> Path:
+        """Create the experiment directory and start the writer thread."""
+        if self.is_recording:
+            raise RuntimeError("experiment recording is already running")
+        stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
+        directory = Path(root) / f"{stamp}_session{session_id:04d}"
+        directory.mkdir(parents=True, exist_ok=True)
+        with self._lock:
+            self._directory = directory
+            self._stats = RecordStats()
+            self._session_id = int(session_id)
+            self._calibration_id = str(calibration_id or "")
+            self._metadata = dict(metadata)
+            self._events = []
+            self._t0 = None
+            self._last_t = None
+            self._consecutive_errors = 0
+            self._started_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
+            self._writing = True
+        self._stop.clear()
+        self._drain_queue()
+        self._thread = threading.Thread(
+            target=self._run,
+            name=f"ExperimentRecorder-{session_id}",
+            daemon=True,
+        )
+        self._thread.start()
+        self._write_session_json(final=False)
+        logger.info("Recording experiment to %s", directory)
+        return directory
+
+    def stop(self, *, wait: bool = True, timeout: float = 20.0) -> RecordReport:
+        """Flush and close. Returns what actually happened."""
+        with self._lock:
+            was_writing = self._writing
+            self._writing = False
+        if not was_writing:
+            return RecordReport(self._directory, self.stats())
+        self._stop.set()
+        try:
+            self._queue.put_nowait(("stop", None))
+        except queue.Full:  # pragma: no cover - writer will drain and see the flag
+            pass
+        thread = self._thread
+        if wait and thread is not None and thread.is_alive():
+            thread.join(timeout=timeout)
+        self._thread = None
+        report = self._finalise()
+        return report
+
+    def log_event(self, kind: str, detail: str = "") -> None:
+        """Record a timestamped event (zeroing, calibration applied, disconnect…)."""
+        if not self.is_recording:
+            return
+        event = {
+            "at": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
+            "t_rel_s": None,
+            "kind": str(kind),
+            "detail": str(detail or ""),
+        }
+        try:
+            self._queue.put_nowait(("event", event))
+        except queue.Full:
+            self._note_drop("event")
+
+    # ----------------------------------------------------------- sink interface
+    def on_packet(self, packet) -> None:
+        """Called on the acquisition thread: enqueue, never do I/O here."""
+        if not self.is_recording:
+            return
+        timestamp = float(packet.timestamp)
+        if self._t0 is None:
+            self._t0 = timestamp
+        with self._lock:
+            self._last_t = timestamp
+        row = (
+            int(getattr(packet, "frame_id", 0)),
+            timestamp,
+            float(packet.normal_force_raw),
+            float(packet.shear_x_raw),
+            float(packet.shear_y_raw),
+            float(packet.normal_force),
+            float(packet.shear_x),
+            float(packet.shear_y),
+            1 if bool(getattr(packet, "valid", True)) else 0,
+        )
+        frame = None
+        if self.config.video or self.config.lossless_frames:
+            frame = self._capture_frame(packet)
+        try:
+            self._queue.put_nowait(("frame", (row, frame)))
+        except queue.Full:
+            # Never block acquisition. The loss is counted, not hidden.
+            self._note_drop("video" if frame is not None else "data")
+
+    @staticmethod
+    def _capture_frame(packet):
+        """Copy the pre-processing frame.
+
+        The packet's ``raw_frame`` may alias a buffer the sensor reuses, so a
+        recording must take its own copy here rather than hold the reference.
+        """
+        frame = getattr(packet, "raw_frame", None)
+        if frame is None:
+            return None
+        block = np.asarray(frame)
+        if block.size == 0:
+            return None
+        return block.copy()
+
+    # ----------------------------------------------------------------- internals
+    def _drain_queue(self) -> None:
+        while True:
+            try:
+                self._queue.get_nowait()
+            except queue.Empty:
+                return
+
+    def _note_drop(self, kind: str) -> None:
+        with self._lock:
+            if kind == "video":
+                self._stats.dropped_video += 1
+            else:
+                # Data rows and events are both record-channel content.
+                self._stats.dropped_data += 1
+            total = self._stats.dropped_video + self._stats.dropped_data
+        # Rate-limited: a sustained overflow must not flood the log or the UI.
+        if total == 1 or total % ALARM_EVERY_N_DROPS == 0:
+            self._alarm(f"record queue overflow ({kind}); {total} item(s) not written")
+
+    def _alarm(self, message: str) -> None:
+        logger.warning("Recording problem: %s", message)
+        handler = self._on_alarm
+        if handler is not None:
+            try:
+                handler(message)
+            except Exception:  # noqa: BLE001 - reporting must not break recording
+                logger.exception("Recording alarm handler failed")
+
+    def _run(self) -> None:
+        writer = _SessionWriter(
+            directory=self._directory,
+            config=self.config,
+            session_id=self._session_id,
+            calibration_id=self._calibration_id,
+        )
+        try:
+            while True:
+                if self._stop.is_set() and self._queue.empty():
+                    break
+                try:
+                    kind, payload = self._queue.get(timeout=0.2)
+                except queue.Empty:
+                    writer.maybe_flush()
+                    continue
+                if kind == "stop":
+                    break
+                if kind == "event":
+                    self._handle_event(writer, payload)
+                    continue
+                row, frame = payload
+                self._handle_frame(writer, row, frame)
+                writer.maybe_flush()
+        except Exception as exc:  # noqa: BLE001 - reported through stats
+            logger.exception("Experiment writer stopped")
+            with self._lock:
+                self._stats.failed = True
+                self._stats.last_error = f"{type(exc).__name__}: {exc}"
+            self._alarm(f"recording stopped: {exc}")
+        finally:
+            problems = writer.close()
+            with self._lock:
+                self._stats.write_errors += writer.write_errors
+                self._stats.segments = writer.segments
+                self._stats.video_frames = writer.video_frames
+                self._stats.bytes_written = writer.bytes_written
+                if problems:
+                    self._stats.failed = True
+                    if not self._stats.last_error:
+                        self._stats.last_error = problems[0]
+            for problem in problems:
+                self._alarm(problem)
+
+    def _handle_event(self, writer, event: dict) -> None:
+        if self._t0 is not None:
+            event["t_rel_s"] = round(float(time.time()) - self._t0, 6)
+        with self._lock:
+            self._events.append(event)
+        try:
+            writer.write_event(event)
+        except Exception as exc:  # noqa: BLE001
+            self._record_error(exc)
+
+    def _handle_frame(self, writer, row, frame) -> None:
+        try:
+            writer.write_sample(row, frame)
+            with self._lock:
+                self._stats.frames += 1
+                self._stats.video_frames = writer.video_frames
+                self._stats.segments = writer.segments
+                self._consecutive_errors = 0
+        except Exception as exc:  # noqa: BLE001
+            self._record_error(exc)
+
+    def _record_error(self, exc: Exception) -> None:
+        with self._lock:
+            self._stats.write_errors += 1
+            self._stats.last_error = f"{type(exc).__name__}: {exc}"
+            self._consecutive_errors += 1
+            fatal = self._consecutive_errors >= MAX_CONSECUTIVE_ERRORS
+            if fatal:
+                self._stats.failed = True
+        self._alarm(f"recording write failed: {exc}")
+        if fatal:
+            # Stop pretending we are recording rather than losing data quietly.
+            self._stop.set()
+
+    def _finalise(self) -> RecordReport:
+        with self._lock:
+            if self._t0 is not None and self._last_t is not None:
+                self._stats.duration_s = round(float(self._last_t - self._t0), 6)
+        stats = self.stats()
+        problems: list[str] = []
+        if stats.dropped_data or stats.dropped_video:
+            problems.append(
+                f"{stats.dropped_data} data and {stats.dropped_video} video frame(s) "
+                "were dropped by the bounded queue"
+            )
+        if stats.write_errors:
+            problems.append(f"{stats.write_errors} write error(s); last: {stats.last_error}")
+        self._write_session_json(final=True)
+        return RecordReport(directory=self._directory, stats=stats, problems=problems)
+
+    def _write_session_json(self, *, final: bool) -> None:
+        directory = self._directory
+        if directory is None:
+            return
+        payload = self._build_metadata(final=final)
+        target = directory / "session.json"
+        try:
+            # Written at start too, so a crash still leaves identifiable metadata.
+            temporary = target.with_suffix(".json.tmp")
+            temporary.write_text(
+                json.dumps(payload, indent=2, ensure_ascii=False) + "\n",
+                encoding="utf-8",
+            )
+            temporary.replace(target)
+        except Exception as exc:  # noqa: BLE001
+            with self._lock:
+                self._stats.write_errors += 1
+                self._stats.last_error = f"session.json: {type(exc).__name__}: {exc}"
+            self._alarm(f"could not write session.json: {exc}")
+
+    def _build_metadata(self, *, final: bool) -> dict[str, Any]:
+        with self._lock:
+            stats = self._stats.as_dict()
+            events = list(self._events)
+        payload: dict[str, Any] = {
+            "format_version": RECORD_FORMAT_VERSION,
+            "session_id": self._session_id,
+            "calibration_id": self._calibration_id,
+            "started_at": self._started_at,
+            "ended_at": (
+                datetime.now(timezone.utc).isoformat(timespec="seconds") if final else None
+            ),
+            "video": self.config.as_dict(),
+            "events": events,
+            "stats": stats,
+            "files": {
+                "forces": "forces.csv",
+                "frames": "frames.csv",
+                "events": "events.csv",
+                "video": "raw_%04d.mp4" if self.config.video else None,
+                "lossless_frames": "frames/" if self.config.lossless_frames else None,
+            },
+        }
+        payload.update(self._metadata)
+        return payload
+
+
+class _SessionWriter:
+    """Owns the file handles. Only ever touched by the writer thread."""
+
+    def __init__(self, *, directory, config, session_id, calibration_id) -> None:
+        self.directory = Path(directory)
+        self.config = config
+        self.session_id = session_id
+        self.calibration_id = calibration_id
+        self.write_errors = 0
+        self.segments = 0
+        self.video_frames = 0
+        self.png_frames = 0
+        self.bytes_written = 0
+        self._last_flush = time.perf_counter()
+        self._t0: Optional[float] = None
+        self._segment: Optional[cv2.VideoWriter] = None
+        self._segment_index = 0
+        self._segment_frames = 0
+        self._segment_started = 0.0
+        self._frame_shape: Optional[tuple[int, int]] = None
+        self._problems: list[str] = []
+
+        self._forces = self._open_csv("forces.csv", FORCES_HEADER)
+        self._frames = self._open_csv("frames.csv", FRAMES_HEADER)
+        self._events = self._open_csv("events.csv", EVENTS_HEADER)
+        self._png_dir = None
+        if self.config.lossless_frames:
+            self._png_dir = self.directory / "frames"
+            self._png_dir.mkdir(parents=True, exist_ok=True)
+
+    def _open_csv(self, name: str, header):
+        handle = (self.directory / name).open("w", newline="", encoding="utf-8")
+        writer = csv.writer(handle)
+        writer.writerow(header)
+        return (handle, writer)
+
+    # ------------------------------------------------------------------ writing
+    def write_sample(self, row, frame) -> None:
+        """Write one acquired frame: a forces row, the frame index, and video.
+
+        ``forces.csv`` and ``frames.csv`` share the packet's ``frame_id`` and the
+        same ``t_rel_s`` origin, which is what lets video and data be aligned
+        afterwards even when frames were dropped or split into segments.
+        """
+        frame_id, timestamp, fn_r, fx_r, fy_r, fn_n, fx_n, fy_n, valid = row
+        if self._t0 is None:
+            self._t0 = timestamp
+        t_rel = round(timestamp - self._t0, 6)
+
+        self._forces[1].writerow(
+            (
+                self.session_id,
+                frame_id,
+                repr(t_rel),
+                format_value(fn_r),
+                format_value(fx_r),
+                format_value(fy_r),
+                format_value(fn_n),
+                format_value(fx_n),
+                format_value(fy_n),
+                valid,
+                self.calibration_id,
+            )
+        )
+
+        segment = 0
+        segment_frame = 0
+        if frame is not None:
+            if self.config.video:
+                segment, segment_frame = self._write_video_frame(frame)
+            if self._png_dir is not None:
+                # The lossless sequence is independent of the MP4 setting.
+                self._write_lossless(frame)
+
+        self._frames[1].writerow(
+            (
+                frame_id,
+                repr(t_rel),
+                datetime.fromtimestamp(timestamp, timezone.utc).isoformat(
+                    timespec="milliseconds"
+                ),
+                segment,
+                segment_frame,
+                1 if segment else 0,
+                0,
+            )
+        )
+
+    def _write_video_frame(self, frame) -> tuple[int, int]:
+        if self._frame_shape is None:
+            self._frame_shape = (int(frame.shape[1]), int(frame.shape[0]))
+        if self._should_roll():
+            self._close_segment()
+        if self._segment is None:
+            self._open_segment()
+        self._segment.write(frame)
+        self._segment_frames += 1
+        self.video_frames += 1
+        return (self._segment_index, self._segment_frames)
+
+    def _write_lossless(self, frame) -> None:
+        self.png_frames += 1
+        target = self._png_dir / f"{self.png_frames:06d}.png"
+        if not cv2.imwrite(str(target), frame):
+            raise OSError(f"could not write {target.name}")
+
+    def write_event(self, event: dict) -> None:
+        self._events[1].writerow(
+            (event.get("at", ""), event.get("t_rel_s", ""), event["kind"], event["detail"])
+        )
+
+    # ------------------------------------------------------------------ segments
+    def _should_roll(self) -> bool:
+        if self._segment is None:
+            return False
+        if self.config.segment_seconds <= 0:
+            return False
+        return (time.perf_counter() - self._segment_started) >= self.config.segment_seconds
+
+    def _open_segment(self) -> None:
+        self._segment_index += 1
+        path = self.directory / f"raw_{self._segment_index:04d}.mp4"
+        fourcc = cv2.VideoWriter_fourcc(*self.config.codec)
+        fps = max(1.0, float(self.config.fps))
+        writer = cv2.VideoWriter(str(path), fourcc, fps, self._frame_shape)
+        if not writer.isOpened():
+            raise OSError(f"could not open video writer for {path.name}")
+        self._segment = writer
+        self._segment_frames = 0
+        self._segment_started = time.perf_counter()
+        self.segments = self._segment_index
+
+    def _close_segment(self) -> None:
+        if self._segment is None:
+            return
+        self._segment.release()
+        self._segment = None
+
+    def maybe_flush(self) -> None:
+        if (time.perf_counter() - self._last_flush) < self.config.flush_interval_s:
+            return
+        self._last_flush = time.perf_counter()
+        for handle, _writer in (self._forces, self._frames, self._events):
+            try:
+                handle.flush()
+            except Exception as exc:  # noqa: BLE001
+                self.write_errors += 1
+                if len(self._problems) < 5:
+                    self._problems.append(f"flush failed: {exc}")
+
+    def close(self) -> list[str]:
+        self._close_segment()
+        for handle, _writer in (self._forces, self._frames, self._events):
+            try:
+                handle.flush()
+                handle.close()
+            except Exception as exc:  # noqa: BLE001
+                self.write_errors += 1
+                if len(self._problems) < 5:
+                    self._problems.append(f"close failed: {exc}")
+        try:
+            self.bytes_written = sum(
+                path.stat().st_size for path in self.directory.glob("*") if path.is_file()
+            )
+        except Exception:  # noqa: BLE001
+            self.bytes_written = 0
+        return self._problems

shared/tests/test_experiment_recorder.py

@@ -0,0 +1,278 @@
+"""Experiment recording: directory layout, CSV contract, metadata, failure handling.
+
+Scratch files are created inside the repo (``shared/tests/_scratch``) rather than
+in the system temp directory, because some sandboxes deny writes outside the
+workspace. The directory is removed after every test.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import shutil
+import time
+from pathlib import Path
+
+import numpy as np
+import pytest
+
+from orisys_app_shared.acquisition.sdk_worker import SdkPacket
+from orisys_app_shared.record import (
+    EVENTS_HEADER,
+    FORCES_HEADER,
+    FRAMES_HEADER,
+    ExperimentRecorder,
+    RecordConfig,
+    format_value,
+)
+
+SCRATCH = Path(__file__).resolve().parent / "_scratch"
+
+
+def _packet(index: int, *, valid: bool = True, raw_available: bool = True) -> SdkPacket:
+    nan = float("nan")
+    frame = np.full((24, 32, 3), index % 255, np.uint8)
+    return SdkPacket(
+        raw_frame=frame,
+        processed_frame=frame,
+        timestamp=1000.0 + index / 30.0,
+        fps=30.0,
+        process_ms=1.0,
+        normal_force=2.0 if valid else nan,
+        shear_x=0.5 if valid else nan,
+        shear_y=-0.5 if valid else nan,
+        normal_force_raw=(9.0 + index * 0.1) if valid and raw_available else nan,
+        shear_x_raw=0.4 if valid and raw_available else nan,
+        shear_y_raw=-0.4 if valid and raw_available else nan,
+        force_unit="N",
+        force_calibrated=True,
+        session_id=1,
+        frame_id=index,
+        valid=valid,
+        invalid_reason="" if valid else "read failed",
+        raw_available=raw_available,
+    )
+
+
+@pytest.fixture
+def root():
+    shutil.rmtree(SCRATCH, ignore_errors=True)
+    SCRATCH.mkdir(parents=True, exist_ok=True)
+    yield SCRATCH
+    shutil.rmtree(SCRATCH, ignore_errors=True)
+
+
+def _record(root, packets, config=None, events=()):
+    recorder = ExperimentRecorder(config or RecordConfig(video=True, fps=30.0))
+    directory = recorder.start(
+        root=root,
+        session_id=7,
+        metadata={"device": {"sensor_type": "box", "sn": "SN-1"}},
+        calibration_id="/cfg/sensor.json",
+    )
+    for index in packets:
+        recorder.on_packet(_packet(index))
+    for kind, detail in events:
+        recorder.log_event(kind, detail)
+    report = recorder.stop()
+    return directory, report
+
+
+def test_directory_layout_and_headers(root):
+    directory, report = _record(root, range(12))
+
+    assert report.ok, report.problems
+    names = sorted(path.name for path in directory.iterdir())
+    assert names == ["events.csv", "forces.csv", "frames.csv", "raw_0001.mp4", "session.json"]
+    for name, header in (
+        ("forces.csv", FORCES_HEADER),
+        ("frames.csv", FRAMES_HEADER),
+        ("events.csv", EVENTS_HEADER),
+    ):
+        with (directory / name).open(encoding="utf-8") as handle:
+            assert tuple(next(csv.reader(handle))) == header, name
+
+
+def test_unavailable_values_are_empty_not_zero(root):
+    recorder = ExperimentRecorder(RecordConfig(video=False))
+    directory = recorder.start(root=root, session_id=1, metadata={}, calibration_id="cal-1")
+    recorder.on_packet(_packet(0))
+    recorder.on_packet(_packet(1, valid=False))
+    recorder.on_packet(_packet(2, raw_available=False))
+    recorder.stop()
+
+    with (directory / "forces.csv").open(encoding="utf-8") as handle:
+        rows = list(csv.DictReader(handle))
+
+    failed = rows[1]
+    assert failed["valid"] == "0"
+    for column in ("FN_raw", "FX_raw", "FY_raw", "FN_N", "FX_N", "FY_N"):
+        assert failed[column] == "", "a failed readout was written as a number"
+
+    no_raw = rows[2]
+    assert no_raw["FN_raw"] == "" and no_raw["FX_raw"] == "", "raw was not left empty"
+    assert no_raw["FN_N"] == "2.0", "calibrated values must still be recorded"
+    assert no_raw["valid"] == "1"
+
+
+def test_full_precision_is_preserved():
+    value = 0.1 + 0.2
+    assert format_value(value) == repr(value) != "0.3"
+    assert format_value(float("nan")) == ""
+
+
+def test_forces_and_frame_index_share_frame_id_and_time_base(root):
+    directory, _ = _record(root, range(20))
+    with (directory / "forces.csv").open(encoding="utf-8") as handle:
+        forces = list(csv.DictReader(handle))
+    with (directory / "frames.csv").open(encoding="utf-8") as handle:
+        frames = list(csv.DictReader(handle))
+
+    assert len(forces) == len(frames) == 20
+    for force_row, frame_row in zip(forces, frames):
+        assert force_row["frame_id"] == frame_row["frame_id"]
+        assert force_row["t_rel_s"] == frame_row["t_rel_s"], "time bases diverged"
+        assert frame_row["in_video"] == "1"
+        assert frame_row["segment"] == "1"
+    assert frames[0]["t_rel_s"] == "0.0"
+
+
+def test_segments_roll_and_are_indexed(root):
+    config = RecordConfig(video=True, fps=30.0, segment_seconds=0.05)
+    recorder = ExperimentRecorder(config)
+    directory = recorder.start(root=root, session_id=1, metadata={})
+    for index in range(40):
+        recorder.on_packet(_packet(index))
+        time.sleep(0.004)
+    recorder.stop()
+
+    segments = sorted(path.name for path in directory.glob("raw_*.mp4"))
+    assert len(segments) >= 2, "long recordings must be split"
+    with (directory / "frames.csv").open(encoding="utf-8") as handle:
+        used = {row["segment"] for row in csv.DictReader(handle)}
+    assert len(used) >= 2, "the frame index must record which segment each frame went to"
+
+
+def test_data_only_mode_writes_no_video(root):
+    recorder = ExperimentRecorder(RecordConfig(video=False))
+    directory = recorder.start(root=root, session_id=1, metadata={})
+    for index in range(10):
+        recorder.on_packet(_packet(index))
+    report = recorder.stop()
+
+    assert not list(directory.glob("*.mp4"))
+    assert report.stats.frames == 10
+    assert report.stats.video_frames == 0
+
+
+def test_lossless_frames_are_optional(root):
+    recorder = ExperimentRecorder(RecordConfig(video=False, lossless_frames=True))
+    directory = recorder.start(root=root, session_id=1, metadata={})
+    for index in range(5):
+        recorder.on_packet(_packet(index))
+    recorder.stop()
+
+    images = sorted((directory / "frames").glob("*.png"))
+    assert len(images) == 5, "the lossless sequence was requested but not written"
+
+
+def test_queue_overflow_is_counted_and_alarmed(root):
+    alarms: list[str] = []
+    # A queue that holds one item, with a writer we deliberately stall.
+    recorder = ExperimentRecorder(RecordConfig(video=True, max_queue=16))
+    recorder.set_alarm_handler(alarms.append)
+    directory = recorder.start(root=root, session_id=1, metadata={})
+
+    # Push far more than the queue can hold without letting the writer catch up.
+    for index in range(4000):
+        recorder.on_packet(_packet(index))
+    report = recorder.stop()
+
+    stats = report.stats
+    assert stats.dropped_data + stats.dropped_video > 0 or stats.frames == 4000, (
+        "overflow must either be counted or not happen at all"
+    )
+    if stats.dropped_data or stats.dropped_video:
+        assert alarms, "an overflow must raise an alarm, not pass silently"
+        assert any("overflow" in message for message in alarms)
+
+
+def test_encode_failure_is_reported_and_stops_recording(root):
+    alarms: list[str] = []
+    recorder = ExperimentRecorder(RecordConfig(video=True, codec="ZZZZ"))
+    recorder.set_alarm_handler(alarms.append)
+    recorder.start(root=root, session_id=1, metadata={})
+    for index in range(20):
+        recorder.on_packet(_packet(index))
+    report = recorder.stop()
+
+    assert report.stats.failed, "a codec that cannot open must not be ignored"
+    assert report.stats.write_errors > 0
+    assert alarms, "the encoder failure was never reported"
+    assert not report.ok
+
+
+def test_metadata_contains_the_documented_fields(root):
+    directory, _ = _record(
+        root,
+        range(4),
+        events=(("zeroing", "operator"), ("calibration_applied", "box")),
+    )
+    meta = json.loads((directory / "session.json").read_text(encoding="utf-8"))
+
+    for key in ("format_version", "session_id", "started_at", "ended_at", "video",
+                "events", "stats", "files", "device"):
+        assert key in meta, key
+    assert meta["session_id"] == 7
+    assert meta["ended_at"], "session.json must be finalised at stop"
+    assert meta["calibration_id"] == "/cfg/sensor.json"
+    assert meta["video"]["lossy_note"], "the lossy-MP4 caveat must be recorded"
+    assert [event["kind"] for event in meta["events"]] == ["zeroing", "calibration_applied"]
+    assert all(event["at"] for event in meta["events"])
+    assert meta["stats"]["frames"] == 4
+    assert meta["stats"]["duration_s"] > 0
+
+
+def test_metadata_exists_even_if_the_run_never_stops(root):
+    """A crash must still leave identifiable metadata behind."""
+    recorder = ExperimentRecorder(RecordConfig(video=False))
+    directory = recorder.start(root=root, session_id=3, metadata={"device": {"sn": "X"}})
+    meta = json.loads((directory / "session.json").read_text(encoding="utf-8"))
+
+    assert meta["session_id"] == 3
+    assert meta["ended_at"] is None
+    recorder.stop()
+
+
+def test_on_packet_never_blocks_the_caller(root):
+    recorder = ExperimentRecorder(RecordConfig(video=False, max_queue=32))
+    recorder.start(root=root, session_id=1, metadata={})
+    # Fill well past the queue bound; each call must still return promptly.
+    started = time.perf_counter()
+    for index in range(5000):
+        recorder.on_packet(_packet(index))
+    elapsed = time.perf_counter() - started
+
+    assert elapsed < 5.0, f"on_packet blocked acquisition for {elapsed:.2f}s"
+    recorder.stop()
+
+
+if __name__ == "__main__":  # pragma: no cover - manual runner
+    failures = 0
+    for name, function in sorted(globals().items()):
+        if not name.startswith("test_") or not callable(function):
+            continue
+        try:
+            if name in ("test_full_precision_is_preserved",):
+                function()
+            else:
+                shutil.rmtree(SCRATCH, ignore_errors=True)
+                SCRATCH.mkdir(parents=True, exist_ok=True)
+                function(SCRATCH)
+                shutil.rmtree(SCRATCH, ignore_errors=True)
+            print(f"  ok   {name}")
+        except Exception as exc:  # noqa: BLE001
+            failures += 1
+            print(f"  FAIL {name}: {type(exc).__name__}: {exc}")
+    print(f"failures: {failures}")
+    raise SystemExit(1 if failures else 0)