From 38d288186c1c9c61021ce8adb4e1973460619d04 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Thu, 2 Jul 2026 23:46:04 -0400
Subject: [PATCH 01/13] fix(sessions): stream Range responses in 1MiB chunks;
 416 on inverted ranges

---
 backend/routers/sessions.py          | 38 +++++++++++++++++++++++-----
 tests/backend/test_sessions_video.py | 20 +++++++++++++++
 2 files changed, 51 insertions(+), 7 deletions(-)

diff --git a/backend/routers/sessions.py b/backend/routers/sessions.py
index fecf119..5e6c26b 100644
--- a/backend/routers/sessions.py
+++ b/backend/routers/sessions.py
@@ -64,8 +64,18 @@ def get_session(session_id: str) -> dict:
     return summary
 
 
+_RANGE_CHUNK = 1024 * 1024  # 1 MiB per read: bounded memory however large the range
+
+
 def _serve_range(file_path: Path, range_header: str) -> Response:
-    """Parse a Range header and return a 206 Partial Content response."""
+    """Parse a Range header and return a 206 Partial Content response.
+
+    Streams the range in chunks instead of materializing it — browsers
+    open <video> with ``bytes=0-``, which previously buffered the whole
+    recording into memory per request (and again on every scrub).
+    """
+    from fastapi.responses import StreamingResponse
+
     size = file_path.stat().st_size
     spec = range_header[len("bytes="):].split(",", 1)[0].strip()
     start_str, _, end_str = spec.partition("-")
@@ -82,19 +92,33 @@ def _serve_range(file_path: Path, range_header: str) -> Response:
         if start < 0 or start >= size:
             raise HTTPException(416, "range out of bounds")
     end = min(end, size - 1)
+    if end < start:
+        # Inverted range (e.g. bytes=500-100): previously produced a
+        # negative length and a read-to-EOF body that contradicted
+        # Content-Length. 416 per RFC 9110.
+        raise HTTPException(416, "range out of bounds")
     length = end - start + 1
-    with open(file_path, "rb") as f:
-        f.seek(start)
-        data = f.read(length)
-    return Response(
-        content=data,
+
+    def _iter():
+        remaining = length
+        with open(file_path, "rb") as f:
+            f.seek(start)
+            while remaining > 0:
+                chunk = f.read(min(_RANGE_CHUNK, remaining))
+                if not chunk:
+                    break
+                remaining -= len(chunk)
+                yield chunk
+
+    return StreamingResponse(
+        _iter(),
         status_code=206,
         headers={
             "Content-Range": f"bytes {start}-{end}/{size}",
             "Accept-Ranges": "bytes",
             "Content-Length": str(length),
-            "Content-Type": "video/mp4",
         },
+        media_type="video/mp4",
     )
 
 
diff --git a/tests/backend/test_sessions_video.py b/tests/backend/test_sessions_video.py
index ba4f6a4..8a1ce22 100644
--- a/tests/backend/test_sessions_video.py
+++ b/tests/backend/test_sessions_video.py
@@ -56,3 +56,23 @@ def test_video_missing_returns_404(client, sessions_root_with_video):
     (s / "video.mp4").unlink()
     r = client.get("/api/sessions/2026-01-01_12-00-00/video")
     assert r.status_code == 404
+
+
+def test_inverted_range_returns_416(client, sessions_root_with_video):
+    r = client.get(
+        "/api/sessions/2026-01-01_12-00-00/video",
+        headers={"Range": "bytes=500-100"},
+    )
+    assert r.status_code == 416
+
+
+def test_open_ended_range_streams_correct_bytes(client, sessions_root_with_video):
+    full = client.get("/api/sessions/2026-01-01_12-00-00/video").content
+    r = client.get(
+        "/api/sessions/2026-01-01_12-00-00/video",
+        headers={"Range": "bytes=10-"},
+    )
+    assert r.status_code == 206
+    assert r.headers["Content-Range"] == f"bytes 10-{len(full)-1}/{len(full)}"
+    assert int(r.headers["Content-Length"]) == len(full) - 10
+    assert r.content == full[10:]

From e89f304b693ecdadde816247c34e8c1b4bd6d628 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Thu, 2 Jul 2026 23:51:38 -0400
Subject: [PATCH 02/13] perf(analyze): copy uploads off the event loop; reuse
 detector across same-config items

---
 backend/routers/analyze.py                   | 32 +++++++++++++++---
 tests/backend/test_analyze_detector_cache.py | 34 ++++++++++++++++++++
 2 files changed, 62 insertions(+), 4 deletions(-)
 create mode 100644 tests/backend/test_analyze_detector_cache.py

diff --git a/backend/routers/analyze.py b/backend/routers/analyze.py
index 0a8b429..6afe862 100644
--- a/backend/routers/analyze.py
+++ b/backend/routers/analyze.py
@@ -87,8 +87,14 @@ async def add_to_queue(
     # a path-traversal ("../") into the saved location.
     safe_name = Path(file.filename or "upload").name or "upload"
     saved = _UPLOAD_DIR / f"{int(time.time() * 1000)}_{safe_name}"
-    with open(saved, "wb") as out:
-        shutil.copyfileobj(file.file, out)
+    # The upload can be a multi-GB video: copy it in the default executor
+    # so the event loop keeps serving /api/live/frame etc. meanwhile.
+    def _save() -> None:
+        with open(saved, "wb") as out:
+            shutil.copyfileobj(file.file, out)
+
+    await asyncio.get_running_loop().run_in_executor(None, _save)
+    await file.close()
 
     try:
         item = AnalyzeQueueItem(
@@ -258,6 +264,24 @@ async def stop_run(request: Request) -> dict:
     return {"status": "stopped"}
 
 
+async def _get_or_build_detector(app, cfg: DetectorConfig):
+    """Build a detector, reusing the previous one when the config matches.
+
+    Model load is multi-second and memory-spiky; batch-running N items with
+    the same preset previously paid it N times. Size-1 cache: a config
+    change drops the old detector (freeing its weights) before building.
+    DetectorConfig is a frozen dataclass, so equality comparison is exact.
+    """
+    cached = getattr(app.state, "analyze_detector_cache", None)
+    if cached is not None and cached[0] == cfg:
+        return cached[1]
+    app.state.analyze_detector_cache = None  # release old weights first
+    loop = asyncio.get_running_loop()
+    detector = await loop.run_in_executor(None, build_detector, cfg)
+    app.state.analyze_detector_cache = (cfg, detector)
+    return detector
+
+
 async def _runner_loop(app, req: RunRequest) -> None:
     """Drain the queue one item at a time on the asyncio loop.
 
@@ -276,7 +300,7 @@ async def _runner_loop(app, req: RunRequest) -> None:
 
         loop = asyncio.get_running_loop()
         # Build a fresh detector per item so different items can use
-        # different model configs. (Future: cache by config hash.)
+        # different model configs.
         cfg = DetectorConfig(
             detector_type=item.pipeline.detector_type,
             face_model=item.pipeline.face_model,
@@ -286,7 +310,7 @@ async def _runner_loop(app, req: RunRequest) -> None:
             identity_model=item.pipeline.identity_model,
             device=req.compute,
         )
-        detector = await loop.run_in_executor(None, build_detector, cfg)
+        detector = await _get_or_build_detector(app, cfg)
 
         events: asyncio.Queue = asyncio.Queue()
 
diff --git a/tests/backend/test_analyze_detector_cache.py b/tests/backend/test_analyze_detector_cache.py
new file mode 100644
index 0000000..898eb81
--- /dev/null
+++ b/tests/backend/test_analyze_detector_cache.py
@@ -0,0 +1,34 @@
+"""Consecutive queue items with the same config reuse one detector build."""
+
+import asyncio
+
+import pytest
+
+from pyfeatlive_core.detector import DetectorConfig
+
+
+def test_same_config_builds_once(client, monkeypatch):
+    from backend.routers import analyze as analyze_router
+
+    builds: list[DetectorConfig] = []
+
+    def _fake_build(cfg):
+        builds.append(cfg)
+        return object()
+
+    monkeypatch.setattr(analyze_router, "build_detector", _fake_build)
+
+    app = client.app
+    cfg = DetectorConfig(device="cpu")
+
+    async def _run():
+        d1 = await analyze_router._get_or_build_detector(app, cfg)
+        d2 = await analyze_router._get_or_build_detector(app, cfg)
+        assert d1 is d2
+        d3 = await analyze_router._get_or_build_detector(
+            app, DetectorConfig(device="cpu", detector_type="MPDetector"),
+        )
+        assert d3 is not d1
+
+    asyncio.run(_run())
+    assert len(builds) == 2  # cpu Detectorv2 once, MPDetector once

From 6414492c9174a4a2d9f335e2683177e124ded9c3 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Thu, 2 Jul 2026 23:54:47 -0400
Subject: [PATCH 03/13] docs(analyze): record the two-resident-detectors
 tradeoff on the cache

---
 backend/routers/analyze.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/backend/routers/analyze.py b/backend/routers/analyze.py
index 6afe862..7f69fb4 100644
--- a/backend/routers/analyze.py
+++ b/backend/routers/analyze.py
@@ -271,6 +271,12 @@ async def _get_or_build_detector(app, cfg: DetectorConfig):
     the same preset previously paid it N times. Size-1 cache: a config
     change drops the old detector (freeing its weights) before building.
     DetectorConfig is a frozen dataclass, so equality comparison is exact.
+
+    ACCEPTED TRADEOFF: the cache persists after the run finishes (so a
+    re-run with the same preset skips the reload), which means a user who
+    analyzes and then uses the Live page has TWO detectors resident
+    (this one + live.detector). Acceptable on the target machines; if it
+    ever bites, evict here on queue_idle or from live /configure.
     """
     cached = getattr(app.state, "analyze_detector_cache", None)
     if cached is not None and cached[0] == cfg:

From 7b4333a4e957326d6af04eb3e6a09b710ca6ec57 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Thu, 2 Jul 2026 23:56:56 -0400
Subject: [PATCH 04/13] fix(analyze): reap WS subscribers on disconnect; never
 drop terminal events

---
 backend/routers/analyze.py       | 40 ++++++++++++++++++++++++++------
 tests/backend/test_analyze_ws.py | 31 +++++++++++++++++++++++++
 2 files changed, 64 insertions(+), 7 deletions(-)
 create mode 100644 tests/backend/test_analyze_ws.py

diff --git a/backend/routers/analyze.py b/backend/routers/analyze.py
index 7f69fb4..05d3d22 100644
--- a/backend/routers/analyze.py
+++ b/backend/routers/analyze.py
@@ -354,7 +354,14 @@ def _broadcast(app, payload: dict) -> None:
         try:
             sub.put_nowait(payload)
         except asyncio.QueueFull:
-            pass
+            # Evict the oldest event to make room: dropping a stale
+            # `progress` is harmless, dropping a terminal `done`/`failed`/
+            # `queue_idle` leaves the UI showing a stuck-running row.
+            try:
+                sub.get_nowait()
+                sub.put_nowait(payload)
+            except (asyncio.QueueEmpty, asyncio.QueueFull):
+                pass
 
 
 # ----- WS ------------------------------------------------------------
@@ -371,12 +378,31 @@ async def analyze_ws(ws: WebSocket) -> None:
             "type": "snapshot",
             "items": [_item_to_dict(i) for i in ws.app.state.analyze_queue.items()],
         })
-        while True:
-            ev = await q.get()
-            try:
-                await ws.send_json(ev)
-            except WebSocketDisconnect:
-                break
+        # Race a receive against the event queue. The client never sends
+        # application messages, so a completed receive means close/disconnect
+        # — without it, a dead client parks this coroutine on q.get()
+        # forever and every reconnect leaks another subscriber queue.
+        recv_task = asyncio.create_task(ws.receive())
+        try:
+            while True:
+                get_task = asyncio.create_task(q.get())
+                done, _ = await asyncio.wait(
+                    {recv_task, get_task}, return_when=asyncio.FIRST_COMPLETED,
+                )
+                if recv_task in done:
+                    get_task.cancel()
+                    break  # disconnect (or any client frame): exit + clean up
+                ev = get_task.result()
+                try:
+                    await ws.send_json(ev)
+                except Exception:
+                    # Dead socket surfaces as RuntimeError/ConnectionClosed,
+                    # NOT WebSocketDisconnect — treat any send failure as gone.
+                    break
+        finally:
+            recv_task.cancel()
+    except WebSocketDisconnect:
+        pass
     finally:
         try:
             ws.app.state.analyze_subscribers.remove(q)
diff --git a/tests/backend/test_analyze_ws.py b/tests/backend/test_analyze_ws.py
new file mode 100644
index 0000000..5572f13
--- /dev/null
+++ b/tests/backend/test_analyze_ws.py
@@ -0,0 +1,31 @@
+"""Analyze WS: subscribers are reaped on disconnect; terminal events survive overflow."""
+
+import asyncio
+
+
+def test_subscriber_removed_after_disconnect(client):
+    app = client.app
+    assert len(app.state.analyze_subscribers) == 0
+    with client.websocket_connect("/api/analyze/ws") as ws:
+        snap = ws.receive_json()
+        assert snap["type"] == "snapshot"
+        assert len(app.state.analyze_subscribers) == 1
+    # Context exit closes the socket; the handler must observe the close
+    # (via its receive task) and remove the subscriber promptly.
+    assert len(app.state.analyze_subscribers) == 0
+
+
+def test_broadcast_overflow_keeps_newest(client):
+    from backend.routers import analyze as analyze_router
+
+    app = client.app
+    q: asyncio.Queue = asyncio.Queue(maxsize=2)
+    app.state.analyze_subscribers.append(q)
+    try:
+        analyze_router._broadcast(app, {"type": "progress", "n": 1})
+        analyze_router._broadcast(app, {"type": "progress", "n": 2})
+        analyze_router._broadcast(app, {"type": "done"})  # queue full here
+        drained = [q.get_nowait(), q.get_nowait()]
+        assert {"type": "done"} in drained  # newest survived (oldest evicted)
+    finally:
+        app.state.analyze_subscribers.remove(q)

From 03e1fac0c7fbbf72964b6ac695b42b863c0c0ca3 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:03:41 -0400
Subject: [PATCH 05/13] =?UTF-8?q?fix(analyze):=20scale=20detect=20batch=20?=
 =?UTF-8?q?by=20resolution=20=E2=80=94=204K=20sources=20no=20longer=20OOM?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pyfeatlive_core/analyze_runner.py | 20 +++++++++++++++++++-
 tests/core/test_analyze_runner.py | 17 +++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/pyfeatlive_core/analyze_runner.py b/pyfeatlive_core/analyze_runner.py
index 9f78685..9fb30f9 100644
--- a/pyfeatlive_core/analyze_runner.py
+++ b/pyfeatlive_core/analyze_runner.py
@@ -26,6 +26,20 @@
 _VIDEO_SUFFIXES = {".mp4", ".mov", ".avi", ".mkv"}
 _IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png"}
 
+# Pixel budget per detect batch: what the tuned default costs (8 frames of
+# 720p). detect.py stacks the batch as full-res float32 and Detectorv2
+# immediately copies it again — an unscaled 8x4K batch is ~1.5GB of
+# transient tensors, a real OOM on 16GB machines.
+_BATCH_PIXEL_BUDGET = 8 * 1280 * 720
+
+
+def _effective_batch_size(requested: int, width: int, height: int) -> int:
+    """Clamp the batch size so total batch pixels stay within budget."""
+    pixels = width * height
+    if pixels <= 0:
+        return requested
+    return max(1, min(requested, _BATCH_PIXEL_BUDGET // pixels))
+
 
 def _load_rgb_image(src: Path) -> Image.Image:
     """Load an image as RGB, closing the source file handle (``.convert``
@@ -207,6 +221,9 @@ def run_item(
     # thread + open file leak and an empty session dir is orphaned).
     recorder = None
     recorder_closed = False
+    # Bound on every path (video and single-image) so it's never
+    # unassigned by the time the batch-flush check below runs.
+    effective_batch = batch_size
     try:
         if is_video:
             total = _count_video_frames(src)
@@ -216,6 +233,7 @@ def run_item(
             with Image.open(src) as im:
                 vid_w, vid_h = im.size
             vid_fps = 1.0
+        effective_batch = _effective_batch_size(batch_size, vid_w, vid_h)
         item.total_frames = total
         yield {"type": "started", "item_id": item.id, "total_frames": total}
 
@@ -292,7 +310,7 @@ def _cancelled() -> bool:
                 break
             batch.append(img)
             offsets.append(idx)
-            if len(batch) >= batch_size:
+            if len(batch) >= effective_batch:
                 _drain_batch(batch, offsets)
                 frames_done += len(batch)
                 item.progress_frames = frames_done
diff --git a/tests/core/test_analyze_runner.py b/tests/core/test_analyze_runner.py
index 733f0cb..5421e8a 100644
--- a/tests/core/test_analyze_runner.py
+++ b/tests/core/test_analyze_runner.py
@@ -129,3 +129,20 @@ def _boom(*args, **kwargs):
     # close() ran → the empty session dir was removed, none orphaned.
     leftover = list(run_root.iterdir()) if run_root.exists() else []
     assert leftover == [], f"orphaned session dir after error: {leftover}"
+
+
+def test_effective_batch_size_scales_with_resolution():
+    from pyfeatlive_core.analyze_runner import _effective_batch_size
+
+    # 720p and below: the requested batch stands.
+    assert _effective_batch_size(8, 1280, 720) == 8
+    assert _effective_batch_size(8, 640, 360) == 8
+    # 4K (3840x2160 = 9x the 720p pixel count): 8 -> 1 to keep the
+    # float32 batch tensor within the tuned 720p budget.
+    assert _effective_batch_size(8, 3840, 2160) == 1
+    # 1440p (~2.7x): 8 -> 2.
+    assert _effective_batch_size(8, 2560, 1440) == 2
+    # Never below 1; never above the request; degenerate dims are safe.
+    assert _effective_batch_size(8, 7680, 4320) == 1
+    assert _effective_batch_size(2, 640, 360) == 2
+    assert _effective_batch_size(8, 0, 0) == 8

From 478dd104cb2ba0f527bb7571e4b5ef15ee01f63b Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:07:31 -0400
Subject: [PATCH 06/13] perf(analyze): estimate frame counts without decoding;
 seek to clip_start

---
 pyfeatlive_core/analyze_runner.py | 51 +++++++++++++++++++++++++++----
 tests/core/test_analyze_runner.py | 50 ++++++++++++++++++++++++++++++
 2 files changed, 95 insertions(+), 6 deletions(-)

diff --git a/pyfeatlive_core/analyze_runner.py b/pyfeatlive_core/analyze_runner.py
index 9fb30f9..9a65dee 100644
--- a/pyfeatlive_core/analyze_runner.py
+++ b/pyfeatlive_core/analyze_runner.py
@@ -62,16 +62,31 @@ def _iter_video_frames(
     try:
         stream = container.streams.video[0]
         fps = float(stream.average_rate or 30)
-        start_pts = (vp.clip_start or 0) * fps
-        end_pts = float("inf") if vp.clip_end is None else vp.clip_end * fps
         step = max(1, int(vp.skip_frames))
+        start_idx = int((vp.clip_start or 0) * fps)
+        end_idx = float("inf") if vp.clip_end is None else vp.clip_end * fps
+        tb = stream.time_base
+        if vp.clip_start and tb:
+            # Jump to the nearest keyframe at/before clip_start instead of
+            # decoding (and discarding) everything from t=0 — a clip_start
+            # 10 minutes in previously decoded ~18k dead frames.
+            try:
+                container.seek(int(vp.clip_start / tb), stream=stream)
+            except Exception:
+                pass  # unusual container: fall back to decode-from-start
+        pos = None  # positional fallback when frames carry no PTS
         try:
-            for i, frame in enumerate(container.decode(stream)):
-                if i < start_pts:
+            for frame in container.decode(stream):
+                if frame.pts is not None and tb:
+                    i = int(round(float(frame.pts * tb) * fps))
+                else:
+                    pos = (pos + 1) if pos is not None else 0
+                    i = pos
+                if i < start_idx:
                     continue
-                if i > end_pts:
+                if i > end_idx:
                     break
-                if (i - int(start_pts)) % step != 0:
+                if (i - start_idx) % step != 0:
                     continue
                 try:
                     img = frame.to_image()
@@ -166,12 +181,36 @@ def _transcode_to_h264(src: Path, dst: Path) -> None:
             dst.unlink(missing_ok=True)
 
 
+def _estimate_frames(stream) -> int | None:
+    """Frame-count estimate from duration x rate — no decode.
+
+    Returns None when the container reports neither duration nor rate.
+    Used for the progress denominator only, so an off-by-a-few estimate
+    is fine; the alternative (full decode just to count) doubled job time
+    on containers that omit stream.frames.
+    """
+    rate = stream.average_rate
+    if not rate:
+        return None
+    dur = stream.duration
+    if dur is not None and stream.time_base is not None:
+        return int(float(dur * stream.time_base) * float(rate))
+    cdur = getattr(stream.container, "duration", None)
+    if cdur:
+        import av as _av
+        return int((cdur / _av.time_base) * float(rate))
+    return None
+
+
 def _count_video_frames(path: Path) -> int:
     container = av.open(str(path))
     try:
         stream = container.streams.video[0]
         if stream.frames and stream.frames > 0:
             return int(stream.frames)
+        estimate = _estimate_frames(stream)
+        if estimate is not None and estimate > 0:
+            return estimate
         n = 0
         for _ in container.decode(stream):
             n += 1
diff --git a/tests/core/test_analyze_runner.py b/tests/core/test_analyze_runner.py
index 5421e8a..8f6a37e 100644
--- a/tests/core/test_analyze_runner.py
+++ b/tests/core/test_analyze_runner.py
@@ -146,3 +146,53 @@ def test_effective_batch_size_scales_with_resolution():
     assert _effective_batch_size(8, 7680, 4320) == 1
     assert _effective_batch_size(2, 640, 360) == 2
     assert _effective_batch_size(8, 0, 0) == 8
+
+
+import av as _av
+
+
+def _make_video(path, n_frames=60, fps=30, size=(64, 64)):
+    import numpy as np
+    container = _av.open(str(path), "w")
+    stream = container.add_stream("h264", rate=fps)
+    stream.width, stream.height = size
+    stream.pix_fmt = "yuv420p"
+    for i in range(n_frames):
+        arr = np.full((size[1], size[0], 3), (i * 4) % 255, dtype=np.uint8)
+        frame = _av.VideoFrame.from_ndarray(arr, format="rgb24")
+        for pkt in stream.encode(frame):
+            container.mux(pkt)
+    for pkt in stream.encode():
+        container.mux(pkt)
+    container.close()
+    return path
+
+
+def test_estimate_frames_close_to_actual(tmp_path):
+    from pyfeatlive_core.analyze_runner import _estimate_frames
+    p = _make_video(tmp_path / "v.mp4", n_frames=60, fps=30)
+    c = _av.open(str(p))
+    try:
+        est = _estimate_frames(c.streams.video[0])
+    finally:
+        c.close()
+    assert est is not None and abs(est - 60) <= 3
+
+
+def test_clip_start_seeks_not_decodes(tmp_path):
+    from pyfeatlive_core.analyze_runner import _iter_video_frames
+    from pyfeatlive_core.analyze_queue import VideoParams
+    p = _make_video(tmp_path / "v.mp4", n_frames=90, fps=30)
+    vp = VideoParams(skip_frames=1, clip_start=2.0, clip_end=None,
+                     track_identities=False)
+    frames = list(_iter_video_frames(p, vp))
+    # First yielded source index must be ~frame 60 (2.0s * 30fps); the
+    # seek lands on the nearest PRIOR keyframe, so indices before 60 must
+    # still be filtered out, and indexing must reflect SOURCE positions.
+    assert frames, "no frames yielded"
+    first_idx = frames[0][0]
+    assert 58 <= first_idx <= 62
+    # Frame indices remain source-referenced and increasing.
+    idxs = [i for i, _ in frames]
+    assert idxs == sorted(idxs)
+    assert idxs[-1] <= 90

From 0e588ce4c0c4efacd227acb9711b17ff69591711 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:17:06 -0400
Subject: [PATCH 07/13] fix(analyze): compare clip_start boundary against the
 untruncated frame position

---
 pyfeatlive_core/analyze_runner.py | 10 +++++++---
 tests/core/test_analyze_runner.py | 21 +++++++++++++++++++++
 2 files changed, 28 insertions(+), 3 deletions(-)

diff --git a/pyfeatlive_core/analyze_runner.py b/pyfeatlive_core/analyze_runner.py
index 9a65dee..288743c 100644
--- a/pyfeatlive_core/analyze_runner.py
+++ b/pyfeatlive_core/analyze_runner.py
@@ -63,7 +63,11 @@ def _iter_video_frames(
         stream = container.streams.video[0]
         fps = float(stream.average_rate or 30)
         step = max(1, int(vp.skip_frames))
-        start_idx = int((vp.clip_start or 0) * fps)
+        # Boundary must compare against the UNTRUNCATED float (old
+        # semantics): int()-truncating admitted frames up to one early on
+        # fractional rates (23.976fps) and phase-shifted the stride.
+        start_f = (vp.clip_start or 0) * fps
+        start_anchor = int(start_f)  # stride phase anchor, as before
         end_idx = float("inf") if vp.clip_end is None else vp.clip_end * fps
         tb = stream.time_base
         if vp.clip_start and tb:
@@ -82,11 +86,11 @@ def _iter_video_frames(
                 else:
                     pos = (pos + 1) if pos is not None else 0
                     i = pos
-                if i < start_idx:
+                if i < start_f:
                     continue
                 if i > end_idx:
                     break
-                if (i - start_idx) % step != 0:
+                if (i - start_anchor) % step != 0:
                     continue
                 try:
                     img = frame.to_image()
diff --git a/tests/core/test_analyze_runner.py b/tests/core/test_analyze_runner.py
index 8f6a37e..12754c5 100644
--- a/tests/core/test_analyze_runner.py
+++ b/tests/core/test_analyze_runner.py
@@ -196,3 +196,24 @@ def test_clip_start_seeks_not_decodes(tmp_path):
     idxs = [i for i, _ in frames]
     assert idxs == sorted(idxs)
     assert idxs[-1] <= 90
+
+
+def test_clip_start_fractional_fps_boundary(tmp_path):
+    """23.976-style rates: the first admitted frame must be the first with
+    index >= clip_start*fps (truncation admitted one early + shifted stride)."""
+    import math
+    from fractions import Fraction
+    from pyfeatlive_core.analyze_runner import _iter_video_frames
+    from pyfeatlive_core.analyze_queue import VideoParams
+
+    p = _make_video(tmp_path / "ntsc.mp4", n_frames=72, fps=Fraction(24000, 1001))
+    c = _av.open(str(p))
+    try:
+        fps = float(c.streams.video[0].average_rate)
+    finally:
+        c.close()
+    vp = VideoParams(skip_frames=1, clip_start=1.0, clip_end=None,
+                     track_identities=False)
+    frames = list(_iter_video_frames(p, vp))
+    assert frames, "no frames yielded"
+    assert frames[0][0] == math.ceil(1.0 * fps)

From db857456a6ebf81d8f975e6657b1689124f75b84 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:20:41 -0400
Subject: [PATCH 08/13] perf(sessions): mtime-keyed caches for thumbnail bboxes
 and frame times

---
 backend/routers/sessions.py          | 95 ++++++++++++++++++++--------
 tests/backend/test_face_thumbnail.py | 72 +++++++++++++++++++++
 2 files changed, 140 insertions(+), 27 deletions(-)

diff --git a/backend/routers/sessions.py b/backend/routers/sessions.py
index 5e6c26b..fc9ef8c 100644
--- a/backend/routers/sessions.py
+++ b/backend/routers/sessions.py
@@ -22,6 +22,72 @@
 router = APIRouter(prefix="/api/sessions", tags=["sessions"])
 
 
+# (session_id-independent) mtime-keyed caches. Keyed by (path, mtime) so
+# any rewrite of the underlying file invalidates naturally; bounded by
+# keeping only the most recent entry per path.
+_BBOX_CACHE: dict[str, tuple[float, dict[tuple[int, int], tuple[float, float, float, float]]]] = {}
+_FRAME_TIMES_CACHE: dict[str, tuple[float, list[float]]] = {}
+
+
+def _bbox_index(fex_path: Path) -> dict[tuple[int, int], tuple[float, float, float, float]]:
+    """(frame, face_idx) -> bbox for every row, parsed once per fex mtime.
+
+    The Viewer's identity strip requests one thumbnail per identity; each
+    previously re-scanned the whole CSV (k full parses for k identities).
+    """
+    key = str(fex_path)
+    mtime = fex_path.stat().st_mtime
+    hit = _BBOX_CACHE.get(key)
+    if hit is not None and hit[0] == mtime:
+        return hit[1]
+    index: dict[tuple[int, int], tuple[float, float, float, float]] = {}
+    with open(fex_path, newline="") as f:
+        for row in _csv.DictReader(f):
+            try:
+                index[(int(row["frame"]), int(row["face_idx"]))] = (
+                    float(row["FaceRectX"]), float(row["FaceRectY"]),
+                    float(row["FaceRectWidth"]), float(row["FaceRectHeight"]),
+                )
+            except (KeyError, ValueError):
+                continue
+    _BBOX_CACHE[key] = (mtime, index)
+    return index
+
+
+def _frame_times_cached(video_path: Path) -> list[float]:
+    """Sorted per-frame presentation timestamps, demuxed once per video mtime.
+
+    Live recordings are written with VARIABLE wall-clock PTS at the detection
+    rate, so a fixed-fps `time = frame / fps` mapping drifts and the overlay
+    desyncs from the video. The fex rows are written lock-step with the video
+    frames (one per encoded frame, in order), so the Viewer aligns fex frame K
+    to the K-th timestamp here and maps video.currentTime <-> frame by actual
+    time. Demuxes packets (no full decode).
+    """
+    import av
+
+    key = str(video_path)
+    mtime = video_path.stat().st_mtime
+    hit = _FRAME_TIMES_CACHE.get(key)
+    if hit is not None and hit[0] == mtime:
+        return hit[1]
+
+    times: list[float] = []
+    container = av.open(str(video_path))
+    try:
+        stream = container.streams.video[0]
+        tb = stream.time_base
+        for packet in container.demux(stream):
+            if packet.pts is None:
+                continue
+            times.append(float(packet.pts * tb) if tb else float(packet.pts))
+    finally:
+        container.close()
+    times.sort()
+    _FRAME_TIMES_CACHE[key] = (mtime, times)
+    return times
+
+
 def _list_session_dirs() -> list[Path]:
     """Return all session subdirectories in the configured root."""
     root = default_sessions_root()
@@ -163,18 +229,7 @@ def face_thumbnail(session_id: str, frame: int, face_idx: int) -> Response:
     if not fex_path.exists():
         raise HTTPException(404, "no fex.csv in session")
 
-    bbox = None
-    with open(fex_path, newline="") as f:
-        for row in _csv.DictReader(f):
-            try:
-                if int(row["frame"]) == frame and int(row["face_idx"]) == face_idx:
-                    bbox = (
-                        float(row["FaceRectX"]), float(row["FaceRectY"]),
-                        float(row["FaceRectWidth"]), float(row["FaceRectHeight"]),
-                    )
-                    break
-            except (KeyError, ValueError):
-                continue
+    bbox = _bbox_index(fex_path).get((frame, face_idx))
     if bbox is None:
         raise HTTPException(404, "face not found for that (frame, face_idx)")
 
@@ -196,26 +251,12 @@ def frame_times(session_id: str) -> dict[str, list[float]]:
     time. Demuxes packets (no full decode) and returns them in presentation
     order.
     """
-    import av
-
     d = _resolve_session(session_id)
     video_path = d / VIDEO_FILENAME
     if not video_path.exists():
         raise HTTPException(404, "no video in session")
 
-    times: list[float] = []
-    container = av.open(str(video_path))
-    try:
-        stream = container.streams.video[0]
-        tb = stream.time_base
-        for packet in container.demux(stream):
-            if packet.pts is None:
-                continue
-            times.append(float(packet.pts * tb) if tb else float(packet.pts))
-    finally:
-        container.close()
-    times.sort()
-    return {"times": times}
+    return {"times": _frame_times_cached(video_path)}
 
 
 @router.post("/{session_id}/reveal")
diff --git a/tests/backend/test_face_thumbnail.py b/tests/backend/test_face_thumbnail.py
index c48db0e..cbfc8a3 100644
--- a/tests/backend/test_face_thumbnail.py
+++ b/tests/backend/test_face_thumbnail.py
@@ -136,3 +136,75 @@ def test_returns_96x96_png_for_valid_frame(client, tmp_path, monkeypatch):
     # Verify it decodes to a 96×96 image
     img = Image.open(io.BytesIO(body))
     assert img.size == (96, 96), f"Expected 96×96 but got {img.size}"
+
+
+# ---------------------------------------------------------------------------
+# Cache tests — bbox index is parsed once per fex.csv mtime, not per request
+# ---------------------------------------------------------------------------
+
+def test_bbox_index_cached_by_mtime(client, tmp_path, monkeypatch):
+    from backend.routers import sessions as sessions_router
+
+    monkeypatch.setattr(
+        "backend.routers.sessions.default_sessions_root", lambda: tmp_path,
+    )
+    sess = tmp_path / "sess_cache"
+    sess.mkdir()
+
+    n_frames = 30
+    width, height = 160, 120
+    _write_minimal_mp4(sess / "video.mp4", n_frames=n_frames, width=width, height=height)
+    _write_fex_csv(sess, [
+        {"frame": 0, "face_idx": 0,
+         "FaceRectX": 20, "FaceRectY": 20,
+         "FaceRectWidth": 60, "FaceRectHeight": 60},
+    ])
+
+    calls = {"n": 0}
+    real_reader = sessions_router._csv.DictReader
+
+    def _counting_reader(*a, **k):
+        calls["n"] += 1
+        return real_reader(*a, **k)
+
+    monkeypatch.setattr(sessions_router._csv, "DictReader", _counting_reader)
+    sessions_router._BBOX_CACHE.clear()
+
+    r1 = client.get("/api/sessions/sess_cache/face-thumbnail/0/0")
+    r2 = client.get("/api/sessions/sess_cache/face-thumbnail/0/0")
+    assert r1.status_code == r2.status_code == 200
+    assert calls["n"] == 1  # second request served from the mtime-keyed index
+
+
+# ---------------------------------------------------------------------------
+# Cache test — frame-times demux runs once per video mtime, not per request
+# ---------------------------------------------------------------------------
+
+def test_frame_times_cached_by_mtime(client, tmp_path, monkeypatch):
+    from backend.routers import sessions as sessions_router
+
+    monkeypatch.setattr(
+        "backend.routers.sessions.default_sessions_root", lambda: tmp_path,
+    )
+    sess = tmp_path / "sess_frame_times"
+    sess.mkdir()
+
+    n_frames = 30
+    width, height = 160, 120
+    _write_minimal_mp4(sess / "video.mp4", n_frames=n_frames, width=width, height=height)
+
+    calls = {"n": 0}
+    real_open = av.open
+
+    def _counting_open(*a, **k):
+        calls["n"] += 1
+        return real_open(*a, **k)
+
+    monkeypatch.setattr(av, "open", _counting_open)
+    sessions_router._FRAME_TIMES_CACHE.clear()
+
+    r1 = client.get("/api/sessions/sess_frame_times/frame-times")
+    r2 = client.get("/api/sessions/sess_frame_times/frame-times")
+    assert r1.status_code == r2.status_code == 200
+    assert r1.json() == r2.json()
+    assert calls["n"] == 1  # second request served from the mtime-keyed cache

From d339ca3c6d6dd3618ec6ae45426e5d563db509c2 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:24:35 -0400
Subject: [PATCH 09/13] fix(sessions): keep first-occurrence-wins for duplicate
 bbox rows in the index

---
 backend/routers/sessions.py | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git a/backend/routers/sessions.py b/backend/routers/sessions.py
index fc9ef8c..7f5aa9b 100644
--- a/backend/routers/sessions.py
+++ b/backend/routers/sessions.py
@@ -44,7 +44,12 @@ def _bbox_index(fex_path: Path) -> dict[tuple[int, int], tuple[float, float, flo
     with open(fex_path, newline="") as f:
         for row in _csv.DictReader(f):
             try:
-                index[(int(row["frame"]), int(row["face_idx"]))] = (
+                pair = (int(row["frame"]), int(row["face_idx"]))
+                if pair in index:
+                    continue  # first occurrence wins — the pre-index scan
+                              # broke on first match; keep that contract for
+                              # any file with duplicate (frame, face_idx) rows
+                index[pair] = (
                     float(row["FaceRectX"]), float(row["FaceRectY"]),
                     float(row["FaceRectWidth"]), float(row["FaceRectHeight"]),
                 )

From 4bf26d2f7592d3560fe81300c7b64e95ecf66c1f Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:29:20 -0400
Subject: [PATCH 10/13] =?UTF-8?q?fix(identities):=20per-session=20write=20?=
 =?UTF-8?q?locks=20+=20unique=20temp=20files=20=E2=80=94=20no=20more=20los?=
 =?UTF-8?q?t=20click-to-label=20updates?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/routers/annotations.py            |  61 ++--
 backend/routers/identities.py             | 365 +++++++++++-----------
 backend/session_locks.py                  |  23 ++
 pyfeatlive_core/annotations.py            |   7 +-
 pyfeatlive_core/identities.py             |  22 +-
 tests/backend/test_identity_write_race.py |  36 +++
 6 files changed, 303 insertions(+), 211 deletions(-)
 create mode 100644 backend/session_locks.py
 create mode 100644 tests/backend/test_identity_write_race.py

diff --git a/backend/routers/annotations.py b/backend/routers/annotations.py
index e32008a..d4a8610 100644
--- a/backend/routers/annotations.py
+++ b/backend/routers/annotations.py
@@ -18,6 +18,8 @@
 )
 from pyfeatlive_core.recorder import default_sessions_root
 
+from backend.session_locks import session_lock
+
 
 router = APIRouter(tags=["annotations"])
 
@@ -65,15 +67,16 @@ class CreateAnnotationRequest(BaseModel):
 @router.post("/api/sessions/{session_id}/annotations", status_code=201)
 def create_annotation(session_id: str, req: CreateAnnotationRequest) -> dict:
     d = _session_dir(session_id)
-    existing = read_annotations(d)
-    ann = Annotation(
-        annotation_id=new_annotation_id(),
-        kind=Kind(req.kind),
-        start_frame=req.start_frame, end_frame=req.end_frame,
-        label=req.label, tag=req.tag,
-        created_at=time.time(), source=req.source,
-    )
-    write_annotations(d, existing + [ann])
+    with session_lock(session_id):
+        existing = read_annotations(d)
+        ann = Annotation(
+            annotation_id=new_annotation_id(),
+            kind=Kind(req.kind),
+            start_frame=req.start_frame, end_frame=req.end_frame,
+            label=req.label, tag=req.tag,
+            created_at=time.time(), source=req.source,
+        )
+        write_annotations(d, existing + [ann])
     return _to_dict(ann)
 
 
@@ -89,20 +92,21 @@ def patch_annotation(
     session_id: str, annotation_id: str, req: PatchAnnotationRequest,
 ) -> dict:
     d = _session_dir(session_id)
-    existing = read_annotations(d)
-    out = []
-    found = None
-    for a in existing:
-        if a.annotation_id == annotation_id:
-            if req.label is not None: a.label = req.label
-            if req.tag is not None: a.tag = req.tag
-            if req.start_frame is not None: a.start_frame = req.start_frame
-            if req.end_frame is not None: a.end_frame = req.end_frame
-            found = a
-        out.append(a)
-    if found is None:
-        raise HTTPException(404, "annotation not found")
-    write_annotations(d, out)
+    with session_lock(session_id):
+        existing = read_annotations(d)
+        out = []
+        found = None
+        for a in existing:
+            if a.annotation_id == annotation_id:
+                if req.label is not None: a.label = req.label
+                if req.tag is not None: a.tag = req.tag
+                if req.start_frame is not None: a.start_frame = req.start_frame
+                if req.end_frame is not None: a.end_frame = req.end_frame
+                found = a
+            out.append(a)
+        if found is None:
+            raise HTTPException(404, "annotation not found")
+        write_annotations(d, out)
     return _to_dict(found)
 
 
@@ -111,9 +115,10 @@ def patch_annotation(
 )
 def delete_annotation(session_id: str, annotation_id: str) -> None:
     d = _session_dir(session_id)
-    existing = read_annotations(d)
-    kept = [a for a in existing if a.annotation_id != annotation_id]
-    if len(kept) == len(existing):
-        raise HTTPException(404, "annotation not found")
-    write_annotations(d, kept)
+    with session_lock(session_id):
+        existing = read_annotations(d)
+        kept = [a for a in existing if a.annotation_id != annotation_id]
+        if len(kept) == len(existing):
+            raise HTTPException(404, "annotation not found")
+        write_annotations(d, kept)
     return None
diff --git a/backend/routers/identities.py b/backend/routers/identities.py
index 8e1cee3..9c45222 100644
--- a/backend/routers/identities.py
+++ b/backend/routers/identities.py
@@ -23,6 +23,8 @@
 )
 from pyfeatlive_core.recorder import default_sessions_root
 
+from backend.session_locks import session_lock
+
 
 router = APIRouter(tags=["identities"])
 
@@ -76,14 +78,15 @@ class CreateIdentityRequest(BaseModel):
 @router.post("/api/sessions/{session_id}/identities", status_code=201)
 def create_identity(session_id: str, req: CreateIdentityRequest) -> dict:
     d = _session_dir(session_id)
-    existing = read_identities(d)
-    ident = Identity(
-        identity_id=new_identity_id(),
-        name=req.name, color=req.color,
-        created_at=time.time(), source=req.source,
-    )
-    write_identities(d, existing + [ident])
-    apply_identity_labels_to_fex(d)
+    with session_lock(session_id):
+        existing = read_identities(d)
+        ident = Identity(
+            identity_id=new_identity_id(),
+            name=req.name, color=req.color,
+            created_at=time.time(), source=req.source,
+        )
+        write_identities(d, existing + [ident])
+        apply_identity_labels_to_fex(d)
     return _identity_to_dict(ident)
 
 
@@ -97,23 +100,24 @@ def patch_identity(
     session_id: str, identity_id: str, req: PatchIdentityRequest,
 ) -> dict:
     d = _session_dir(session_id)
-    existing = read_identities(d)
-    out = []
-    found = None
-    for ident in existing:
-        if ident.identity_id == identity_id:
-            if req.name is not None:
-                ident.name = req.name
-            if req.color is not None:
-                ident.color = req.color
-            found = ident
-        out.append(ident)
-    if found is None:
-        raise HTTPException(404, "identity not found")
-    write_identities(d, out)
-    # Rename propagates to the fex.csv IdentityLabel column so
-    # downstream analysis sees the updated name.
-    apply_identity_labels_to_fex(d)
+    with session_lock(session_id):
+        existing = read_identities(d)
+        out = []
+        found = None
+        for ident in existing:
+            if ident.identity_id == identity_id:
+                if req.name is not None:
+                    ident.name = req.name
+                if req.color is not None:
+                    ident.color = req.color
+                found = ident
+            out.append(ident)
+        if found is None:
+            raise HTTPException(404, "identity not found")
+        write_identities(d, out)
+        # Rename propagates to the fex.csv IdentityLabel column so
+        # downstream analysis sees the updated name.
+        apply_identity_labels_to_fex(d)
     return _identity_to_dict(found)
 
 
@@ -122,15 +126,16 @@ def patch_identity(
 )
 def delete_identity(session_id: str, identity_id: str) -> None:
     d = _session_dir(session_id)
-    existing = read_identities(d)
-    kept = [i for i in existing if i.identity_id != identity_id]
-    if len(kept) == len(existing):
-        raise HTTPException(404, "identity not found")
-    write_identities(d, kept)
-    # Drop assignments that referenced this identity_id
-    assignments = [a for a in read_assignments(d) if a.identity_id != identity_id]
-    write_assignments(d, assignments)
-    apply_identity_labels_to_fex(d)
+    with session_lock(session_id):
+        existing = read_identities(d)
+        kept = [i for i in existing if i.identity_id != identity_id]
+        if len(kept) == len(existing):
+            raise HTTPException(404, "identity not found")
+        write_identities(d, kept)
+        # Drop assignments that referenced this identity_id
+        assignments = [a for a in read_assignments(d) if a.identity_id != identity_id]
+        write_assignments(d, assignments)
+        apply_identity_labels_to_fex(d)
     return None
 
 
@@ -161,82 +166,83 @@ def auto_init_identities(session_id: str) -> dict:
     """
     import csv as _csv
     d = _session_dir(session_id)
-    existing = read_identities(d)
-    if existing:
-        # Already initialized — return as-is.
+    with session_lock(session_id):
+        existing = read_identities(d)
+        if existing:
+            # Already initialized — return as-is.
+            return {
+                "identities": [_identity_to_dict(i) for i in existing],
+                "created": 0, "assignments": len(read_assignments(d)),
+            }
+
+        fex_path = d / "fex.csv"
+        if not fex_path.exists() or fex_path.stat().st_size == 0:
+            return {"identities": [], "created": 0, "assignments": 0}
+
+        # Decide which column to group rows by.
+        with open(fex_path, newline="", encoding="utf-8") as f:
+            reader = _csv.DictReader(f)
+            rows = list(reader)
+            fieldnames = reader.fieldnames or []
+        group_col = "Identity" if "Identity" in fieldnames else "face_idx"
+
+        # Group: cluster_value -> [(frame, face_idx), ...]
+        groups: dict[str, list[tuple[int, int]]] = {}
+        for row in rows:
+            raw = row.get(group_col)
+            if raw is None or raw == "":
+                continue
+            try:
+                frame = int(row["frame"])
+                face_idx = int(row["face_idx"])
+            except (KeyError, ValueError):
+                continue
+            groups.setdefault(str(raw), []).append((frame, face_idx))
+
+        if not groups:
+            return {"identities": [], "created": 0, "assignments": 0}
+
+        # Create one identity per group with auto-assigned HSL-spread color.
+        n = len(groups)
+        new_idents: list[Identity] = []
+        cluster_to_id: dict[str, str] = {}
+        label_prefix = "Person" if group_col == "Identity" else "Face"
+        for idx, (cluster_value, _) in enumerate(sorted(groups.items())):
+            hue = int((idx * 360 / max(1, n)) % 360)
+            color = f"hsl({hue}, 70%, 55%)"
+            ident = Identity(
+                identity_id=new_identity_id(),
+                name=f"{label_prefix} {idx}",
+                color=color,
+                created_at=time.time(),
+                source="auto",
+            )
+            new_idents.append(ident)
+            cluster_to_id[cluster_value] = ident.identity_id
+
+        write_identities(d, new_idents)
+
+        # Bulk-write all assignments.
+        new_assignments = []
+        for cluster_value, pairs in groups.items():
+            iid = cluster_to_id[cluster_value]
+            for frame, face_idx in pairs:
+                new_assignments.append(IdentityAssignment(
+                    frame=frame, face_idx=face_idx, identity_id=iid,
+                ))
+        write_assignments(d, new_assignments)
+
+        # Stamp fex.csv with the new identity labels so downstream tools
+        # see them without joining tables.
+        apply_identity_labels_to_fex(d)
+
         return {
-            "identities": [_identity_to_dict(i) for i in existing],
-            "created": 0, "assignments": len(read_assignments(d)),
+            "identities": [_identity_to_dict(i) for i in new_idents],
+            "created": len(new_idents),
+            "assignments": len(new_assignments),
+            "grouped_by": group_col,
         }
 
-    fex_path = d / "fex.csv"
-    if not fex_path.exists() or fex_path.stat().st_size == 0:
-        return {"identities": [], "created": 0, "assignments": 0}
-
-    # Decide which column to group rows by.
-    with open(fex_path, newline="", encoding="utf-8") as f:
-        reader = _csv.DictReader(f)
-        rows = list(reader)
-        fieldnames = reader.fieldnames or []
-    group_col = "Identity" if "Identity" in fieldnames else "face_idx"
-
-    # Group: cluster_value -> [(frame, face_idx), ...]
-    groups: dict[str, list[tuple[int, int]]] = {}
-    for row in rows:
-        raw = row.get(group_col)
-        if raw is None or raw == "":
-            continue
-        try:
-            frame = int(row["frame"])
-            face_idx = int(row["face_idx"])
-        except (KeyError, ValueError):
-            continue
-        groups.setdefault(str(raw), []).append((frame, face_idx))
-
-    if not groups:
-        return {"identities": [], "created": 0, "assignments": 0}
-
-    # Create one identity per group with auto-assigned HSL-spread color.
-    n = len(groups)
-    new_idents: list[Identity] = []
-    cluster_to_id: dict[str, str] = {}
-    label_prefix = "Person" if group_col == "Identity" else "Face"
-    for idx, (cluster_value, _) in enumerate(sorted(groups.items())):
-        hue = int((idx * 360 / max(1, n)) % 360)
-        color = f"hsl({hue}, 70%, 55%)"
-        ident = Identity(
-            identity_id=new_identity_id(),
-            name=f"{label_prefix} {idx}",
-            color=color,
-            created_at=time.time(),
-            source="auto",
-        )
-        new_idents.append(ident)
-        cluster_to_id[cluster_value] = ident.identity_id
-
-    write_identities(d, new_idents)
-
-    # Bulk-write all assignments.
-    new_assignments = []
-    for cluster_value, pairs in groups.items():
-        iid = cluster_to_id[cluster_value]
-        for frame, face_idx in pairs:
-            new_assignments.append(IdentityAssignment(
-                frame=frame, face_idx=face_idx, identity_id=iid,
-            ))
-    write_assignments(d, new_assignments)
-
-    # Stamp fex.csv with the new identity labels so downstream tools
-    # see them without joining tables.
-    apply_identity_labels_to_fex(d)
-
-    return {
-        "identities": [_identity_to_dict(i) for i in new_idents],
-        "created": len(new_idents),
-        "assignments": len(new_assignments),
-        "grouped_by": group_col,
-    }
-
 
 class ClusterRequest(BaseModel):
     threshold: float = 0.8
@@ -250,43 +256,44 @@ def cluster_identities(session_id: str, req: ClusterRequest) -> dict:
     similarity matrix so the UI can suggest merges.
     """
     d = _session_dir(session_id)
-    try:
-        result = cluster_session(d, threshold=req.threshold)
-    except ValueError as e:
-        raise HTTPException(400, str(e)) from e
-
-    # Replace existing identities with one per cluster
-    n = result["n_clusters"]
-    new_idents: list[Identity] = []
-    cluster_id_to_identity: dict[int, str] = {}
-    for idx, cid in enumerate(result["cluster_ids"]):
-        hue = int((idx * 360 / max(1, n)) % 360)
-        ident = Identity(
-            identity_id=new_identity_id(),
-            name=f"Person {idx}",
-            color=f"hsl({hue}, 70%, 55%)",
-            created_at=time.time(),
-            source="cluster",
-        )
-        new_idents.append(ident)
-        cluster_id_to_identity[cid] = ident.identity_id
-
-    new_assignments = [
-        IdentityAssignment(
-            frame=f, face_idx=fi,
-            identity_id=cluster_id_to_identity[c],
-        )
-        for (f, fi, c) in result["cluster_assignments"]
-    ]
-    # Atomic-with-rollback write of all 3 files. If any step fails,
-    # the previous identities + assignments + fex labels are restored.
-    write_identities_and_assignments(d, new_idents, new_assignments)
+    with session_lock(session_id):
+        try:
+            result = cluster_session(d, threshold=req.threshold)
+        except ValueError as e:
+            raise HTTPException(400, str(e)) from e
+
+        # Replace existing identities with one per cluster
+        n = result["n_clusters"]
+        new_idents: list[Identity] = []
+        cluster_id_to_identity: dict[int, str] = {}
+        for idx, cid in enumerate(result["cluster_ids"]):
+            hue = int((idx * 360 / max(1, n)) % 360)
+            ident = Identity(
+                identity_id=new_identity_id(),
+                name=f"Person {idx}",
+                color=f"hsl({hue}, 70%, 55%)",
+                created_at=time.time(),
+                source="cluster",
+            )
+            new_idents.append(ident)
+            cluster_id_to_identity[cid] = ident.identity_id
+
+        new_assignments = [
+            IdentityAssignment(
+                frame=f, face_idx=fi,
+                identity_id=cluster_id_to_identity[c],
+            )
+            for (f, fi, c) in result["cluster_assignments"]
+        ]
+        # Atomic-with-rollback write of all 3 files. If any step fails,
+        # the previous identities + assignments + fex labels are restored.
+        write_identities_and_assignments(d, new_idents, new_assignments)
 
-    return {
-        "identities": [_identity_to_dict(i) for i in new_idents],
-        "similarity": result["similarity"],
-        "n_clusters": n,
-    }
+        return {
+            "identities": [_identity_to_dict(i) for i in new_idents],
+            "similarity": result["similarity"],
+            "n_clusters": n,
+        }
 
 
 @router.post("/api/sessions/{session_id}/identities/{keep_id}/merge/{absorb_id}")
@@ -304,30 +311,31 @@ def merge_identities(
     if keep_id == absorb_id:
         raise HTTPException(400, "keep_id and absorb_id must differ")
     d = _session_dir(session_id)
-    existing = read_identities(d)
-    ids = {i.identity_id for i in existing}
-    if keep_id not in ids:
-        raise HTTPException(404, "keep identity not found")
-    if absorb_id not in ids:
-        raise HTTPException(404, "absorb identity not found")
-
-    # Retag assignments
-    assignments = read_assignments(d)
-    new_assignments = [
-        IdentityAssignment(
-            frame=a.frame,
-            face_idx=a.face_idx,
-            identity_id=keep_id if a.identity_id == absorb_id else a.identity_id,
-        )
-        for a in assignments
-    ]
-    # Drop the absorbed identity from the catalog
-    kept = [i for i in existing if i.identity_id != absorb_id]
-    # Atomic-with-rollback write of all 3 files. If anything fails
-    # the prior identities + assignments + fex labels are restored.
-    write_identities_and_assignments(d, kept, new_assignments)
-
-    return {"identities": [_identity_to_dict(i) for i in kept]}
+    with session_lock(session_id):
+        existing = read_identities(d)
+        ids = {i.identity_id for i in existing}
+        if keep_id not in ids:
+            raise HTTPException(404, "keep identity not found")
+        if absorb_id not in ids:
+            raise HTTPException(404, "absorb identity not found")
+
+        # Retag assignments
+        assignments = read_assignments(d)
+        new_assignments = [
+            IdentityAssignment(
+                frame=a.frame,
+                face_idx=a.face_idx,
+                identity_id=keep_id if a.identity_id == absorb_id else a.identity_id,
+            )
+            for a in assignments
+        ]
+        # Drop the absorbed identity from the catalog
+        kept = [i for i in existing if i.identity_id != absorb_id]
+        # Atomic-with-rollback write of all 3 files. If anything fails
+        # the prior identities + assignments + fex labels are restored.
+        write_identities_and_assignments(d, kept, new_assignments)
+
+        return {"identities": [_identity_to_dict(i) for i in kept]}
 
 
 @router.post("/api/sessions/{session_id}/identities/{identity_id}/assign")
@@ -335,14 +343,15 @@ def assign_identity(
     session_id: str, identity_id: str, req: AssignRequest,
 ) -> dict:
     d = _session_dir(session_id)
-    # Validate identity exists
-    if not any(i.identity_id == identity_id for i in read_identities(d)):
-        raise HTTPException(404, "identity not found")
-    upsert_assignment(
-        d, frame=req.frame, face_idx=req.face_idx, identity_id=identity_id,
-    )
-    apply_identity_labels_to_fex(d)
-    return {
-        "frame": req.frame, "face_idx": req.face_idx,
-        "identity_id": identity_id,
-    }
+    with session_lock(session_id):
+        # Validate identity exists
+        if not any(i.identity_id == identity_id for i in read_identities(d)):
+            raise HTTPException(404, "identity not found")
+        upsert_assignment(
+            d, frame=req.frame, face_idx=req.face_idx, identity_id=identity_id,
+        )
+        apply_identity_labels_to_fex(d)
+        return {
+            "frame": req.frame, "face_idx": req.face_idx,
+            "identity_id": identity_id,
+        }
diff --git a/backend/session_locks.py b/backend/session_locks.py
new file mode 100644
index 0000000..32f97f0
--- /dev/null
+++ b/backend/session_locks.py
@@ -0,0 +1,23 @@
+"""Per-session write locks.
+
+Identity/annotation endpoints are sync ``def`` routes, so Starlette runs
+them on a threadpool (up to 40 threads) — two rapid click-to-label calls
+would otherwise both read-modify-write the same CSV and silently lose one
+update. One process-wide lock per session id serializes those windows.
+"""
+
+from __future__ import annotations
+
+import threading
+
+_REGISTRY: dict[str, threading.Lock] = {}
+_REGISTRY_LOCK = threading.Lock()
+
+
+def session_lock(session_id: str) -> threading.Lock:
+    with _REGISTRY_LOCK:
+        lock = _REGISTRY.get(session_id)
+        if lock is None:
+            lock = threading.Lock()
+            _REGISTRY[session_id] = lock
+        return lock
diff --git a/pyfeatlive_core/annotations.py b/pyfeatlive_core/annotations.py
index ba9272f..1c2b57e 100644
--- a/pyfeatlive_core/annotations.py
+++ b/pyfeatlive_core/annotations.py
@@ -11,6 +11,8 @@
 from __future__ import annotations
 
 import csv
+import os
+import threading
 import time
 import uuid
 from dataclasses import dataclass, asdict
@@ -83,7 +85,10 @@ def write_annotations(
     """Replace the annotations file atomically."""
     p = annotations_path(session_dir)
     p.parent.mkdir(parents=True, exist_ok=True)
-    tmp = p.with_suffix(".csv.tmp")
+    # Unique per writer: a FIXED tmp name let two concurrent writers
+    # interleave into one file before replace(). (The rename itself is
+    # atomic; the shared scratch path was the hazard.)
+    tmp = p.with_suffix(f".csv.tmp.{os.getpid()}.{threading.get_ident()}")
     with open(tmp, "w", newline="", encoding="utf-8") as f:
         writer = csv.DictWriter(f, fieldnames=_HEADER)
         writer.writeheader()
diff --git a/pyfeatlive_core/identities.py b/pyfeatlive_core/identities.py
index d9dfa27..3bccc14 100644
--- a/pyfeatlive_core/identities.py
+++ b/pyfeatlive_core/identities.py
@@ -18,6 +18,8 @@
 from __future__ import annotations
 
 import csv
+import os
+import threading
 import uuid
 from dataclasses import dataclass, asdict
 from pathlib import Path
@@ -75,7 +77,10 @@ def write_identities(session_dir: Path, identities: Iterable[Identity]) -> None:
     """Replace the identities catalog atomically."""
     p = identities_path(session_dir)
     p.parent.mkdir(parents=True, exist_ok=True)
-    tmp = p.with_suffix(".csv.tmp")
+    # Unique per writer: a FIXED tmp name let two concurrent writers
+    # interleave into one file before replace(). (The rename itself is
+    # atomic; the shared scratch path was the hazard.)
+    tmp = p.with_suffix(f".csv.tmp.{os.getpid()}.{threading.get_ident()}")
     with open(tmp, "w", newline="", encoding="utf-8") as f:
         writer = csv.DictWriter(f, fieldnames=_IDENTITY_HEADER)
         writer.writeheader()
@@ -132,7 +137,10 @@ def write_assignments(
     """Replace the assignments file atomically."""
     p = assignments_path_v2(session_dir)
     p.parent.mkdir(parents=True, exist_ok=True)
-    tmp = p.with_suffix(".csv.tmp")
+    # Unique per writer: a FIXED tmp name let two concurrent writers
+    # interleave into one file before replace(). (The rename itself is
+    # atomic; the shared scratch path was the hazard.)
+    tmp = p.with_suffix(f".csv.tmp.{os.getpid()}.{threading.get_ident()}")
     with open(tmp, "w", newline="", encoding="utf-8") as f:
         writer = csv.DictWriter(f, fieldnames=_ASSIGNMENT_HEADER_V2)
         writer.writeheader()
@@ -226,7 +234,10 @@ def apply_identity_labels_to_fex(session_dir: Path) -> int:
         if "IdentityLabel" not in df.columns:
             return len(df)
         df["IdentityLabel"] = ""
-        tmp = fex_path.with_suffix(".csv.tmp")
+        # Unique per writer: a FIXED tmp name let two concurrent writers
+        # interleave into one file before replace(). (The rename itself is
+        # atomic; the shared scratch path was the hazard.)
+        tmp = fex_path.with_suffix(f".csv.tmp.{os.getpid()}.{threading.get_ident()}")
         df.to_csv(tmp, index=False)
         tmp.replace(fex_path)
         return len(df)
@@ -256,7 +267,10 @@ def apply_identity_labels_to_fex(session_dir: Path) -> int:
     else:
         df["IdentityLabel"] = ""
 
-    tmp = fex_path.with_suffix(".csv.tmp")
+    # Unique per writer: a FIXED tmp name let two concurrent writers
+    # interleave into one file before replace(). (The rename itself is
+    # atomic; the shared scratch path was the hazard.)
+    tmp = fex_path.with_suffix(f".csv.tmp.{os.getpid()}.{threading.get_ident()}")
     df.to_csv(tmp, index=False)
     tmp.replace(fex_path)
     return len(df)
diff --git a/tests/backend/test_identity_write_race.py b/tests/backend/test_identity_write_race.py
new file mode 100644
index 0000000..7a3ea77
--- /dev/null
+++ b/tests/backend/test_identity_write_race.py
@@ -0,0 +1,36 @@
+"""Concurrent assignment writes must not lose updates."""
+
+from concurrent.futures import ThreadPoolExecutor
+
+
+def test_concurrent_assigns_all_land(client, tmp_path, monkeypatch):
+    # Point sessions at tmp and fabricate a minimal session directory.
+    monkeypatch.setattr(
+        "backend.routers.identities.default_sessions_root", lambda: tmp_path,
+    )
+    monkeypatch.setattr(
+        "backend.routers.sessions.default_sessions_root", lambda: tmp_path,
+    )
+    sid = "2026-01-01_00-00-00"
+    (tmp_path / sid).mkdir()
+
+    # One identity to assign everything to.
+    r = client.post(f"/api/sessions/{sid}/identities",
+                    json={"name": "A", "color": "#ff0000"})
+    assert r.status_code == 201, r.text
+    ident = r.json()["identity_id"]
+
+    def _assign(frame: int):
+        return client.post(
+            f"/api/sessions/{sid}/identities/{ident}/assign",
+            json={"frame": frame, "face_idx": 0},
+        )
+
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        results = list(pool.map(_assign, range(24)))
+    assert all(r.status_code == 200 for r in results), \
+        [r.status_code for r in results]
+
+    listed = client.get(f"/api/sessions/{sid}/identities/assignments").json()
+    frames = sorted(a["frame"] for a in listed)
+    assert frames == list(range(24))  # pre-lock: read-modify-write lost some

From 8960c09c63b17d2112836b68164cecd37b3e694f Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:46:01 -0400
Subject: [PATCH 11/13] fix(analyze): positional frame indices for VFR sources;
 ns mtime keys; retrieve WS recv errors

---
 backend/routers/analyze.py        |  8 +++++
 backend/routers/sessions.py       | 16 +++++----
 pyfeatlive_core/analyze_runner.py | 48 ++++++++++++++++++++++-----
 tests/core/test_analyze_runner.py | 54 +++++++++++++++++++++++++++++--
 4 files changed, 109 insertions(+), 17 deletions(-)

diff --git a/backend/routers/analyze.py b/backend/routers/analyze.py
index 05d3d22..c881b74 100644
--- a/backend/routers/analyze.py
+++ b/backend/routers/analyze.py
@@ -391,6 +391,14 @@ async def analyze_ws(ws: WebSocket) -> None:
                 )
                 if recv_task in done:
                     get_task.cancel()
+                    try:
+                        recv_task.exception()  # retrieve, if any: disconnects
+                                                # and transport errors both
+                                                # mean "gone" — this just
+                                                # avoids an "exception was
+                                                # never retrieved" log spam
+                    except asyncio.CancelledError:
+                        pass
                     break  # disconnect (or any client frame): exit + clean up
                 ev = get_task.result()
                 try:
diff --git a/backend/routers/sessions.py b/backend/routers/sessions.py
index 7f5aa9b..75ec9e4 100644
--- a/backend/routers/sessions.py
+++ b/backend/routers/sessions.py
@@ -22,11 +22,13 @@
 router = APIRouter(prefix="/api/sessions", tags=["sessions"])
 
 
-# (session_id-independent) mtime-keyed caches. Keyed by (path, mtime) so
-# any rewrite of the underlying file invalidates naturally; bounded by
-# keeping only the most recent entry per path.
-_BBOX_CACHE: dict[str, tuple[float, dict[tuple[int, int], tuple[float, float, float, float]]]] = {}
-_FRAME_TIMES_CACHE: dict[str, tuple[float, list[float]]] = {}
+# (session_id-independent) mtime-keyed caches. Keyed by (path, mtime_ns) —
+# nanosecond mtime, not float st_mtime, so two rewrites within the same
+# float-precision instant still invalidate correctly — so any rewrite of
+# the underlying file invalidates naturally; bounded by keeping only the
+# most recent entry per path.
+_BBOX_CACHE: dict[str, tuple[int, dict[tuple[int, int], tuple[float, float, float, float]]]] = {}
+_FRAME_TIMES_CACHE: dict[str, tuple[int, list[float]]] = {}
 
 
 def _bbox_index(fex_path: Path) -> dict[tuple[int, int], tuple[float, float, float, float]]:
@@ -36,7 +38,7 @@ def _bbox_index(fex_path: Path) -> dict[tuple[int, int], tuple[float, float, flo
     previously re-scanned the whole CSV (k full parses for k identities).
     """
     key = str(fex_path)
-    mtime = fex_path.stat().st_mtime
+    mtime = fex_path.stat().st_mtime_ns
     hit = _BBOX_CACHE.get(key)
     if hit is not None and hit[0] == mtime:
         return hit[1]
@@ -72,7 +74,7 @@ def _frame_times_cached(video_path: Path) -> list[float]:
     import av
 
     key = str(video_path)
-    mtime = video_path.stat().st_mtime
+    mtime = video_path.stat().st_mtime_ns
     hit = _FRAME_TIMES_CACHE.get(key)
     if hit is not None and hit[0] == mtime:
         return hit[1]
diff --git a/pyfeatlive_core/analyze_runner.py b/pyfeatlive_core/analyze_runner.py
index 288743c..2a404f7 100644
--- a/pyfeatlive_core/analyze_runner.py
+++ b/pyfeatlive_core/analyze_runner.py
@@ -48,6 +48,21 @@ def _load_rgb_image(src: Path) -> Image.Image:
         return im.convert("RGB")
 
 
+def _pts_positions(container, stream) -> dict[int, int]:
+    """Packet pts -> positional frame index (presentation order).
+
+    A cheap demux (no decode) — the same cost the frame-times endpoint
+    pays. Used only on the seek path: deriving indices as round(pts*fps)
+    corrupted them for VFR sources (this app's own Live recordings are
+    wall-clock-PTS), producing duplicate/gapped fex frame values.
+    """
+    ptses = sorted(
+        p.pts for p in container.demux(stream) if p.pts is not None
+    )
+    container.seek(0)
+    return {pts: i for i, pts in enumerate(ptses)}
+
+
 def _iter_video_frames(
     path: Path, vp: VideoParams,
 ) -> Iterator[tuple[int, Image.Image]]:
@@ -57,6 +72,15 @@ def _iter_video_frames(
     frame). frame_index is the source index in the original stream, not
     the post-skip count — so downstream Fex rows reference real positions
     and stay aligned with the full source video copied into the session.
+
+    Indexing is POSITIONAL (decode order), never derived from pts*fps:
+    this app's own Live recordings are VFR with wall-clock-ms PTS, so
+    round(pts*fps) produces duplicate/gapped/out-of-range indices. When
+    there's no seek, we just count frames as we decode them (identical to
+    the pre-seek-support contract). When clip_start forces a seek, plain
+    positional counting can't survive the jump — a cheap demux pass maps
+    each packet's pts to its presentation-order position first, so the
+    post-seek decode can still recover the true positional index.
     """
     container = av.open(str(path))
     try:
@@ -70,21 +94,29 @@ def _iter_video_frames(
         start_anchor = int(start_f)  # stride phase anchor, as before
         end_idx = float("inf") if vp.clip_end is None else vp.clip_end * fps
         tb = stream.time_base
+        pts_pos: dict[int, int] | None = None
         if vp.clip_start and tb:
-            # Jump to the nearest keyframe at/before clip_start instead of
-            # decoding (and discarding) everything from t=0 — a clip_start
-            # 10 minutes in previously decoded ~18k dead frames.
+            # Seek path: index by packet POSITION (pts -> presentation-order
+            # index), then jump. Positional enumerate can't survive a seek,
+            # and round(pts*fps) corrupts VFR indices.
+            pts_pos = _pts_positions(container, stream)
             try:
+                # Jump to the nearest keyframe at/before clip_start instead
+                # of decoding (and discarding) everything from t=0 — a
+                # clip_start 10 minutes in previously decoded ~18k dead
+                # frames.
                 container.seek(int(vp.clip_start / tb), stream=stream)
             except Exception:
-                pass  # unusual container: fall back to decode-from-start
-        pos = None  # positional fallback when frames carry no PTS
+                pts_pos = None  # unusual container: decode from start
+        pos = -1  # positional counter for the no-seek path
         try:
             for frame in container.decode(stream):
-                if frame.pts is not None and tb:
-                    i = int(round(float(frame.pts * tb) * fps))
+                if pts_pos is not None and frame.pts is not None:
+                    i = pts_pos.get(frame.pts)
+                    if i is None:
+                        continue  # packet unseen in the demux pass: skip
                 else:
-                    pos = (pos + 1) if pos is not None else 0
+                    pos += 1
                     i = pos
                 if i < start_f:
                     continue
diff --git a/tests/core/test_analyze_runner.py b/tests/core/test_analyze_runner.py
index 12754c5..a226cbe 100644
--- a/tests/core/test_analyze_runner.py
+++ b/tests/core/test_analyze_runner.py
@@ -179,6 +179,54 @@ def test_estimate_frames_close_to_actual(tmp_path):
     assert est is not None and abs(est - 60) <= 3
 
 
+def _make_vfr_video(path, n_frames=40, size=(64, 64)):
+    """Encode an mp4 with explicit, jittered PTS — mimicking this app's own
+    Live recorder, which writes wall-clock-ms PTS rather than a constant
+    frame rate. Gaps cycle deterministically over 80-219ms."""
+    import numpy as np
+    from fractions import Fraction
+
+    container = _av.open(str(path), "w")
+    stream = container.add_stream("h264", rate=30)
+    stream.width, stream.height = size
+    stream.pix_fmt = "yuv420p"
+    stream.codec_context.time_base = Fraction(1, 1000)
+    pts_ms = 0
+    for i in range(n_frames):
+        arr = np.full((size[1], size[0], 3), (i * 4) % 255, dtype=np.uint8)
+        frame = _av.VideoFrame.from_ndarray(arr, format="rgb24")
+        frame.pts = pts_ms
+        frame.time_base = Fraction(1, 1000)
+        for pkt in stream.encode(frame):
+            container.mux(pkt)
+        pts_ms += 80 + (i * 37) % 140  # jittered wall-clock-style gap
+    for pkt in stream.encode():
+        container.mux(pkt)
+    container.close()
+    return path
+
+
+def test_vfr_source_yields_positional_indices(tmp_path):
+    """Regression: a VFR source (wall-clock-ms PTS, like this app's own
+    Live recordings) must not have its fex frame indices derived from
+    round(pts*fps) — that corrupts them into duplicate/gapped values on
+    non-CFR streams. With no clip_start (no seek), indexing must be plain
+    positional decode order."""
+    from pyfeatlive_core.analyze_runner import _iter_video_frames
+    from pyfeatlive_core.analyze_queue import VideoParams
+
+    n = 40
+    p = _make_vfr_video(tmp_path / "vfr.mp4", n_frames=n)
+    vp = VideoParams(skip_frames=1, clip_start=None, clip_end=None,
+                     track_identities=False)
+    frames = list(_iter_video_frames(p, vp))
+    idxs = [i for i, _ in frames]
+    assert idxs == list(range(len(idxs))), (
+        f"VFR source produced non-positional indices: {idxs}"
+    )
+    assert len(idxs) == n
+
+
 def test_clip_start_seeks_not_decodes(tmp_path):
     from pyfeatlive_core.analyze_runner import _iter_video_frames
     from pyfeatlive_core.analyze_queue import VideoParams
@@ -192,9 +240,11 @@ def test_clip_start_seeks_not_decodes(tmp_path):
     assert frames, "no frames yielded"
     first_idx = frames[0][0]
     assert 58 <= first_idx <= 62
-    # Frame indices remain source-referenced and increasing.
+    # Frame indices remain source-referenced and STRICTLY increasing (no
+    # duplicates — a duplicate here would mean two decoded frames mapped
+    # to the same pts->position, corrupting downstream fex.csv rows).
     idxs = [i for i, _ in frames]
-    assert idxs == sorted(idxs)
+    assert all(b > a for a, b in zip(idxs, idxs[1:]))
     assert idxs[-1] <= 90
 
 

From 6d92d21ca96843fda96ab84f6d169022e5c6b87c Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:51:14 -0400
Subject: [PATCH 12/13] fix(analyze): skip PTS-less frames on the seek path
 instead of mis-indexing them

---
 pyfeatlive_core/analyze_runner.py | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/pyfeatlive_core/analyze_runner.py b/pyfeatlive_core/analyze_runner.py
index 2a404f7..09c1587 100644
--- a/pyfeatlive_core/analyze_runner.py
+++ b/pyfeatlive_core/analyze_runner.py
@@ -111,7 +111,13 @@ def _iter_video_frames(
         pos = -1  # positional counter for the no-seek path
         try:
             for frame in container.decode(stream):
-                if pts_pos is not None and frame.pts is not None:
+                if pts_pos is not None:
+                    if frame.pts is None:
+                        # Post-seek, a PTS-less frame has no place in the
+                        # position map; falling into the positional counter
+                        # (which starts from 0 as if unseeked) could admit
+                        # it with a bogus low index. Skip, like map misses.
+                        continue
                     i = pts_pos.get(frame.pts)
                     if i is None:
                         continue  # packet unseen in the demux pass: skip

From 3ae2863addf83043909127975fae34deee17e794 Mon Sep 17 00:00:00 2001
From: ljchang <lukejchang@gmail.com>
Date: Fri, 3 Jul 2026 00:51:55 -0400
Subject: [PATCH 13/13] release: bump app version to 0.8.22

---
 tauri/src-tauri/Cargo.lock      | 2 +-
 tauri/src-tauri/Cargo.toml      | 2 +-
 tauri/src-tauri/tauri.conf.json | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/tauri/src-tauri/Cargo.lock b/tauri/src-tauri/Cargo.lock
index 81a82b1..73c94ec 100644
--- a/tauri/src-tauri/Cargo.lock
+++ b/tauri/src-tauri/Cargo.lock
@@ -2658,7 +2658,7 @@ dependencies = [
 
 [[package]]
 name = "pyfeatlive-tauri"
-version = "0.8.21"
+version = "0.8.22"
 dependencies = [
  "block2",
  "env_logger",
diff --git a/tauri/src-tauri/Cargo.toml b/tauri/src-tauri/Cargo.toml
index 0c36603..e118980 100644
--- a/tauri/src-tauri/Cargo.toml
+++ b/tauri/src-tauri/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "pyfeatlive-tauri"
-version = "0.8.21"
+version = "0.8.22"
 description = "Native desktop wrapper for pyfeatlive"
 authors = ["cosanlab"]
 license = "MIT"
diff --git a/tauri/src-tauri/tauri.conf.json b/tauri/src-tauri/tauri.conf.json
index bceada1..90a6167 100644
--- a/tauri/src-tauri/tauri.conf.json
+++ b/tauri/src-tauri/tauri.conf.json
@@ -1,7 +1,7 @@
 {
   "$schema": "https://schema.tauri.app/config/2",
   "productName": "Py-feat Live",
-  "version": "0.8.21",
+  "version": "0.8.22",
   "identifier": "com.cosanlab.pyfeatlive",
   "build": {
     "frontendDist": "../dist",