chore: update KPN — unconditional push wake, and the start() lock it exposed

Two commits on top of c9aa246.

6802328 fixes a lost wake in the channel itself: push fired push_callback_
only on the empty->non-empty edge, computed from a head_ sampled before the
item was published, so a pop landing in that window left the item in the ring
with the consumer idle and no wake outstanding. Absorbing, because every later
push then sees a non-empty ring and the edge never fires again. Firing
unconditionally is correct by construction — the callback runs after the
publishing store — and the redundant wakes are cheap, since on_input_ready
re-checks the level and the submit gate collapses a wake arriving mid-firing.

771b9f8 is the gap that change exposed. abbb2d4 guarded ThreadPool::submit
against stop(), which clears queues_, but missed that start() rebuilds the
same vector under no lock. A submission genuinely lands there — a network
starts nodes one at a time, and an already-started node fires into the next
one's channel, whose callback submits — and push_back can reallocate under a
reader that has already indexed it. Unobserved before; 4 races in one run
after callbacks became frequent enough during startup.

Verified: KPN 148/148, this repo 149/149, ThreadSanitizer clean across five
unit runs and two stress runs.

Not re-measured on a clip. The GPU has a driver/library version mismatch
(kernel module 610.43.03 against NVML 610.57 — an update without a reboot),
so scene_analyze aborts in cuInit before it decodes anything. The SuperHero-1
measurement recorded in c599e07 — exit 0, outran() 0, scenes.json and presence
byte-identical — stands against c9aa246; both commits added since are
concurrency fixes with no effect on pipeline semantics, but a re-run is worth
doing once the driver is sorted.

TRACES: AR-004 | SR-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) ────────────────

docs/SPEC.md

@@ -128,10 +128,43 @@ Two consequences worth stating:
   depends on timing. The same command run twice can produce different dumps, and
   a golden fixture cannot be built on that.
 
-**Current:** fixed in KPN — node data outputs use `push_blocking`, sentinels
-remain out-of-band so EOF can always overtake a stalled data path. Verified on
-the same clip: 385 of 385 sampled frames written, zero drops, and two
-consecutive runs byte-identical where previously they were not.
+**Current:** fixed in KPN. Node data outputs *park* on a full channel — the
+value is held in a one-slot buffer, the worker is released, and the channel's
+space callback resubmits the node once the consumer drains. That replaced
+`push_blocking`, which slept inside the push and, with one thread per node,
+stopped that node draining its own input. Sentinels remain out-of-band so EOF
+can always overtake a stalled data path. Verified on the same clip: 385 of 385
+sampled frames written, zero drops, and two consecutive runs byte-identical
+where previously they were not.
+
+A later audit found the losslessness was still incomplete in three places, all
+now closed and each pinned by a regression case in the KPN suite:
+
+- **`FilterNode` and `RouterNode`** were the last data paths still using the
+  throwing `push()` with the exception swallowed. A full output discarded the
+  value, and that included the **EOF sentinel**. The decimator passes EOF by
+  predicate but its output is reliably full — the embedder is the slowest node
+  in the chain — so the token was discarded, nothing downstream shut down, and
+  the run had to be killed. This was the wedge.
+- **The sentinel could arrive ahead of a value still queued behind it.** `pop()`
+  observed the ring empty and then took the sentinel; a producer can push a
+  value *and* publish the sentinel inside that window, so a consumer treating
+  EOF as a hard stop loses the tail.
+- **Two firings of one node could overlap**, because the submit gate was
+  released before the firing had finished with the node's state. That breaks the
+  one-slot park itself: a parked value can be overwritten by the other firing,
+  with no drop recorded anywhere.
+
+**New constraint:** a channel carries at most one undelivered sentinel. A second
+offered before the first is taken is refused and reported, never queued and
+never overwritten — two control tokens on one channel means the stream ended
+twice. Single-shot EOF is what everything does today; this becomes live the
+moment a pipeline is reused for a second input.
+
+**Consequence:** a lossless decimator is a backpressure point, not a relief
+valve. The source now throttles to the face branch rather than quietly thinning
+it. That is what this requirement asks for, but it changes the shape of a loaded
+run and has not yet been benchmarked.
 
 It also ran *faster* (29 s → 17 s). A dropped frame has already cost its decode,
 and the overflow exception cost more — so the lossy path was paying for work it

docs/requirements.md

@@ -31,7 +31,7 @@ Status: `Done` · `In Progress` · `Planned` · `TBD` · `Withdrawn`
 | AR-001 | Detect faces in sampled frames; emit bbox, confidence, 5-point landmarks in original pixel space | SR-002 | High | Done |
 | AR-002 | Minimum face size **40×40 px** (VR-013 measured end to end; VR-005's 32 px is an embedder-only upper bound), expressed in **original** resolution (decoupled from `dense_scale`) | SR-002 | High | **Done** — `FaceDetectorFunc::drop_undersized()`. The threshold is divided by `bbox_upscale` rather than every box multiplied, which keeps the comparison on the detector's own numbers and means turning `dense_scale` on cannot silently raise the minimum face the pipeline accepts. Verified at the threshold and at `dense_scale` 0.5 (UT-002), and end to end on the fixture (IT-001) — the superhero dump's smallest side is *exactly* its recorded 32 px, so the filter is binding there rather than vacuously satisfied |
 | AR-003 | No fixed per-frame face cap — crowd scenes must not lose background cast | SR-002 | Medium | **Done** — `max_faces` defaults to 0 (no cap); the matcher batches through its GEMM buffer instead of throwing |
-| AR-004 | Backpressure: unbounded faces/frame absorbed by slowing, never by dropping or throwing | SR-002 | High | **Mostly** — node outputs *park* on a full channel: the value is held, the worker released, and a channel space-callback resumes the node. Replaces `push_blocking`, which parked a scheduler worker inside the push and, with one thread per node, stopped that node draining its own input. Verified: 385/385 frames, 0 drops. Two remaining holes now closed: **(a)** `FanoutNode` dropped on overflow rather than waiting, so the AR-010 scene join shed frames exactly when the dense branch fell behind — measured at **9 of 2192 items delivered** to the slower of two branches, now lossless with the fast branch throttled to within its buffering; **(b)** the residual hang, recorded as ~1 run in 20 at a 300 s timeout, was a **startup** lost wake, not a mid-stream one — `start()` enables a node's inputs several statements before it installs the push callback, and a producer firing into that gap is accepted by the ring while waking nobody, since `Channel::push` signals only the empty→non-empty edge. Signature is zero items delivered, never a partial stall. Reproduced 7 times in 24 under CPU contention and 0 in 10 without; `start()` now closes with the level-triggered `on_input_ready()`, giving 0 in 24 on the same harness. **Consequence to hold onto:** a lossless fanout makes join depth a correctness precondition — one branch can now run ahead of another only by the slower branch's buffering, so `kSceneJoinDepth` must exceed the TransNetV2 window. **Gap:** capacity is still counted in *items*, not bytes, so a crowd frame carrying 60 crops occupies one slot exactly as an empty one does — the memory ceiling the plan asks for is unenforced |
+| AR-004 | Backpressure: unbounded faces/frame absorbed by slowing, never by dropping or throwing | SR-002 | High | **Mostly** — node outputs *park* on a full channel: the value is held, the worker released, and a channel space-callback resumes the node. Replaces `push_blocking`, which parked a scheduler worker inside the push and, with one thread per node, stopped that node draining its own input. Verified: 385/385 frames, 0 drops. Holes closed since, in the order they surfaced: **(a)** `FanoutNode` dropped on overflow rather than waiting, so the AR-010 scene join shed frames exactly when the dense branch fell behind — measured at **9 of 2192 items delivered** to the slower of two branches, now lossless with the fast branch throttled to within its buffering; **(b)** the residual hang, recorded as ~1 run in 20 at a 300 s timeout, was a **startup** lost wake, not a mid-stream one — `start()` enables a node's inputs several statements before it installs the push callback, and a producer firing into that gap is accepted by the ring while waking nobody, since `Channel::push` signals only the empty→non-empty edge. Signature is zero items delivered, never a partial stall. Reproduced 7 times in 24 under CPU contention and 0 in 10 without; `start()` now closes with the level-triggered `on_input_ready()`, giving 0 in 24 on the same harness — though the *cause* was narrower than recorded there and is fixed properly in **(e)**; **(c)** `FilterNode` and `RouterNode` were the last data paths still using the throwing `push()` with the exception swallowed, so a full output discarded the value — including the **EOF sentinel**. The decimator passes EOF by predicate (`if (f.eof) return true;`) but its output is reliably full, the embedder being the slowest node in the chain, so the token was discarded, nothing downstream ever shut down, and the run had to be killed. **This is the wedge.** Both now route sentinels out-of-band and retry data until taken; the regression case delivers 6 of 40 values and never sets `saw_eof` before, 40 and terminating after; **(d)** the sentinel could be delivered *ahead of* a value still queued behind it — `pop()` observed the ring empty and then took the sentinel, and a producer can push a value *and* publish the sentinel inside that window, so any consumer treating EOF as a hard stop loses the tail. `take_sentinel` now re-checks emptiness *after* observing `has_eof_`, which is sound because the sentinel is published with a release store after the ring pushes. ~1 run in 15 before, 0 in 25 after; **(e)** two `fire_once` invocations for one node could overlap, because the submit gate was released before the firing had finished touching node state. That breaks the one-slot park the whole scheme rests on — a parked value can be overwritten by the other firing, with no drop recorded anywhere. ThreadSanitizer caught it as a race on `pending_done_`; the release is now the last act of a firing. The same sweep found the callbacks themselves being written while a running neighbour read them (ten TSan races), which is the *actual* cause of the startup lost wake in **(b)** — callbacks are now installed in a `prepare()` pass before any node starts. **New constraint:** a channel carries at most **one undelivered sentinel**; a second offered before the first is taken is refused and reported, never queued and never overwritten, since two control tokens on one channel means the stream ended twice. Single-shot EOF today, live the moment a pipeline is reused for a second input. **Consequence to hold onto:** a lossless fanout makes join depth a correctness precondition — one branch can now run ahead of another only by the slower branch's buffering, so `kSceneJoinDepth` must exceed the TransNetV2 window. Making the decimator lossless also makes it a backpressure point rather than a relief valve: the source now throttles to the face branch instead of quietly thinning it. Correct under this requirement, but it changes the shape of a loaded run and is **not yet benchmarked**. **Gap:** capacity is still counted in *items*, not bytes, so a crowd frame carrying 60 crops occupies one slot exactly as an empty one does — the memory ceiling the plan asks for is unenforced |
 | AR-005 | Align to 112×112 via ArcFace 5-point similarity transform, fitted by **Umeyama least squares over all five points** (as InsightFace does) — never a robust fit, which would discard the landmarks AR-030 reads | SR-002 | High | **Done** — `umeyama_similarity()`. The RANSAC fit it replaces disagreed by a median 17 source px on 400 headshots, 83.5% of crops embedding below cos 0.99, and was unstable and RNG-driven: rebuilding caught 1614 near-duplicates against the original build's ~100. **All galleries rebuilt** (2456 actors, 10254 embeddings); measured separation gain is small (0.583 → 0.590), so recorded accuracy figures should be re-run but are not expected to move far |
 | AR-006 | 512-d L2-normalised embeddings, batched | SR-002 | High | Done |
 | AR-007 | Associate detections by IoU + embedding, with **frame-dependent** weighting | SR-002 | High | **Done** — `track_alpha` is the base for ordinary frames; drops to embedding-only on cut/boundary and for dormant tracks |
@@ -314,7 +314,7 @@ because it will be trusted.
 | AR-001 | T3 | Detector returns plausible boxes on a known frame | — smoke only |
 | AR-002 | T2 | Faces below 40 px (original res) are dropped | Exactly at threshold; with `dense_scale` 0.5 — the interaction that motivated the requirement |
 | AR-003 | T2 | No cap applied; a 40-face frame yields 40 | Crowd frame |
-| AR-004 | T1 | Saturated input blocks rather than drops or throws | Bounded queue at capacity; **byte-based** limit with large crops; SIGTERM mid-block. Two cases the KPN suite now pins, both of which failed before being written: a fanout feeding an unequal pair loses nothing *and* throttles the fast branch (either assertion alone passes on a broken implementation); and a node started with data already in its input still fires — the startup lost wake, which needs no contention to reproduce once the state is constructed directly |
+| AR-004 | T1 | Saturated input blocks rather than drops or throws | Bounded queue at capacity; **byte-based** limit with large crops; SIGTERM mid-block. Cases the KPN suite now pins, each of which failed before being written: a fanout feeding an unequal pair loses nothing *and* throttles the fast branch (either assertion alone passes on a broken implementation); a filter delivers EOF into a saturated output; a sentinel is never delivered ahead of a queued value; a twice-parked value keeps its payload; and a node started with data already in its input still fires — the startup lost wake, which needs no contention to reproduce once the state is constructed directly |
 | AR-005 | T1 | Known landmarks → expected 112×112 warp; the fit never mirrors | Landmarks near frame edge; degenerate/collinear points; a mirrored set — SVD returns a reflection unless the determinant guard rejects it |
 | AR-006 | T3 | Embeddings are unit-norm | Batch smaller than, equal to, larger than `embed_batch_size` |
 | AR-007 | T2 | Association picks the right track | Two faces crossing paths; one leaving frame as another enters |

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");

src/main.cpp

@@ -71,6 +71,7 @@
 #include "nodes/face_tracker_node.hpp"
 #include "nodes/identity_matcher_node.hpp"
 #include "nodes/frame_annotation_node.hpp"
+#include "inference/scene_detector.hpp"   // ISceneDetector::kWindow, for the join-depth derivation
 #include "nodes/scene_detector_node.hpp"
 #include "scene_boundaries.hpp"
 #include "nodes/scene_boundary_annotator_node.hpp"
@@ -84,8 +85,10 @@
 
 #include <opencv2/core/utility.hpp>   // cv::setNumThreads (SAE_CV_THREADS)
 
+#include <algorithm>
 #include <atomic>
 #include <chrono>
+#include <cmath>
 #include <csignal>
 #include <cstdlib>
 #include <cstring>
@@ -101,12 +104,66 @@
 // ── CLI parsing ───────────────────────────────────────────────────────────────
 
 /// TRACES: AR-010, AR-004 | SR-002
-/// How deeply the sampled branch is buffered behind the dense one. TransNetV2
-/// needs kWindow (100) dense frames before it can score any of them, so the face
-/// branch must lag by at least that much or it asks about frames nobody has
-/// looked at yet. Backpressure turns depth into lag: the fanout blocks on the
-/// slower branch rather than dropping, so the detector simply runs ahead.
-static constexpr std::size_t kSceneJoinDepth = 256;
+/// Depth of the dense branch's own input queue. Part of how far behind the
+/// fanout head TransNetV2 can be, and therefore an input to the join depth.
+static constexpr std::size_t kSceneInputDepth = 128;
+
+/// TRACES: AR-010, AR-004 | SR-002
+/// How far the sampled branch must trail the dense one, in seconds of film.
+///
+/// TransNetV2 needs kWindow (100) dense frames before it can score any of
+/// them, and its input queue can hold kSceneInputDepth more, so in the worst
+/// case it has scored only up to (kSceneInputDepth + kWindow) frames behind
+/// whatever the fanout has just delivered. The face branch must be at least
+/// that far behind, or `scene_annotate` asks about frames nobody has looked at
+/// yet. Backpressure turns depth into lag: the fanout blocks on the slower
+/// branch rather than dropping, so the dense branch simply runs ahead.
+///
+/// Divided by a *lower bound* on native frame rate, because a slower source
+/// makes the same frame count span more film — 24 fps is the floor for the
+/// material this runs on, so it is the conservative choice.
+static constexpr double kMinNativeFps    = 24.0;
+static constexpr double kSceneJoinLagSec =
+    (kSceneInputDepth + ISceneDetector::kWindow) / kMinNativeFps;   // ~9.5 s
+
+/// Margin over that minimum, for jitter in TransNetV2's inference time.
+static constexpr double kSceneJoinSafety = 2.0;
+
+/// TRACES: AR-004 | SR-002
+/// Slots the sampled branch needs to hold `kSceneJoinLagSec` of film.
+///
+/// This used to be a constant 256, which is the whole bug: the requirement is a
+/// span of *film*, and the slots needed to hold it depend on `sample_fps`.
+/// Pinned at 256 it was ~256 s of lag at 1 fps — 27x what the join needs — and
+/// nothing recomputed it if `sample_fps` changed, so the one number the join's
+/// correctness rests on drifted silently with an unrelated knob.
+///
+/// It is also the largest single memory item in the pipeline. Every message
+/// embeds `Frame source`, so a slot on this branch holds a full decoded image:
+/// 256 of them is ~1.5 GB at 1080p, against ~110 MB for the derived depth at
+/// 1 fps. See AR-004 — capacity is counted in items, and only the byte figure
+/// (now correct, see types.hpp) shows what a slot really costs.
+static std::size_t scene_join_depth(float sample_fps) {
+    const double slots = kSceneJoinSafety * kSceneJoinLagSec * sample_fps;
+    // Floor of 16: below that the queue stops absorbing ordinary jitter and
+    // starts throttling the fanout, which would slow the dense branch it
+    // exists to let run ahead.
+    return std::max<std::size_t>(16, static_cast<std::size_t>(std::ceil(slots)));
+}
+
+/// TRACES: AR-004 | SR-002
+/// The decimator's input, on the *full-rate* stream.
+///
+/// This was also kSceneJoinDepth, which put a 256-slot buffer of full-rate
+/// frames in front of the decimator — and at 1 fps against 24 fps native, 23 of
+/// every 24 of those frames exist only to be discarded a moment later. Holding
+/// ~1.5 GB of decoded images for frames the very next node throws away is the
+/// worst available use of the memory budget.
+///
+/// A filter is a pass-through, not a reservoir: the lag belongs *after*
+/// decimation, where a slot buys `1/sample_fps` seconds of film instead of
+/// `1/native_fps`. Sized only to keep the decimator fed.
+static constexpr std::size_t kDecimatorInputDepth = 16;
 
 /// Set when the scene branch is built, so shutdown can report whether the join
 /// actually worked.
@@ -529,7 +586,7 @@ int main(int argc, char** argv) {
         auto boundaries = std::make_shared<SceneBoundaries>();
         scene_fn.set_boundaries(boundaries);
         kpn::ObjectNode<SceneDetectorFunc, kpn::in<"dense">, kpn::out<>, "scene_detector", 0>
-            scene_node(scene_fn, 128);
+            scene_node(scene_fn, kSceneInputDepth);
 
         // Decimator: keep frames on the sample_fps cadence, drop the rest.
         // eof always passes so downstream shuts down cleanly. Stateful — one
@@ -544,7 +601,7 @@ int main(int argc, char** argv) {
                     return true;
                 }
                 return false;
-            }, kSceneJoinDepth);
+            }, kDecimatorInputDepth);
 
         /// TRACES: AR-010 | SR-002
         // Stamp is_scene_boundary from the detector's published verdict. tol is
@@ -559,7 +616,7 @@ int main(int argc, char** argv) {
         // otherwise look exactly like "no boundary here".
         SceneBoundaryAnnotatorFunc annotate_fn{boundaries, 0.5 / cfg.sample_fps};
         kpn::ObjectNode<SceneBoundaryAnnotatorFunc, kpn::in<"frame">, kpn::out<"frame">,
-                        "scene_annotate", 0> annotate(annotate_fn, kSceneJoinDepth);
+                        "scene_annotate", 0> annotate(annotate_fn, scene_join_depth(cfg.sample_fps));
 
         // Reported at shutdown: without this the join is unverifiable, and an
         // annotator that never fired looks identical to footage with no

src/nodes/scene_detector_node.hpp

@@ -6,6 +6,8 @@
 #include <memory>
 #include "inference/scene_detector.hpp"
 
+#include <opencv2/imgproc.hpp>   // cv::resize, for to_model_input
+
 #include <nlohmann/json.hpp>
 #include <algorithm>
 #include <atomic>
@@ -76,7 +78,7 @@ struct SceneDetectorFunc {
         }
         prev_ts_ = f.timestamp_sec;
 
-        images_.push_back(f.image);
+        images_.push_back(to_model_input(f.image));
         times_.push_back(f.timestamp_sec);
 
         // Once we have a full window, score it and slide forward by `stride`.
@@ -90,6 +92,48 @@ struct SceneDetectorFunc {
         }
     }
 
+    /// TRACES: AR-004, AR-010 | SR-002
+    /// Reduce a decoded frame to exactly what TransNetV2 consumes, once.
+    ///
+    /// The window used to hold the frames as decoded — full resolution — and
+    /// leave the downscale to the backend. But the model's input is 48x27
+    /// (`ISceneDetector::kFrameW/H`; the config note for `dense_scale` says so
+    /// outright: "TransNetV2 downsamples to 48x27 regardless"), so the buffer
+    /// held ~590 MB at 1080p to feed something that needs ~380 KB. That is not
+    /// a channel capacity, so no amount of tuning channel depths would ever
+    /// have found it.
+    ///
+    /// It is also redundant work. Windows overlap by `kWindow - stride`, so a
+    /// frame appears in several of them and was re-downscaled once per window;
+    /// now it is downscaled once, when it arrives.
+    ///
+    /// **This must reproduce the backends' preprocessing exactly**, because the
+    /// project invariant is that every model gets the input it was trained for
+    /// — a model run off-distribution returns confident, plausible, wrong
+    /// output, and here that means fabricated shot boundaries. Both
+    /// ort_backend.cpp and trt_backend.cpp guard mis-sized input with, in this
+    /// order, `convertTo(CV_8UC3)` then
+    /// `cv::resize(..., {kFrameW, kFrameH}, 0, 0, cv::INTER_AREA)`. The same
+    /// two operations are done here, so the tensor the model receives is
+    /// unchanged; the backend guard then sees a correctly-sized frame and does
+    /// nothing. The interface has always specified this shape as the caller's
+    /// job ("Each frame must already be kFrameW x kFrameH, BGR, CV_8UC3"), so
+    /// this makes the node meet a contract it was already given.
+    static cv::Mat to_model_input(const cv::Mat& src) {
+        cv::Mat typed;
+        if (src.type() != CV_8UC3) src.convertTo(typed, CV_8UC3);
+        else                       typed = src;
+
+        if (typed.cols == ISceneDetector::kFrameW &&
+            typed.rows == ISceneDetector::kFrameH)
+            return typed;
+
+        cv::Mat small;
+        cv::resize(typed, small, {ISceneDetector::kFrameW, ISceneDetector::kFrameH},
+                   0, 0, cv::INTER_AREA);
+        return small;
+    }
+
     /// TRACES: AR-011 | SR-002
     // How close two boundaries have to be before they are the same boundary,
     // derived from the cadence the detector was actually fed.

src/types.hpp

@@ -184,3 +184,111 @@ struct ActorGallery {
     bool     calib_valid{false};
     uint64_t calib_hash{0};
 };
+
+// ── Channel byte accounting ───────────────────────────────────────────────────
+/// TRACES: AR-004 | SR-002
+///
+/// KPN measures a channel's occupancy in *items* and its bandwidth in bytes,
+/// and gets the byte figure from `kpn::ChannelDataSize<T>`. That primary
+/// template returns `sizeof(T)` — right for a POD, badly wrong for every type
+/// below, each of which is a handful of vectors and a `cv::Mat` header owning
+/// megabytes on the heap.
+///
+/// Unspecialised, the diagnostics reported roughly 200 bytes for a message
+/// carrying a full decoded frame — off by four orders of magnitude at 1080p.
+/// That is not merely a cosmetic stat: it is the one instrument for choosing
+/// channel capacities against a memory ceiling, which is the open half of
+/// AR-004, and it was reading fiction.
+///
+/// **What the number means.** `cv::Mat` is reference-counted, so one decoded
+/// frame referenced from several messages is counted once per reference. The
+/// sum is therefore an upper bound on distinct bytes, and the right bound for
+/// the question being asked: how much would this channel keep alive if nothing
+/// else held it.
+///
+/// Declared against a forward declaration rather than including
+/// `<kpn/channel.hpp>` here, so the message definitions keep no dependency on
+/// the framework that carries them — and so any translation unit that can see
+/// these types also sees their sizes, which is what stops one channel being
+/// instantiated with the default and another with the specialisation.
+
+namespace kpn { template<typename T> struct ChannelDataSize; }
+
+namespace sae::bytes {
+
+inline std::size_t of(const cv::Mat& m) {
+    return m.empty() ? 0u : m.total() * m.elemSize();
+}
+inline std::size_t of(const std::vector<cv::Mat>& v) {
+    std::size_t n = 0;
+    for (const auto& m : v) n += of(m);
+    return n;
+}
+inline std::size_t of(const Frame& f) { return sizeof(Frame) + of(f.image); }
+
+inline std::size_t of(const std::vector<IdentifiedActor>& v) {
+    std::size_t n = v.size() * sizeof(IdentifiedActor);
+    for (const auto& a : v) {
+        n += of(a.crop);
+        // The id strings are short but there is one set per actor per frame,
+        // and a crowd frame carries dozens.
+        n += a.name.capacity() + a.imdb_id.capacity()
+           + a.tmdb_id.capacity() + a.jellyfin_id.capacity();
+    }
+    return n;
+}
+
+}  // namespace sae::bytes
+
+template<> struct kpn::ChannelDataSize<Frame> {
+    static std::size_t bytes(const Frame& f) { return sae::bytes::of(f); }
+};
+
+template<> struct kpn::ChannelDataSize<SceneFrame> {
+    static std::size_t bytes(const SceneFrame& v) {
+        return sizeof(SceneFrame) + sae::bytes::of(v.source)
+             + v.faces.size() * sizeof(DetectedFace);
+    }
+};
+
+template<> struct kpn::ChannelDataSize<AlignedSceneFrame> {
+    static std::size_t bytes(const AlignedSceneFrame& v) {
+        return sizeof(AlignedSceneFrame) + sae::bytes::of(v.source)
+             + v.faces.size() * sizeof(DetectedFace)
+             + sae::bytes::of(v.crops);
+    }
+};
+
+template<> struct kpn::ChannelDataSize<EmbeddedSceneFrame> {
+    static std::size_t bytes(const EmbeddedSceneFrame& v) {
+        return sizeof(EmbeddedSceneFrame) + sae::bytes::of(v.source)
+             + v.faces.size() * sizeof(DetectedFace)
+             + sae::bytes::of(v.crops)
+             + v.embeddings.size() * sizeof(Embedding);
+    }
+};
+
+template<> struct kpn::ChannelDataSize<TrackedSceneFrame> {
+    static std::size_t bytes(const TrackedSceneFrame& v) {
+        return sizeof(TrackedSceneFrame) + sae::bytes::of(v.source)
+             + v.faces.size() * sizeof(DetectedFace)
+             + sae::bytes::of(v.crops)
+             + v.track_ids.size() * sizeof(int)
+             + v.embeddings.size() * sizeof(Embedding);
+    }
+};
+
+template<> struct kpn::ChannelDataSize<MatchedSceneFrame> {
+    static std::size_t bytes(const MatchedSceneFrame& v) {
+        return sizeof(MatchedSceneFrame) + sae::bytes::of(v.source)
+             + sae::bytes::of(v.actors);
+    }
+};
+
+template<> struct kpn::ChannelDataSize<SceneAnnotation> {
+    static std::size_t bytes(const SceneAnnotation& v) {
+        return sizeof(SceneAnnotation) + sae::bytes::of(v.visible_actors);
+    }
+};
+
+// CutEvent owns nothing on the heap, so the default sizeof(T) is already right.

tests/CMakeLists.txt

@@ -28,6 +28,7 @@ add_executable(sae_tests
     test_embedding_dump.cpp
     test_audio_signature.cpp
     test_benchmark.cpp
+    test_channel_bytes.cpp
     ${CMAKE_SOURCE_DIR}/src/backends/gemm_backend.cpp
     ${CMAKE_SOURCE_DIR}/src/gallery/gallery_store.cpp
     ${CMAKE_SOURCE_DIR}/src/audio_signature.cpp

tests/test_channel_bytes.cpp

@@ -0,0 +1,109 @@
+// Channel byte accounting for the pipeline message types.
+//
+// TRACES: AR-004 | SR-002
+//
+// kpn::ChannelDataSize<T> is what a channel reports as bytes pushed, and its
+// primary template returns sizeof(T). Every message type here is a handful of
+// vectors and a cv::Mat header owning megabytes on the heap, so unspecialised
+// the diagnostics reported ~200 bytes for a message carrying a full decoded
+// frame — off by four orders of magnitude at 1080p.
+//
+// That is the instrument for choosing channel capacities against a memory
+// ceiling, which is the open half of AR-004. These cases assert it measures the
+// payload rather than the header, because a stat that is quietly wrong is worse
+// than no stat: it was read as evidence.
+#include <catch2/catch_test_macros.hpp>
+
+#include <kpn/channel.hpp>
+
+#include "types.hpp"
+
+namespace {
+
+Frame frame_with_image(int w, int h) {
+    Frame f;
+    f.image = cv::Mat(h, w, CV_8UC3, cv::Scalar(0, 0, 0));
+    f.timestamp_sec = 1.0;
+    return f;
+}
+
+}  // namespace
+
+TEST_CASE("frame bytes count the decoded image, not the header", "[channel_bytes]") {
+    const Frame f = frame_with_image(1920, 1080);
+    const std::size_t got = kpn::ChannelDataSize<Frame>::bytes(f);
+
+    // 1920 * 1080 * 3 = 6,220,800 payload bytes.
+    REQUIRE(got >= 1920u * 1080u * 3u);
+    // The header is a rounding error next to it; this is the assertion that
+    // fails on the unspecialised default.
+    CHECK(got > 100u * sizeof(Frame));
+}
+
+TEST_CASE("an empty frame costs only its header", "[channel_bytes]") {
+    // The eof sentinel carries no image, and must not be charged for one.
+    Frame eof;
+    eof.eof = true;
+    CHECK(kpn::ChannelDataSize<Frame>::bytes(eof) == sizeof(Frame));
+}
+
+TEST_CASE("crops and embeddings are counted on top of the frame", "[channel_bytes]") {
+    // The case AR-003 created: a crowd frame occupies one slot exactly as an
+    // empty one does, and only the byte figure distinguishes them.
+    EmbeddedSceneFrame v;
+    v.source = frame_with_image(640, 360);
+    const std::size_t bare = kpn::ChannelDataSize<EmbeddedSceneFrame>::bytes(v);
+
+    constexpr int kFaces = 60;
+    for (int i = 0; i < kFaces; ++i) {
+        v.faces.push_back({});
+        v.crops.emplace_back(112, 112, CV_8UC3, cv::Scalar(0, 0, 0));
+        v.embeddings.emplace_back();
+    }
+    const std::size_t crowded = kpn::ChannelDataSize<EmbeddedSceneFrame>::bytes(v);
+
+    // 60 crops at 112*112*3 = 2,257,920 bytes, plus 60 * 2 KiB of embeddings.
+    CHECK(crowded - bare >= kFaces * (112u * 112u * 3u + sizeof(Embedding)));
+    // And the crowd frame really is the multiple of the empty one that the
+    // item-count capacity cannot see: 640x360x3 is ~691 KB, the crops ~2.26 MB.
+    CHECK(crowded > 3 * bare);
+}
+
+TEST_CASE("every message type on a channel measures its payload", "[channel_bytes]") {
+    // A specialisation missing for any one of these silently reverts that
+    // channel to sizeof(T), which is exactly how this went unnoticed.
+    const Frame f = frame_with_image(320, 240);
+    const std::size_t img = 320u * 240u * 3u;
+
+    SceneFrame sf;        sf.source = f;
+    AlignedSceneFrame af; af.source = f;
+    EmbeddedSceneFrame ef; ef.source = f;
+    TrackedSceneFrame tf; tf.source = f;
+    MatchedSceneFrame mf; mf.source = f;
+
+    CHECK(kpn::ChannelDataSize<SceneFrame>::bytes(sf)         >= img);
+    CHECK(kpn::ChannelDataSize<AlignedSceneFrame>::bytes(af)  >= img);
+    CHECK(kpn::ChannelDataSize<EmbeddedSceneFrame>::bytes(ef) >= img);
+    CHECK(kpn::ChannelDataSize<TrackedSceneFrame>::bytes(tf)  >= img);
+    CHECK(kpn::ChannelDataSize<MatchedSceneFrame>::bytes(mf)  >= img);
+
+    // SceneAnnotation carries no source frame — only the actors it identified,
+    // each with its own crop.
+    SceneAnnotation sa;
+    sa.visible_actors.push_back({});
+    sa.visible_actors.back().crop = cv::Mat(112, 112, CV_8UC3, cv::Scalar(0, 0, 0));
+    CHECK(kpn::ChannelDataSize<SceneAnnotation>::bytes(sa) >= 112u * 112u * 3u);
+}
+
+TEST_CASE("a shared image is charged to each message holding it", "[channel_bytes]") {
+    // cv::Mat is reference-counted, so a frame referenced from several messages
+    // is counted once per reference. The sum is an upper bound on distinct
+    // bytes, and the right bound for "what would this channel keep alive if
+    // nothing else held it" — which is the question a capacity answers.
+    const Frame f = frame_with_image(320, 240);
+    SceneFrame a; a.source = f;
+    SceneFrame b; b.source = f;   // shares the same pixel buffer
+
+    CHECK(kpn::ChannelDataSize<SceneFrame>::bytes(a)
+          == kpn::ChannelDataSize<SceneFrame>::bytes(b));
+}

tests/test_scene_detector_node.cpp

@@ -14,6 +14,9 @@
 
 #include "nodes/scene_detector_node.hpp"
 
+#include <opencv2/imgproc.hpp>
+#include <utility>
+
 #include <vector>
 
 namespace {
@@ -84,3 +87,105 @@ TEST_CASE("too few frames to have a cadence yields an inert window",
 TEST_CASE("a single observed interval is enough", "[scene][AR-011]") {
     CHECK(SceneDetectorFunc::dedup_window_sec({1.0 / 24.0}) == 0.5 / 24.0);
 }
+
+// ── AR-004 — the window stores the model's input, not the decoded frame ───────
+//
+// TRACES: AR-004, AR-010 | SR-002 | UT-003
+//
+// The rolling window held frames as decoded, at full resolution, and left the
+// downscale to the backend — ~590 MB at 1080p to feed a model whose input is
+// 48x27, about 380 KB. Not a channel capacity, so no amount of tuning channel
+// depths would have found it.
+//
+// The risk in fixing it is the project invariant: every model gets the input it
+// was trained for. A model run off-distribution returns confident, plausible,
+// wrong output, and here that means fabricated shot boundaries — which would be
+// indistinguishable from a real cut in the output.
+//
+// So these cases do not check that the frames got smaller. They check that the
+// pixels are *identical* to what the backend would have produced from the full
+// frame, by performing the backend's own two operations independently and
+// comparing byte for byte. Both ort_backend.cpp and trt_backend.cpp guard
+// mis-sized input with convertTo(CV_8UC3) then
+// cv::resize(..., {kFrameW, kFrameH}, 0, 0, cv::INTER_AREA), in that order.
+namespace {
+
+cv::Mat gradient(int w, int h) {
+    // Structured content, not a flat fill: INTER_AREA averages, so a constant
+    // image would compare equal under almost any resize and prove nothing.
+    cv::Mat m(h, w, CV_8UC3);
+    for (int y = 0; y < h; ++y)
+        for (int x = 0; x < w; ++x)
+            m.at<cv::Vec3b>(y, x) = cv::Vec3b(
+                static_cast<uchar>((x * 7 + y * 3) % 256),
+                static_cast<uchar>((x * 13 + y * 5) % 256),
+                static_cast<uchar>((x * 3 + y * 11) % 256));
+    return m;
+}
+
+bool identical(const cv::Mat& a, const cv::Mat& b) {
+    if (a.size() != b.size() || a.type() != b.type()) return false;
+    cv::Mat diff;
+    cv::absdiff(a, b, diff);
+    return cv::countNonZero(diff.reshape(1)) == 0;
+}
+
+}  // namespace
+
+TEST_CASE("the window frame is what the backend would have produced",
+          "[scene][AR-004]") {
+    for (auto [w, h] : {std::pair{1920, 1080}, std::pair{640, 360}, std::pair{720, 480}}) {
+        INFO("source " << w << "x" << h);
+        const cv::Mat full = gradient(w, h);
+
+        // The backend's own guard, performed here independently.
+        cv::Mat expected;
+        cv::resize(full, expected, {ISceneDetector::kFrameW, ISceneDetector::kFrameH},
+                   0, 0, cv::INTER_AREA);
+
+        const cv::Mat got = SceneDetectorFunc::to_model_input(full);
+
+        REQUIRE(got.cols == ISceneDetector::kFrameW);
+        REQUIRE(got.rows == ISceneDetector::kFrameH);
+        REQUIRE(got.type() == CV_8UC3);
+        CHECK(identical(got, expected));
+    }
+}
+
+TEST_CASE("a frame already at model size is passed through untouched",
+          "[scene][AR-004]") {
+    // The backend skips its guard for a correctly-sized frame, so this path must
+    // not resize either — resampling an already-48x27 image would change it.
+    const cv::Mat exact = gradient(ISceneDetector::kFrameW, ISceneDetector::kFrameH);
+    CHECK(identical(SceneDetectorFunc::to_model_input(exact), exact));
+}
+
+TEST_CASE("conversion happens before the resize, as the backend does it",
+          "[scene][AR-004]") {
+    // Order matters: converting a 4-channel frame after downscaling averages
+    // alpha into the colour channels and gives different pixels. The backends
+    // convert first, so this must too.
+    cv::Mat four(360, 640, CV_8UC4, cv::Scalar(10, 20, 30, 255));
+    cv::Mat typed;
+    four.convertTo(typed, CV_8UC3);
+    cv::Mat expected;
+    cv::resize(typed, expected, {ISceneDetector::kFrameW, ISceneDetector::kFrameH},
+               0, 0, cv::INTER_AREA);
+
+    CHECK(identical(SceneDetectorFunc::to_model_input(four), expected));
+}
+
+TEST_CASE("the window's memory is bounded by the model input, not the source",
+          "[scene][AR-004]") {
+    // The point of the change, stated as a number: a full window of 1080p
+    // frames is ~590 MB as decoded and ~380 KB as model input.
+    const cv::Mat full = gradient(1920, 1080);
+    const cv::Mat small = SceneDetectorFunc::to_model_input(full);
+
+    const std::size_t decoded = full.total()  * full.elemSize();
+    const std::size_t stored  = small.total() * small.elemSize();
+
+    INFO("decoded " << decoded << " B, stored " << stored << " B");
+    CHECK(stored * 1000 < decoded);          // three orders of magnitude
+    CHECK(stored == ISceneDetector::kFrameW * ISceneDetector::kFrameH * 3u);
+}