feat(replay): the whole replay chain is C++, including the sink

The sae_kpn module has not compiled since the AR-007/AR-008 tracker redesign,
and was switched off at the build rather than patched because the fix is a
restructuring. Two failures, one cause.

It did not compile: `add_face_tracker` built FaceTrackerFunc from a Config
alone, and the tracker has required a TrackRegistry and a calibration since
association moved into probability space.

And presence was rebuilt in Python. `replay.py::build_minimal` merged
per-frame detections into windows by annealing gaps, which is what the
pipeline did before AR-012. The sink builds a window from a TrackRegistry
claim instead — the extent of a track an actor owned, starting when they
appeared rather than when recognition first succeeded. Those answer different
questions, so every sweep was tuning against a contract the shipped code had
stopped honouring.

Both follow from the seam being a factory per node. The chain has a
construction order — the matcher fits the calibration, the registry needs a
discounter built from it, the tracker needs both, and the sink needs the
registry's claims — and independent factories cannot express it, so the
tracker kept being built against a signature that no longer existed. One
`add_pipeline` mirrors main.cpp exactly and is now the only way to build the
chain, so the ordering cannot be got wrong again from Python. DP-001 is the
requirement behind it: a replay harness is a front-end, and its job is to
supply frames and read the result, not to re-derive presence.

Lifetimes needed a home. ResultSinkFunc holds `const Config&` and
`std::atomic<bool>&`, which under main() are locals in a frame outliving the
pipeline; there is no such frame when the network is built and torn down from
Python. ReplaySession owns both for the network's lifetime, keyed by network
and released explicitly — a sweep builds one network per replay and the sink
retains every annotation, so holding them forever would grow with films x
configs. Getting this wrong presented as an empty output_path: the sink
announced `[result_sink] writing ` and wrote nothing.

test_sae_kpn.py is ported rather than left behind. It called all three removed
factories and asserted on SceneAnnotations read back per frame; neither half
survives, so it now waits on pipeline_done and asserts on the file the sink
writes. Verified against gallery_lvface.h5: three frames through the real
chain, timestamps 0/1/2, truth file written. EOF is a control token the sink
flushes on and does not record, so three inputs give three frames, never four.

SAE_BUILD_KPN_BINDINGS goes back to ON.

TRACES: VR-011, VR-002 | DP-001 | PR-002

CMakeLists.txt

@@ -326,32 +326,17 @@ target_link_libraries(sae_embed PRIVATE sae_gallery)
 # network (KPN_BUILD_PYTHON is enabled per-TU inside the .cpp). Powers the
 # threshold-sweep optimizer in scripts/optimizer/.
 #
-# OFF by default, and this is a statement of fact rather than a preference: the
-# module HAS NOT COMPILED since the AR-007/AR-008 tracker redesign. FaceTrackerFunc
-# now requires a TrackRegistry and a calibration at construction, and the binding
-# still builds it from a Config alone. The .so in a stale build/ directory
-# predates that change.
-#
-# Fixing it is VR-011's job, not a patch: the tracker needs the calibration, the
-# calibration comes from the matcher, and the matcher is added to the network
-# afterwards -- so the seam has to be restructured, exactly as main.cpp already
-# is (matcher first, then registry, then tracker). Presence claims do not cross
-# the seam at all today, which is the other half of the same rewrite.
-#
-# Recorded as a switch rather than left as a build error so that `cmake --build`
-# succeeds and the breakage is attributed instead of rediscovered. Turning it on
-# reproduces the failure immediately, which is the point.
-#
 # TRACES: VR-011 | PR-002
-option(SAE_BUILD_KPN_BINDINGS
-       "Build the sae_kpn Python module (BROKEN pending VR-011)" OFF)
+# ON again. It was OFF for one commit because it had not compiled since the
+# AR-007/AR-008 tracker redesign -- the binding built FaceTrackerFunc from a
+# Config alone, and the tracker had required a registry and a calibration since.
+# VR-011 replaced the three per-node factories with one `add_pipeline` that
+# builds the chain in main.cpp's order, which is the only order that satisfies
+# those dependencies, so the failure mode cannot recur from Python.
+option(SAE_BUILD_KPN_BINDINGS "Build the sae_kpn Python module" ON)
 if(SAE_BUILD_KPN_BINDINGS)
     nanobind_add_module(sae_kpn src/kpn_bindings.cpp)
     target_link_libraries(sae_kpn PRIVATE sae_gallery)
-else()
-    message(STATUS
-        "sae_kpn: SKIPPED (SAE_BUILD_KPN_BINDINGS=OFF). The Python replay "
-        "bindings do not compile against the post-AR-012 tracker; see VR-011.")
 endif()
 
 # ── sae_audio — Python module: the v1 audio signature (IR-004) ────────────────

scripts/docs/run_holdout_all_models.py

@@ -79,10 +79,9 @@ def main():
                 "--dump", str(dump), "--gallery", str(gallery),
                 "--out", str(pred_path),
                 "--prob-threshold", str(cfg["prob_threshold"]),
-                # anneal_sec is replay-local now (it configures replay.py's
-                # own windowing, not the pipeline). extinction_sec is gone
-                # entirely with SceneTrackerFunc -- see AR-012/AR-013.
-                "--anneal-sec", str(cfg.get("anneal_sec", 10.0)),
+                # anneal_sec and extinction_sec are both gone: presence is
+                # the registry's, built from track extents (AR-012/AR-013), and
+                # replay.py no longer windows anything itself (VR-011).
                 "--expand-gallery",
             ]
             print(f"RUN {model}/{film['slug']}...", file=sys.stderr)

scripts/optimizer/optimize.py

@@ -17,7 +17,7 @@ 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 track_alpha:0:1 \
+      --params prob_threshold:0.5:0.999 ownership_logodds:0.5:4 track_alpha:0:1 \
       --popsize 20 --maxiter 25 --trajectory traj.json
 """
 from __future__ import annotations
@@ -239,7 +239,7 @@ def main():
             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.get('anneal_sec', float('nan')):.0f}  → "
+                  f"own={cfg.get('ownership_logodds', float('nan')):.2f}  → "
                   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)

scripts/optimizer/replay.py

@@ -2,18 +2,22 @@
 """
 replay.py — replay a dumped embedding HDF5 through the real KPN downstream nodes.
 
-TRACES: VR-002 | PR-002
+TRACES: VR-002, VR-011 | PR-002
 
 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 → frame_annotation, 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]].
+face_tracker → identity_matcher → frame_annotation → result_sink, and reads back
+the truth file that sink wrote. No decode, no GPU embedding — only the cheap
+downstream tail runs, so a sweep can vary Config knobs freely.
+
+The sink is part of the network, not a Python reimplementation of it. That is
+VR-011: presence comes from TrackRegistry claims, so a replayed window and a
+scene_analyze window are produced by the same code rather than by two functions
+that agreed once. 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-sec 10] ...
+      --out replayed.json [--prob-threshold 0.99] [--track-extinction-sec 5] ...
 """
 from __future__ import annotations
 
@@ -108,17 +112,32 @@ def load_frames(dump_path: str, min_conf: float = 0.0):
     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.
+def replay(dump_path: str, gallery: str, cfg: dict, build_dir: str,
+           out_path: str, stop: bool = True, raw_out: str | None = None,
+           eof_timeout: float = 300.0) -> dict:
+    """Run the dump through the real KPN chain and return the truth file it wrote.
 
-    cfg may include "detector_conf" to prune dumped detections below that confidence
-    (upward-only from the 0.5 dump floor) before matching.
+    TRACES: VR-011, VR-002 | PR-002
 
-    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."""
+    `out_path` is where the C++ sink writes. That is the change VR-011 makes:
+    the presence windows in that file are built by ResultSinkFunc from
+    TrackRegistry claims -- the extent of a track an actor owned (AR-012),
+    ending at the last sighting (AR-013) -- and are byte-for-byte the same
+    construction scene_analyze ships. This function used to build them itself,
+    in Python, by annealing gaps between per-frame detections, which is what the
+    pipeline did BEFORE AR-012. A sweep tuned against that was tuning a contract
+    the shipped code had stopped honouring.
+
+    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 per-frame annotations as JSON lines for the
+    montage renderers. Derived from the truth file's own `frames` array rather
+    than tapped separately out of the network -- see write_raw_frames.
+
+    eof_timeout: how long to wait for the sink to write. A replay that never
+    reaches EOF is a wedged pipeline, and returning an empty result would look
+    like a film with no cast rather than like a failure."""
     sys.path.insert(0, build_dir)
     import sae_kpn
 
@@ -161,120 +180,104 @@ def replay(dump_path: str, gallery: str, cfg: dict, build_dir: str, stop: bool =
     # 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,
-                                 stamp["model_name"], stamp["model_sha256"])
-    sae_kpn.add_frame_annotation(net, "scene", cap)
+
+    # TRACES: VR-011, VR-002 | DP-001 | PR-002
+    # One call builds tracker -> matcher -> annotation -> sink in the only order
+    # that works (the matcher fits the calibration the tracker needs, and the
+    # sink needs the registry's claims). This used to be three factory calls
+    # assembled here, which is how the seam broke: the ordering constraint could
+    # not be expressed, so the tracker was built from a Config alone long after
+    # it had started requiring a registry and a calibration.
+    cfg = dict(cfg)
+    cfg["output_path"] = out_path
+    cfg["movie_path"]  = movie
+    cfg["sample_fps"]  = fps
+    # Verbosity 1 (standard) adds the per-frame array; only pay for it when the
+    # caller wants raw frames, since it retains every annotation in memory.
+    cfg["verbosity"] = 1 if raw_out else 0
+    sae_kpn.add_pipeline(net, gallery, cfg, cap,
+                         stamp["model_name"], stamp["model_sha256"])
+
     net.connect("replay", 0, "tracker", 0)
     net.connect("tracker", 0, "matcher", 0)
-    net.connect("matcher", 0, "scene", 0)
+    net.connect("matcher", 0, "annotation", 0)
+    net.connect("annotation", 0, "sink", 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
+    # The sink writes on the EOF annotation. Wait for it rather than reading
+    # anything back through the seam: presence is the registry's answer, and the
+    # registry lives entirely on the C++ side.
+    #
+    # This replaces a read loop that pulled one SceneAnnotation per input frame
+    # and rebuilt windows in Python. That loop needed a heuristic -- "keep
+    # reading past eof until we've collected all n_frames annotations, or hit a
+    # run of 8 consecutive eofs" -- to work around a tail it was losing. None of
+    # that exists now: nothing is read per frame, so nothing can be lost per
+    # frame.
+    deadline = time.time() + eof_timeout
+    while not sae_kpn.pipeline_done(net):
+        if time.time() > deadline:
+            sae_kpn.release_pipeline(net)
+            raise TimeoutError(
+                f"replay did not finish within {eof_timeout}s "
+                f"({len(frames) - 1} frames); the sink never saw EOF")
+        time.sleep(0.02)
 
-    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()
+    sae_kpn.release_pipeline(net)
+
+    with open(out_path) as f:
+        result = json.load(f)
+
+    if raw_out:
+        write_raw_frames(result, raw_out)
     return result
 
 
-def build_minimal(annotations, movie, fps, cfg) -> dict:
-    """Per-actor [start,end] windows, built by annealing per-frame detections.
+def write_raw_frames(truth: dict, raw_out: str) -> None:
+    """Per-frame annotations as JSONL, for the montage/error-frame renderers.
 
     TRACES: VR-011 | PR-002
 
-    This NO LONGER mirrors ResultSinkFunc, and the docstring used to claim it
-    did. The sink builds a window from a TrackRegistry claim -- the extent
-    [first_seen, last_seen] of a track an actor owned (AR-012) -- so a window
-    starts when the actor appeared rather than when recognition first
-    succeeded, and interior gaps are absorbed by the track surviving them.
-    This function still bridges gaps between isolated accepted frames, which is
-    what anneal_sec did before AR-012/AR-013 withdrew it.
+    Derived from the truth file's own `frames` array (verbosity 1) rather than
+    from a second stream tapped out of the network. One producer, one set of
+    numbers: a bbox drawn on a montage is now provably the bbox the sink
+    recorded, which it was not when Python read annotations separately.
 
-    So a replayed window and a pipeline window are answers to different
-    questions, and a sweep tuned against this one is not tuning the shipped
-    behaviour. That is VR-011's job -- "rewrite the replay harness for the
-    post-AR-012 output contract" -- and it is a rewrite, not an edit, because
-    the registry's claims do not cross the Python seam at all today.
-
-    `anneal_sec` is therefore replay-local now: it configures THIS function and
-    is no longer forwarded to the C++ Config, which has no such field.
+    The shape is the legacy one -- {timestamp_sec, visible_actors:[...]} with
+    actor_idx/bbox/name/similarity -- because dump_scene_montage.py and
+    dump_error_frames.py read exactly those fields, and rewriting them is not
+    what this requirement is about.
     """
-    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}
+    with open(raw_out, "w") as f:
+        for fr in truth.get("frames", []):
+            visible = []
+            for a in fr.get("identified", []):
+                visible.append({
+                    "actor_idx":  0,          # >= 0 means "known"; the renderers
+                                              # test the sign, never the value
+                    "name":       a.get("name", ""),
+                    "imdb_id":    a.get("imdb_id", ""),
+                    "tmdb_id":    a.get("tmdb_id", ""),
+                    "jellyfin_id": a.get("jellyfin_id", ""),
+                    "similarity": a.get("similarity", 0.0),
+                    "track_id":   a.get("track_id", -1),
+                    "bbox":       a.get("bbox", [0, 0, 0, 0]),
+                })
+            for u in fr.get("unknowns", []):
+                visible.append({
+                    "actor_idx":  -1,
+                    "name":       "",
+                    "similarity": u.get("confidence", 0.0),
+                    "track_id":   u.get("track_id", -1),
+                    "bbox":       u.get("bbox", [0, 0, 0, 0]),
+                })
+            f.write(json.dumps({"timestamp_sec": fr.get("t", 0.0),
+                                "visible_actors": visible}) + "\n")
 
 
-# TRACES: AR-024 | SR-002
-# Keys the C++ Config actually still has. Seven names were removed here, all of
-# them accepted silently for months after the fields behind them were deleted:
-#
-#   match_threshold, match_ratio, match_ratio_ceil  — the raw-cosine accept
-#     fallback, retired with AR-024's enforcement.
-#   track_max_embed_dist, cut_revive_sim            — raw cosines, retired
-#     earlier by AR-024 when association moved into probability space.
-#   track_max_frames_missing, cut_inactive_max_frames — frame counts whose
-#     meaning changed with sample_fps, retired by AR-008/AR-013 in favour of
-#     track_extinction_sec.
-#
-# A sweep that varied one of these was measuring nothing, and reported a
-# perfectly ordinary-looking F1 for its trouble. kpn_bindings.cpp reads config
-# keys with a contains() check, so an unknown key is not an error — which makes
-# a stale entry here silently inert rather than loudly wrong.
 CFG_KEYS = ["detector_conf", "prob_threshold", "match_prior",
             "track_alpha", "track_min_iou", "track_assoc_min_prob",
             "track_extinction_sec",
@@ -284,10 +287,12 @@ CFG_KEYS = ["detector_conf", "prob_threshold", "match_prior",
             "ownership_logodds", "evidence_rho_max", "evidence_admit_below",
             "evidence_max_views"]
 
-# Swept like a Config key but consumed entirely in Python, by build_minimal.
-# Kept separate so nobody has to guess which of these the pipeline actually
-# reads: everything in CFG_KEYS crosses the seam, and nothing here does.
-REPLAY_LOCAL_KEYS = ["anneal_sec"]
+# TRACES: VR-011 | PR-002
+# REPLAY_LOCAL_KEYS is gone with build_minimal. It held anneal_sec, the last
+# parameter this harness applied itself -- and the only reason it needed a
+# separate list was that the harness was still doing windowing the pipeline had
+# stopped doing. Every key is a Config key now, because every decision is the
+# pipeline's.
 
 
 def main():
@@ -298,7 +303,7 @@ def main():
     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 + REPLAY_LOCAL_KEYS:
+    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.
@@ -310,10 +315,7 @@ def main():
     p.add_argument("--require-gallery-stamp", action="store_true")
     args = p.parse_args()
 
-    # Both lists go into one dict: config_from_dict reads C++ keys with a
-    # contains() check and ignores the rest, and build_minimal reads its own.
-    cfg = {k: getattr(args, k)
-           for k in CFG_KEYS + REPLAY_LOCAL_KEYS if getattr(args, k) is not None}
+    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
     if args.require_gallery_stamp:
@@ -322,9 +324,13 @@ def main():
     # 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))
+    result = replay(args.dump, args.gallery, cfg, args.build_dir,
+                    out_path=args.out, stop=True, raw_out=args.raw_out)
+    # NOT rewritten here: the sink already wrote args.out, and that file is the
+    # artifact. Dumping `result` back over it would make this script the last
+    # writer of a file it did not produce -- and any formatting difference would
+    # be a diff between the replayed truth file and a scene_analyze one that is
+    # this script's doing rather than the pipeline's.
     print(f"[replay] {len(result['actors'])} actors → {args.out}", file=sys.stderr)
 
 

scripts/optimizer/test_sae_kpn.py

@@ -1,17 +1,30 @@
 #!/usr/bin/env python3
 """
-Smoke test for the sae_kpn module: assemble the real downstream pipeline nodes
-(face_tracker → identity_matcher → frame_annotation) in a Python-driven KPN network,
-fed by a no-input Python source node, and verify SceneAnnotations flow out.
+Smoke test for the sae_kpn module: assemble the real downstream pipeline
+(tracker → matcher → annotation → sink) in a Python-driven KPN network, fed by a
+no-input Python source node, and verify the sink writes a truth file.
+
+TRACES: VR-011 | PR-002
 
 Proves the KPN-native replay path works without any numpy port of node logic.
+
+Rewritten for `add_pipeline`. It previously called three node factories and read
+SceneAnnotations back through the seam, asserting on what came out per frame.
+Neither half of that survives VR-011: the factories are gone because the chain
+has a construction order Python could not express, and presence is now the C++
+sink's answer, derived from TrackRegistry claims. Nothing is read per frame, so
+the assertions are on the file the sink writes.
+
 Run:  python scripts/optimizer/test_sae_kpn.py [gallery.json] [build_dir]
 """
+import json
 import sys
-import queue
-import numpy as np
+import tempfile
+import time
 from pathlib import Path
 
+import numpy as np
+
 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")
@@ -31,7 +44,6 @@ def make_frame(t, n):
 def main():
     net = sae_kpn.Network()
     sae_kpn._register_types(net)
-    cfg = {"prob_threshold": 0.99, "track_extinction_sec": 5.0}
 
     frames = [make_frame(float(t), 1) for t in range(3)]
     frames.append({"timestamp_sec": 3.0, "eof": True})
@@ -39,36 +51,67 @@ def main():
     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.
+        # Emit each frame once, then keep returning EOF so the node thread stays
+        # responsive to stop(). The sleep matters: a no-input source is called in
+        # a tight loop, and hot-spinning EOFs pegs a core and floods the channel.
         i = idx[0]
         idx[0] += 1
-        return frames[i] if i < len(frames) else eof_frame
+        if i < len(frames):
+            return frames[i]
+        time.sleep(0.05)
+        return 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_frame_annotation(net, "scene", 16)
-    net.connect("replay", 0, "tracker", 0)
-    net.connect("tracker", 0, "matcher", 0)
-    net.connect("matcher", 0, "scene", 0)
-    net.build()
-    net.start()
+    with tempfile.TemporaryDirectory() as tmp:
+        out_path = str(Path(tmp) / "truth.json")
+        cfg = {
+            "prob_threshold": 0.99,
+            "track_extinction_sec": 5.0,
+            "output_path": out_path,
+            "movie_path": "sae_kpn smoke test",
+            "sample_fps": 1.0,
+            # Standard verbosity emits the per-frame array this test asserts on.
+            # At 0 the file carries only the actor epochs, and three random
+            # embeddings against a real gallery need not produce any.
+            "verbosity": 1,
+        }
 
-    got = []
-    for _ in range(4):
-        sa = net.read("scene", 0)
-        got.append(sa)
-        if sa.get("eof"):
-            break
-    net.stop()
+        sae_kpn.add_node_python(net, "replay", source, [], ["EmbeddedSceneFrame"], 16)
+        # No embedder stamp: these embeddings are random, not the output of any
+        # model, so there is nothing truthful to claim. That warns rather than
+        # failing, and would be fatal under SAE_REQUIRE_GALLERY_STAMP — which is
+        # correct, since an unverifiable binding is exactly what it guards.
+        sae_kpn.add_pipeline(net, GAL, cfg, 16)
 
-    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")
+        net.connect("replay", 0, "tracker", 0)
+        net.connect("tracker", 0, "matcher", 0)
+        net.connect("matcher", 0, "annotation", 0)
+        net.connect("annotation", 0, "sink", 0)
+        net.build()
+        net.start()
+
+        # The sink writes on the EOF annotation. Wait for that rather than
+        # reading anything back: presence lives entirely on the C++ side.
+        deadline = time.time() + 30.0
+        while not sae_kpn.pipeline_done(net):
+            if time.time() > deadline:
+                sae_kpn.release_pipeline(net)
+                raise TimeoutError("sink never saw EOF within 30s")
+            time.sleep(0.02)
+
+        net.stop()
+        sae_kpn.release_pipeline(net)
+
+        with open(out_path) as f:
+            truth = json.load(f)
+
+    per_frame = truth.get("frames", [])
+    assert "actors" in truth, "truth file has no actors array"
+    assert len(per_frame) == 3, f"expected 3 frames, got {len(per_frame)}"
+    # EOF is a control token, not an observation: the sink flushes on it and does
+    # not record it, so three inputs give three frames and never four.
+    assert [f["t"] for f in per_frame] == [0.0, 1.0, 2.0], "timestamps wrong"
+    assert all("identified" in f for f in per_frame), "missing identified"
+    print(f"OK: {len(per_frame)} frames through the real KPN chain, sink wrote its truth file")
 
 
 if __name__ == "__main__":

src/kpn_bindings.cpp

@@ -1,11 +1,38 @@
-// sae_kpn — run the real downstream pipeline nodes (face_tracker, identity_matcher,
-// frame_annotation) inside a Python-assembled KPN network, fed by a Python HDF5 replay
-// source. Lets a parameter sweep re-run the exact C++ matching/tracking logic over
-// dumped embeddings — no video decode, no GPU — with different Config knobs each run.
+// sae_kpn — run the real downstream pipeline inside a Python-assembled KPN
+// network, fed by a Python HDF5 replay source. Lets a parameter sweep re-run the
+// exact C++ tracking/matching/presence logic over dumped embeddings — no video
+// decode, no GPU — with different Config knobs each run.
+//
+/// TRACES: VR-011, VR-002 | PR-002
+//
+// **The whole chain is C++, including the sink.** That is the VR-011 change and
+// it is the point of the requirement: replay must drive the real nodes, not a
+// reimplementation. Two things were wrong before.
+//
+//   1. It did not compile. `add_face_tracker` built `FaceTrackerFunc` from a
+//      Config alone, and the tracker has required a TrackRegistry and a
+//      calibration since AR-007/AR-008 moved association into probability
+//      space. Any .so in a stale build/ predates that.
+//
+//   2. Presence was rebuilt in Python. `replay.py::build_minimal` merged
+//      per-frame detections into windows by annealing gaps — which is what the
+//      pipeline did before AR-012. The sink now builds a window from a
+//      TrackRegistry claim: the extent of a track an actor owned, starting when
+//      they appeared rather than when recognition first succeeded. Those answer
+//      different questions, so every sweep was tuning against a contract the
+//      shipped code had stopped honouring.
+//
+// Both had the same root cause, which is why this is one binding and not three.
+// The chain has a construction ORDER — the matcher fits the calibration, the
+// registry needs a discounter built from it, the tracker needs both, and the
+// sink needs the registry's claims — and a factory-per-node API cannot express
+// it. `add_pipeline` mirrors main.cpp exactly and is the only way to build the
+// chain, so the ordering cannot be got wrong again from Python.
 //
 // Boundary types (cross the Python seam):
 //   EmbeddedSceneFrame  IN   (built by the Python replay source from HDF5 arrays)
-//   SceneAnnotation     OUT  (read by the Python sink → presence JSON)
+//   SceneAnnotation     OUT  (optional tee for per-frame debug rendering only —
+//                             the presence output is written by the C++ sink)
 // Intermediate types (TrackedSceneFrame, MatchedSceneFrame) flow C++→C++ only, but
 // still need channel factories + converters registered so PyNetwork can wire them.
 
@@ -20,6 +47,9 @@
 #include "nodes/face_tracker_node.hpp"
 #include "nodes/identity_matcher_node.hpp"
 #include "nodes/frame_annotation_node.hpp"
+#include "nodes/result_sink_node.hpp"
+#include "track_registry.hpp"
+#include "evidence_discount.hpp"
 
 #include <nanobind/nanobind.h>
 #include <nanobind/ndarray.h>
@@ -27,6 +57,8 @@
 #include <nanobind/stl/vector.h>
 #include <nanobind/stl/map.h>
 
+#include <atomic>
+#include <map>
 #include <memory>
 #include <optional>
 #include <variant>
@@ -34,10 +66,52 @@
 namespace nb = nanobind;
 using namespace nb::literals;
 
+// ── ReplaySession ─────────────────────────────────────────────────────────────
+/// TRACES: VR-011 | PR-002
+/// State the network's nodes reference but do not own.
+///
+/// ResultSinkFunc holds `std::atomic<bool>&`, exactly as it does under main(),
+/// where it is a stack local in a function that outlives the pipeline. There is
+/// no such frame here -- the network is built and torn down from Python -- so
+/// the flag lives in a session held for the network's lifetime and released
+/// explicitly. The registry is here for the same reason: the sink's claim
+/// callback captures it.
+struct ReplaySession {
+    /// Owns the Config, and must. ResultSinkFunc holds `const Config&` -- under
+    /// main() that is a stack local in a frame which outlives the pipeline, so
+    /// the reference is fine there. There is no such frame here: the network is
+    /// built inside a binding call and torn down from Python, so a Config local
+    /// to add_pipeline dies the moment it returns and the sink is left reading
+    /// freed memory. It presented as an empty output_path -- the sink announced
+    /// `[result_sink] writing ` and wrote nothing.
+    Config                         cfg;
+    std::atomic<bool>              done{false};
+    std::shared_ptr<TrackRegistry> registry;
+};
+
+// Function-local static so ordering against other translation units cannot bite.
+inline std::map<void*, std::shared_ptr<ReplaySession>>& sessions() {
+    static std::map<void*, std::shared_ptr<ReplaySession>> s;
+    return s;
+}
+
 // The variant spanning every type that flows on a channel in the replay chain.
 using SaeVariant = std::variant<EmbeddedSceneFrame, TrackedSceneFrame,
                                 MatchedSceneFrame, SceneAnnotation>;
 
+// ── Node wrapper aliases ──────────────────────────────────────────────────────
+// Named once so add_pipeline and the runtime setters cannot disagree about a
+// node's port names: a mismatch there is a dynamic_cast that returns null, i.e.
+// a runtime setter that silently does nothing.
+using MatcherWrap = kpn::ObjectVariantNodeWrapper<
+    IdentityMatcherFunc, SaeVariant, kpn::in<"tracked">,  kpn::out<"matched">>;
+using TrackerWrap = kpn::ObjectVariantNodeWrapper<
+    FaceTrackerFunc,     SaeVariant, kpn::in<"embedded">, kpn::out<"tracked">>;
+using AnnotWrap   = kpn::ObjectVariantNodeWrapper<
+    FrameAnnotationFunc, SaeVariant, kpn::in<"matched">,  kpn::out<"annotation">>;
+using SinkWrap    = kpn::ObjectVariantNodeWrapper<
+    ResultSinkFunc,      SaeVariant, kpn::in<"annotation">, kpn::out<>>;
+
 // ── Converters ─────────────────────────────────────────────────────────────────
 // Only EmbeddedSceneFrame (in) and SceneAnnotation (out) actually cross the seam;
 // the two intermediates get identity-ish stubs (never converted in practice) so the
@@ -183,6 +257,28 @@ static Config config_from_dict(nb::dict d) {
     /// TRACES: GR-004 | SR-001
     if (d.contains("require_gallery_stamp"))
         cfg.require_gallery_stamp = nb::cast<bool>(d["require_gallery_stamp"]);
+
+    /// TRACES: VR-011, IR-001 | PR-002, SR-003
+    // The sink is a real node in this network now, so it needs the two things
+    // that decide what it writes and where. Both used to be irrelevant here
+    // because the replay never had a sink -- Python rebuilt presence instead,
+    // which is the reimplementation VR-002 forbids and VR-011 removes.
+    if (d.contains("output_path"))
+        cfg.output_path = nb::cast<std::string>(d["output_path"]);
+    if (d.contains("verbosity")) {
+        const int v = nb::cast<int>(d["verbosity"]);
+        cfg.verbosity = v == 2 ? Verbosity::xray
+                      : v == 1 ? Verbosity::standard
+                               : Verbosity::minimal;
+    }
+    // Reported verbatim in the truth file's extraction block, so a replayed
+    // manifest says which gallery scope produced it (IR-002).
+    if (d.contains("gallery_scope"))
+        cfg.gallery_scope = nb::cast<std::string>(d["gallery_scope"]);
+    if (d.contains("sample_fps"))
+        cfg.sample_fps = nb::cast<float>(d["sample_fps"]);
+    if (d.contains("movie_path"))
+        cfg.movie_path = nb::cast<std::string>(d["movie_path"]);
     return cfg;
 }
 
@@ -222,38 +318,51 @@ NB_MODULE(sae_kpn, m) {
                             std::move(outs), cap);
     }, "net"_a, "name"_a, "callable"_a, "inputs"_a, "outputs"_a, "capacity"_a = 5);
 
-    // ── Real node factories ─────────────────────────────────────────────────────
-    m.def("add_face_tracker", [](Net& net, std::string name, nb::dict cfg_dict, std::size_t cap) {
+    // ── The pipeline ────────────────────────────────────────────────────────────
+    /// TRACES: VR-011, VR-002 | DP-001 | PR-002, PR-004
+    ///
+    /// One call builds the whole downstream chain, in the one order that works:
+    ///
+    ///     matcher (fits the calibration)
+    ///       -> registry (needs a discounter built from it)
+    ///         -> tracker (needs both)
+    ///           -> frame_annotation
+    ///             -> result_sink (needs the registry's claims)
+    ///
+    /// This replaces add_face_tracker / add_identity_matcher / add_frame_annotation.
+    /// They were separate because the network is assembled node by node from
+    /// Python -- and that is exactly how the seam broke: the tracker's dependency
+    /// on a calibration that only exists once the matcher is built cannot be
+    /// expressed as three independent factories, so the tracker factory kept
+    /// constructing FaceTrackerFunc{cfg} against a signature that no longer
+    /// existed. A binding that cannot represent the order will eventually be
+    /// called in the wrong one.
+    ///
+    /// DP-001 -- "modes are front-ends and must not fork pipeline logic" -- is
+    /// the requirement this serves. The replay harness is a front-end. Its job is
+    /// to supply frames and read the result, not to re-derive presence.
+    m.def("add_pipeline", [](Net& net, std::string gallery_path, nb::dict cfg_dict,
+                             std::size_t cap, std::string embedder_model,
+                             std::string embedder_sha256) {
         Config cfg = config_from_dict(cfg_dict);
-        auto node = std::make_shared<kpn::ObjectVariantNodeWrapper<
-            FaceTrackerFunc, SaeVariant, kpn::in<"embedded">, kpn::out<"tracked">>>(cap, cfg);
-        net.add(std::move(name), std::move(node));
-    }, "net"_a, "name"_a, "config"_a, "capacity"_a = 16);
+        cfg.gallery_path = gallery_path;  // so a refreshed calibration persists back
 
-    /// TRACES: GR-004 | SR-001
-    // embedder_model / embedder_sha256 identify whatever produced the embeddings
-    // that will be fed in. In a replay those come from the dump's own stamp (see
-    // scripts/optimizer/SCHEMA.md), because there is no live embedder in the
-    // network — the dump *is* the embedder as far as this gallery is concerned.
-    // Passing neither leaves the binding unverifiable, which warns loudly and is
-    // fatal under SAE_REQUIRE_GALLERY_STAMP.
-    m.def("add_identity_matcher", [](Net& net, std::string name, std::string gallery_path,
-                                     nb::dict cfg_dict, std::size_t cap,
-                                     std::string embedder_model,
-                                     std::string embedder_sha256) {
-        Config cfg = config_from_dict(cfg_dict);
-        cfg.gallery_path = gallery_path;  // needed to persist refreshed calibration back
         // Cache loaded galleries by path so a threshold sweep (many networks, same
-        // gallery) pays the ~24s JSON parse only once. The matcher holds a const
-        // ref; the cache keeps the gallery alive for the process lifetime.
+        // gallery) pays the parse once. The matcher holds a const ref; the cache
+        // keeps the gallery alive for the process lifetime.
         static std::map<std::string, std::shared_ptr<ActorGallery>> cache;
         auto it = cache.find(gallery_path);
         if (it == cache.end())
             it = cache.emplace(gallery_path,
                      std::make_shared<ActorGallery>(load_gallery(gallery_path))).first;
 
-        // Checked on every construction, not only on the cache miss: the same
-        // process may replay several dumps against one cached gallery.
+        /// TRACES: GR-004 | SR-001
+        // embedder_model / embedder_sha256 identify whatever produced the
+        // embeddings that will be fed in. In a replay those come from the dump's
+        // own stamp: there is no live embedder here, so the dump *is* the
+        // embedder as far as this gallery is concerned. Checked on every
+        // construction, not only on a cache miss -- one process may replay
+        // several dumps against one cached gallery.
         EmbedderStamp feeding;
         feeding.model_name   = std::move(embedder_model);
         feeding.model_sha256 = std::move(embedder_sha256);
@@ -263,31 +372,74 @@ NB_MODULE(sae_kpn, m) {
                                    : feeding.model_name,
                                cfg.require_gallery_stamp);
 
-        auto node = std::make_shared<kpn::ObjectVariantNodeWrapper<
-            IdentityMatcherFunc, SaeVariant, kpn::in<"tracked">, kpn::out<"matched">>>(
-            cap, *it->second, cfg);
-        net.add(std::move(name), std::move(node));
-    }, "net"_a, "name"_a, "gallery"_a, "config"_a, "capacity"_a = 16,
+        // 1. Matcher first: its constructor fits (or loads) the calibration.
+        auto matcher = std::make_shared<MatcherWrap>(cap, *it->second, cfg);
+
+        // 2. The calibration every other stage must decide in (AR-024).
+        auto same_person = same_person_probability(matcher->functor().calibration());
+
+        // 3. Registry + discounter, from Config (AR-025).
+        TrackRegistry::Config reg_cfg;
+        reg_cfg.track_extinction_sec = cfg.track_extinction_sec;
+        reg_cfg.ownership_logodds    = cfg.ownership_logodds;
+        EvidenceDiscounter::Config disc_cfg;
+        disc_cfg.max_views   = cfg.evidence_max_views;
+        disc_cfg.admit_below = cfg.evidence_admit_below;
+        disc_cfg.rho_max     = cfg.evidence_rho_max;
+        auto registry = std::make_shared<TrackRegistry>(
+            reg_cfg, EvidenceDiscounter(same_person, disc_cfg));
+        matcher->functor().set_registry(registry);
+
+        // 4. Tracker, which needs both.
+        auto tracker = std::make_shared<TrackerWrap>(cap, cfg, registry, same_person);
+
+        // 5. Projection, stateless.
+        auto annot = std::make_shared<AnnotWrap>(cap);
+
+        // 6. The real sink. `done` outlives the network via the session below;
+        //    ResultSinkFunc holds it by reference, as it does in main.cpp.
+        auto session = std::make_shared<ReplaySession>();
+        session->cfg      = cfg;          // the sink holds this by reference
+        session->registry = registry;
+        auto sink = std::make_shared<SinkWrap>(cap, session->cfg, session->done);
+
+        /// TRACES: AR-012, AR-016 | IR-003 | SR-002
+        // The claim path, identical to main.cpp's. Without the flush hook every
+        // track still live at EOF is silently dropped -- which in a replay is
+        // most of the closing scene, and reads as a recognition miss rather than
+        // as a missing wire.
+        ResultSinkFunc& sink_fn = sink->functor();
+        registry->on_track_dead([&sink_fn](const DeadTrack& d) { sink_fn.add_claim(d); });
+        sink_fn.set_pre_write_hook([registry](double last_ts) { registry->flush(last_ts); });
+
+        net.add("tracker", tracker);
+        net.add("matcher", matcher);
+        net.add("annotation", annot);
+        net.add("sink", sink);
+
+        // Keyed by network so release_pipeline can free it. Not a leak-by-design:
+        // a sweep builds one network per replay, and the sink accumulates every
+        // annotation, so holding these forever would grow with films x configs.
+        sessions()[&net] = session;
+    }, "net"_a, "gallery"_a, "config"_a, "capacity"_a = 16,
        "embedder_model"_a = "", "embedder_sha256"_a = "");
 
-    /// TRACES: AR-012, AR-013 | SR-002
-    // Was add_scene_tracker, backed by the extinction-timer state machine. The
-    // node is gone (see frame_annotation_node.hpp) and so is the timer; this
-    // projects a matched frame into the same SceneAnnotation the Python sink
-    // already reads, so the seam's output type is unchanged. It takes no config
-    // because it has no state to configure -- which is the point.
-    m.def("add_frame_annotation", [](Net& net, std::string name, std::size_t cap) {
-        auto node = std::make_shared<kpn::ObjectVariantNodeWrapper<
-            FrameAnnotationFunc, SaeVariant, kpn::in<"matched">, kpn::out<"annotation">>>(cap);
-        net.add(std::move(name), std::move(node));
-    }, "net"_a, "name"_a, "capacity"_a = 16);
+    /// Drop the session for a network. Idempotent. Call after net.stop(); not
+    /// calling it holds one registry and one sink's accumulated frames per
+    /// replay, which a long sweep will notice.
+    m.def("release_pipeline", [](Net& net) { sessions().erase(&net); }, "net"_a);
+
+    /// True once the sink has written its output. The sink flushes on the EOF
+    /// annotation, so a caller that reads the file before this is racing it.
+    m.def("pipeline_done", [](Net& net) {
+        auto it = sessions().find(&net);
+        return it != sessions().end()
+            && it->second->done.load(std::memory_order_acquire);
+    }, "net"_a);
 
     // ── Runtime setters (persistent-pipeline reuse across a threshold sweep) ─────
     // Build the network once, then change thresholds between replays — no rebuild,
     // no teardown (which is where the ROCm deadlock lives), no gallery reload.
-    using MatcherWrap = kpn::ObjectVariantNodeWrapper<
-        IdentityMatcherFunc, SaeVariant, kpn::in<"tracked">, kpn::out<"matched">>;
-
     m.def("set_prob_threshold", [](Net& net, std::string name, float t) {
         auto* w = dynamic_cast<MatcherWrap*>(net.node_ptr(name));
         if (!w) throw std::runtime_error("set_prob_threshold: '" + name + "' is not an identity_matcher");