Add an offline analysis pass and result cache for replay

Replay re-ran the SDK chain on every seek: 0.48-0.51 wall-seconds per video
second, about 4.8 minutes for a ten-minute recording. The new
orisys_app_shared.analysis package runs that pass once and stores the flow
samples a replay draws - 4,034 B per frame at the shipped finger grid instead of
the chain's 1,176,000 B; the result contract is in
apps/studio/docs/replay-analysis.md.

Verified: shared/tests 416 passed / 1 skipped (15 new) and apps/studio/tests
unchanged at 425 items / 0 failed / 18 skipped; tools/lint_gate.py clean; a
264-frame pass costs 0.5036 wall-s per video-s against 0.5052 for the bare chain
measured the same day.

apps/studio/docs/replay-analysis.md

@@ -0,0 +1,117 @@
+# 回放分析结果（Part A）
+
+本文定义 `orisys_app_shared.analysis` 这个包：一次离线分析 pass 存下什么、怎么编码、
+什么时候作废、播放端怎么读。它是 [record-channel.md](record-channel.md) 所描述的记录格式的
+下游消费者。播放端（时间轴、跳转、单步、倍速）是它的续篇
+[replay-analysis-part-b.md](replay-analysis-part-b.md)：结果里没有的播放语义都在那一篇。
+
+回放过去等于每移动一次光标就重跑一遍 SDK 链路：**每个视频秒 0.48–0.51 秒墙钟**，十分钟录像
+一次约 4.8 分钟。Part A 把这一遍跑一次，只存回放真正要画的东西。
+
+## 1. 结果模型
+
+原始输出每帧 **1,176,000 B**（`flow` 672,000 + `depth_map` 336,000 + `valid_mask` 84,000 +
+灰度 `img` 84,000），30 fps 十分钟 21.2 GB —— 不可缓存。结果里没有这些字段，只有共享选择器
+`select_flow_samples` 选出的样本。
+
+**每帧记录**（`frame_id`、`t_rel_s`、网格 keyframe 编号、候选数、深度统计、`dx`、`dy`、
+`magnitude`、位打包 `active`、掩码摘要）在 shipped finger 网格下是 **4,034 B/帧**：
+`24 + 330×12 + ⌈330/8⌉ + 8`；十分钟 30 fps 约 70 MB。选这个编码而不是实测的另外两种
+（float16 的 2,022 B/帧、丢掉 `magnitude` 的 4,002 B/帧），是因为往返必须**逐位相等**，而
+`magnitude` 正是"同一份结果换个阈值也能用"的依据（第 3 节）。`dx`/`dy`/`magnitude` 存
+float32，与实时选择器比对时后者也要转成 float32 才是精确比较。
+
+**不存的**：力（`forces.csv` 已是图表读的内容，180.9 B/帧）；深度图只存每帧一个
+`depth_nonzero` 计数（4 B）加汇总统计 —— 这段素材上深度几乎全零（264 帧中 241 帧全零，最大
+336 px / 84,000），统计记下来"深度视图是空的"才有解释。
+
+`provenance` 随结果存：算法版本、传感器配置及其内容哈希、标定 id、单位、参考帧 id、首个输出
+帧 id、解码偏移、逐通道的"重算/记录"标记（`flow`、`depth` 重算，`forces` 记录）、`complete`
+标志与计数。
+
+## 2. Keyframe：候选位置与有效掩码
+
+330 个候选**位置**在整段录像上逐字节相同（两个视频、327 帧共用一个摘要），所以它是
+**keyframe**，只存一次（int16 rows + int16 cols 共 1,320 B，规则网格 zlib 后 132 B）；
+有效掩码同样静态（327 帧中 0 帧变化），也存 keyframe（位打包 10,500 B，zlib level 6 后
+687 B）。两者加两个 36 B 的 blob 头，`keyframes.bin` 一共 891 B。
+
+每帧仍带 8 字节掩码摘要：某一帧的掩码若真的变了，读端能**发现**，而不是拿错误的掩码把这一帧
+静默重建出来。测试用 Step 0 记录的 blake2b-16 摘要 `d76349a3b901aea50915eb1308901c2e`
+校验存回来的掩码，所以 keyframe 是被验证过的副本。
+
+box 链路没有 `valid_mask`，`max_arrows=400` 的上限因此按幅值排序，位置集会逐帧移动。格式对此
+只有一条代码路径：**每个位置集都是 keyframe，按摘要去重**。finger 一段录像一个；box 的 33 帧
+录像 26 个，每帧 1,600 B —— 正好等于把位置内联存储的代价。
+
+## 3. 失效键
+
+键是五个分量的摘要（`model.result_key`）：**录像身份、算法版本、传感器配置、标定、参考帧**；
+目录名就是这个摘要。
+
+**显示设置刻意不在键里。** 阈值（`flow_on_px`/`flow_off_px`）变化时读端从存下来的
+`magnitude` 重新推导 `active`，不需要重建结果；只有采样**网格**（`flow_grid_spacing`、
+`flow_edge_margin`、`max_arrows`）变化才会移动候选位置，那份策略记在 provenance 里供比对而
+不塞进键里。SDK 版本同样不在键里（补丁版本不该让键变化），它记在 provenance 的 `versions`，
+由 `AnalysisReader.stale_reasons()` 报出。
+
+## 4. 参考帧规则
+
+`Sensor.reset()` 清空光流跟踪器，所以**重置在指定的那一帧处理之前施加**：参考帧本身就是新参考
+的第一帧，位移恰好为零。实测在真实录像第 36 帧重置后该帧 `|flow|.mean()` = 0.000000000，而
+上一帧是 0.5354；第 36 帧之前的每一帧与不重置的那一遍逐字节相同。
+
+provenance 记 `reference_frame_id`（跟踪器被清空的那一帧）与 `first_output_frame_id`（结果
+第一帧）；没有显式重置时两者相等，所以这个字段永远不是"未知"，`reference_is_explicit` 区分
+两种情况。
+
+帧 id 是采集会话自己的计数器，**不从 0 或 1 开始**：证据里的 box 录像是 20…52。所以请求一个
+录像里不存在的参考帧会**拒绝运行**（`ValueError`，不留下目录），而不是默默按"没有重置"跑完
+—— 那会存下一个与要求不符的 `reference_frame_id`。
+
+## 5. 两个时间基准
+
+**视频时间 ≠ 信号时间。** 容器 33 帧 @30 fps 是 1.100 s，这 33 帧对应的信号是 1.920 s（实测
+16.664 fps）。`t_rel_s` 来自 `frames.csv`，是**信号时间**，也是读端 `frame_at()` 使用的唯一
+时间轴；容器时长永远不派生自它，也不与它互相换算。有序段索引由 `session.json` 的
+`files.video`、`stats.segments`、`prereoll_segment` 加 `RecordingReader.segments()` 的
+`raw_*.mp4` 排序推导；两者不一致时 pass **不修复**：声明了却缺失的段让结果 `complete=false`。
+
+## 6. box 的偏移陷阱
+
+在视频上构造 box `Sensor` 时，拼接探测**悄悄吃掉最前面的 7 个已解码帧**（33 帧视频只剩 26 帧
+可处理，调用方读到的第一帧是视频第 8 帧）；finger 链路吃掉 0 帧。偏移不是猜的：构造后直接读
+底层 capture 的 `CAP_PROP_POS_FRAMES`（box = 7，finger = 0），记进 provenance 的
+`decode_order_offsets`。因此**帧身份只来自录像自己的 `frames.csv`**：可读帧 d 对应 `in_video`
+行 `d + offset`，`frame_id`、`t_rel_s`、`segment` 全部取自该行。后备项（后端报不出位置时按常量
+7）把 `decode_order_offset_source` 记成 `declared`。计数把"少掉的帧"分开记：`skipped`（被偏移
+吃掉的索引行）、`unmapped`（解码出来却没有索引行）、`frames_short`（有索引行却没有解码帧）。
+
+**构造 box `Sensor` 还会往进程当前目录写约 2 MB 的 `config/compound_eye.png`**，所以每次开
+链路都在临时工作目录里、由进程级锁串行化：这是本设计唯一无法消除的全局副作用，SDK 没有提供
+改写这个路径的入口。
+
+## 7. 完整性与淘汰
+
+结果目录里有一个显式 `INCOMPLETE` 标记，从写入第一个字节起存在，直到元数据声明完成之后才
+删除。默认读端只接受同时满足"元数据 `complete=true`"、"provenance 也 `complete=true`"、
+"没有 `INCOMPLETE` 标记"的目录，三者在读取时互相校验，不一致直接报格式错误。
+
+取消、SDK 失败、任意异常走同一条路：写完已完成的块与 keyframe，写上 `complete=false` 和停止
+原因，保留标记 —— 被中断的一遍是**可查看的**（`allow_incomplete=True` 能读到已写完的帧），但
+绝不会被默认当作有效结果。一帧的链路调用失败就结束整遍，而不是把过期或空测量写进缓存。
+
+写入分块：帧先攒在内存里，满了整块落到 `frames-*.bin.part` 再原子改名，内存上限是
+`chunk_frames × 每帧记录大小`（256 帧约 1.0 MB）。淘汰按明确的容量上限整份删除结果（默认
+1 GiB），先删不可用的未完成结果，再按时间从旧到新，永不触碰调用方 `keep` 的那一份。
+
+## 8. 读端
+
+`AnalysisReader` 按帧 id 和按时间戳随机访问：`meta.json` 的块清单定位块，块尾偏移表定位帧，
+一次读一帧记录。seek 不需要重跑 pass，也不需要录像、SDK 或 Qt —— 测试把结果目录整体复制到
+别处再读，逐帧比对相同。这一遍不比裸链路贵：264 帧 / 8.8 s 上三次计时均值 0.5036 墙钟秒每
+视频秒，同一天用 Step 0 自己的 `bench_chain.py` 重测裸链路为 0.5052。
+
+两个建议给 SDK（本文档不实施）：给外部帧模式一个公开的逐帧读取方法，替代
+`Sensor._read_frame`；给拼接预览路径一个可配置目标，替代相对当前目录的
+`config/compound_eye.png`。

shared/src/orisys_app_shared/analysis/__init__.py

@@ -0,0 +1,27 @@
+"""Offline analysis of a recording: run the chain once, read the result many times.
+
+Why this exists: replaying a recording used to mean re-running the SDK chain every time
+the operator moved the cursor, which costs 0.48–0.51 wall-seconds per video-second —
+about 4.8 minutes for a ten-minute recording, every time. Part A runs that pass **once**
+and stores the flow samples a replay draws, so seeking is a file read.
+
+What is stored is deliberately *not* the chain's output. One frame of raw output is
+1,176,000 B (``flow`` [300,280,2] float32, ``depth_map``, ``valid_mask``, the gray
+image), which is 21.2 GB for ten minutes at 30 fps. The selected samples cost 4,034 B
+per frame. The rest is recomputed — that is what makes the cache a cache.
+
+The three modules have one job each:
+
+``model``
+    The result and its byte encoding: the key, the provenance, the frame record.
+``cache``
+    Writing it without ever leaving a partial result that looks finished, and evicting
+    whole results against a size cap.
+``reader``
+    Serving it: by frame id, by timestamp, without replaying anything.
+``service``
+    The pass itself — the only place that imports the SDK.
+
+Nothing here imports Qt, so the pass and the reader are testable headless, and a
+caller that only reads a stored result never loads the SDK at all.
+"""

shared/src/orisys_app_shared/analysis/cache.py

@@ -0,0 +1,624 @@
+"""Writing and evicting analysis results: chunked, atomic, and never half-served.
+
+A pass over a ten-minute recording produces tens of thousands of frame records. Two
+properties matter more than speed:
+
+* **A reader must never see a partial result as a finished one.** Every file is built
+  under a temporary name and moved into place with :func:`os.replace`, the metadata
+  that declares the result complete is written *last*, and an explicit ``INCOMPLETE``
+  marker exists from the first byte written until that move. A cancelled or crashed
+  pass leaves a directory that refuses to open by default.
+* **Memory must not grow with the recording.** Frames accumulate into a fixed-size
+  chunk and the chunk is written once. The bound is
+  ``chunk_frames * frame_payload_bytes(max candidates)`` — 256 frames of the shipped
+  finger grid is about 1.0 MB — and ``chunk_frames`` is clamped so a caller cannot
+  raise it without bound.
+
+Storage is not free, so :func:`evict` removes whole results against a stated size cap.
+It only ever deletes a direct child of the cache root that carries a ``meta.json``, so
+a wrong path cannot turn into a recursive delete of something else.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import shutil
+import struct
+import time
+import zlib
+from collections.abc import Iterable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from orisys_app_shared.analysis.model import (
+    ANALYSIS_FORMAT,
+    ANALYSIS_FORMAT_VERSION,
+    CHUNK_HEADER,
+    CHUNK_MAGIC,
+    CHUNK_TRAILER,
+    CODEC_RAW,
+    CODEC_ZLIB,
+    KEYFRAME_GRID,
+    KEYFRAME_HEADER,
+    KEYFRAME_MAGIC,
+    KEYFRAME_MASK,
+    RESULT_DIRNAME,
+    KeyframeRef,
+    Provenance,
+    ResultKey,
+    digest8,
+    keyframe_payload,
+)
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "DEFAULT_MAX_BYTES",
+    "INCOMPLETE_MARKER",
+    "META_NAME",
+    "AnalysisWriter",
+    "ResultCache",
+    "ResultSummary",
+    "evict",
+    "scan_results",
+]
+
+#: The file that declares a finished result. Its presence is not enough — the
+#: provenance inside must also say ``complete``.
+META_NAME = "meta.json"
+
+#: Written before the first data byte and removed after the last. A reader refuses a
+#: directory that has one, whatever the metadata claims.
+INCOMPLETE_MARKER = "INCOMPLETE"
+
+CHUNK_PREFIX = "frames-"
+CHUNK_SUFFIX = ".bin"
+KEYFRAMES_NAME = "keyframes.bin"
+
+DEFAULT_CHUNK_FRAMES = 256
+
+#: A chunk is built in memory, so this is a real memory bound, not a tuning knob.
+MAX_CHUNK_FRAMES = 4096
+
+#: Stated cap for :func:`evict`. A ten-minute finger recording is about 70 MB, so this
+#: holds roughly fourteen of them; the operator can lower it per call.
+DEFAULT_MAX_BYTES = 1 << 30
+
+#: Above this many remembered grid digests the writer stops deduplicating and gives a
+#: deviating frame its own keyframe. Correct either way; the map exists to keep the
+#: common case (one grid for the whole recording) at one keyframe.
+MAX_GRID_DEDUP = 4096
+
+_KIND_NAMES = {KEYFRAME_GRID: "grid", KEYFRAME_MASK: "mask"}
+_KIND_CODES = {name: code for code, name in _KIND_NAMES.items()}
+_CODEC_NAMES = {CODEC_RAW: "raw", CODEC_ZLIB: "zlib"}
+_CODEC_CODES = {name: code for code, name in _CODEC_NAMES.items()}
+
+
+def _atomic_write_json(path: Path, payload: Any) -> None:
+    """Write JSON through a temporary sibling, so a reader sees all of it or none."""
+    text = json.dumps(
+        payload, indent=2, sort_keys=False, ensure_ascii=False, default=str
+    )
+    temporary = path.with_name(path.name + ".part")
+    with temporary.open("w", encoding="utf-8", newline="\n") as handle:
+        handle.write(text)
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(temporary, path)
+
+
+def _atomic_replace_file(path: Path, data: bytes | bytearray) -> None:
+    temporary = path.with_name(path.name + ".part")
+    with temporary.open("wb") as handle:
+        handle.write(data)
+        handle.flush()
+        os.fsync(handle.fileno())
+    os.replace(temporary, path)
+
+
+@dataclass(frozen=True)
+class ResultSummary:
+    """One result directory as :func:`scan_results` found it."""
+
+    directory: Path
+    key: str
+    complete: bool
+    bytes_used: int
+    modified: float
+    stopped_reason: str = ""
+
+
+def _directory_bytes(directory: Path) -> int:
+    total = 0
+    for path in directory.rglob("*"):
+        try:
+            if path.is_file():
+                total += path.stat().st_size
+        except OSError:
+            # A file that vanished between listing and sizing is not a reason to stop
+            # measuring the rest of the directory.
+            logger.debug("could not size %s", path, exc_info=True)
+    return total
+
+
+def scan_results(root: str | Path) -> list[ResultSummary]:
+    """Every result directly under ``root``, newest last. Unreadable entries are listed
+    as incomplete rather than skipped: a directory nobody can read is worth seeing."""
+    base = Path(root)
+    found: list[ResultSummary] = []
+    if not base.is_dir():
+        return found
+    for child in sorted(base.iterdir()):
+        if not child.is_dir():
+            continue
+        meta_path = child / META_NAME
+        complete = False
+        key = child.name
+        reason = ""
+        modified = 0.0
+        if meta_path.is_file():
+            try:
+                meta = json.loads(meta_path.read_text(encoding="utf-8"))
+                complete = bool(meta.get("complete"))
+                key = str(meta.get("key") or child.name)
+                reason = str(meta.get("stopped_reason") or "")
+                modified = meta_path.stat().st_mtime
+            except (OSError, ValueError):
+                logger.warning(
+                    "unreadable result metadata at %s", meta_path, exc_info=True
+                )
+        if not modified:
+            try:
+                modified = child.stat().st_mtime
+            except OSError:
+                modified = 0.0
+        if (child / INCOMPLETE_MARKER).exists():
+            complete = False
+        found.append(
+            ResultSummary(
+                directory=child,
+                key=key,
+                complete=complete,
+                bytes_used=_directory_bytes(child),
+                modified=modified,
+                stopped_reason=reason,
+            )
+        )
+    return found
+
+
+def evict(
+    root: str | Path,
+    *,
+    max_bytes: int = DEFAULT_MAX_BYTES,
+    keep: Iterable[str | Path] = (),
+    dry_run: bool = False,
+) -> list[Path]:
+    """Remove whole results under ``root`` until the total fits ``max_bytes``.
+
+    Incomplete results go first — they are already unreadable by default, so they are
+    the cheapest thing to lose — and then complete ones oldest first. Any directory in
+    ``keep`` is never touched, which is how a caller protects the result it is using.
+
+    Only a direct child of ``root`` carrying a ``meta.json`` is a candidate, so a bad
+    argument cannot delete an arbitrary tree.
+    """
+    base = Path(root).resolve()
+    protected = {Path(item).resolve() for item in keep}
+    limit = max(0, int(max_bytes))
+    summaries = scan_results(base)
+    total = sum(item.bytes_used for item in summaries)
+    removed: list[Path] = []
+    order = sorted(
+        (item for item in summaries if item.directory.resolve() not in protected),
+        key=lambda item: (item.complete, item.modified),
+    )
+    for item in order:
+        if total <= limit:
+            break
+        if not (item.directory / META_NAME).is_file():
+            continue
+        if item.directory.parent.resolve() != base:
+            continue
+        removed.append(item.directory)
+        total -= item.bytes_used
+        if not dry_run:
+            shutil.rmtree(item.directory, ignore_errors=False)
+    if removed:
+        logger.info(
+            "analysis cache: %s %d result(s), %d bytes used against a %d byte cap",
+            "would remove" if dry_run else "removed",
+            len(removed),
+            total,
+            limit,
+        )
+    return removed
+
+
+class AnalysisWriter:
+    """Append-only writer for one result directory.
+
+    Usage is a fixed order — ``begin``, any number of keyframes and frames, then
+    either ``commit`` or ``abandon`` — and every exit path leaves the directory
+    honest: ``commit`` is the only thing that produces a result a reader will serve.
+    """
+
+    def __init__(
+        self,
+        directory: str | Path,
+        *,
+        key: ResultKey,
+        chunk_frames: int = DEFAULT_CHUNK_FRAMES,
+    ) -> None:
+        self.directory = Path(directory)
+        self.key = key
+        self.chunk_frames = max(1, min(int(chunk_frames), MAX_CHUNK_FRAMES))
+        self._keyframe_file = None
+        self._keyframe_offset = 0
+        self._keyframes: list[KeyframeRef] = []
+        self._grid_ids: dict[bytes, int] = {}
+        self._mask_id: int | None = None
+        self._next_keyframe_id = 0
+        self._chunk_buffer = bytearray()
+        self._chunk_offsets: list[int] = []
+        self._chunk_ids: list[int] = []
+        self._chunk_times: list[float] = []
+        self._chunk_index = 0
+        self._chunks: list[dict[str, Any]] = []
+        self._frame_count = 0
+        self.bytes_written = 0
+        self._committed = False
+        self._abandoned = False
+        self._started_at = ""
+
+    # ------------------------------------------------------------------- lifecycle
+    def begin(self) -> None:
+        """Claim the directory: drop an older result, then mark this one incomplete.
+
+        The marker is written *before* the old data is removed, so an interruption
+        between the two steps still leaves a directory that will not be served.
+        """
+        self.directory.mkdir(parents=True, exist_ok=True)
+        self._started_at = time.strftime("%Y-%m-%dT%H:%M:%S%z")
+        self._write_marker("running")
+        for stale in list(self.directory.glob(CHUNK_PREFIX + "*" + CHUNK_SUFFIX)):
+            stale.unlink()
+        for stale in list(self.directory.glob("*.part")):
+            stale.unlink()
+        keyframes = self.directory / KEYFRAMES_NAME
+        if keyframes.exists():
+            keyframes.unlink()
+        self._keyframe_file = (self.directory / (KEYFRAMES_NAME + ".part")).open("wb")
+        self._write_meta(complete=False, provenance=None, counts={}, reason="running")
+
+    def _write_marker(self, reason: str) -> None:
+        _atomic_write_json(
+            self.directory / INCOMPLETE_MARKER,
+            {"key": self.key.digest, "started_at": self._started_at, "reason": reason},
+        )
+
+    def _write_meta(
+        self,
+        *,
+        complete: bool,
+        provenance: Provenance | None,
+        counts: dict[str, Any],
+        reason: str,
+    ) -> None:
+        record = dict(provenance.to_dict()) if provenance is not None else {}
+        record["complete"] = bool(complete)
+        record["stopped_reason"] = str(reason)
+        _atomic_write_json(
+            self.directory / META_NAME,
+            {
+                "format": ANALYSIS_FORMAT,
+                "format_version": ANALYSIS_FORMAT_VERSION,
+                "key": self.key.digest,
+                "key_components": self.key.components,
+                "complete": bool(complete),
+                "stopped_reason": str(reason),
+                "provenance": record,
+                "keyframes": [item.to_dict() for item in self._keyframes],
+                "chunks": list(self._chunks),
+                "counts": dict(counts),
+            },
+        )
+
+    def commit(self, *, provenance: Provenance, counts: dict[str, Any]) -> Path:
+        """Finish the result: close the chunk and keyframes, then declare it complete.
+
+        The completion flag is written before the marker is removed, so an interruption
+        in between leaves a *complete-looking* directory that still refuses to open.
+        Failing closed is the only acceptable direction.
+        """
+        if self._committed:
+            raise RuntimeError("this result was already committed")
+        if self._abandoned:
+            raise RuntimeError("this result was abandoned and cannot be committed")
+        self._flush_chunk()
+        if self._keyframe_file is not None:
+            self._keyframe_file.flush()
+            os.fsync(self._keyframe_file.fileno())
+            self._keyframe_file.close()
+            self._keyframe_file = None
+            part = self.directory / (KEYFRAMES_NAME + ".part")
+            if part.exists():
+                os.replace(part, self.directory / KEYFRAMES_NAME)
+        self._write_meta(complete=True, provenance=provenance, counts=counts, reason="")
+        marker = self.directory / INCOMPLETE_MARKER
+        if marker.exists():
+            marker.unlink()
+        self._committed = True
+        return self.directory
+
+    def abandon(self, reason: str, *, provenance: Provenance | None = None) -> Path:
+        """Stop without a result. Whatever was written stays, and stays unusable.
+
+        The buffered chunk and the open keyframes are closed into real files rather than
+        dropped, so an interrupted pass is *inspectable* (a caller passing
+        ``allow_incomplete=True`` can read the frames that were finished) while still
+        being refused by every default reader. Keeping the frames costs one chunk and
+        buys the only evidence of how far the pass got; discarding them would leave a
+        directory whose metadata claims five frames and whose files hold none.
+
+        The provenance is kept when the caller has one: a cancelled pass still has to say
+        which input it was working on.
+        """
+        if self._committed:
+            raise RuntimeError("this result was already committed")
+        self._abandoned = True
+        try:
+            self._flush_chunk()
+        except OSError:
+            logger.warning(
+                "could not write the partial chunk of a stopped pass", exc_info=True
+            )
+        if self._keyframe_file is not None:
+            try:
+                self._keyframe_file.flush()
+                os.fsync(self._keyframe_file.fileno())
+                self._keyframe_file.close()
+            except OSError:
+                logger.debug("could not close the keyframe file", exc_info=True)
+            self._keyframe_file = None
+            part = self.directory / (KEYFRAMES_NAME + ".part")
+            try:
+                if part.exists():
+                    os.replace(part, self.directory / KEYFRAMES_NAME)
+            except OSError:
+                logger.debug("could not publish the partial keyframes", exc_info=True)
+        if self.directory.is_dir():
+            self._write_marker(reason or "failed")
+            counts = {"frames_written": self._frame_count}
+            if provenance is not None:
+                counts = dict(provenance.counts)
+            self._write_meta(
+                complete=False,
+                provenance=provenance,
+                counts=counts,
+                reason=reason or "failed",
+            )
+        return self.directory
+
+    def __enter__(self) -> AnalysisWriter:
+        self.begin()
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> bool:
+        if exc_type is not None and not self._committed:
+            reason = f"{exc_type.__name__}: {exc}" if exc else "failed"
+            self.abandon(reason)
+        return False
+
+    # ------------------------------------------------------------------- keyframes
+    def _append_keyframe(
+        self, kind: int, payload: bytes, *, count: int, height: int, width: int
+    ) -> int:
+        digest = digest8(payload)
+        codec = CODEC_ZLIB
+        stored = zlib.compress(payload, 6)
+        if len(stored) >= len(payload):
+            codec = CODEC_RAW
+            stored = payload
+        identifier = self._next_keyframe_id
+        self._next_keyframe_id += 1
+        header = KEYFRAME_HEADER.pack(
+            KEYFRAME_MAGIC,
+            kind,
+            codec,
+            int(count),
+            int(height),
+            int(width),
+            0,
+            0,
+            len(payload),
+            digest,
+        )
+        if self._keyframe_file is None:
+            raise RuntimeError("the writer is not open; call begin() first")
+        self._keyframe_file.write(header)
+        self._keyframe_file.write(stored)
+        shape = (int(count),) if kind == KEYFRAME_GRID else (int(height), int(width))
+        self._keyframes.append(
+            KeyframeRef(
+                id=identifier,
+                kind=_KIND_NAMES[kind],
+                digest=digest.hex(),
+                shape=shape,
+                codec=_CODEC_NAMES[codec],
+                raw_bytes=len(payload),
+                stored_bytes=len(stored),
+                offset=self._keyframe_offset,
+            )
+        )
+        self._keyframe_offset += len(header) + len(stored)
+        self.bytes_written += len(header) + len(stored)
+        return identifier
+
+    def grid_keyframe(self, rows: np.ndarray, cols: np.ndarray) -> int:
+        """The candidate location set as a keyframe, reused when it repeats.
+
+        The finger chain produces one of these for a whole recording because its
+        validity mask is static; a chain whose candidate set moves gets one per
+        distinct set, which is the same cost as storing the locations inline.
+        """
+        rows = np.ascontiguousarray(rows, dtype="<i2").reshape(-1)
+        cols = np.ascontiguousarray(cols, dtype="<i2").reshape(-1)
+        if rows.size != cols.size:
+            raise ValueError(f"rows/cols length mismatch: {rows.size} vs {cols.size}")
+        payload = keyframe_payload(KEYFRAME_GRID, np.concatenate([rows, cols]))
+        digest = digest8(payload)
+        known = self._grid_ids.get(digest)
+        if known is not None:
+            return known
+        if len(self._grid_ids) >= MAX_GRID_DEDUP:
+            for stale in list(self._grid_ids)[: MAX_GRID_DEDUP // 2]:
+                del self._grid_ids[stale]
+        identifier = self._append_keyframe(
+            KEYFRAME_GRID, payload, count=int(rows.size), height=0, width=0
+        )
+        self._grid_ids[digest] = identifier
+        return identifier
+
+    def mask_keyframe(self, mask: np.ndarray, digest: bytes) -> int:
+        """The validity mask as a keyframe. One per result: it is static when it is not."""
+        array = np.asarray(mask, dtype=bool)
+        if array.ndim != 2:
+            raise ValueError(f"a validity mask is 2-D, got shape {array.shape}")
+        payload = keyframe_payload(KEYFRAME_MASK, array)
+        stored_digest = digest8(payload)
+        if stored_digest != digest:
+            raise ValueError("the mask digest does not describe the stored mask")
+        if self._mask_id is None:
+            self._mask_id = self._append_keyframe(
+                KEYFRAME_MASK,
+                payload,
+                count=0,
+                height=int(array.shape[0]),
+                width=int(array.shape[1]),
+            )
+        return self._mask_id
+
+    # ---------------------------------------------------------------------- frames
+    def add_frame(self, record: bytes, *, frame_id: int, t_rel_s: float) -> None:
+        """Append one encoded frame. The chunk is written when it is full."""
+        if self._committed or self._abandoned:
+            raise RuntimeError("this result is closed")
+        # Absolute from the start of the file, like the offset table below it: a reader
+        # seeks once and reads one record, with no header arithmetic in between.
+        self._chunk_offsets.append(CHUNK_HEADER.size + len(self._chunk_buffer))
+        self._chunk_ids.append(int(frame_id))
+        self._chunk_times.append(float(t_rel_s))
+        self._chunk_buffer += record
+        self._frame_count += 1
+        if len(self._chunk_ids) >= self.chunk_frames:
+            self._flush_chunk()
+
+    def _flush_chunk(self) -> None:
+        if not self._chunk_ids:
+            return
+        offsets = struct.pack(f"<{len(self._chunk_offsets)}Q", *self._chunk_offsets)
+        ids = struct.pack(f"<{len(self._chunk_ids)}i", *self._chunk_ids)
+        times = struct.pack(f"<{len(self._chunk_times)}d", *self._chunk_times)
+        trailer = offsets + ids + times
+        # Absolute, from the start of the file: the reader seeks to it directly.
+        offset_table_offset = CHUNK_HEADER.size + len(self._chunk_buffer)
+        body = bytes(self._chunk_buffer)
+        header = CHUNK_HEADER.pack(
+            CHUNK_MAGIC,
+            ANALYSIS_FORMAT_VERSION,
+            0,
+            self._chunk_index,
+            len(self._chunk_ids),
+            CHUNK_HEADER.size,
+        )
+        payload = (
+            header
+            + body
+            + trailer
+            + CHUNK_TRAILER.pack(offset_table_offset, len(self._chunk_ids), CHUNK_MAGIC)
+        )
+        name = f"{CHUNK_PREFIX}{self._chunk_index:04d}{CHUNK_SUFFIX}"
+        _atomic_replace_file(self.directory / name, payload)
+        self._chunks.append(
+            {
+                "index": self._chunk_index,
+                "file": name,
+                "frames": len(self._chunk_ids),
+                "first_frame_id": int(self._chunk_ids[0]),
+                "last_frame_id": int(self._chunk_ids[-1]),
+                "bytes": len(payload),
+            }
+        )
+        self.bytes_written += len(payload)
+        self._chunk_index += 1
+        self._chunk_buffer = bytearray()
+        self._chunk_offsets = []
+        self._chunk_ids = []
+        self._chunk_times = []
+
+    @property
+    def frames_written(self) -> int:
+        return self._frame_count
+
+
+class ResultCache:
+    """The results belonging to one recording, addressed by key.
+
+    ``root`` is usually ``<recording>/analysis``: a result then travels with the
+    recording it describes, and deleting the recording deletes its cache. Nothing here
+    writes to the current working directory — that matters because opening a *box*
+    ``Sensor`` drops a ~2 MB ``config/compound_eye.png`` relative to it (see
+    ``service.py``).
+    """
+
+    def __init__(self, root: str | Path, *, max_bytes: int = DEFAULT_MAX_BYTES) -> None:
+        self.root = Path(root)
+        self.max_bytes = int(max_bytes)
+
+    @classmethod
+    def for_recording(cls, recording_dir: str | Path, **kwargs: Any) -> ResultCache:
+        return cls(Path(recording_dir) / RESULT_DIRNAME, **kwargs)
+
+    def directory(self, key: ResultKey) -> Path:
+        return self.root / key.digest
+
+    def exists(self, key: ResultKey) -> bool:
+        return (self.directory(key) / META_NAME).is_file()
+
+    def is_complete(self, key: ResultKey) -> bool:
+        directory = self.directory(key)
+        if (directory / INCOMPLETE_MARKER).exists():
+            return False
+        meta = directory / META_NAME
+        if not meta.is_file():
+            return False
+        try:
+            return bool(json.loads(meta.read_text(encoding="utf-8")).get("complete"))
+        except (OSError, ValueError):
+            return False
+
+    def reader(self, key: ResultKey, *, allow_incomplete: bool = False):
+        """Open the stored result. Imported lazily so this module stays import-light."""
+        from orisys_app_shared.analysis.reader import AnalysisReader
+
+        return AnalysisReader(self.directory(key), allow_incomplete=allow_incomplete)
+
+    def writer(
+        self, key: ResultKey, *, chunk_frames: int = DEFAULT_CHUNK_FRAMES
+    ) -> AnalysisWriter:
+        return AnalysisWriter(self.directory(key), key=key, chunk_frames=chunk_frames)
+
+    def evict(
+        self, *, keep: Iterable[str | Path] = (), dry_run: bool = False
+    ) -> list[Path]:
+        return evict(self.root, max_bytes=self.max_bytes, keep=keep, dry_run=dry_run)
+
+    def total_bytes(self) -> int:
+        return sum(item.bytes_used for item in scan_results(self.root))

shared/src/orisys_app_shared/analysis/model.py

@@ -0,0 +1,526 @@
+"""What one offline analysis pass stores, and the byte encoding it uses.
+
+The pass re-runs the SDK chain over the recording's ``raw_*.mp4`` segments and keeps
+only what a replay draws: the flow samples the shared selector returns, not the flow
+field they came from. The fields a frame produces are far too large to cache — the
+measured cost is 1,176,000 B per frame (``flow`` 672,000 + ``depth_map`` 336,000 +
+``valid_mask`` 84,000 + the gray ``img`` 84,000), which is 21.2 GB for ten minutes at
+30 fps. The samples themselves cost 4,034 B per frame at the shipped finger grid.
+
+Three structural facts decide the layout, and each one is a measured result rather
+than a guess:
+
+* the candidate **location** set is byte-identical on every frame of a finger
+  recording (330 candidates, one digest across two videos), so it is a **keyframe**
+  and the per-frame record carries only values;
+* the **validity mask** is likewise static (0 of 327 frames differed), so it is a
+  keyframe too — and the per-frame record still carries an 8-byte digest of the mask
+  it was computed with, so a frame that *did* change is detectable instead of being
+  silently reconstructed against the wrong mask;
+* a location set that is *not* static (the box chain has no valid mask, so its
+  ``max_arrows`` cap orders by magnitude and moves frame to frame) gets its own
+  keyframe, deduplicated by digest. The format has one code path, and "stored once"
+  is true by construction wherever it can be.
+
+Display settings are deliberately **not** part of the invalidation key. The record
+keeps ``magnitude`` as well as the packed ``active`` bits, so a threshold change is
+served by re-deriving activity from magnitude; only a change to the sample *grid*
+(``flow_grid_spacing``, ``flow_edge_margin``, ``max_arrows``) moves the candidate
+locations, and that policy is recorded in the provenance for comparison instead of
+being baked into the key.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import struct
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
+from typing import Any
+
+import numpy as np
+
+from orisys_app_shared.scene3d.visual_policy import FlowSamples
+
+__all__ = [
+    "ALGORITHM_VERSION",
+    "ANALYSIS_FORMAT",
+    "ANALYSIS_FORMAT_VERSION",
+    "CHUNK_HEADER",
+    "CHUNK_MAGIC",
+    "CHUNK_TRAILER",
+    "CODEC_RAW",
+    "CODEC_ZLIB",
+    "FRAME_FLAG_DROPPED",
+    "FRAME_FLAG_VALIDITY",
+    "FRAME_HEADER",
+    "KEYFRAME_GRID",
+    "KEYFRAME_HEADER",
+    "KEYFRAME_MAGIC",
+    "KEYFRAME_MASK",
+    "UNIT",
+    "FramePayload",
+    "KeyframeRef",
+    "Provenance",
+    "ResultKey",
+    "decode_frame",
+    "encode_frame",
+    "frame_payload_bytes",
+    "keyframe_payload",
+    "recording_identity",
+    "result_key",
+    "unpack_mask",
+]
+
+#: Directory name a result lives under, beside the recording's ``frames.csv``.
+RESULT_DIRNAME = "analysis"
+
+#: On-disk result format. Bump only for an incompatible layout change; a change to
+#: what the *chain computes* belongs in :data:`ALGORITHM_VERSION` instead.
+ANALYSIS_FORMAT = "orisys-replay-analysis"
+ANALYSIS_FORMAT_VERSION = 1
+
+#: Identity of the analysis the pass runs: the SDK calls it makes, in order.
+#: ``compute_deformation(decompose=True, check_motion=False)`` then
+#: ``compute_contact()`` — the same pair the live acquisition loop and the offline
+#: depth replay use. A change to either call, or to the chain they drive, must change
+#: this string or two different results would share one cache entry.
+ALGORITHM_VERSION = "flow-hhd-contact-1"
+
+#: Unit of ``dx``/``dy``/``magnitude``: UV pixels of arrow-tip displacement, the same
+#: unit :class:`~orisys_app_shared.scene3d.visual_policy.FlowSamples` documents.
+UNIT = "uv_px"
+
+#: ``frame_id``, ``t_rel_s``, grid keyframe id, flags, candidate count, depth statistic.
+#: The four trailing pad bytes keep the value arrays 8-byte aligned inside a record.
+FRAME_HEADER = struct.Struct("<idHBHI3x")
+FRAME_FLAG_VALIDITY = 0x01
+FRAME_FLAG_DROPPED = 0x02
+
+#: magic, kind, codec, count, height, width, flags, reserved, raw bytes, digest
+KEYFRAME_HEADER = struct.Struct("<4sBBIIIBBQ8s")
+KEYFRAME_MAGIC = b"ORAK"
+KEYFRAME_GRID = 0
+KEYFRAME_MASK = 1
+
+CODEC_RAW = 0
+CODEC_ZLIB = 1
+
+#: magic, version, reserved, chunk index, frame count, header size
+CHUNK_HEADER = struct.Struct("<4sHHIII")
+CHUNK_MAGIC = b"ORAC"
+#: offset-table offset, frame count, magic
+CHUNK_TRAILER = struct.Struct("<QI4s")
+
+
+def digest8(*parts: bytes) -> bytes:
+    """An 8-byte blake2b over the concatenation of ``parts``."""
+    hasher = hashlib.blake2b(digest_size=8)
+    for part in parts:
+        hasher.update(part)
+    return hasher.digest()
+
+
+def canonical_digest8(payload: Any) -> bytes:
+    """An 8-byte digest of a JSON-representable structure, independent of key order."""
+    text = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
+    return digest8(text.encode("utf-8"))
+
+
+# --------------------------------------------------------------------------- the key
+@dataclass(frozen=True)
+class ResultKey:
+    """The five facts that decide whether a cached result still describes this input.
+
+    Display settings are absent on purpose: they are a property of the viewer, not of
+    the measurement, and the record keeps enough to serve a differently-thresholded
+    view (see the module docstring).
+    """
+
+    recording: str
+    algorithm: str
+    sensor_config: str
+    calibration: str
+    reference_frame: int | None
+
+    @property
+    def components(self) -> dict[str, str]:
+        return {
+            "recording": self.recording,
+            "algorithm": self.algorithm,
+            "sensor_config": self.sensor_config,
+            "calibration": self.calibration,
+            "reference_frame": ""
+            if self.reference_frame is None
+            else str(self.reference_frame),
+        }
+
+    @property
+    def digest(self) -> str:
+        """The directory name. Short because it is a name, not a security boundary."""
+        return canonical_digest8(self.components).hex()
+
+
+def result_key(
+    *,
+    recording: str,
+    sensor_config: str,
+    calibration: str,
+    reference_frame: int | None = None,
+    algorithm: str = ALGORITHM_VERSION,
+) -> ResultKey:
+    """Build the invalidation key. A changed argument is a different key, by construction."""
+    return ResultKey(
+        recording=str(recording),
+        algorithm=str(algorithm),
+        sensor_config=str(sensor_config),
+        calibration=str(calibration),
+        reference_frame=None if reference_frame is None else int(reference_frame),
+    )
+
+
+def recording_identity(
+    session: Mapping[str, Any], segments: Sequence[tuple[str, int]]
+) -> str:
+    """Identity of one recorded session, from its own metadata and segment sizes.
+
+    Deliberately not the directory path: a copied recording directory is the same
+    recording, and a re-used path is not. Segment *sizes* are included so a truncated
+    or extended ``raw_*.mp4`` cannot inherit an earlier result; modification times are
+    not, because copying a tree rewrites them.
+    """
+    device = dict(session.get("device") or {})
+    stats = dict(session.get("stats") or {})
+    payload = {
+        "session_id": session.get("session_id"),
+        "started_at": session.get("started_at"),
+        "ended_at": session.get("ended_at"),
+        "sensor_type": device.get("sensor_type"),
+        "config_ref": device.get("config_ref"),
+        "config_sha256": device.get("config_sha256"),
+        "frames": stats.get("frames"),
+        "video_frames": stats.get("video_frames"),
+        "segments": stats.get("segments"),
+        "files": sorted((str(name), int(size)) for name, size in segments),
+    }
+    return canonical_digest8(payload).hex()
+
+
+# ------------------------------------------------------------------- frame encoding
+def encode_frame(
+    *,
+    frame_id: int,
+    t_rel_s: float,
+    grid_id: int,
+    dx: np.ndarray,
+    dy: np.ndarray,
+    magnitude: np.ndarray,
+    active: np.ndarray,
+    validity_digest: bytes | None = None,
+    depth_nonzero: int = 0,
+    dropped: bool = False,
+) -> bytes:
+    """One frame record: a fixed header, then the value arrays.
+
+    Arrays are little-endian regardless of host order, so a result written on one
+    machine reads back identically on another. ``dx``/``dy``/``magnitude`` are stored
+    as float32 — the precision the display needs and the reason a stored payload is
+    compared to a live selection *cast to float32* rather than in float64.
+    """
+    dx = np.ascontiguousarray(dx, dtype="<f4").reshape(-1)
+    dy = np.ascontiguousarray(dy, dtype="<f4").reshape(-1)
+    magnitude = np.ascontiguousarray(magnitude, dtype="<f4").reshape(-1)
+    active = np.ascontiguousarray(active, dtype=bool).reshape(-1)
+    count = int(dx.size)
+    if not (dy.size == magnitude.size == active.size == count):
+        raise ValueError(
+            "dx/dy/magnitude/active must have the same length, got "
+            f"{dx.size}/{dy.size}/{magnitude.size}/{active.size}"
+        )
+    if not 0 <= int(grid_id) <= 0xFFFF:
+        raise ValueError(f"grid id {grid_id} does not fit the frame header")
+    if count > 0xFFFF:
+        raise ValueError(f"{count} candidates do not fit the frame header")
+    flags = 0
+    if validity_digest is not None:
+        flags |= FRAME_FLAG_VALIDITY
+    if dropped:
+        flags |= FRAME_FLAG_DROPPED
+    parts = [
+        FRAME_HEADER.pack(
+            int(frame_id),
+            float(t_rel_s),
+            int(grid_id),
+            flags,
+            count,
+            max(0, int(depth_nonzero)),
+        ),
+        dx.tobytes(),
+        dy.tobytes(),
+        magnitude.tobytes(),
+        np.packbits(active).tobytes(),
+    ]
+    if validity_digest is not None:
+        digest = bytes(validity_digest)
+        if len(digest) != 8:
+            raise ValueError(f"validity digest must be 8 bytes, got {len(digest)}")
+        parts.append(digest)
+    return b"".join(parts)
+
+
+@dataclass(frozen=True)
+class DecodedFrame:
+    """The raw fields of one frame record, before the grid keyframe is applied."""
+
+    frame_id: int
+    t_rel_s: float
+    grid_id: int
+    count: int
+    depth_nonzero: int
+    dropped: bool
+    dx: np.ndarray
+    dy: np.ndarray
+    magnitude: np.ndarray
+    active: np.ndarray
+    validity_digest: bytes | None
+
+
+def decode_frame(
+    buffer: bytes | bytearray | memoryview, offset: int = 0
+) -> DecodedFrame:
+    """Decode one frame record starting at ``offset``. Ignores any trailing bytes."""
+    view = memoryview(buffer)
+    frame_id, t_rel, grid_id, flags, count, depth_nonzero = FRAME_HEADER.unpack_from(
+        view, offset
+    )
+    cursor = offset + FRAME_HEADER.size
+    arrays = []
+    for _ in range(3):
+        end = cursor + count * 4
+        arrays.append(np.frombuffer(view[cursor:end], dtype="<f4").copy())
+        cursor = end
+    bits_end = cursor + (count + 7) // 8
+    active = np.unpackbits(
+        np.frombuffer(view[cursor:bits_end], dtype=np.uint8), count=count
+    )
+    cursor = bits_end
+    validity = None
+    if flags & FRAME_FLAG_VALIDITY:
+        validity = bytes(view[cursor : cursor + 8])
+    return DecodedFrame(
+        frame_id=int(frame_id),
+        t_rel_s=float(t_rel),
+        grid_id=int(grid_id),
+        count=int(count),
+        depth_nonzero=int(depth_nonzero),
+        dropped=bool(flags & FRAME_FLAG_DROPPED),
+        dx=arrays[0],
+        dy=arrays[1],
+        magnitude=arrays[2],
+        active=active.astype(bool),
+        validity_digest=validity,
+    )
+
+
+def frame_payload_bytes(candidates: int, *, with_validity: bool = True) -> int:
+    """Stored bytes of one frame record at ``candidates`` sample locations."""
+    total = FRAME_HEADER.size + int(candidates) * 12 + (int(candidates) + 7) // 8
+    return total + 8 if with_validity else total
+
+
+def keyframe_payload(kind: int, array: np.ndarray) -> bytes:
+    """The uncompressed payload of a grid or mask keyframe.
+
+    A grid is ``rows`` (int16) followed by ``cols`` (int16); a mask is bit-packed
+    along its flattened order. Both are the exact bytes the digest is taken over, so
+    a reader can verify a keyframe before trusting it.
+    """
+    if kind == KEYFRAME_GRID:
+        return array.astype("<i2").tobytes()
+    if kind == KEYFRAME_MASK:
+        return np.packbits(np.asarray(array, dtype=bool).reshape(-1)).tobytes()
+    raise ValueError(f"unknown keyframe kind {kind}")
+
+
+def unpack_mask(payload: bytes, height: int, width: int) -> np.ndarray:
+    """Bit-packed mask payload back to a ``(height, width)`` boolean array."""
+    count = int(height) * int(width)
+    bits = np.unpackbits(np.frombuffer(payload, dtype=np.uint8), count=count)
+    return bits.astype(bool).reshape(int(height), int(width))
+
+
+# ------------------------------------------------------------------- decoded frames
+@dataclass(frozen=True)
+class FramePayload:
+    """One analysed frame, as the reader serves it.
+
+    ``rows``/``cols`` come from the frame's grid keyframe, so they are identical for
+    every frame that shares one; ``dx``/``dy``/``magnitude``/``active`` are this
+    frame's own values. Arrays are float32 and int32 because that is what is stored —
+    a comparison against a live selection must cast to float32 to be exact.
+    """
+
+    frame_id: int
+    t_rel_s: float
+    grid_id: int
+    rows: np.ndarray
+    cols: np.ndarray
+    dx: np.ndarray
+    dy: np.ndarray
+    magnitude: np.ndarray
+    active: np.ndarray
+    validity_digest: bytes | None = None
+    depth_nonzero: int = 0
+    dropped: bool = False
+
+    @property
+    def candidates(self) -> int:
+        return int(self.rows.size)
+
+    @property
+    def active_count(self) -> int:
+        return int(np.count_nonzero(self.active))
+
+    def samples(self, *, threshold: float | None = None) -> FlowSamples:
+        """This frame as the shared selector's own type.
+
+        ``threshold=None`` returns the activity the pass recorded. Any other value
+        re-derives it from ``magnitude``, which is what lets a display-settings change
+        be served without invalidating the result.
+        """
+        active = self.active
+        if threshold is not None:
+            active = np.asarray(self.magnitude >= float(threshold), dtype=bool)
+        return FlowSamples(
+            rows=self.rows.copy(),
+            cols=self.cols.copy(),
+            dx=self.dx.copy(),
+            dy=self.dy.copy(),
+            magnitude=self.magnitude.copy(),
+            active=active.copy(),
+        )
+
+    def canonical_bytes(self) -> bytes:
+        """A byte image of the payload, for exact comparison between two results."""
+        return encode_frame(
+            frame_id=self.frame_id,
+            t_rel_s=self.t_rel_s,
+            grid_id=self.grid_id,
+            dx=self.dx,
+            dy=self.dy,
+            magnitude=self.magnitude,
+            active=self.active,
+            validity_digest=self.validity_digest,
+            depth_nonzero=self.depth_nonzero,
+            dropped=self.dropped,
+        )
+
+
+@dataclass(frozen=True)
+class KeyframeRef:
+    """Inventory entry for one keyframe blob inside ``keyframes.bin``."""
+
+    id: int
+    kind: str
+    digest: str
+    shape: tuple[int, ...]
+    codec: str
+    raw_bytes: int
+    stored_bytes: int
+    offset: int
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id,
+            "kind": self.kind,
+            "digest": self.digest,
+            "shape": list(self.shape),
+            "codec": self.codec,
+            "raw_bytes": self.raw_bytes,
+            "stored_bytes": self.stored_bytes,
+            "offset": self.offset,
+        }
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, Any]) -> KeyframeRef:
+        return cls(
+            id=int(data["id"]),
+            kind=str(data["kind"]),
+            digest=str(data["digest"]),
+            shape=tuple(int(v) for v in data.get("shape") or ()),
+            codec=str(data["codec"]),
+            raw_bytes=int(data["raw_bytes"]),
+            stored_bytes=int(data["stored_bytes"]),
+            offset=int(data["offset"]),
+        )
+
+
+# ------------------------------------------------------------------- provenance
+@dataclass
+class Provenance:
+    """Everything needed to say what a stored result is and what it is not.
+
+    ``complete`` is duplicated from the result's top-level metadata on purpose: one is
+    the gate a reader checks, the other travels with the record. They are written from
+    a single source and compared on read, so they cannot disagree silently.
+    """
+
+    format_version: int = ANALYSIS_FORMAT_VERSION
+    algorithm: str = ALGORITHM_VERSION
+    unit: str = UNIT
+    key: str = ""
+    key_components: dict[str, str] = field(default_factory=dict)
+    recording: str = ""
+    recording_identity: str = ""
+    session_id: int = 0
+    sensor_type: str = ""
+    chain: str = ""
+    sensor_config_ref: str = ""
+    sensor_config_sha256: str = ""
+    calibration_id: str = ""
+    #: The frame whose processing started from a clean tracker reference. Equal to
+    #: ``first_output_frame_id`` when no explicit reset was requested.
+    reference_frame_id: int | None = None
+    reference_is_explicit: bool = False
+    first_output_frame_id: int | None = None
+    last_output_frame_id: int | None = None
+    #: Decoded frames the chain consumed *before* the first frame a caller can read.
+    #: Measured per segment; ``None`` when the checked segments disagree.
+    decode_order_offset: int | None = None
+    decode_order_offsets: dict[str, int] = field(default_factory=dict)
+    #: ``measured`` when the offset came from the capture's own decode position,
+    #: ``declared`` when the backend could not report one and the chain's constant was
+    #: used instead. A result built on a declared offset is weaker evidence.
+    decode_order_offset_source: str = "measured"
+    segments: list[int] = field(default_factory=list)
+    segments_declared: int = 0
+    segments_missing: list[int] = field(default_factory=list)
+    #: The selection policy used while recording samples. Not part of the key.
+    policy: dict[str, Any] = field(default_factory=dict)
+    #: Per channel: ``recomputed`` (re-run from the video) or ``recorded`` (read from a
+    #: CSV the recorder wrote). No channel is both.
+    sources: dict[str, str] = field(default_factory=dict)
+    versions: dict[str, str] = field(default_factory=dict)
+    config: dict[str, Any] = field(default_factory=dict)
+    open_working_directory: str = ""
+    debug_files_written: list[str] = field(default_factory=list)
+    started_at: str = ""
+    finished_at: str = ""
+    elapsed_s: float = 0.0
+    #: Wall-clock cost of opening the SDK chain, summed over segments. It is separate
+    #: from ``elapsed_s`` because a box open spends 1.6 s before the first frame and
+    #: that time is not throughput.
+    open_seconds: float = 0.0
+    warnings: list[str] = field(default_factory=list)
+    complete: bool = False
+    stopped_reason: str = ""
+    counts: dict[str, Any] = field(default_factory=dict)
+
+    def to_dict(self) -> dict[str, Any]:
+        return dict(self.__dict__)
+
+    @classmethod
+    def from_dict(cls, data: Mapping[str, Any]) -> Provenance:
+        known = {f for f in cls().__dict__}
+        return cls(**{key: value for key, value in dict(data).items() if key in known})

shared/src/orisys_app_shared/analysis/reader.py

@@ -0,0 +1,567 @@
+"""Reading a stored result: by frame id, by timestamp, and nothing else.
+
+A seek must not re-run the pass. Two levels of index make that true:
+
+* ``meta.json`` lists the chunks in order with their first and last frame id, so a
+  frame id picks a chunk by binary search;
+* each chunk ends with a trailer holding the frame's own byte offset, frame id and
+  timestamp, so a position inside that chunk is another binary search and one read.
+
+The trailer is read the first time a chunk is needed and then cached, which costs
+20 bytes per frame of memory — 360 KB for a ten-minute recording at 30 fps. Everything
+else (the sample grid, the validity mask) is loaded from its keyframe on demand and
+verified against the digest recorded for it.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import zlib
+from collections import OrderedDict
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from orisys_app_shared.analysis.cache import INCOMPLETE_MARKER, META_NAME
+from orisys_app_shared.analysis.model import (
+    ANALYSIS_FORMAT,
+    ANALYSIS_FORMAT_VERSION,
+    CHUNK_HEADER,
+    CHUNK_MAGIC,
+    CHUNK_TRAILER,
+    CODEC_RAW,
+    CODEC_ZLIB,
+    KEYFRAME_GRID,
+    KEYFRAME_HEADER,
+    KEYFRAME_MAGIC,
+    KEYFRAME_MASK,
+    FramePayload,
+    KeyframeRef,
+    Provenance,
+    decode_frame,
+    digest8,
+    unpack_mask,
+)
+from orisys_app_shared.scene3d.visual_policy import FlowSamples
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "AnalysisFormatError",
+    "AnalysisIncompleteError",
+    "AnalysisNotFoundError",
+    "AnalysisReader",
+    "AnalysisResultError",
+]
+
+#: How many keyframe payloads to keep decoded. One grid and one mask is the norm; the
+#: box chain needs more, and re-reading a 1.3 KB blob is cheap when it falls out.
+_GRID_CACHE = 16
+
+
+class AnalysisResultError(RuntimeError):
+    """Base for every reason a stored result cannot be served."""
+
+
+class AnalysisNotFoundError(AnalysisResultError):
+    """There is no result at that path."""
+
+
+class AnalysisIncompleteError(AnalysisResultError):
+    """The result exists but was cancelled, failed, or never finished.
+
+    Raised by default rather than served with a warning: a partial result that looks
+    finished is worse than no result, because every chart drawn from it would be
+    missing an unknown number of frames without saying so.
+    """
+
+
+class AnalysisFormatError(AnalysisResultError):
+    """The result is not in a format this build can read, or it is corrupt."""
+
+
+@dataclass
+class _ChunkIndex:
+    """One chunk's trailer, loaded on first use."""
+
+    index: int
+    path: Path
+    offsets: np.ndarray  # uint64, per frame, relative to the file start
+    frame_ids: np.ndarray  # int32
+    times: np.ndarray  # float64
+    sizes: np.ndarray  # uint64, per frame record length
+    first_frame_id: int
+    last_frame_id: int
+
+
+class AnalysisReader:
+    """Random access into one analysis result directory."""
+
+    def __init__(
+        self, directory: str | Path, *, allow_incomplete: bool = False
+    ) -> None:
+        self.directory = Path(directory)
+        self.allow_incomplete = bool(allow_incomplete)
+        meta_path = self.directory / META_NAME
+        if not meta_path.is_file():
+            raise AnalysisNotFoundError(f"no analysis result at {self.directory}")
+        try:
+            meta = json.loads(meta_path.read_text(encoding="utf-8"))
+        except (OSError, ValueError) as exc:
+            raise AnalysisFormatError(f"{meta_path} is unreadable: {exc}") from exc
+        if str(meta.get("format") or "") != ANALYSIS_FORMAT:
+            raise AnalysisFormatError(
+                f"{meta_path} is not an {ANALYSIS_FORMAT} result "
+                f"(format={meta.get('format')!r})"
+            )
+        version = int(meta.get("format_version") or 0)
+        if version != ANALYSIS_FORMAT_VERSION:
+            raise AnalysisFormatError(
+                f"result format version {version} cannot be read by this build "
+                f"(it reads {ANALYSIS_FORMAT_VERSION})"
+            )
+        self.meta: dict[str, Any] = dict(meta)
+        self.key: str = str(meta.get("key") or self.directory.name)
+        self.key_components: dict[str, str] = dict(meta.get("key_components") or {})
+        self.provenance = Provenance.from_dict(meta.get("provenance") or {})
+        marker = self.directory / INCOMPLETE_MARKER
+        declared = bool(meta.get("complete"))
+        if declared != bool(self.provenance.complete):
+            raise AnalysisFormatError(
+                f"{meta_path} disagrees with its own provenance about completeness"
+            )
+        if marker.exists() or not declared:
+            if not self.allow_incomplete:
+                reason = self.provenance.stopped_reason or "unfinished"
+                raise AnalysisIncompleteError(
+                    f"{self.directory} is not a complete result ({reason}); "
+                    "pass allow_incomplete=True only to inspect a partial one"
+                )
+            self.complete = False
+        else:
+            self.complete = True
+        self.counts: dict[str, Any] = dict(meta.get("counts") or {})
+        self.keyframes: dict[int, KeyframeRef] = {
+            item.id: item
+            for item in (
+                KeyframeRef.from_dict(entry) for entry in meta.get("keyframes") or []
+            )
+        }
+        self.chunks: list[dict[str, Any]] = [
+            dict(entry) for entry in meta.get("chunks") or []
+        ]
+        self._chunk_index: dict[int, _ChunkIndex] = {}
+        self._mask_cache: np.ndarray | None = None
+        self._mask_loaded = False
+        self._grid_cache: OrderedDict[int, tuple[np.ndarray, np.ndarray]] = (
+            OrderedDict()
+        )
+        self._handle = None
+        self._handle_chunk = -1
+
+    # ------------------------------------------------------------------- lifecycle
+    def close(self) -> None:
+        if self._handle is not None:
+            try:
+                self._handle.close()
+            finally:
+                self._handle = None
+                self._handle_chunk = -1
+
+    def __enter__(self) -> AnalysisReader:
+        return self
+
+    def __exit__(self, exc_type, exc, tb) -> bool:
+        self.close()
+        return False
+
+    # -------------------------------------------------------------------- indexing
+    def __len__(self) -> int:
+        return int(sum(int(entry.get("frames") or 0) for entry in self.chunks))
+
+    @property
+    def frame_ids(self) -> np.ndarray:
+        """Every stored frame id, in stored order."""
+        parts = [self._chunk(chunk["index"]).frame_ids for chunk in self.chunks]
+        if not parts:
+            return np.empty(0, dtype=np.int64)
+        return np.concatenate(parts).astype(np.int64)
+
+    @property
+    def times(self) -> np.ndarray:
+        """Every stored ``t_rel_s``, in stored order."""
+        parts = [self._chunk(chunk["index"]).times for chunk in self.chunks]
+        if not parts:
+            return np.empty(0, dtype=np.float64)
+        return np.concatenate(parts)
+
+    @property
+    def frame_record_bytes(self) -> int:
+        """Stored bytes of the frame records alone.
+
+        Excludes the chunk offset tables, the keyframes and the metadata, so a caller can
+        compare it with :func:`~orisys_app_shared.analysis.model.frame_payload_bytes`
+        instead of guessing what the overhead was.
+        """
+        return int(
+            sum(int(self._chunk(entry["index"]).sizes.sum()) for entry in self.chunks)
+        )
+
+    @property
+    def bytes_per_frame(self) -> float:
+        """Frame-record bytes divided by frames — the number to quote per frame."""
+        count = len(self)
+        return 0.0 if not count else self.frame_record_bytes / count
+
+    @property
+    def keyframe_counts(self) -> dict[str, int]:
+        """How many keyframes of each kind the result holds."""
+        counts: dict[str, int] = {}
+        for reference in self.keyframes.values():
+            counts[reference.kind] = counts.get(reference.kind, 0) + 1
+        return counts
+
+    def _chunk(self, index: int) -> _ChunkIndex:
+        cached = self._chunk_index.get(index)
+        if cached is not None:
+            return cached
+        entry = next(
+            (item for item in self.chunks if int(item.get("index", -1)) == int(index)),
+            None,
+        )
+        if entry is None:
+            raise AnalysisFormatError(f"chunk {index} is not in the result inventory")
+        path = self.directory / str(entry["file"])
+        if not path.is_file():
+            raise AnalysisFormatError(f"chunk file {path} is missing")
+        data = path.read_bytes()
+        if len(data) < CHUNK_HEADER.size + CHUNK_TRAILER.size:
+            raise AnalysisFormatError(f"{path} is shorter than a chunk header")
+        magic, version, _reserved, chunk_index, count, header_size = (
+            CHUNK_HEADER.unpack_from(data)
+        )
+        if magic != CHUNK_MAGIC:
+            raise AnalysisFormatError(f"{path} is not a frame chunk")
+        if version != ANALYSIS_FORMAT_VERSION or int(chunk_index) != int(index):
+            raise AnalysisFormatError(
+                f"{path} declares version {version} chunk {chunk_index}, expected "
+                f"{ANALYSIS_FORMAT_VERSION} and {index}"
+            )
+        declared = int(entry.get("frames") or 0)
+        if declared != int(count):
+            raise AnalysisFormatError(
+                f"{path} holds {count} frames, the inventory says {declared}"
+            )
+        table_offset, trailer_count, trailer_magic = CHUNK_TRAILER.unpack_from(
+            data, len(data) - CHUNK_TRAILER.size
+        )
+        if trailer_magic != CHUNK_MAGIC or int(trailer_count) != int(count):
+            raise AnalysisFormatError(f"{path} has a damaged trailer")
+        offsets = np.frombuffer(
+            data, dtype="<u8", count=count, offset=int(table_offset)
+        ).copy()
+        cursor = int(table_offset) + count * 8
+        frame_ids = np.frombuffer(data, dtype="<i4", count=count, offset=cursor).copy()
+        cursor += count * 4
+        times = np.frombuffer(data, dtype="<f8", count=count, offset=cursor).copy()
+        if int(header_size) > int(table_offset):
+            raise AnalysisFormatError(
+                f"{path} declares a header past its own frame data"
+            )
+        sizes = np.empty(count, dtype=np.uint64)
+        if count:
+            sizes[:-1] = offsets[1:] - offsets[:-1]
+            sizes[-1] = np.uint64(int(table_offset) - int(offsets[-1]))
+        record = _ChunkIndex(
+            index=int(index),
+            path=path,
+            offsets=offsets,
+            frame_ids=frame_ids,
+            times=times,
+            sizes=sizes,
+            first_frame_id=int(frame_ids[0]) if count else 0,
+            last_frame_id=int(frame_ids[-1]) if count else 0,
+        )
+        self._chunk_index[int(index)] = record
+        return record
+
+    def _chunk_for_frame(self, frame_id: int) -> _ChunkIndex | None:
+        """The chunk holding ``frame_id``.
+
+        The inventory is ordered by frame id for every result this code writes, so the
+        binary search answers in ``log(chunks)``. A hand-edited or foreign result whose
+        ranges are not monotonic falls back to a scan rather than answering with the
+        wrong chunk.
+        """
+        low, high = 0, len(self.chunks) - 1
+        while low <= high:
+            middle = (low + high) // 2
+            record = self._chunk(int(self.chunks[middle]["index"]))
+            if not len(record.frame_ids):
+                low = middle + 1
+                continue
+            if frame_id < record.first_frame_id:
+                high = middle - 1
+            elif frame_id > record.last_frame_id:
+                low = middle + 1
+            else:
+                return record
+        for entry in self.chunks:
+            record = self._chunk(int(entry["index"]))
+            if len(record.frame_ids) and bool(
+                np.any(record.frame_ids == int(frame_id))
+            ):
+                return record
+        return None
+
+    # ------------------------------------------------------------------ keyframes
+    def _keyframe_payload(self, reference: KeyframeRef) -> bytes:
+        path = self.directory / "keyframes.bin"
+        if not path.is_file():
+            raise AnalysisFormatError("keyframes.bin is missing from the result")
+        with path.open("rb") as handle:
+            handle.seek(reference.offset)
+            header = handle.read(KEYFRAME_HEADER.size)
+            if len(header) != KEYFRAME_HEADER.size:
+                raise AnalysisFormatError(f"keyframe {reference.id} is truncated")
+            (
+                magic,
+                kind,
+                codec,
+                count,
+                height,
+                width,
+                _flags,
+                _reserved,
+                raw_bytes,
+                digest,
+            ) = KEYFRAME_HEADER.unpack(header)
+            if magic != KEYFRAME_MAGIC:
+                raise AnalysisFormatError(f"keyframe {reference.id} has no magic")
+            stored = handle.read(reference.stored_bytes)
+        if len(stored) != reference.stored_bytes:
+            raise AnalysisFormatError(f"keyframe {reference.id} is truncated")
+        payload = zlib.decompress(stored) if codec == CODEC_ZLIB else stored
+        if len(payload) != int(raw_bytes):
+            raise AnalysisFormatError(
+                f"keyframe {reference.id} decoded to {len(payload)} bytes, "
+                f"its header says {raw_bytes}"
+            )
+        if digest8(payload).hex() != reference.digest:
+            raise AnalysisFormatError(
+                f"keyframe {reference.id} does not match its recorded digest"
+            )
+        expected_kind = {KEYFRAME_GRID: "grid", KEYFRAME_MASK: "mask"}.get(kind)
+        if expected_kind != reference.kind:
+            raise AnalysisFormatError(
+                f"keyframe {reference.id} is a {expected_kind}, the inventory says "
+                f"{reference.kind}"
+            )
+        if codec not in (CODEC_RAW, CODEC_ZLIB):
+            raise AnalysisFormatError(f"keyframe {reference.id} uses codec {codec}")
+        declared = (int(height), int(width)) if kind == KEYFRAME_MASK else (int(count),)
+        if reference.kind == "grid" and declared != reference.shape:
+            raise AnalysisFormatError(
+                f"keyframe {reference.id} shape disagrees with it"
+            )
+        return payload
+
+    def grid(self, grid_id: int) -> tuple[np.ndarray, np.ndarray]:
+        """The candidate locations of one grid keyframe, as ``(rows, cols)`` int32."""
+        cached = self._grid_cache.get(grid_id)
+        if cached is not None:
+            self._grid_cache.move_to_end(grid_id)
+            return cached
+        reference = self.keyframes.get(int(grid_id))
+        if reference is None or reference.kind != "grid":
+            raise AnalysisFormatError(f"no grid keyframe {grid_id} in this result")
+        payload = self._keyframe_payload(reference)
+        count = int(reference.shape[0]) if reference.shape else 0
+        if len(payload) != count * 4:
+            raise AnalysisFormatError(
+                f"grid keyframe {grid_id} holds {len(payload)} bytes for {count} locations"
+            )
+        rows = np.frombuffer(payload, dtype="<i2", count=count, offset=0).astype(
+            np.int32
+        )
+        cols = np.frombuffer(
+            payload, dtype="<i2", count=count, offset=count * 2
+        ).astype(np.int32)
+        rows.flags.writeable = False
+        cols.flags.writeable = False
+        self._grid_cache[grid_id] = (rows, cols)
+        if len(self._grid_cache) > _GRID_CACHE:
+            self._grid_cache.popitem(last=False)
+        return rows, cols
+
+    def mask(self) -> np.ndarray | None:
+        """The validity mask this result was built with, or ``None`` when there was none."""
+        if self._mask_loaded:
+            return self._mask_cache
+        self._mask_loaded = True
+        self._mask_cache = None
+        for reference in self.keyframes.values():
+            if reference.kind != "mask":
+                continue
+            payload = self._keyframe_payload(reference)
+            height = int(reference.shape[0]) if len(reference.shape) > 0 else 0
+            width = int(reference.shape[1]) if len(reference.shape) > 1 else 0
+            self._mask_cache = unpack_mask(payload, height, width)
+            break
+        return self._mask_cache
+
+    @property
+    def mask_digest(self) -> str | None:
+        for reference in self.keyframes.values():
+            if reference.kind == "mask":
+                return reference.digest
+        return None
+
+    # ----------------------------------------------------------------- frame access
+    def _read_record(self, chunk: _ChunkIndex, slot: int) -> bytes:
+        if self._handle_chunk != chunk.index or self._handle is None:
+            if self._handle is not None:
+                self._handle.close()
+            self._handle = chunk.path.open("rb")
+            self._handle_chunk = chunk.index
+        self._handle.seek(int(chunk.offsets[slot]))
+        return self._handle.read(int(chunk.sizes[slot]))
+
+    def _slot_for(self, chunk: _ChunkIndex, frame_id: int) -> int:
+        ids = chunk.frame_ids
+        if ids.size and bool(np.all(np.diff(ids) > 0)):
+            return int(np.searchsorted(ids, int(frame_id)))
+        hits = np.flatnonzero(ids == int(frame_id))
+        if not hits.size:
+            raise AnalysisNotFoundError(
+                f"frame {frame_id} is not in chunk {chunk.path.name}"
+            )
+        return int(hits[0])
+
+    def has(self, frame_id: int) -> bool:
+        return self._chunk_for_frame(int(frame_id)) is not None
+
+    def frame(self, frame_id: int) -> FramePayload:
+        """One frame by its recorded id. Random access; no replay of the pass."""
+        chunk = self._chunk_for_frame(int(frame_id))
+        if chunk is None:
+            raise AnalysisNotFoundError(f"frame {frame_id} is not in this result")
+        slot = self._slot_for(chunk, int(frame_id))
+        decoded = decode_frame(self._read_record(chunk, slot))
+        if decoded.frame_id != int(frame_id):
+            raise AnalysisFormatError(
+                f"frame {frame_id} resolved to a record for {decoded.frame_id}"
+            )
+        rows, cols = self.grid(decoded.grid_id)
+        if rows.size != decoded.count:
+            raise AnalysisFormatError(
+                f"frame {frame_id} has {decoded.count} values but grid "
+                f"{decoded.grid_id} has {rows.size} locations"
+            )
+        return FramePayload(
+            frame_id=decoded.frame_id,
+            t_rel_s=decoded.t_rel_s,
+            grid_id=decoded.grid_id,
+            rows=rows,
+            cols=cols,
+            dx=decoded.dx,
+            dy=decoded.dy,
+            magnitude=decoded.magnitude,
+            active=decoded.active,
+            validity_digest=decoded.validity_digest,
+            depth_nonzero=decoded.depth_nonzero,
+            dropped=decoded.dropped,
+        )
+
+    def frame_at(self, t_rel_s: float) -> FramePayload | None:
+        """The stored frame at or just before ``t_rel_s``, or ``None`` for an empty result.
+
+        Uses the recording's own time base, which is *signal* time: the container
+        duration and the recorded duration are different numbers and neither is
+        derived from the other.
+        """
+        times = self.times
+        if times.size == 0:
+            return None
+        if bool(np.all(np.diff(times) >= 0)):
+            slot = int(np.searchsorted(times, float(t_rel_s), side="right")) - 1
+            slot = max(0, min(slot, times.size - 1))
+        else:
+            candidates = np.flatnonzero(times <= float(t_rel_s))
+            slot = int(candidates[-1]) if candidates.size else 0
+        ids = self.frame_ids
+        return self.frame(int(ids[slot]))
+
+    def samples(self, frame_id: int, *, policy: Any | None = None) -> FlowSamples:
+        """The frame's flow samples, as the shared selector would return them.
+
+        A ``policy`` is honoured only for its activation threshold: the candidate
+        *locations* are the ones recorded, because those are what the pass selected.
+        Change the grid and this result no longer describes the view — compare
+        :attr:`provenance.policy` rather than expecting the reader to re-grid.
+        """
+        threshold = None if policy is None else float(policy.normalised().flow_on_px)
+        return self.frame(int(frame_id)).samples(threshold=threshold)
+
+    def frames(
+        self, *, start: int | None = None, stop: int | None = None
+    ) -> Iterator[FramePayload]:
+        """Frames in stored order, optionally bounded by frame id (both inclusive)."""
+        for entry in self.chunks:
+            chunk = self._chunk(int(entry["index"]))
+            for slot in range(len(chunk.frame_ids)):
+                frame_id = int(chunk.frame_ids[slot])
+                if start is not None and frame_id < int(start):
+                    continue
+                if stop is not None and frame_id > int(stop):
+                    continue
+                yield self.frame(frame_id)
+
+    # ------------------------------------------------------------------- freshness
+    def stale_reasons(
+        self,
+        *,
+        sensor_config: str = "",
+        algorithm: str = "",
+        calibration: str = "",
+        sdk_version: str = "",
+    ) -> list[str]:
+        """Why this result may no longer describe what would be computed now.
+
+        The key covers the recording, the algorithm, the sensor configuration, the
+        calibration and the reference frame. The SDK version is deliberately **not** in
+        the key — a patch release does not change the key, so it would otherwise be
+        served silently. It is reported here instead, next to the caller's other
+        warnings, exactly as the replay already reports a changed recorded config.
+        """
+        reasons: list[str] = []
+        recorded = self.key_components
+        if (
+            sensor_config
+            and recorded.get("sensor_config")
+            and sensor_config != recorded["sensor_config"]
+        ):
+            reasons.append("sensor_config_changed")
+        if (
+            algorithm
+            and recorded.get("algorithm")
+            and algorithm != recorded["algorithm"]
+        ):
+            reasons.append("algorithm_changed")
+        if (
+            calibration
+            and recorded.get("calibration")
+            and calibration != recorded["calibration"]
+        ):
+            reasons.append("calibration_changed")
+        if sdk_version:
+            stored = str(self.provenance.versions.get("sdk") or "")
+            if stored and stored != sdk_version:
+                reasons.append("sdk_changed")
+        if not self.complete:
+            reasons.append("incomplete")
+        return reasons

shared/tests/test_replay_analysis.py

@@ -0,0 +1,615 @@
+"""Part A: the offline analysis pass and its result cache, checked on real footage.
+
+Three things are being defended here, and each one is a way the feature could be
+silently wrong rather than visibly broken:
+
+* the stored payload is **what the live selector returns for that frame** — not a
+  close approximation of it, and not a different frame;
+* a result is either complete or unreadable, so a cancelled pass can never be drawn;
+* the pass still costs what it cost when it was measured, so a cache that is correct
+  but slower than re-running the chain is caught.
+
+The inputs are real: the 33-frame box recording under ``dist/`` (its own ``frames.csv``,
+``session.json`` and ``raw_0001.mp4``) and the shipped finger configuration and asset
+bundle. The pass is driven with the **box** chain on it for the offset and round-trip
+checks, and with the **finger** chain on the same footage for the keyframe, reset and
+size checks — the finger chain over box footage is exactly what the Step 0 measurement
+did, and it is the only way to get a validity mask (and therefore a static candidate
+grid) out of this recording.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import inspect
+import json
+import os
+import shutil
+import threading
+from pathlib import Path
+
+import numpy as np
+import pytest
+from orisys_app_shared.analysis import model as analysis_model
+from orisys_app_shared.analysis.cache import (
+    INCOMPLETE_MARKER,
+    META_NAME,
+    ResultCache,
+    evict,
+    scan_results,
+)
+from orisys_app_shared.analysis.model import frame_payload_bytes, result_key
+from orisys_app_shared.analysis.reader import (
+    AnalysisIncompleteError,
+    AnalysisReader,
+)
+from orisys_app_shared.analysis.service import (
+    AnalysisService,
+    ChainSpec,
+    chain_spec_for,
+)
+from orisys_app_shared.scene3d.visual_policy import VisualPolicy
+
+REPO = Path(__file__).resolve().parents[2]
+STUDIO = REPO / "dist" / "OrisysStudio"
+RECORDING = (
+    STUDIO
+    / "outputs"
+    / "recordings"
+    / "20260930T015028Z_session0002_40cb8e7441434feab48bb646afdba011"
+)
+FINGER_CONFIG = STUDIO / "config" / "sensors" / "finger" / "sensor.json"
+FINGER_EXPORT = STUDIO / "config" / "sensors" / "finger" / "assets" / "30fps-0703"
+
+#: Step 0's digest of the finger validity mask, over ``np.packbits(mask)`` with
+#: blake2b-16: one value across three videos and 327 frames. Reproduced here from the
+#: stored keyframe, which is what makes the mask keyframe a *verified* copy rather than
+#: a plausible-looking one. (The result stores a blake2b-8 digest of the same bytes; the
+#: longer digest is the comparison because that is the one on record.)
+STEP0_MASK_DIGEST = "d76349a3b901aea50915eb1308901c2e"
+
+#: The frame id the reset is applied at. The recording's own ids run 20…52, so this is
+#: deliberately not a small number: an off-by-one in the reset rule would land on a
+#: frame that does not exist and be invisible.
+REFERENCE_FRAME = 36
+
+
+def _require_inputs() -> None:
+    missing = [
+        str(path)
+        for path in (
+            RECORDING / "frames.csv",
+            RECORDING / "raw_0001.mp4",
+            FINGER_CONFIG,
+        )
+        if not path.exists()
+    ]
+    if missing:
+        pytest.skip("analysis inputs are not present: " + ", ".join(missing))
+    pytest.importorskip("orisys", reason="the SDK is needed to run the analysis pass")
+
+
+def _finger_chain() -> ChainSpec:
+    return ChainSpec(
+        sensor_type="finger", config_ref=FINGER_CONFIG, export_dir=FINGER_EXPORT
+    )
+
+
+def _box_service(root: Path, **kwargs) -> AnalysisService:
+    from orisys_app_shared.record.playback import RecordingReader
+
+    chain = chain_spec_for(RecordingReader(RECORDING).session())
+    return AnalysisService(RECORDING, chain=chain, cache=ResultCache(root), **kwargs)
+
+
+def _finger_service(root: Path, **kwargs) -> AnalysisService:
+    return AnalysisService(
+        RECORDING, chain=_finger_chain(), cache=ResultCache(root), **kwargs
+    )
+
+
+def _directory_bytes(directory: Path) -> int:
+    return sum(path.stat().st_size for path in directory.rglob("*") if path.is_file())
+
+
+# --------------------------------------------------------------------------- fixtures
+@pytest.fixture(scope="module")
+def box_pass(tmp_path_factory):
+    _require_inputs()
+    root = tmp_path_factory.mktemp("analysis-box")
+    service = _box_service(root)
+    outcome = service.run()
+    return service, outcome
+
+
+@pytest.fixture(scope="module")
+def finger_pass(tmp_path_factory):
+    _require_inputs()
+    root = tmp_path_factory.mktemp("analysis-finger")
+    service = _finger_service(root)
+    outcome = service.run()
+    return service, outcome
+
+
+@pytest.fixture(scope="module")
+def finger_pass_again(tmp_path_factory):
+    """A second pass over the same input, in its own cache root."""
+    _require_inputs()
+    root = tmp_path_factory.mktemp("analysis-finger-again")
+    service = _finger_service(root)
+    return service, service.run()
+
+
+@pytest.fixture(scope="module")
+def finger_reset_pass(tmp_path_factory):
+    _require_inputs()
+    root = tmp_path_factory.mktemp("analysis-finger-reset")
+    service = _finger_service(root, reference_frame=REFERENCE_FRAME)
+    return service, service.run()
+
+
+# --------------------------------------------------------------- frame identity
+def test_the_box_chain_consumes_frames_before_the_first_one_a_caller_can_read(box_pass):
+    """The offset is measured, recorded, and applied — not assumed.
+
+    Constructing a box ``Sensor`` reads seven frames through its stitch probe, so the
+    first frame the pass can read is the recording's *eighth* video frame. Nothing about
+    the file says so, which is why the offset comes from the capture's own decode
+    position and why the resulting frame identity is checked against ``frames.csv``.
+    """
+    _, outcome = box_pass
+    assert outcome.complete, outcome.reason
+    assert outcome.provenance.decode_order_offsets == {"1": 7}
+    assert outcome.provenance.decode_order_offset == 7
+    assert outcome.provenance.decode_order_offset_source == "measured"
+    assert outcome.provenance.counts["skipped"] == 7
+
+    frames = _box_service(Path(outcome.directory).parent).reader().info().frames
+    rows = sorted(
+        (frame for frame in frames if frame.in_video and frame.segment == 1),
+        key=lambda frame: frame.segment_frame,
+    )
+    assert len(rows) == 33
+    assert outcome.provenance.first_output_frame_id == rows[7].frame_id == 27
+    assert outcome.provenance.last_output_frame_id == rows[-1].frame_id == 52
+    assert outcome.frames_written == 26
+    assert outcome.frames_expected == 26
+
+
+def test_the_box_result_round_trips_against_a_fresh_run_of_the_chain(box_pass):
+    """Frame for frame, the store equals what the chain computes now."""
+    service, outcome = box_pass
+    assert outcome.complete, outcome.reason
+    reader = AnalysisReader(outcome.directory)
+    try:
+        checked = 0
+        for live in service.recompute():
+            stored = reader.frame(live.row.frame_id)
+            assert stored.t_rel_s == live.row.t_rel_s
+            assert live.payload_matches(stored), f"frame {live.row.frame_id} differs"
+            checked += 1
+        assert checked == outcome.frames_written == len(reader) == 26
+    finally:
+        reader.close()
+
+
+# ------------------------------------------------------------------- keyframes
+def test_the_finger_result_stores_the_candidate_grid_and_the_mask_once(finger_pass):
+    """330 candidates on every frame, from one grid keyframe and one mask keyframe.
+
+    This is the measurement the format rests on: the candidate *location* set does not
+    move, so it is stored once and the per-frame record carries only values.
+    """
+    _, outcome = finger_pass
+    assert outcome.complete, outcome.reason
+    reader = AnalysisReader(outcome.directory)
+    try:
+        assert reader.keyframe_counts == {"grid": 1, "mask": 1}
+        payloads = list(reader.frames())
+        assert len(payloads) == 33
+        assert {payload.candidates for payload in payloads} == {330}
+        assert len({payload.grid_id for payload in payloads}) == 1
+
+        mask = reader.mask()
+        assert mask is not None and mask.shape == (300, 280)
+        packed = np.packbits(mask, axis=None).tobytes()
+        assert hashlib.blake2b(packed, digest_size=16).hexdigest() == STEP0_MASK_DIGEST
+
+        assert outcome.provenance.counts["candidates_min"] == 330
+        assert outcome.provenance.counts["candidates_max"] == 330
+        assert outcome.provenance.counts["active_zero_frames"] == 5
+        assert outcome.provenance.counts["active_total"] == 152
+    finally:
+        reader.close()
+
+
+def test_every_stored_frame_reconstructs_what_the_live_selector_returns(finger_pass):
+    """The keyframe path is exact: the grid is the one the selector chose, per frame."""
+    service, outcome = finger_pass
+    assert outcome.complete, outcome.reason
+    reader = AnalysisReader(outcome.directory)
+    try:
+        policy = service.policy
+        assert policy.flow_grid_spacing == 10 and policy.max_arrows == 400
+        for live in service.recompute():
+            stored = reader.frame(live.row.frame_id)
+            assert live.payload_matches(stored), f"frame {live.row.frame_id} differs"
+            rebuilt = reader.samples(live.row.frame_id)
+            assert np.array_equal(rebuilt.rows, np.asarray(live.samples.rows, np.int32))
+            assert np.array_equal(rebuilt.cols, np.asarray(live.samples.cols, np.int32))
+            assert np.array_equal(
+                np.asarray(rebuilt.magnitude, np.float32),
+                np.asarray(live.samples.magnitude, np.float32),
+            )
+            assert np.array_equal(rebuilt.active, np.asarray(live.samples.active, bool))
+    finally:
+        reader.close()
+
+
+def test_a_threshold_change_is_served_without_rebuilding_the_result(finger_pass):
+    """Display settings are not in the key because the record can answer a new threshold.
+
+    Activity is stored, but magnitude is stored too — so a viewer with a different
+    ``flow_on_px`` gets a correct answer from the same file.
+    """
+    _, outcome = finger_pass
+    reader = AnalysisReader(outcome.directory)
+    try:
+        frame_id = int(reader.frame_ids[0])
+        recorded = reader.samples(frame_id)
+        impossible = reader.samples(frame_id, policy=VisualPolicy(flow_on_px=1e9))
+        assert impossible.active_count == 0
+        permissive = reader.samples(
+            frame_id, policy=VisualPolicy(flow_on_px=0.0, flow_off_px=0.0)
+        )
+        assert permissive.active_count >= recorded.active_count
+        assert np.array_equal(permissive.magnitude, recorded.magnitude)
+        assert np.array_equal(
+            permissive.active, np.asarray(permissive.magnitude >= 0.0, bool)
+        )
+    finally:
+        reader.close()
+
+
+# ------------------------------------------------------------------ determinism
+def test_two_passes_over_the_same_input_store_byte_identical_payloads(
+    finger_pass, finger_pass_again
+):
+    """Re-running the pass produces the same bytes, so a cache hit cannot drift.
+
+    The determinism contract is the **frame records** and the **keyframe digests**. The
+    metadata is deliberately outside it: ``elapsed_s``, ``open_seconds``, ``started_at``
+    and ``finished_at`` describe the run, not the measurement, and two runs of the same
+    pass cannot share them.
+    """
+    _, first = finger_pass
+    _, second = finger_pass_again
+    assert first.complete and second.complete
+    reader_a = AnalysisReader(first.directory)
+    reader_b = AnalysisReader(second.directory)
+    try:
+        ids = reader_a.frame_ids
+        assert np.array_equal(ids, reader_b.frame_ids)
+        assert reader_a.frame_record_bytes == reader_b.frame_record_bytes
+        for frame_id in ids:
+            assert (
+                reader_a.frame(int(frame_id)).canonical_bytes()
+                == reader_b.frame(int(frame_id)).canonical_bytes()
+            ), f"frame {frame_id} is not reproducible"
+        assert {key: value.digest for key, value in reader_a.keyframes.items()} == {
+            key: value.digest for key, value in reader_b.keyframes.items()
+        }
+        assert reader_a.mask_digest == reader_b.mask_digest
+    finally:
+        reader_a.close()
+        reader_b.close()
+
+
+# ------------------------------------------------------------------ cancellation
+def test_an_interrupted_pass_is_incomplete_and_is_not_served(tmp_path):
+    """A cancelled pass leaves a directory that refuses to open by default."""
+    _require_inputs()
+    root = tmp_path / "cache"
+    cancelled = threading.Event()
+    seen: list[int] = []
+
+    def on_progress(progress):
+        seen.append(progress.frame_id)
+        if len(seen) >= 5:
+            cancelled.set()
+
+    service = _finger_service(root, on_progress=on_progress, cancelled=cancelled)
+    outcome = service.run()
+
+    assert outcome.complete is False
+    assert outcome.reason == "cancelled"
+    assert 0 < outcome.frames_written < 33
+    assert (outcome.directory / INCOMPLETE_MARKER).is_file()
+    meta = json.loads((outcome.directory / META_NAME).read_text(encoding="utf-8"))
+    assert meta["complete"] is False
+    assert meta["stopped_reason"] == "cancelled"
+    assert meta["provenance"]["complete"] is False
+    assert [item.complete for item in scan_results(root)] == [False]
+
+    with pytest.raises(AnalysisIncompleteError):
+        AnalysisReader(outcome.directory)
+
+    # The partial data is still inspectable on purpose, and says what it is.
+    partial = AnalysisReader(outcome.directory, allow_incomplete=True)
+    try:
+        assert partial.complete is False
+        assert len(partial) == outcome.frames_written
+        assert "incomplete" in partial.stale_reasons()
+    finally:
+        partial.close()
+
+
+def test_a_declared_segment_that_is_missing_makes_the_result_incomplete(tmp_path):
+    """A segment the session declares and the directory lacks is a failure, not a short read."""
+    _require_inputs()
+    recording = tmp_path / "recording"
+    recording.mkdir()
+    shutil.copy2(RECORDING / "frames.csv", recording / "frames.csv")
+    shutil.copy2(RECORDING / "forces.csv", recording / "forces.csv")
+    shutil.copy2(RECORDING / "events.csv", recording / "events.csv")
+    session = json.loads((RECORDING / "session.json").read_text(encoding="utf-8"))
+    session["stats"]["segments"] = 2
+    (recording / "session.json").write_text(json.dumps(session), encoding="utf-8")
+    try:
+        os.link(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4")
+    except OSError:
+        shutil.copy2(RECORDING / "raw_0001.mp4", recording / "raw_0001.mp4")
+
+    service = AnalysisService(
+        recording, chain=_finger_chain(), cache=ResultCache(tmp_path / "cache")
+    )
+    outcome = service.run()
+    assert outcome.complete is False
+    assert outcome.reason == "segments_missing"
+    assert outcome.frames_written == 0
+    assert outcome.provenance.segments == [1]
+    assert outcome.provenance.segments_declared == 2
+    assert outcome.provenance.segments_missing == [2]
+    with pytest.raises(AnalysisIncompleteError):
+        AnalysisReader(outcome.directory)
+
+
+def test_a_reference_frame_outside_the_recording_is_refused(tmp_path):
+    """The reset is applied at a named frame or the pass does not run.
+
+    Frame ids come from the acquisition session and need not start at zero or one: this
+    recording's run 20…52. A requested reference of 16 therefore names nothing, and
+    ignoring that would store a result whose ``reference_frame_id`` is not the frame
+    that was asked for.
+    """
+    _require_inputs()
+    service = _finger_service(tmp_path / "cache", reference_frame=16)
+    with pytest.raises(ValueError, match="reference frame 16"):
+        service.run()
+    assert not (tmp_path / "cache").exists(), (
+        "a refused pass must not leave a directory"
+    )
+
+
+# ------------------------------------------------------------------ reference reset
+def test_resetting_the_reference_records_the_frame_and_clears_the_tracker(
+    finger_pass, finger_reset_pass
+):
+    """The reset is exact at the named frame and touches nothing before it.
+
+    ``Sensor.reset`` clears the optical-flow tracker, so the frame the reset is applied
+    at measures against an empty reference: its displacement is exactly zero — the
+    Step 0 signature. The frames before it are byte-identical to a pass with no reset,
+    which is what makes "reference frame" a statement about a position rather than about
+    a slightly different run.
+    """
+    _, plain = finger_pass
+    _, reset = finger_reset_pass
+    assert reset.complete, reset.reason
+    assert reset.provenance.reference_is_explicit is True
+    assert reset.provenance.reference_frame_id == REFERENCE_FRAME
+    assert reset.provenance.first_output_frame_id == 20
+    assert reset.provenance.counts["reset_count"] == 1
+    assert reset.provenance.counts["reset_seconds"] >= 0.0
+    # With no explicit reset the reference is the pass's own clean start.
+    assert plain.provenance.reference_is_explicit is False
+    assert plain.provenance.reference_frame_id == plain.provenance.first_output_frame_id
+
+    reader = AnalysisReader(plain.directory)
+    reset_reader = AnalysisReader(reset.directory)
+    try:
+        before = int(REFERENCE_FRAME) - 1
+        assert reset_reader.frame(REFERENCE_FRAME).magnitude.max() == 0.0
+        assert reader.frame(REFERENCE_FRAME).magnitude.max() > 0.0
+        assert reset_reader.frame(before).magnitude.max() > 0.0
+
+        for frame_id in reader.frame_ids:
+            if int(frame_id) < REFERENCE_FRAME:
+                assert (
+                    reader.frame(int(frame_id)).canonical_bytes()
+                    == reset_reader.frame(int(frame_id)).canonical_bytes()
+                ), f"frame {frame_id} before the reference was disturbed by the reset"
+        assert (
+            reader.frame(int(REFERENCE_FRAME)).canonical_bytes()
+            != reset_reader.frame(int(REFERENCE_FRAME)).canonical_bytes()
+        )
+    finally:
+        reader.close()
+        reset_reader.close()
+
+
+# ------------------------------------------------------------------ the key
+def test_the_key_follows_the_inputs_that_change_the_result_and_no_others(finger_pass):
+    """Five components, and display settings are deliberately not one of them."""
+    base = dict(
+        recording="rec-a",
+        sensor_config="config-a",
+        calibration="cal-a",
+        reference_frame=36,
+    )
+    digest = result_key(**base).digest
+    assert result_key(**base).digest == digest
+    assert result_key(**{**base, "sensor_config": "config-b"}).digest != digest
+    assert result_key(**{**base, "algorithm": "other-chain"}).digest != digest
+    assert result_key(**{**base, "calibration": "cal-b"}).digest != digest
+    assert result_key(**{**base, "reference_frame": None}).digest != digest
+    assert result_key(**{**base, "recording": "rec-b"}).digest != digest
+
+    # Nothing in the signature can carry a display setting, so a change to one cannot
+    # move the key even by accident.
+    parameters = set(inspect.signature(result_key).parameters)
+    assert not {name for name in parameters if "policy" in name or "display" in name}
+    assert "policy" not in parameters
+
+    # And the concrete consequence: two services that differ only in the policy the
+    # samples are selected with address the same stored result.
+    service = _finger_service(finger_pass[1].directory.parent)
+    other = _finger_service(finger_pass[1].directory.parent)
+    other.policy = VisualPolicy(flow_on_px=1.5, max_arrows=1)
+    assert (
+        service.dependency_report()["key"].digest
+        == other.dependency_report()["key"].digest
+        == finger_pass[1].key.digest
+    )
+    recorded = finger_pass[1].provenance.policy
+    assert recorded["flow_on_px"] == 0.4 and recorded["max_arrows"] == 400
+
+
+# ------------------------------------------------------------------ size
+def test_the_stored_size_is_the_measured_encoding(box_pass, finger_pass):
+    """State the encoding, then hold it: 4,034 B per frame at the finger grid.
+
+    Implicit candidate grid (a keyframe) + float32 ``dx``/``dy``/``magnitude`` +
+    bit-packed ``active`` + a 24-byte header + the 8-byte validity digest:
+    ``24 + 330*12 + 42 + 8 = 4,034`` B. The alternative measured encodings are float16
+    ``dxy``+magnitude (2,022 B) and dropping magnitude (4,002 B with the grid, no
+    header); float32 is chosen because the round trip has to be *exact*, and the
+    magnitude is what serves a changed threshold.
+    """
+    assert frame_payload_bytes(330, with_validity=True) == 4034
+    assert frame_payload_bytes(330, with_validity=False) == 4026
+    assert frame_payload_bytes(400, with_validity=False) == 4874
+
+    _, finger = finger_pass
+    reader = AnalysisReader(finger.directory)
+    try:
+        assert reader.bytes_per_frame == 4034.0
+        assert reader.frame_record_bytes == 33 * 4034
+        assert reader.bytes_per_frame <= 4096, "the stated per-frame bound"
+        total_per_frame = _directory_bytes(finger.directory) / len(reader)
+        assert total_per_frame <= 4600, "keyframes and metadata included"
+    finally:
+        reader.close()
+
+    _, box = box_pass
+    box_reader = AnalysisReader(box.directory)
+    try:
+        # The box chain has no validity mask, so there is no digest and every frame
+        # carries its own grid keyframe: 4,874 B of record per frame at 400 candidates.
+        assert box_reader.frame_record_bytes == 26 * 4874
+        assert box_reader.keyframe_counts == {"grid": 26}
+        assert box_reader.mask() is None
+    finally:
+        box_reader.close()
+
+
+# ------------------------------------------------------------------ reader
+def test_a_stored_result_is_served_without_the_recording_or_a_replay(
+    finger_pass, tmp_path
+):
+    """A seek is a file read: the result answers with the recording and SDK out of reach."""
+    _, outcome = finger_pass
+    copy = tmp_path / "copied-result"
+    shutil.copytree(outcome.directory, copy)
+
+    original = AnalysisReader(outcome.directory)
+    copied = AnalysisReader(copy)
+    try:
+        assert np.array_equal(original.frame_ids, copied.frame_ids)
+        for frame_id in original.frame_ids:
+            assert (
+                original.frame(int(frame_id)).canonical_bytes()
+                == copied.frame(int(frame_id)).canonical_bytes()
+            )
+        # Random access by frame id and by timestamp both land on the right frame.
+        times = copied.times
+        middle = len(times) // 2
+        assert copied.frame(int(copied.frame_ids[middle])).t_rel_s == times[middle]
+        assert copied.frame_at(float(times[middle])).frame_id == int(
+            copied.frame_ids[middle]
+        )
+        assert copied.frame_at(float(times[middle]) + 1e-6).frame_id == int(
+            copied.frame_ids[middle]
+        )
+        assert copied.frame_at(-1.0).frame_id == int(copied.frame_ids[0])
+        assert copied.frame_at(float(times[-1]) + 100.0).frame_id == int(
+            copied.frame_ids[-1]
+        )
+        window = list(
+            copied.frames(start=int(copied.frame_ids[1]), stop=int(copied.frame_ids[2]))
+        )
+        assert [payload.frame_id for payload in window] == [
+            int(copied.frame_ids[1]),
+            int(copied.frame_ids[2]),
+        ]
+        assert copied.has(int(copied.frame_ids[0]))
+        assert not copied.has(10_000)
+    finally:
+        original.close()
+        copied.close()
+
+
+def test_eviction_removes_whole_results_against_the_cap(tmp_path):
+    """The cap is enforced by removing complete results, oldest first, never a mistake."""
+    root = tmp_path / "cache"
+    root.mkdir()
+    made: list[Path] = []
+    for index in range(3):
+        directory = root / f"result-{index}"
+        directory.mkdir()
+        (directory / META_NAME).write_text(
+            json.dumps({"complete": True, "key": f"result-{index}", "format": "x"}),
+            encoding="utf-8",
+        )
+        (directory / "blob.bin").write_bytes(b"\0" * 4096)
+        os.utime(directory / META_NAME, (1_600_000_000 + index, 1_600_000_000 + index))
+        made.append(directory)
+
+    def size_of(path: Path) -> int:
+        return sum(item.stat().st_size for item in path.rglob("*") if item.is_file())
+
+    kept = made[2]
+    # A cap that one removal satisfies: the oldest complete result goes, and the
+    # protected one is never a candidate whatever the arithmetic says.
+    cap = sum(size_of(item) for item in made) - size_of(made[0])
+    removed = evict(root, max_bytes=cap, keep=[kept])
+    assert removed == [made[0]]
+    assert made[0].exists() is False
+    assert kept.is_dir() and made[1].is_dir()
+    assert [item.directory.name for item in scan_results(root)] == [
+        "result-1",
+        "result-2",
+    ]
+
+    # A dry run reports without deleting, and still never proposes the protected result.
+    assert evict(root, max_bytes=1, dry_run=True, keep=[kept]) == [made[1]]
+    assert made[1].is_dir()
+    assert kept.is_dir()
+
+
+# ------------------------------------------------------------------ incompleteness gate
+def test_a_result_without_metadata_is_never_served(tmp_path):
+    """The only thing that opens is a directory whose metadata says it finished."""
+    directory = tmp_path / "empty-result"
+    directory.mkdir()
+    from orisys_app_shared.analysis.reader import AnalysisNotFoundError
+
+    with pytest.raises(AnalysisNotFoundError):
+        AnalysisReader(directory)
+
+    (directory / META_NAME).write_text(
+        json.dumps({"format": analysis_model.ANALYSIS_FORMAT, "format_version": 1}),
+        encoding="utf-8",
+    )
+    with pytest.raises(AnalysisIncompleteError):
+        AnalysisReader(directory)