feat(tooling): X-Ray threshold optimizer, gallery utilities, artifact registry, docs build

Optimizer (scripts/optimizer/): replay.py runs the real C++ tracker/matcher/
scene_tracker chain over a dumped-embeddings HDF5 via sae_kpn, so a threshold
sweep never re-decodes video or re-embeds faces. optimize.py drives scipy's
differential_evolution over the knob space, with DE-level parallelism
(multiple population candidates evaluated concurrently via a ThreadPoolExecutor)
on top of per-film replay parallelism. second_score.py is the per-second X-Ray
scoring metric (TPI/FPI/FN, out-of-cast misID weighted 10x, fair recall masked
to gallery-known cast) that superseded an earlier scene-union metric.
dump_error_frames.py / dump_scene_montage.py extract annotated video frames
(bounding boxes, TPI/FPI/FN captions, onscreen-vs-offscreen split) for visual
review of a replay against ground truth. Gallery utilities: cast_restrict.py,
gallery_membership.py, fetch_missing_actors.py, reembed_gallery.py.

scripts/validation/: X-Ray ground-truth loading and provider-agnostic identity
matching (identity.py's keys_for — an actor is the union of every id we can
derive, since pipeline output and ground truth don't share one id space).

scripts/artifacts/: push/pull scripts for the Gitea generic package registry —
galleries, montage frames, and experiment data (manifests/trajectories/results)
are pushed there instead of committed, since none are needed to run the app,
only benchmarks. Versioned by git short-SHA.

scripts/docs/: MkDocs site build (build_site.sh) and the calibration-curve
comparison chart (calibration_chart.py, matplotlib, reads each gallery's
embedded calibration).

Gallery-building scripts (make_jellyfin_gallery.py, make_gallery.py,
filter_gallery.py, run_from_jellyfin.py, movienet_eval.py, movienet_prep.py,
sae_gallery.py) updated to read/write HDF5 galleries exclusively, matching the
engine-side format switch. run_from_jellyfin.py and the optimizer no longer
carry movie source paths in shared manifests (some source filenames include
scene-release tags) — resolved locally via a gitignored file-lut.json instead.

scripts/optimizer/SCHEMA.md

@@ -0,0 +1,50 @@
+# Embedding-dump HDF5 schema (v1)
+
+One file per analysed title. Captures the pipeline state at the `EmbeddedSceneFrame`
+channel — i.e. after decode → detect → align → embed, but **before** tracking and
+identity matching. Everything downstream (face tracker, identity matcher, scene
+tracker/anneal) is cheap CPU math, so replaying from this file lets a parameter
+sweep re-run the whole downstream tail thousands of times with no GPU and no video.
+
+Written by the C++ dump sink (`--dump-embeddings out.h5`); read by
+`scripts/optimizer/replay.py`.
+
+## Layout
+
+The dump is **flat/ragged**: all faces across all frames are concatenated into
+per-face arrays, with a per-frame index table pointing into them. This avoids
+variable-length HDF5 types and reads straight into numpy.
+
+```
+/                                     (root)
+  attrs:
+    schema_version : int   = 1
+    movie          : str   (source video path)
+    sample_fps     : float
+    embed_dim      : int   = 512
+
+  frames/                             group — one row per sampled frame
+    timestamp_sec  : float64 [F]
+    frame_idx      : int64   [F]
+    is_cut         : uint8   [F]      (histogram intra-scene cut)
+    is_scene_boundary : uint8 [F]     (TransNetV2 boundary; 0 if scene_detect off)
+    face_offset    : int64   [F]      start index into faces/* for this frame
+    face_count     : int32   [F]      number of faces in this frame
+
+  faces/                              group — one row per detected face, concatenated
+    embedding      : float32 [N, 512] L2-normalised ArcFace embedding
+    bbox           : float32 [N, 4]   x, y, w, h in original video pixels
+    landmarks      : float32 [N, 10]  5 (x,y) pairs, SCRFD/ArcFace order
+    confidence     : float32 [N]      detector confidence
+```
+
+`F` = number of sampled frames, `N` = total faces (= sum of face_count).
+Frame *i*'s faces are `faces/*[ face_offset[i] : face_offset[i]+face_count[i] ]`.
+
+## Invariants
+- `embedding` rows are unit-norm (cosine == dot product against the gallery).
+- `face_offset[0] == 0`; `face_offset[i+1] == face_offset[i] + face_count[i]`.
+- `bbox` is already mapped to original resolution (bbox_upscale applied at dump time),
+  matching what the identity matcher would emit.
+- A frame with no faces has `face_count == 0` (still gets a row, so timestamps stay dense).
+- EOF sentinel frames are NOT written.

scripts/optimizer/cast_restrict.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""
+cast_restrict.py — produce a per-film gallery restricted to its credited cast.
+
+Benchmark arm: instead of matching a face against the WHOLE gallery (2418 actors,
+risking cross-film misIDs like naming Archie Yates in a film he's not in), restrict
+the matcher's candidate set to the title's credited cast (from Jellyfin — the top
+~15 billed actors, exactly what run_from_jellyfin.py does in production).
+
+Filters a gallery to actors whose jellyfin_id is in the film's cast set, writing a
+small gallery JSON the replay can load. Actors are kept if their jellyfin_id (or, as
+a fallback, normalized name) matches the cast.
+
+Used by the full-vs-restricted bake-off. Cached per (gallery, film) so a DE sweep
+reuses the restricted gallery.
+"""
+from __future__ import annotations
+
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts" / "validation"))
+from identity import norm_name  # noqa: E402
+
+_CACHE: dict = {}
+
+
+def restricted_gallery_path(gallery_path: str, cast_jellyfin_ids: set[str],
+                            cast_names: set[str] | None = None) -> str:
+    """Write (once, cached) a gallery filtered to the film's credited cast; return path.
+
+    Matches gallery actors to the cast by jellyfin_id first, then normalized name."""
+    key = (gallery_path, frozenset(cast_jellyfin_ids))
+    if key in _CACHE:
+        return _CACHE[key]
+
+    gal = json.loads(Path(gallery_path).read_text())
+    names = {norm_name(n) for n in (cast_names or set())}
+    kept = []
+    for a in gal["actors"]:
+        jid = a.get("jellyfin_id", "")
+        if (jid and jid in cast_jellyfin_ids) or (names and norm_name(a["name"]) in names):
+            kept.append(a)
+
+    tf = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False,
+                                     prefix="castgal_")
+    json.dump({"actors": kept}, tf)
+    tf.close()
+    _CACHE[key] = tf.name
+    return tf.name
+
+
+def load_casts(casts_json: str) -> dict[str, list[str]]:
+    """film name → [jellyfin person id, ...] from jellyfin_casts.json."""
+    return json.loads(Path(casts_json).read_text())

scripts/optimizer/dump_error_frames.py

@@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+"""
+dump_error_frames.py — extract example video frames for visual inspection of a
+replayed prediction vs X-Ray ground truth: best-agreement seconds, FPI (false
+identification) seconds, and FN (missed cast) seconds.
+
+Reuses second_score.py's per-second timeline/prediction loading, but keeps the
+per-second classification (score_seconds only returns aggregates) and picks
+representative timestamps in each bucket, then pulls single frames from the
+source video via ffmpeg -ss (nearest keyframe-independent seek + decode).
+
+If --raw (the JSONL from `replay.py --raw-out`) is given, also draws each visible
+actor's bounding box + name/similarity on the extracted frame — green for
+identified, orange for unknown — matching debug_renderer_node.hpp's colour
+convention. Without --raw, frames are saved unannotated.
+
+Usage:
+  python scripts/optimizer/dump_error_frames.py \
+      --pred pred.json --raw raw.jsonl \
+      --xray experiments/xray/.../900_The_Many_Saints_Of_Newark \
+      --movie "/mnt/movies/The Many Saints Of Newark (2021)/....mp4" \
+      --gallery experiments/galleries/gallery_LVFace-B_Glint360K.h5 \
+      --out-dir experiments/dump_review/many_saints --n-per-bucket 6
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import cv2
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts" / "optimizer"))
+sys.path.insert(0, str(REPO / "scripts" / "validation"))
+
+from second_score import load_second_timeline, load_pred_intervals, _match  # noqa: E402
+from sample_eval import load_gallery_keys  # noqa: E402
+from identity import keys_for  # noqa: E402
+
+
+def per_second_detail(pred_json: dict, xray_dir: str, gallery_keys: set | None):
+    """Like second_score.score_seconds, but yields one record per sampled second
+    instead of collapsing to aggregates."""
+    timeline, film_cast, duration = load_second_timeline(xray_dir)
+    pred = load_pred_intervals(pred_json)
+
+    name_by_keys = {}
+    for a in pred_json.get("actors", []):
+        k = frozenset(keys_for(imdb_id=a.get("imdb_id"), tmdb_id=a.get("tmdb_id"),
+                                jellyfin_id=a.get("jellyfin_id"), name=a.get("name")))
+        name_by_keys[k] = a.get("name", "?")
+
+    records = []
+    for t in sorted(timeline):
+        G = [set(a) for a in timeline[t]]
+        P_all = [(k, set(k)) for k, wins in pred if any(w0 <= t <= w1 for w0, w1 in wins)]
+        if gallery_keys is not None:
+            G = [g for g in G if g & gallery_keys]
+
+        P = [p for _, p in P_all]
+        tp, matched = _match(P, G)
+        fp_names, fn_names = [], []
+        for key, pa in P_all:
+            if not any(pa & ga for ga in G):
+                fp_names.append(name_by_keys.get(key, "?"))
+        for j, ga in enumerate(G):
+            if not matched[j]:
+                fn_names.append("|".join(sorted(x for x in ga if not x.startswith("imdb:") and not x.startswith("tmdb:"))) or "?")
+
+        union = tp + len(fp_names) + len(fn_names)
+        jaccard = (tp / union) if union else 1.0
+        records.append({"t": t, "tp": tp, "fp": fp_names, "fn": fn_names, "jaccard": jaccard})
+    return records
+
+
+def pick_timestamps(records, n_per_bucket):
+    best = sorted(records, key=lambda r: (-r["jaccard"], -r["tp"]))
+    best = [r for r in best if r["tp"] > 0][:n_per_bucket]
+
+    fpi = [r for r in records if r["fp"]]
+    fpi = sorted(fpi, key=lambda r: -len(r["fp"]))[:n_per_bucket]
+
+    fn = [r for r in records if r["fn"]]
+    fn = sorted(fn, key=lambda r: -len(r["fn"]))[:n_per_bucket]
+
+    return {"best": best, "fpi": fpi, "fn": fn}
+
+
+def pick_by_interval(records, interval_sec):
+    """One best (highest jaccard) and one worst (lowest jaccard) second per
+    interval_sec-second window across the whole film, e.g. --interval-sec 600 for
+    a per-10-minute best/worst sweep. Windows with no sampled seconds are skipped
+    (X-Ray timelines only cover scenes, so gaps between/after scenes are common)."""
+    windows: dict[int, list] = {}
+    for r in records:
+        windows.setdefault(r["t"] // interval_sec, []).append(r)
+
+    buckets: dict[str, list] = {}
+    for w in sorted(windows):
+        wr = windows[w]
+        best = max(wr, key=lambda r: (r["jaccard"], r["tp"]))
+        worst = min(wr, key=lambda r: (r["jaccard"], -max(len(r["fp"]), len(r["fn"]))))
+        buckets[f"w{w:03d}_best"] = [best]
+        buckets[f"w{w:03d}_worst"] = [worst]
+    return buckets
+
+
+def load_raw_annotations(raw_path: str):
+    """second (int, floor) -> list of visible_actors dicts (last frame wins if
+    several fall in the same second, which is the common case at 1fps sampling)."""
+    by_second = {}
+    with open(raw_path) as f:
+        for line in f:
+            sa = json.loads(line)
+            if sa.get("eof"):
+                continue
+            by_second[int(sa["timestamp_sec"])] = sa.get("visible_actors", [])
+    return by_second
+
+
+def draw_annotations(frame_path: Path, actors: list):
+    img = cv2.imread(str(frame_path))
+    if img is None:
+        return
+    for a in actors:
+        known = a.get("actor_idx", -1) >= 0
+        colour = (60, 200, 0) if known else (220, 100, 0)  # BGR: green / orange
+        x, y, w, h = a["bbox"]
+        x, y, w, h = int(x), int(y), int(w), int(h)
+        cv2.rectangle(img, (x, y), (x + w, y + h), colour, 2)
+
+        label = f"{a['name']} {a['similarity']*100:.0f}%" if known else f"unknown {a['similarity']*100:.0f}%"
+        (tw, th), baseline = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1)
+        strip_y0 = max(0, y - th - 4)
+        cv2.rectangle(img, (x, strip_y0), (x + tw + 4, y), colour, cv2.FILLED)
+        cv2.putText(img, label, (x + 2, y - 2), cv2.FONT_HERSHEY_SIMPLEX, 0.5,
+                    (255, 255, 255), 1, cv2.LINE_AA)
+    cv2.imwrite(str(frame_path), img)
+
+
+def extract_frame(movie: str, t: float, out_path: Path):
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    subprocess.run(
+        ["ffmpeg", "-y", "-ss", str(t), "-i", movie, "-frames:v", "1",
+         "-q:v", "2", str(out_path)],
+        check=True, capture_output=True)
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--pred", required=True)
+    p.add_argument("--raw", help="raw per-frame annotations JSONL (replay.py --raw-out); "
+                                  "draws bboxes + names on extracted frames if given")
+    p.add_argument("--xray", required=True)
+    p.add_argument("--movie", required=True)
+    p.add_argument("--gallery")
+    p.add_argument("--out-dir", required=True)
+    p.add_argument("--n-per-bucket", type=int, default=6)
+    p.add_argument("--interval-sec", type=int,
+                   help="instead of global best/fpi/fn buckets, pick one best + one "
+                        "worst (by jaccard) second per interval-sec window across "
+                        "the whole film, e.g. 600 for per-10-minute best/worst")
+    args = p.parse_args()
+
+    pred_json = json.loads(Path(args.pred).read_text())
+    gk = load_gallery_keys(args.gallery) if args.gallery else None
+    records = per_second_detail(pred_json, args.xray, gk)
+    buckets = (pick_by_interval(records, args.interval_sec) if args.interval_sec
+              else pick_timestamps(records, args.n_per_bucket))
+    raw_by_second = load_raw_annotations(args.raw) if args.raw else None
+
+    out_dir = Path(args.out_dir)
+    manifest = []
+    for bucket, recs in buckets.items():
+        for r in recs:
+            fname = f"{bucket}_t{r['t']:05d}.jpg"
+            out_path = out_dir / bucket / fname
+            try:
+                extract_frame(args.movie, r["t"], out_path)
+                ok = True
+                if raw_by_second is not None:
+                    draw_annotations(out_path, raw_by_second.get(r["t"], []))
+            except subprocess.CalledProcessError as e:
+                ok = False
+                print(f"[dump_error_frames] ffmpeg failed at t={r['t']}: {e}", file=sys.stderr)
+            manifest.append({"bucket": bucket, "t": r["t"], "tp": r["tp"],
+                             "fp": r["fp"], "fn": r["fn"], "jaccard": round(r["jaccard"], 3),
+                             "file": str(out_path.relative_to(out_dir)) if ok else None})
+            print(f"[{bucket}] t={r['t']}s tp={r['tp']} fp={r['fp']} fn={r['fn']}", file=sys.stderr)
+
+    (out_dir / "manifest.json").write_text(json.dumps(manifest, indent=2, ensure_ascii=False))
+    print(f"[dump_error_frames] wrote {len(manifest)} frames + manifest.json to {out_dir}",
+          file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/dump_scene_montage.py

@@ -0,0 +1,383 @@
+#!/usr/bin/env python3
+"""
+dump_scene_montage.py — one BEST and one WORST frame per X-Ray scene, split into
+onscreen vs. offscreen actor identification (TPI / FPI / FN).
+
+For each X-Ray scene (scenes.csv span), scores every sampled second by a simple
+per-second Jaccard agreement (TPI / (TPI+FPI+FN), same spirit as second_score.py)
+and picks the single best-agreement and single worst-agreement second. Each gets
+one output frame: the full frame (not a face crop) with a solid box drawn for
+every currently-active TPI/FPI actor who has a REAL detection backing them, plus a
+black caption panel below with two columns — Onscreen (has a real detection) and
+Offscreen (no real detection: FN misses, and "ghost" detections where the tracker
+is re-emitting a frozen last-known bbox with nothing there — see
+docs/rep4-optimizer-results.md) — names colour-coded by bucket, with a legend.
+
+A predicted bbox is checked against the dump's OWN raw per-frame face detections
+(IoU) to tell a real detection from a ghost. Ghosts are NEVER drawn as boxes (they
+have no real screen position); they only appear as a name in the Offscreen column.
+
+Frames where at least one FPI name isn't in the film's cast AT ALL (an out-of-cast
+misID, not just a right-actor/wrong-scene timing slip) are also copied into
+<out-dir>/out_of_cast_fpi/ for quick review of the most confident wrong answers.
+
+Requires the raw per-frame annotations from `replay.py --raw-out` (bboxes aren't
+in the merged pred.json) and the film's HDF5 dump (for ghost-checking against real
+detections).
+
+Usage:
+  python scripts/optimizer/dump_scene_montage.py \
+      --raw raw.jsonl --dump experiments/dumps/.../dump_X.h5 \
+      --xray experiments/xray/.../900_The_Many_Saints_Of_Newark \
+      --movie "/mnt/movies/.../X.mp4" \
+      --gallery experiments/galleries/gallery_LVFace-B_Glint360K.h5 \
+      --out-dir experiments/results/holdout/montage/many_saints --scene 5
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import cv2
+import h5py
+import numpy as np
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts" / "optimizer"))
+sys.path.insert(0, str(REPO / "scripts" / "validation"))
+
+from sample_eval import load_gallery_keys  # noqa: E402
+from identity import keys_for  # noqa: E402
+
+COLOUR_TPI = (60, 200, 0)     # green, BGR
+COLOUR_FPI = (0, 60, 220)     # red, BGR
+CAPTION_H = 28  # px per line in the bottom strip
+
+
+def load_scene_spans(xray_dir: str):
+    """scene_id -> (t0_sec, t1_sec), from scenes.csv (ms)."""
+    spans = {}
+    with open(Path(xray_dir) / "scenes.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            sn = (r.get("scene") or "").strip()
+            try:
+                spans[sn] = (float(r["start"]) / 1000.0, float(r["end"]) / 1000.0)
+            except (KeyError, ValueError):
+                continue
+    return spans
+
+
+def load_film_cast(xray_dir: str) -> set:
+    """Every actor key X-Ray credits ANYWHERE in the film — used to tell an
+    out-of-cast misID (named someone who isn't even in this film) apart from an
+    in-cast timing slip (right actor, wrong scene), same distinction as
+    second_score.py's FPI_misid vs FPI_incast."""
+    keys = set()
+    with open(Path(xray_dir) / "people.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            nm = (r.get("name_id") or "").strip()
+            person = (r.get("person") or "").strip()
+            if nm or person:
+                keys |= keys_for(imdb_id=nm, name=person)
+    return keys
+
+
+def load_scene_cast(xray_dir: str):
+    """scene_id -> set of actor key-frozensets X-Ray lists as present."""
+    id_to_name = {}
+    with open(Path(xray_dir) / "people.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            nm = (r.get("name_id") or "").strip()
+            if nm:
+                id_to_name[nm] = (r.get("person") or "").strip()
+
+    scene_cast: dict[str, set] = {}
+    with open(Path(xray_dir) / "people_in_scenes.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            sn = (r.get("scene") or "").strip()
+            nm = (r.get("name_id") or "").strip()
+            if sn and nm:
+                scene_cast.setdefault(sn, set()).add(
+                    frozenset(keys_for(imdb_id=nm, name=id_to_name.get(nm))))
+    return scene_cast
+
+
+def load_raw_by_second(raw_path: str):
+    by_second: dict[int, list] = {}
+    with open(raw_path) as f:
+        for line in f:
+            sa = json.loads(line)
+            if sa.get("eof"):
+                continue
+            by_second[int(sa["timestamp_sec"])] = sa.get("visible_actors", [])
+    return by_second
+
+
+def load_dump_faces_by_second(dump_path: str):
+    """second (int) -> list of raw detected bboxes (x,y,w,h), for ghost-checking.
+    A predicted actor's bbox is real iff it overlaps one of these; a bbox with no
+    overlap at all is a frozen/stale re-emission, not an actual detection."""
+    by_second: dict[int, list] = {}
+    with h5py.File(dump_path, "r") as f:
+        ts = f["frames/timestamp_sec"][:]
+        off = f["frames/face_offset"][:]
+        cnt = f["frames/face_count"][:]
+        bbox = f["faces/bbox"][:]
+        for i in range(len(ts)):
+            s, n = int(off[i]), int(cnt[i])
+            by_second[int(ts[i])] = [tuple(b) for b in bbox[s:s + n]]
+    return by_second
+
+
+def iou(a, b):
+    ax, ay, aw, ah = a
+    bx, by, bw, bh = b
+    ix0, iy0 = max(ax, bx), max(ay, by)
+    ix1, iy1 = min(ax + aw, bx + bw), min(ay + ah, by + bh)
+    iw, ih = max(0.0, ix1 - ix0), max(0.0, iy1 - iy0)
+    inter = iw * ih
+    union = aw * ah + bw * bh - inter
+    return inter / union if union > 0 else 0.0
+
+
+def is_ghost(bbox, real_boxes, iou_thresh=0.3):
+    return not any(iou(bbox, rb) >= iou_thresh for rb in real_boxes)
+
+
+def actor_key(a: dict) -> frozenset:
+    return frozenset(keys_for(imdb_id=a.get("imdb_id"), tmdb_id=a.get("tmdb_id"),
+                              jellyfin_id=a.get("jellyfin_id"), name=a.get("name")))
+
+
+COLOUR_FN = (220, 130, 0)   # blue, BGR
+LEGEND = (("TPI (correct)", COLOUR_TPI), ("FPI (wrong)", COLOUR_FPI),
+         ("FN (missed)", COLOUR_FN))
+
+
+def render_frame(frame_path: Path, t: int, tpi_boxes: list, fpi_boxes: list, entries: list):
+    """entries: list of (name, bucket, onscreen) — bucket in {tpi,fpi,fn},
+    onscreen=True iff a real detected face backs this name at this second. Ghost
+    detections (bucket fpi/tpi but no real face — see is_ghost) are never drawn as
+    boxes: they have no real screen position, they only ever appear in the
+    Offscreen column."""
+    img = cv2.imread(str(frame_path))
+    if img is None:
+        return None
+
+    for name, bbox, sim in tpi_boxes:
+        x, y, w, h = (int(v) for v in bbox)
+        cv2.rectangle(img, (x, y), (x + w, y + h), COLOUR_TPI, 2)
+        _label(img, (x, y), f"{name} {sim*100:.0f}%", COLOUR_TPI)
+    for name, bbox, sim in fpi_boxes:
+        x, y, w, h = (int(v) for v in bbox)
+        cv2.rectangle(img, (x, y), (x + w, y + h), COLOUR_FPI, 2)
+        _label(img, (x, y), f"{name} {sim*100:.0f}%", COLOUR_FPI)
+
+    h_img, w_img = img.shape[:2]
+    bucket_colour = {"tpi": COLOUR_TPI, "fpi": COLOUR_FPI, "fn": COLOUR_FN}
+    onscreen = [(n, bucket_colour[b]) for n, b, on in entries if on]
+    offscreen = [(n, bucket_colour[b]) for n, b, on in entries if not on]
+
+    n_rows = max(len(onscreen), len(offscreen), 1)
+    header_h = 24
+    legend_h = CAPTION_H
+    table_h = header_h + n_rows * CAPTION_H + legend_h + 16
+    canvas = np.zeros((h_img + table_h, w_img, 3), dtype=np.uint8)  # black bg
+    canvas[:h_img] = img
+
+    col_x = (8, w_img // 2 + 8)
+    cv2.putText(canvas, f"t={t}s", (8, 16), cv2.FONT_HERSHEY_SIMPLEX, 0.5,
+                (255, 255, 255), 1, cv2.LINE_AA)
+    y0 = h_img + header_h
+    cv2.putText(canvas, "Onscreen", (col_x[0], y0), cv2.FONT_HERSHEY_SIMPLEX, 0.55,
+                (255, 255, 255), 1, cv2.LINE_AA)
+    cv2.putText(canvas, "Offscreen", (col_x[1], y0), cv2.FONT_HERSHEY_SIMPLEX, 0.55,
+                (255, 255, 255), 1, cv2.LINE_AA)
+    cv2.line(canvas, (col_x[1] - 8, h_img), (col_x[1] - 8, h_img + table_h),
+             (90, 90, 90), 1)
+
+    for i in range(n_rows):
+        y = y0 + CAPTION_H * (i + 1)
+        if i < len(onscreen):
+            name, colour = onscreen[i]
+            cv2.putText(canvas, name, (col_x[0], y), cv2.FONT_HERSHEY_SIMPLEX, 0.5,
+                        colour, 1, cv2.LINE_AA)
+        if i < len(offscreen):
+            name, colour = offscreen[i]
+            cv2.putText(canvas, name, (col_x[1], y), cv2.FONT_HERSHEY_SIMPLEX, 0.5,
+                        colour, 1, cv2.LINE_AA)
+
+    ly = y0 + CAPTION_H * (n_rows + 1) + 4
+    lx = 8
+    for label, colour in LEGEND:
+        (tw, _), _ = cv2.getTextSize(label, cv2.FONT_HERSHEY_SIMPLEX, 0.42, 1)
+        cv2.rectangle(canvas, (lx, ly - 10), (lx + 12, ly + 2), colour, cv2.FILLED)
+        cv2.putText(canvas, label, (lx + 18, ly), cv2.FONT_HERSHEY_SIMPLEX, 0.42,
+                    (200, 200, 200), 1, cv2.LINE_AA)
+        lx += tw + 40
+    return canvas
+
+
+def _label(img, pt, text, colour):
+    x, y = pt
+    (tw, th), _ = cv2.getTextSize(text, cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1)
+    strip_y0 = max(0, y - th - 4)
+    cv2.rectangle(img, (x, strip_y0), (x + tw + 4, y), colour, cv2.FILLED)
+    cv2.putText(img, text, (x + 2, y - 2), cv2.FONT_HERSHEY_SIMPLEX, 0.5,
+                (255, 255, 255), 1, cv2.LINE_AA)
+
+
+def extract_frame(movie: str, t: float, out_path: Path):
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    subprocess.run(
+        ["ffmpeg", "-y", "-ss", str(t), "-i", movie, "-frames:v", "1",
+         "-q:v", "2", str(out_path)],
+        check=True, capture_output=True)
+
+
+def classify_second(t: int, gt_cast: set, film_cast: set, raw_by_second: dict,
+                    dump_faces_by_second: dict):
+    """One second's TPI/FPI/FN classification: (score, tpi_boxes, fpi_boxes,
+    entries, has_outofcast). score = Jaccard-style agreement in [0,1], used to
+    rank seconds for best/worst picking."""
+    actors = raw_by_second.get(t, [])
+    real_boxes = dump_faces_by_second.get(t, [])
+
+    tpi_boxes, fpi_boxes = [], []
+    entries = []  # (name, bucket, onscreen)
+    cur_state: dict[frozenset, str] = {}
+    has_outofcast = False
+
+    for a in actors:
+        if a.get("actor_idx", -1) < 0:
+            continue
+        key = actor_key(a)
+        name = a.get("name", "?")
+        bbox = tuple(a["bbox"])
+        sim = a.get("similarity", 0.0)
+        ghost = is_ghost(bbox, real_boxes)
+        hit = any(key & g for g in gt_cast)
+
+        if ghost:
+            cur_state[key] = "ghost"
+            entries.append((name, "fpi" if not hit else "tpi", False))
+        elif hit:
+            tpi_boxes.append((name, bbox, sim))
+            cur_state[key] = "tpi"
+            entries.append((name, "tpi", True))
+        else:
+            fpi_boxes.append((name, bbox, sim))
+            cur_state[key] = "fpi"
+            entries.append((name, "fpi", True))
+
+        if not hit and not (key & film_cast):
+            has_outofcast = True  # named someone not in the film at all (FPI_misid)
+
+    tpi_keys = [k for k, state in cur_state.items() if state in ("tpi", "ghost")]
+    fn_count = 0
+    for g in gt_cast:
+        if any(g & k for k in tpi_keys):
+            continue
+        nm = next((x.split("name:", 1)[1] for x in g if x.startswith("name:")), None)
+        entries.append((nm or next(iter(g), "?"), "fn", False))
+        fn_count += 1
+
+    tp = sum(1 for _, b, on in entries if b == "tpi" and on)
+    fp = sum(1 for _, b, on in entries if b == "fpi")
+    union = tp + fp + fn_count
+    score = tp / union if union else 1.0  # both-empty = perfect agreement
+    return score, tpi_boxes, fpi_boxes, entries, has_outofcast
+
+
+def process_scene(scene_id: str, t0: float, t1: float, gt_cast: set, film_cast: set,
+                  raw_by_second: dict, dump_faces_by_second: dict,
+                  gallery_keys: set | None, movie: str, out_dir: Path,
+                  outofcast_dir: Path):
+    if gallery_keys is not None:
+        gt_cast = {g for g in gt_cast if g & gallery_keys}
+
+    per_second = {}
+    for t in range(int(t0), int(t1)):
+        per_second[t] = classify_second(t, gt_cast, film_cast, raw_by_second,
+                                        dump_faces_by_second)
+    if not per_second:
+        return []
+
+    best_t = max(per_second, key=lambda t: per_second[t][0])
+    worst_t = min(per_second, key=lambda t: per_second[t][0])
+
+    manifest = []
+    for label, t in (("best", best_t), ("worst", worst_t)):
+        score, tpi_boxes, fpi_boxes, entries, has_outofcast = per_second[t]
+        fname = f"{scene_id}_{label}_t{t:06d}.jpg"
+        out_path = out_dir / fname
+        try:
+            extract_frame(movie, t, out_path)
+            canvas = render_frame(out_path, t, tpi_boxes, fpi_boxes, entries)
+            if canvas is not None:
+                cv2.imwrite(str(out_path), canvas)
+            manifest.append({"label": label, "t": t, "score": round(score, 3),
+                             "entries": entries, "file": fname,
+                             "outofcast": has_outofcast})
+            print(f"[scene {scene_id}] {label} t={t}s score={score:.2f} "
+                  f"entries={entries}", file=sys.stderr)
+            if has_outofcast:
+                outofcast_dir.mkdir(parents=True, exist_ok=True)
+                cv2.imwrite(str(outofcast_dir / fname), cv2.imread(str(out_path)))
+        except subprocess.CalledProcessError as e:
+            print(f"[dump_scene_montage] ffmpeg failed at t={t}: {e}", file=sys.stderr)
+
+    return manifest
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--raw", required=True, help="raw per-frame annotations (replay.py --raw-out)")
+    p.add_argument("--dump", required=True, help="film's HDF5 embedding dump (for ghost-checking)")
+    p.add_argument("--xray", required=True)
+    p.add_argument("--movie", required=True)
+    p.add_argument("--gallery")
+    p.add_argument("--out-dir", required=True)
+    p.add_argument("--scene", help="only process this X-Ray scene id (default: all)")
+    args = p.parse_args()
+
+    spans = load_scene_spans(args.xray)
+    scene_cast = load_scene_cast(args.xray)
+    film_cast = load_film_cast(args.xray)
+    raw_by_second = load_raw_by_second(args.raw)
+    dump_faces_by_second = load_dump_faces_by_second(args.dump)
+    gk = load_gallery_keys(args.gallery) if args.gallery else None
+
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    outofcast_dir = out_dir / "out_of_cast_fpi"
+
+    scene_ids = [args.scene] if args.scene else sorted(spans, key=lambda s: spans[s][0])
+    all_manifest = {}
+    for sn in scene_ids:
+        if sn not in spans:
+            print(f"[dump_scene_montage] unknown scene id: {sn}", file=sys.stderr)
+            continue
+        t0, t1 = spans[sn]
+        gt_cast = scene_cast.get(sn, set())
+        scene_dir = out_dir / f"scene_{sn}"
+        scene_dir.mkdir(parents=True, exist_ok=True)
+        m = process_scene(sn, t0, t1, gt_cast, film_cast, raw_by_second,
+                          dump_faces_by_second, gk, args.movie, scene_dir, outofcast_dir)
+        all_manifest[sn] = m
+
+    (out_dir / "manifest.json").write_text(json.dumps(all_manifest, indent=2, ensure_ascii=False))
+    total = sum(len(v) for v in all_manifest.values())
+    n_outofcast = sum(1 for v in all_manifest.values() for r in v if r.get("outofcast"))
+    print(f"[dump_scene_montage] wrote {total} best/worst frames across "
+          f"{len(all_manifest)} scenes to {out_dir} "
+          f"({n_outofcast} copied to {outofcast_dir})", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/fetch_missing_actors.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+"""
+fetch_missing_actors.py — close the gallery coverage gap.
+
+X-Ray credits ~67% of each film's cast that our gallery never had a reference
+embedding for, making those actors unrecoverable FNs no threshold can fix. This
+fetches images for those missing actors (by IMDb nm id → TMDB profile photos),
+embeds them with the SAME SCRFD+ArcFace models (sae_embed), and writes gallery
+entries. Merge the result into the baseline to make those actors recognisable.
+
+nm → TMDB person → /person/{id}/images profile photos → download → embed.
+
+Usage:
+  python scripts/optimizer/fetch_missing_actors.py \
+      --missing missing_actors.json \
+      --out gallery_missing.json \
+      [--images-per-actor 3] [--build-dir build]
+  # TMDB_API_KEY from env/.env
+
+Then merge:
+  python scripts/optimizer/fetch_missing_actors.py --merge \
+      gallery_arcface_w600k_r50.json gallery_missing.json \
+      --out gallery_augmented.json
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts"))
+import sae_env  # noqa: E402  loads .env
+from sae_tmdb import tmdb_get, tmdb_person_for_imdb, TMDB_IMG  # noqa: E402
+from sae_gallery import download_images, wikidata_image_urls   # noqa: E402
+from sae_embed_loader import load_embedder                     # noqa: E402
+
+
+def profile_urls_for_imdb(imdb_id: str, token: str, n: int) -> tuple[str | None, list[str]]:
+    """(tmdb_person_id, [image_url,...]) via /find then /person/{id}/images."""
+    data = tmdb_get(f"/find/{imdb_id}", token, external_source="imdb_id")
+    people = data.get("person_results", [])
+    if not people:
+        return None, []
+    pid = str(people[0]["id"])
+    imgs = tmdb_get(f"/person/{pid}/images", token)
+    profiles = imgs.get("profiles", [])[:n]
+    return pid, [TMDB_IMG + p["file_path"] for p in profiles if p.get("file_path")]
+
+
+def fetch(missing_path, out_path, token, build_dir, models_dir, arcface,
+          images_per_actor, use_wikidata=False):
+    missing = json.loads(Path(missing_path).read_text())
+    src = "TMDB + Wikidata fallback" if use_wikidata else "TMDB"
+    print(f"[fetch] {len(missing)} missing actors to resolve via {src}", file=sys.stderr)
+    embedder = load_embedder(build_dir, models_dir, arcface)
+
+    img_root = Path(tempfile.mkdtemp(prefix="missing_gallery_"))
+    actors = []
+    n_resolved = n_no_tmdb = n_no_img = n_no_face = 0
+    n_via_wikidata = 0
+
+    for i, m in enumerate(missing, 1):
+        nm, name = m["imdb_id"], m.get("name", "")
+        tmdb_id, urls = None, []
+        try:
+            tmdb_id, urls = profile_urls_for_imdb(nm, token, images_per_actor)
+        except Exception as e:
+            print(f"  [{i}] {name}: TMDB error {e}", file=sys.stderr)
+        # Wikidata fallback: keyed cleanly by IMDb nm (P345→P18 Commons photo),
+        # recovers on-camera character actors TMDB's film-centric DB misses.
+        if (not urls) and use_wikidata:
+            wiki_urls = wikidata_image_urls(nm)[:images_per_actor]
+            if wiki_urls:
+                urls = wiki_urls
+                n_via_wikidata += 1
+        if not urls:
+            if tmdb_id is None:
+                n_no_tmdb += 1
+            else:
+                n_no_img += 1
+            continue
+        dest = img_root / nm
+        dest.mkdir(parents=True, exist_ok=True)
+        paths = download_images(urls, dest, images_per_actor)
+        embeddings = []
+        for p in paths:
+            res = embedder.embed(str(p))
+            if res.ok:
+                embeddings.append(list(res.embedding))
+        if not embeddings:
+            n_no_face += 1
+            continue
+        actors.append({"imdb_id": nm, "tmdb_id": str(tmdb_id) if tmdb_id else "",
+                       "jellyfin_id": "", "name": name,
+                       "embeddings": embeddings, "source_images": []})
+        n_resolved += 1
+        if i % 20 == 0 or i == len(missing):
+            print(f"  [{i}/{len(missing)}] resolved={n_resolved} "
+                  f"(wiki={n_via_wikidata}) no_tmdb={n_no_tmdb} no_img={n_no_img} "
+                  f"no_face={n_no_face}", file=sys.stderr)
+
+    Path(out_path).write_text(json.dumps({"actors": actors}, indent=2))
+    n_emb = sum(len(a["embeddings"]) for a in actors)
+    print(f"\n[fetch] recovered {n_resolved}/{len(missing)} actors "
+          f"({n_via_wikidata} via Wikidata), {n_emb} embeddings → {out_path}",
+          file=sys.stderr)
+    print(f"[fetch] unrecoverable: no_tmdb={n_no_tmdb} no_img={n_no_img} "
+          f"no_face={n_no_face}", file=sys.stderr)
+
+
+def merge(base_path, add_path, out_path):
+    base = json.loads(Path(base_path).read_text())
+    add = json.loads(Path(add_path).read_text())
+    have = {a.get("imdb_id") for a in base["actors"] if a.get("imdb_id")}
+    added = [a for a in add["actors"] if a.get("imdb_id") not in have]
+    base["actors"].extend(added)
+    Path(out_path).write_text(json.dumps(base, indent=2))
+    print(f"[merge] {len(base['actors'])-len(added)} + {len(added)} = "
+          f"{len(base['actors'])} actors → {out_path}", file=sys.stderr)
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--merge", nargs=2, metavar=("BASE", "ADD"),
+                   help="merge ADD gallery into BASE → --out")
+    p.add_argument("--missing")
+    p.add_argument("--out", required=True)
+    p.add_argument("--tmdb-key", default=os.environ.get("TMDB_API_KEY"))
+    p.add_argument("--build-dir", default=str(REPO / "build"))
+    p.add_argument("--models-dir", default=str(REPO / "models"))
+    p.add_argument("--arcface", default=None)
+    p.add_argument("--images-per-actor", type=int, default=3)
+    p.add_argument("--wikidata", action="store_true",
+                   help="fall back to Wikidata (P345→P18 Commons photo) when TMDB has no image")
+    args = p.parse_args()
+
+    if args.merge:
+        merge(args.merge[0], args.merge[1], args.out)
+        return
+    if not args.missing:
+        sys.exit("--missing required (or use --merge)")
+    if not args.tmdb_key:
+        sys.exit("no TMDB key — set TMDB_API_KEY")
+    fetch(args.missing, args.out, args.tmdb_key, args.build_dir, args.models_dir,
+          args.arcface, args.images_per_actor, use_wikidata=args.wikidata)
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/gallery_membership.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""
+gallery_membership.py — definitive per-film gallery coverage of X-Ray cast.
+
+For each film, splits the X-Ray cast (people.csv) into those WITH a gallery reference
+embedding and those WITHOUT. This is the model-independent foundation for honest
+FP/FN rates: because every model's gallery is built from the SAME TMDB source images
+(same actors), the membership list is identical across models — only the embedding
+values differ. So FN can be measured over the recognisable denominator (in-gallery
+cast) and out-of-cast misIDs (predicted actor not in the film at all) are well defined.
+
+Outputs experiments/results/membership.json:
+  { film: {
+      xray_cast: N, in_gallery: M, coverage: M/N,
+      in_gallery_names: [...], missing_names: [...] } }
+
+Usage:
+  python scripts/optimizer/gallery_membership.py \
+      --manifest experiments/manifests/films.json \
+      --gallery  gallery_arcface_w600k_r50.json \
+      --out      experiments/results/membership.json
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts" / "validation"))
+from identity import keys_for  # noqa: E402
+
+
+def gallery_keyset(gallery_path: str) -> set:
+    keys = set()
+    for a in json.loads(Path(gallery_path).read_text())["actors"]:
+        if not a.get("embeddings"):
+            continue  # no embedding = not actually recognisable
+        keys |= keys_for(imdb_id=a.get("imdb_id"), tmdb_id=a.get("tmdb_id"),
+                         jellyfin_id=a.get("jellyfin_id"), name=a.get("name"))
+    return keys
+
+
+def film_cast(xray_dir: str) -> dict[str, str]:
+    """nm_id → person name from a film's X-Ray people.csv."""
+    out = {}
+    with open(Path(xray_dir) / "people.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            nm = (r.get("name_id") or "").strip()
+            if nm:
+                out[nm] = (r.get("person") or "").strip()
+    return out
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--manifest", required=True)
+    p.add_argument("--gallery", required=True)
+    p.add_argument("--out", required=True)
+    args = p.parse_args()
+
+    gkeys = gallery_keyset(args.gallery)
+    films = json.loads(Path(args.manifest).read_text())
+
+    report = {}
+    tot_cast = tot_in = 0
+    print(f"{'film':32s} {'cast':>5s} {'in-gal':>7s} {'cover':>6s}")
+    for f in films:
+        cast = film_cast(f["xray"])
+        in_g, miss = [], []
+        for nm, name in cast.items():
+            if keys_for(imdb_id=nm, name=name) & gkeys:
+                in_g.append(name)
+            else:
+                miss.append(name)
+        n, m = len(cast), len(in_g)
+        tot_cast += n; tot_in += m
+        report[f["name"]] = {"xray_cast": n, "in_gallery": m,
+                             "coverage": round(m / n, 3) if n else 0.0,
+                             "in_gallery_names": sorted(in_g),
+                             "missing_names": sorted(miss)}
+        print(f"{f['name'][:32]:32s} {n:>5d} {m:>7d} {m/n*100 if n else 0:>5.0f}%")
+    print(f"{'TOTAL':32s} {tot_cast:>5d} {tot_in:>7d} {tot_in/tot_cast*100:>5.0f}%")
+
+    Path(args.out).parent.mkdir(parents=True, exist_ok=True)
+    Path(args.out).write_text(json.dumps(report, indent=2))
+    print(f"\n→ {args.out}")
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/optimize.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python3
+"""
+optimize.py — Differential Evolution over pipeline thresholds, scored against X-Ray.
+
+Replaces the coarse grid sweep with scipy's differential_evolution over the
+continuous knob space. Each candidate config is a full-9-film replay (real KPN
+nodes) scored against Amazon X-Ray presence, micro-averaged. The gallery is loaded
+once per process (binding caches by path), so an evaluation is just N cheap replays.
+
+Objective: **maximize micro-F1** (DE minimizes, so we return -F1). NOTE: X-Ray recall
+is a face-vs-cast-in-scene ceiling (see [[xray-validation-results]]), so unconstrained
+F1 tends to push prob_threshold DOWN to recover unreachable recall — trading real
+precision for it. We therefore log precision/recall at every evaluation and print
+them at the optimum so the trade-off is visible and you can pick another operating
+point from the trajectory (--trajectory).
+
+Usage:
+  python scripts/optimizer/optimize.py --manifest films.json \
+      --gallery gallery_arcface_w600k_r50.json \
+      --params prob_threshold:0.5:0.999 anneal_sec:1:30 extinction_sec:1:15 \
+      --popsize 20 --maxiter 25 --trajectory traj.json
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import time
+from pathlib import Path
+
+import numpy as np
+from scipy.optimize import differential_evolution
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts" / "optimizer"))
+sys.path.insert(0, str(REPO / "scripts" / "validation"))
+
+import json as _json
+import os
+import subprocess
+import tempfile
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+# Concurrent per-eval replays. Each replay is an isolated subprocess, so parallelism
+# is deadlock-safe; with 9 films/eval, 8 workers replays nearly all at once. Tune via
+# REPLAY_WORKERS (8 is the measured sweet spot on this 24GB GPU).
+REPLAY_WORKERS = int(os.environ.get("REPLAY_WORKERS", "8"))
+
+# DE-level parallelism: how many population candidates get evaluated concurrently
+# (each spawning its own REPLAY_WORKERS film subprocesses). Total concurrent GPU
+# replay processes ≈ DE_WORKERS × min(REPLAY_WORKERS, n_films). Threads, not
+# multiprocessing — each objective() call just waits on subprocess.run, so threads
+# share the GIL fine and avoid pickling the objective/gallery-key cache.
+DE_WORKERS = int(os.environ.get("DE_WORKERS", "1"))
+
+from second_score import score_seconds  # noqa: E402  uniform per-second TPI/FPI scoring
+from sample_eval import load_gallery_keys  # noqa: E402
+
+_GAL_KEYS: dict = {}  # gallery path → key set (fair-recall FN mask), loaded once
+_REPLAY_TIMEOUT = 45  # seconds per film; a wedged replay is killed, not left to hang
+
+REPLAY_CLI = str(Path(__file__).resolve().parent / "replay.py")
+
+
+def _gallery_keys(path):
+    if path not in _GAL_KEYS:
+        _GAL_KEYS[path] = load_gallery_keys(path)
+    return _GAL_KEYS[path]
+
+
+def _replay_subprocess(dump, gallery, cfg, build_dir):
+    """Run one replay in a SUBPROCESS with a timeout, returning its presence JSON.
+
+    In-process replay intermittently DEADLOCKS at network teardown — a KPN worker
+    stuck mid-rocBLAS GEMM inside the ROCm driver makes ~PyNode's jthread.join() hang
+    forever (root-caused via gdb, 2026-07-15). Isolating each replay means a wedged
+    GPU thread only kills that subprocess; the sweep continues. Returns None on
+    timeout/failure (the caller drops that film from the average)."""
+    with tempfile.NamedTemporaryFile("r", suffix=".json", delete=False) as tf:
+        out = tf.name
+    argv = [sys.executable, REPLAY_CLI, "--dump", dump, "--gallery", gallery,
+            "--out", out, "--build-dir", build_dir]
+    for k, v in cfg.items():
+        if isinstance(v, bool):          # store_true flags: pass the flag, not a value
+            if v:
+                argv.append(f"--{k.replace('_', '-')}")
+        else:
+            argv += [f"--{k.replace('_', '-')}", str(v)]
+    try:
+        subprocess.run(argv, timeout=_REPLAY_TIMEOUT, capture_output=True, check=True)
+        return _json.loads(Path(out).read_text())
+    except (subprocess.TimeoutExpired, subprocess.CalledProcessError,
+            FileNotFoundError, ValueError) as e:
+        print(f"[opt] replay failed for {Path(dump).name}: {type(e).__name__}",
+              file=sys.stderr)
+        return None
+    finally:
+        try:
+            Path(out).unlink()
+        except OSError:
+            pass
+
+
+def evaluate(cfg, films, build_dir, step=None):
+    """Objective = MACRO-mean over films of each film's duration-weighted per-scene F1.
+
+    Each film's replay runs in a subprocess (timeout-guarded) to survive the
+    intermittent ROCm teardown deadlock. A film whose replay times out is dropped
+    from the average rather than hanging the whole sweep.
+
+    UNIFORM PER-SECOND scoring (second_score.py): every second of the film is sampled;
+    GT(t) = the cast of the X-Ray scene containing t, Pred(t) = actors whose presence
+    window covers t. Counts instances — TPI / FPI / FN — with FPI weighted 10× when the
+    named actor isn't in the film's cast at all (a real misID vs a timing slip). FN
+    counts only gallery-known actors (fair recall). Reports agreement_rate = mean
+    per-second Jaccard (the "% of on-screen actors we agree with X-Ray about, over
+    time"). Objective = macro-mean across films of the per-second weighted F1.
+
+    expand_gallery: controlled by env SAE_EXPAND (default on). Set SAE_EXPAND=0 to run
+    the no-expansion arm — the overnight matrix tests both to quantify what expansion buys.
+
+    The 9 films' replays run CONCURRENTLY (REPLAY_WORKERS) — each is an isolated
+    subprocess, so parallelism is safe (a wedged one only kills itself)."""
+    if os.environ.get("SAE_EXPAND", "1") == "1":
+        cfg = {**cfg, "expand_gallery": True}
+
+    def _one(film):
+        pj = _replay_subprocess(film["dump"], film.get("gallery"), cfg, build_dir)
+        if pj is None:
+            return None
+        return score_seconds(pj, film["xray"],
+                             gallery_keys=_gallery_keys(film.get("gallery")))
+
+    with ThreadPoolExecutor(max_workers=REPLAY_WORKERS) as ex:
+        per_film = [m for m in ex.map(_one, films) if m is not None]
+    n = len(per_film)
+    if not n:
+        return {"precision": 0.0, "recall": 0.0, "f1": 0.0, "agreement": 0.0,
+                "TPI": 0, "FPI": 0, "FPI_misid": 0, "FN": 0}
+    return {"precision": sum(m["precision"] for m in per_film) / n,
+            "recall": sum(m["recall"] for m in per_film) / n,
+            "f1": sum(m["f1"] for m in per_film) / n,
+            "agreement": sum(m["agreement_rate"] for m in per_film) / n,
+            "TPI": sum(m["TPI"] for m in per_film),
+            "FPI": sum(m["FPI"] for m in per_film),
+            "FPI_misid": sum(m["FPI_misid"] for m in per_film),
+            "FN": sum(m["FN"] for m in per_film)}
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--manifest", required=True)
+    p.add_argument("--gallery", help="default gallery if not per-film")
+    p.add_argument("--params", nargs="+", required=True,
+                   help="knob:lo:hi (e.g. prob_threshold:0.5:0.999). Int knobs kept float, rounded in cfg.")
+    p.add_argument("--build-dir", default=str(REPO / "build"))
+    p.add_argument("--step", type=float, default=5.0)
+    p.add_argument("--popsize", type=int, default=20)
+    p.add_argument("--maxiter", type=int, default=25)
+    p.add_argument("--seed", type=int, default=0)
+    p.add_argument("--trajectory", help="write every evaluation here (JSON lines)")
+    p.add_argument("--out", help="write best config + metrics")
+    args = p.parse_args()
+
+    films = json.loads(Path(args.manifest).read_text())
+    for f in films:
+        f.setdefault("gallery", args.gallery)
+        if not Path(f["dump"]).exists():
+            sys.exit(f"[opt] missing dump for {f['name']}: {f['dump']}")
+
+    names, bounds = [], []
+    int_knobs = {"track_max_frames_missing", "cut_inactive_max_frames"}
+    for spec in args.params:
+        k, lo, hi = spec.split(":")
+        names.append(k); bounds.append((float(lo), float(hi)))
+    print(f"[opt] DE over {names} bounds={bounds}", file=sys.stderr)
+    print(f"[opt] {len(films)} films, popsize={args.popsize}, maxiter={args.maxiter}", file=sys.stderr)
+
+    traj = []
+    evals = [0]
+    t0 = time.time()
+    traj_lock = threading.Lock()
+
+    def vec_to_cfg(x):
+        cfg = {}
+        for k, v in zip(names, x):
+            cfg[k] = int(round(v)) if k in int_knobs else float(v)
+        return cfg
+
+    def objective(x):
+        cfg = vec_to_cfg(x)
+        m = evaluate(cfg, films, args.build_dir, args.step)
+        with traj_lock:
+            evals[0] += 1
+            rec = {"eval": evals[0], "config": cfg, **m, "t": round(time.time() - t0, 1)}
+            traj.append(rec)
+            print(f"[opt] eval {evals[0]:3d}  thr={cfg['prob_threshold']:.2f} "
+                  f"ann={cfg['anneal_sec']:.0f} ext={cfg['extinction_sec']:.1f}  → "
+                  f"F1={m['f1']*100:.1f}% P={m['precision']*100:.1f}% R={m['recall']*100:.1f}% "
+                  f"agree={m.get('agreement', 0)*100:.1f}% misID={m.get('FPI_misid', 0)}",
+                  file=sys.stderr)
+            if args.trajectory:
+                with open(args.trajectory, "a") as tf:
+                    tf.write(json.dumps(rec) + "\n")
+        return -m["f1"]
+
+    de_kwargs = dict(
+        popsize=args.popsize, maxiter=args.maxiter,
+        seed=args.seed, polish=False, tol=1e-4, mutation=(0.5, 1.0), recombination=0.7,
+        init="sobol")
+    if DE_WORKERS > 1:
+        pool = ThreadPoolExecutor(max_workers=DE_WORKERS)
+        de_kwargs["workers"] = pool.map
+    result = differential_evolution(objective, bounds, **de_kwargs)
+
+    best_cfg = vec_to_cfg(result.x)
+    best = evaluate(best_cfg, films, args.build_dir, args.step)
+    print("\n══ DE optimum (by F1) ═══════════════════════════")
+    print(f"  config : {best_cfg}")
+    print(f"  F1     : {best['f1']*100:.2f}%")
+    print(f"  precision: {best['precision']*100:.2f}%   recall: {best['recall']*100:.2f}%")
+    print(f"  TP/FP/FN: {best['TP']}/{best['FP']}/{best['FN']}")
+    print(f"  evaluations: {evals[0]}  time: {time.time()-t0:.0f}s")
+
+    # Also surface the highest-precision config seen (the ship-safe operating point).
+    if traj:
+        hp = max(traj, key=lambda r: (r["precision"], r["recall"]))
+        print("\n── highest-precision config seen (ship-safe) ──")
+        print(f"  config : {hp['config']}")
+        print(f"  P={hp['precision']*100:.2f}%  R={hp['recall']*100:.2f}%  F1={hp['f1']*100:.2f}%")
+
+    if args.out:
+        Path(args.out).write_text(json.dumps(
+            {"best_by_f1": {"config": best_cfg, **best}, "n_evals": evals[0]}, indent=2))
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/reembed_gallery.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""
+reembed_gallery.py — re-embed an existing gallery's actors with a different model.
+
+For the embedding-model bake-off: take a reference gallery (with all actor ids +
+source_images) and produce a new gallery where every actor's embeddings are computed
+by a DIFFERENT ArcFace/LVFace model from the SAME cached source images. All identity
+keys (imdb/tmdb/jellyfin/name) are preserved, so membership/matching is unchanged —
+only the embedding vectors (and hence the model's similarity space) differ.
+
+Source images live in `--images <root>/<jellyfin_id>_<Name>/NN.jpg` (the gallery build
+cache). Actors are matched to their image dir by jellyfin_id first, then name.
+
+Usage:
+  python scripts/optimizer/reembed_gallery.py \
+      --ref gallery_arcface_w600k_r50.h5 \
+      --images images \
+      --arcface models/arcface_r18.onnx \
+      --out experiments/galleries/gallery_arcface_r18.h5 \
+      [--build-dir build]
+"""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts"))
+from sae_embed_loader import load_embedder  # noqa: E402
+from sae_gallery import load_gallery_hdf5, save_gallery_hdf5  # noqa: E402
+
+
+def find_dir(images_root: Path, jellyfin_id: str, name: str) -> Path | None:
+    if jellyfin_id:
+        d = images_root / f"{jellyfin_id}_{name.replace(' ', '_')}"
+        if d.is_dir():
+            return d
+        # jellyfin_id prefix match (name spelling may differ)
+        hits = list(images_root.glob(f"{jellyfin_id}_*"))
+        if hits:
+            return hits[0]
+    hits = list(images_root.glob(f"*_{name.replace(' ', '_')}"))
+    return hits[0] if hits else None
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--ref", required=True, help="reference gallery.h5 (ids + source imgs)")
+    p.add_argument("--images", required=True, help="image cache root")
+    p.add_argument("--arcface", required=True, help="model ONNX to re-embed with")
+    p.add_argument("--out", required=True)
+    p.add_argument("--build-dir", default=str(REPO / "build"))
+    p.add_argument("--models-dir", default=str(REPO / "models"))
+    args = p.parse_args()
+
+    ref = load_gallery_hdf5(Path(args.ref))
+    images_root = Path(args.images)
+    embedder = load_embedder(args.build_dir, args.models_dir, args.arcface)
+
+    out_actors = []
+    n_ok = n_nodir = n_noemb = 0
+    total = len(ref["actors"])
+    for i, a in enumerate(ref["actors"], 1):
+        d = find_dir(images_root, a.get("jellyfin_id", ""), a["name"])
+        if d is None:
+            n_nodir += 1
+            continue
+        embeddings = []
+        for img in sorted(d.glob("*.jpg")):
+            res = embedder.embed(str(img))
+            if res.ok:
+                embeddings.append(list(res.embedding))
+        if not embeddings:
+            n_noemb += 1
+            continue
+        out_actors.append({"imdb_id": a.get("imdb_id", ""), "tmdb_id": a.get("tmdb_id", ""),
+                           "jellyfin_id": a.get("jellyfin_id", ""), "name": a["name"],
+                           "embeddings": embeddings,
+                           "source_images": [p.name for p in sorted(d.glob("*.jpg"))]})
+        n_ok += 1
+        if i % 200 == 0 or i == total:
+            print(f"  [{i}/{total}] ok={n_ok} no_dir={n_nodir} no_emb={n_noemb}",
+                  file=sys.stderr)
+
+    save_gallery_hdf5({"actors": out_actors}, Path(args.out))
+    n_emb = sum(len(a["embeddings"]) for a in out_actors)
+    print(f"[reembed] {Path(args.arcface).stem}: {n_ok}/{total} actors, {n_emb} embeddings "
+          f"→ {args.out}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/replay.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python3
+"""
+replay.py — replay a dumped embedding HDF5 through the real KPN downstream nodes.
+
+Reads an embedding dump (scripts/optimizer/SCHEMA.md), feeds each frame as an
+EmbeddedSceneFrame into a Python-assembled KPN network wiring the *real* C++
+face_tracker → identity_matcher → scene_tracker, and returns the same presence-window
+JSON that scene_analyze's result_sink produces (minimal schema). No decode, no GPU
+embedding — only the cheap downstream tail runs, so a sweep can vary Config knobs
+freely. See [[kpn-python-replay-optimizer]].
+
+CLI:
+  python scripts/optimizer/replay.py --dump film.h5 --gallery gallery.json \
+      --out replayed.json [--prob-threshold 0.99] [--anneal 10] ...
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import time
+from pathlib import Path
+
+import h5py
+import numpy as np
+
+REPO = Path(__file__).resolve().parent.parent.parent
+
+
+def load_frames(dump_path: str, min_conf: float = 0.0):
+    """Yield EmbeddedSceneFrame dicts from the HDF5 dump, then a trailing EOF.
+
+    `min_conf` drops detections below that detector confidence before they reach the
+    matcher — an UPWARD-only detector_conf sweep on already-dumped faces (the dump was
+    made at detector_conf=0.5, so 0.5 is the floor). Lets us test whether near-threshold
+    detections are real faces (raising min_conf hurts recall) or phantoms (it helps
+    precision at no recall cost)."""
+    with h5py.File(dump_path, "r") as f:
+        ts   = f["frames/timestamp_sec"][:]
+        fidx = f["frames/frame_idx"][:]
+        cut  = f["frames/is_cut"][:]
+        off  = f["frames/face_offset"][:]
+        cnt  = f["frames/face_count"][:]
+        emb  = f["faces/embedding"][:]
+        bbox = f["faces/bbox"][:]
+        lmk  = f["faces/landmarks"][:]
+        conf = f["faces/confidence"][:]
+        movie = f.attrs.get("movie", "")
+        fps   = float(f.attrs.get("sample_fps", 1.0))
+
+    frames = []
+    for i in range(len(ts)):
+        s, n = int(off[i]), int(cnt[i])
+        keep = slice(s, s + n)
+        c = np.ascontiguousarray(conf[keep], dtype=np.float32)
+        if min_conf > 0.0 and n:
+            m = c >= min_conf
+            sel = np.where(m)[0]
+            frames.append({
+                "timestamp_sec": float(ts[i]), "frame_idx": int(fidx[i]),
+                "is_cut": bool(cut[i]), "eof": False,
+                "bbox":       np.ascontiguousarray(bbox[keep][sel], dtype=np.float32),
+                "landmarks":  np.ascontiguousarray(lmk[keep][sel],  dtype=np.float32),
+                "confidence": np.ascontiguousarray(c[sel],          dtype=np.float32),
+                "embeddings": np.ascontiguousarray(emb[keep][sel],  dtype=np.float32),
+            })
+        else:
+            frames.append({
+                "timestamp_sec": float(ts[i]), "frame_idx": int(fidx[i]),
+                "is_cut": bool(cut[i]), "eof": False,
+                "bbox":       np.ascontiguousarray(bbox[keep], dtype=np.float32),
+                "landmarks":  np.ascontiguousarray(lmk[keep],  dtype=np.float32),
+                "confidence": c,
+                "embeddings": np.ascontiguousarray(emb[keep],  dtype=np.float32),
+            })
+    last_ts = float(ts[-1]) if len(ts) else 0.0
+    frames.append({"timestamp_sec": last_ts, "eof": True})
+    return frames, str(movie), fps
+
+
+def replay(dump_path: str, gallery: str, cfg: dict, build_dir: str, stop: bool = True,
+          raw_out: str | None = None) -> dict:
+    """Run the dump through the real KPN chain; return minimal-schema presence JSON.
+
+    cfg may include "detector_conf" to prune dumped detections below that confidence
+    (upward-only from the 0.5 dump floor) before matching.
+
+    raw_out: if set, also write the raw per-frame annotations (timestamp, actor_idx,
+    name, bbox, similarity — one entry per input frame, before merging into windows)
+    as JSON lines to this path. Needed to draw bounding boxes on extracted frames;
+    the merged window schema returned by this function has no per-frame bbox."""
+    sys.path.insert(0, build_dir)
+    import sae_kpn
+
+    frames, movie, fps = load_frames(dump_path, min_conf=float(cfg.get("detector_conf", 0.0)))
+
+    net = sae_kpn.Network()
+    sae_kpn._register_types(net)
+
+    idx = [0]
+    eof = {"timestamp_sec": frames[-1]["timestamp_sec"], "eof": True}
+
+    def source():
+        # A no-input source node's run_loop calls this in a tight loop. Once frames
+        # are exhausted we must NOT hot-spin returning EOF — that pegs a core and
+        # floods the downstream channel with EOFs (livelock that wedged DE). Sleep
+        # briefly after the single real EOF so net.stop() can tear the thread down.
+        i = idx[0]
+        idx[0] += 1
+        if i < len(frames):
+            return frames[i]
+        time.sleep(0.05)
+        return eof
+
+    # Channel capacity must exceed the frame count so the fast source can't overflow
+    # a downstream FIFO before the serial reader drains it — PyNode DROPS on overflow,
+    # which would silently truncate the replay. Size to the whole film + slack.
+    # Every channel gets capacity ≥ the whole film so NOTHING can ever overflow-drop:
+    # the source can push all frames before any downstream node has drained, and a
+    # dropped frame silently corrupts the score. Memory is cheap (a few k pointers);
+    # correctness is not. Generous slack on top.
+    cap = len(frames) * 2 + 64
+    sae_kpn.add_node_python(net, "replay", source, [], ["EmbeddedSceneFrame"], cap)
+    sae_kpn.add_face_tracker(net, "tracker", cfg, cap)
+    sae_kpn.add_identity_matcher(net, "matcher", gallery, cfg, cap)
+    sae_kpn.add_scene_tracker(net, "scene", cfg, cap)
+    net.connect("replay", 0, "tracker", 0)
+    net.connect("tracker", 0, "matcher", 0)
+    net.connect("matcher", 0, "scene", 0)
+    net.build()
+    net.start()
+
+    # Read exactly one annotation per input frame. The source emits EOF as an ordinary
+    # value AFTER the last frame, but the concurrent pipeline lets that EOF OVERTAKE
+    # the last few real frames still flowing tracker→matcher→scene. Breaking on the
+    # first eof therefore dropped a random tail (~0.5–1%, race-dependent). Instead we
+    # keep reading past eof until we've collected all n_frames annotations (or hit a
+    # run of consecutive eofs meaning the pipeline is genuinely drained).
+    n_expected = len(frames) - 1   # excludes the trailing eof frame
+    annotations = []
+    eof_streak = 0
+    max_reads = n_expected * 2 + 32
+    for _ in range(max_reads):
+        sa = net.read("scene", 0)
+        if sa.get("eof"):
+            eof_streak += 1
+            # stragglers can still arrive after an eof; only stop once we've either
+            # got everything or seen several eofs in a row (truly drained).
+            if len(annotations) >= n_expected or eof_streak >= 8:
+                break
+            continue
+        eof_streak = 0
+        annotations.append(sa)
+        if len(annotations) >= n_expected:
+            break
+
+    if raw_out:
+        with open(raw_out, "w") as f:
+            for sa in annotations:
+                f.write(json.dumps(sa) + "\n")
+
+    result = build_minimal(annotations, movie, fps, cfg)
+    if stop:
+        net.stop()
+    return result
+
+
+def build_minimal(annotations, movie, fps, cfg) -> dict:
+    """Reproduce result_sink's minimal schema: per-actor annealed [start,end] windows.
+
+    Mirrors ResultSinkFunc::build_actor_windows — merge each actor's detection
+    timestamps into windows, bridging gaps shorter than anneal_sec.
+    """
+    anneal = float(cfg.get("anneal_sec", 10.0))
+    info = {}           # actor_idx -> identity fields
+    times = {}          # actor_idx -> [timestamps]
+    for sa in annotations:
+        for a in sa["visible_actors"]:
+            if a["actor_idx"] < 0:
+                continue
+            info[a["actor_idx"]] = a
+            times.setdefault(a["actor_idx"], []).append(sa["timestamp_sec"])
+
+    actors = []
+    for idx, ts in times.items():
+        ts.sort()
+        scenes = []
+        ws = we = ts[0]
+        for t in ts[1:]:
+            if t - we > anneal:
+                scenes.append([ws, we])
+                ws = t
+            we = t
+        scenes.append([ws, we])
+        a = info[idx]
+        actors.append({
+            "name": a["name"], "imdb_id": a["imdb_id"], "tmdb_id": a["tmdb_id"],
+            "jellyfin_id": a["jellyfin_id"], "scenes": scenes,
+        })
+
+    return {"schema_version": 1, "movie": movie, "sample_fps": fps,
+            "anneal_sec": anneal, "actors": actors}
+
+
+CFG_KEYS = ["detector_conf", "prob_threshold", "match_prior", "match_threshold", "match_ratio",
+            "match_ratio_ceil", "track_alpha", "track_min_iou", "track_max_embed_dist",
+            "track_max_frames_missing", "cut_revive_sim", "cut_inactive_max_frames",
+            "extinction_sec", "anneal_sec"]
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--dump", required=True, help="embedding HDF5 dump")
+    p.add_argument("--gallery", required=True)
+    p.add_argument("--out", required=True, help="output presence JSON")
+    p.add_argument("--raw-out", help="also write raw per-frame annotations (JSONL, with bboxes) here")
+    p.add_argument("--build-dir", default=str(REPO / "build"))
+    for k in CFG_KEYS:
+        p.add_argument(f"--{k.replace('_','-')}", type=float, default=None)
+    # per-film gallery expansion: promotes pose-varied views of confidently-identified
+    # actors into an in-memory annex, recovering ~+4 recall at no precision cost.
+    p.add_argument("--expand-gallery", action="store_true")
+    args = p.parse_args()
+
+    cfg = {k: getattr(args, k) for k in CFG_KEYS if getattr(args, k) is not None}
+    if args.expand_gallery:
+        cfg["expand_gallery"] = True
+    # stop=True: PyNode::stop() sets stop_flag_ before joining, so the source
+    # thread's run_loop actually exits. stop=False skips that, leaving stop_flag_
+    # false forever — the PyNode destructor's jthread.join() then blocks forever
+    # (verified via gdb: stuck in the source node's run_loop, not the GEMM path).
+    result = replay(args.dump, args.gallery, cfg, args.build_dir, stop=True,
+                    raw_out=args.raw_out)
+    Path(args.out).write_text(json.dumps(result, indent=2))
+    print(f"[replay] {len(result['actors'])} actors → {args.out}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/second_score.py

@@ -0,0 +1,197 @@
+#!/usr/bin/env python3
+"""
+second_score.py — uniform per-second agreement with X-Ray.
+
+Unlike scene_score.py (which unions our detections over a whole X-Ray scene), this
+samples EVERY SECOND of the film and asks: at second t, do we name the same actors
+X-Ray says are on screen?
+
+  GT(t)   = the cast set of the X-Ray scene containing t (scenes.csv + people_in_scenes)
+  Pred(t) = actors whose presence window [start,end] covers t (the pipeline's output)
+
+Per second we count instances:
+  TPI = |Pred ∩ GT|                      true positive instances
+  FPI = |Pred − GT|                      false positive instances, split into:
+          FPI_misid  — actor NOT in the film's cast at all (a real misID, weighted 10×)
+          FPI_incast — actor in the film but not this second (timing/boundary)
+  FN  = |GT − Pred|, counting only gallery-known actors (fair recall — ~67% of X-Ray
+        cast have no reference embedding and can never be recognised)
+  agreement at t = Jaccard |Pred ∩ GT| / |Pred ∪ GT|  — PARTIAL credit, so naming 2
+        of 3 actors scores 2/3, not 0. Averaged over sampled seconds → the
+        "what fraction of the time do we agree with X-Ray" number. (Exact-set match is
+        reported separately as exact_match_rate; it is far harsher and dominated by
+        recall.)
+
+Objective (DE): per-second F1 computed with the WEIGHTED FPI, so naming someone who
+isn't in the film hurts 10× more than a boundary slip.
+
+Reported: TPI, FPI (+split), FN, precision, recall, F1, and agreement_rate — the
+fraction of sampled seconds where we exactly matched X-Ray.
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+import json
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(REPO / "scripts" / "validation"))
+from identity import keys_for  # noqa: E402
+
+
+def load_second_timeline(xray_dir: str):
+    """Return (timeline, film_cast_keys, duration).
+
+    timeline: dict second -> list of actor key-sets on screen per X-Ray.
+    Each second inside a scene [start,end) inherits that scene's cast set.
+    """
+    d = Path(xray_dir)
+    id_to_name = {}
+    with open(d / "people.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            nm = (r.get("name_id") or "").strip()
+            if nm:
+                id_to_name[nm] = (r.get("person") or "").strip()
+
+    film_cast = set()
+    for nm, name in id_to_name.items():
+        film_cast |= keys_for(imdb_id=nm, name=name)
+
+    spans = {}
+    with open(d / "scenes.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            sn = (r.get("scene") or "").strip()
+            try:
+                spans[sn] = (float(r["start"]) / 1000.0, float(r["end"]) / 1000.0)
+            except (KeyError, ValueError):
+                continue
+
+    scene_cast: dict[str, list] = {}
+    with open(d / "people_in_scenes.csv", newline="", encoding="utf-8") as f:
+        for r in csv.DictReader(f):
+            sn = (r.get("scene") or "").strip()
+            nm = (r.get("name_id") or "").strip()
+            if sn in spans and nm:
+                scene_cast.setdefault(sn, []).append(
+                    frozenset(keys_for(imdb_id=nm, name=id_to_name.get(nm))))
+
+    timeline: dict[int, list] = {}
+    duration = 0.0
+    for sn, (t0, t1) in spans.items():
+        duration = max(duration, t1)
+        cast = scene_cast.get(sn, [])
+        for t in range(int(t0), int(t1)):
+            timeline[t] = cast
+    return timeline, film_cast, duration
+
+
+def load_pred_intervals(pred_json: dict):
+    """[(keyset, [(t0,t1),...]), ...] for each actor the pipeline named."""
+    out = []
+    for a in pred_json.get("actors", []):
+        keys = frozenset(keys_for(imdb_id=a.get("imdb_id"), tmdb_id=a.get("tmdb_id"),
+                                  jellyfin_id=a.get("jellyfin_id"), name=a.get("name")))
+        out.append((keys, [(float(t0), float(t1)) for t0, t1 in a.get("scenes", [])]))
+    return out
+
+
+def _match(P, G):
+    """Greedy 1:1 match by key intersection; returns (n_matched, matched_G_mask)."""
+    used = [False] * len(G)
+    n = 0
+    for pa in P:
+        for j, ga in enumerate(G):
+            if not used[j] and (pa & ga):
+                used[j] = True
+                n += 1
+                break
+    return n, used
+
+
+def score_seconds(pred_json: dict, xray_dir: str, gallery_keys: set | None = None,
+                  misid_weight: float = 10.0):
+    timeline, film_cast, duration = load_second_timeline(xray_dir)
+    pred = load_pred_intervals(pred_json)
+
+    TPI = FPI = FN = 0
+    FPI_misid = FPI_incast = 0
+    FPI_w = 0.0
+    jaccard_sum = 0.0   # partial-credit agreement, summed over seconds
+    exact = 0
+    n_sec = 0
+
+    for t in sorted(timeline):
+        G = [set(a) for a in timeline[t]]
+        P = [set(k) for k, wins in pred if any(w0 <= t <= w1 for w0, w1 in wins)]
+        # fair recall: only GT actors we could possibly recognise
+        if gallery_keys is not None:
+            G = [g for g in G if g & gallery_keys]
+
+        tp, matched = _match(P, G)
+        # classify each unmatched prediction
+        fpi_w = 0.0
+        n_fp = 0
+        for pa in P:
+            if any(pa & ga for ga in G):
+                continue
+            n_fp += 1
+            if pa & film_cast:
+                FPI_incast += 1; fpi_w += 1.0
+            else:
+                FPI_misid += 1; fpi_w += misid_weight
+        fn = len(G) - tp
+
+        TPI += tp; FPI += n_fp; FN += fn; FPI_w += fpi_w
+        # partial-credit agreement: |∩| / |∪| at this second
+        union = tp + n_fp + fn
+        if union:
+            jaccard_sum += tp / union
+        else:
+            jaccard_sum += 1.0   # both empty = agreement (nobody on screen)
+        if n_fp == 0 and fn == 0:
+            exact += 1
+        n_sec += 1
+
+    prec = TPI / (TPI + FPI_w) if TPI + FPI_w else 0.0   # weighted (misID hurts 10×)
+    prec_raw = TPI / (TPI + FPI) if TPI + FPI else 0.0
+    rec = TPI / (TPI + FN) if TPI + FN else 0.0
+    f1 = 2 * prec * rec / (prec + rec) if prec + rec else 0.0
+    return {"TPI": TPI, "FPI": FPI, "FPI_misid": FPI_misid, "FPI_incast": FPI_incast,
+            "FN": FN, "precision": prec, "precision_raw": prec_raw, "recall": rec,
+            "f1": f1,
+            # partial-credit: mean per-second Jaccard = "% of actors we agree on, over time"
+            "agreement_rate": jaccard_sum / n_sec if n_sec else 0.0,
+            "exact_match_rate": exact / n_sec if n_sec else 0.0,
+            "n_seconds": n_sec, "duration_sec": duration}
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__,
+                                formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument("--pred", required=True)
+    p.add_argument("--xray", required=True)
+    p.add_argument("--gallery")
+    args = p.parse_args()
+    gk = None
+    if args.gallery:
+        sys.path.insert(0, str(REPO / "scripts" / "validation"))
+        from sample_eval import load_gallery_keys
+        gk = load_gallery_keys(args.gallery)
+    m = score_seconds(json.loads(Path(args.pred).read_text()), args.xray, gk)
+    print(f"seconds sampled : {m['n_seconds']} (film {m['duration_sec']:.0f}s)")
+    print(f"TPI/FPI/FN      : {m['TPI']}/{m['FPI']}/{m['FN']}")
+    print(f"  FPI misID     : {m['FPI_misid']}  (actor not in film — weighted 10x)")
+    print(f"  FPI in-cast   : {m['FPI_incast']}")
+    print(f"precision (w)   : {m['precision']*100:.1f}%   raw {m['precision_raw']*100:.1f}%")
+    print(f"recall          : {m['recall']*100:.1f}%")
+    print(f"F1 (weighted)   : {m['f1']*100:.1f}%")
+    print(f"AGREEMENT       : {m['agreement_rate']*100:.1f}%  (mean per-second % of actors "
+          f"we agree on with X-Ray)")
+    print(f"  exact-set match: {m['exact_match_rate']*100:.1f}% of seconds (harsher, "
+          f"all-or-nothing)")
+
+
+if __name__ == "__main__":
+    main()

scripts/optimizer/test_sae_kpn.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""
+Smoke test for the sae_kpn module: assemble the real downstream pipeline nodes
+(face_tracker → identity_matcher → scene_tracker) in a Python-driven KPN network,
+fed by a no-input Python source node, and verify SceneAnnotations flow out.
+
+Proves the KPN-native replay path works without any numpy port of node logic.
+Run:  python scripts/optimizer/test_sae_kpn.py [gallery.json] [build_dir]
+"""
+import sys
+import queue
+import numpy as np
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent.parent
+GAL = sys.argv[1] if len(sys.argv) > 1 else str(REPO / "gallery_arcface_w600k_r50.json")
+BUILD = sys.argv[2] if len(sys.argv) > 2 else str(REPO / "build")
+sys.path.insert(0, BUILD)
+import sae_kpn  # noqa: E402
+
+
+def make_frame(t, n):
+    e = np.random.randn(n, 512).astype(np.float32)
+    e /= np.linalg.norm(e, axis=1, keepdims=True)
+    return {"timestamp_sec": t, "eof": False,
+            "bbox": np.tile(np.array([10, 10, 50, 50], np.float32), (n, 1)),
+            "landmarks": np.tile(np.arange(10, dtype=np.float32), (n, 1)),
+            "confidence": np.full((n,), 0.9, np.float32), "embeddings": e}
+
+
+def main():
+    net = sae_kpn.Network()
+    sae_kpn._register_types(net)
+    cfg = {"prob_threshold": 0.99, "anneal_sec": 10.0, "extinction_sec": 5.0}
+
+    frames = [make_frame(float(t), 1) for t in range(3)]
+    frames.append({"timestamp_sec": 3.0, "eof": True})
+    idx = [0]
+    eof_frame = {"timestamp_sec": 3.0, "eof": True}
+
+    def source():
+        # Emit each frame once, then keep returning EOF (never block) so the node
+        # thread stays responsive to stop() after the sink has seen EOF.
+        i = idx[0]
+        idx[0] += 1
+        return frames[i] if i < len(frames) else eof_frame
+
+    sae_kpn.add_node_python(net, "replay", source, [], ["EmbeddedSceneFrame"], 8)
+    sae_kpn.add_face_tracker(net, "tracker", cfg, 16)
+    sae_kpn.add_identity_matcher(net, "matcher", GAL, cfg, 16)
+    sae_kpn.add_scene_tracker(net, "scene", cfg, 16)
+    net.connect("replay", 0, "tracker", 0)
+    net.connect("tracker", 0, "matcher", 0)
+    net.connect("matcher", 0, "scene", 0)
+    net.build()
+    net.start()
+
+    got = []
+    for _ in range(4):
+        sa = net.read("scene", 0)
+        got.append(sa)
+        if sa.get("eof"):
+            break
+    net.stop()
+
+    non_eof = [g for g in got if not g.get("eof")]
+    assert len(non_eof) == 3, f"expected 3 annotations, got {len(non_eof)}"
+    assert got[-1].get("eof"), "expected trailing EOF"
+    assert [g["timestamp_sec"] for g in non_eof] == [0.0, 1.0, 2.0], "timestamps wrong"
+    assert all("visible_actors" in g for g in non_eof), "missing visible_actors"
+    print(f"OK: {len(non_eof)} annotations through the real KPN chain, EOF received")
+
+
+if __name__ == "__main__":
+    main()