feat: join the decode butterfly so scene boundaries reach the face branch

AR-010 — is_scene_boundary had no producer: SceneDetectorFunc was a terminal
sink writing scenes.json and never annotating the frames flowing to face
detection. The flag was permanently false, so the boundary half of AR-007's
frame-dependent association was dead code that a test could still exercise
synthetically and appear to verify.

The topology already forks after decode — dense frames to TransNetV2, sampled
frames to face detection — so this is a fork-join. SceneBoundaries is the join:
the detector publishes each window's verdict with a watermark, and an annotator
on the sampled branch stamps the flag.

The watermark is the part that matters. TransNetV2 buffers 100 frames before it
can score any of them, so at any instant it has an opinion up to some time T and
none after. Without recording T a consumer cannot tell "no boundary" from "not
scored yet", and those demand opposite behaviour — treating unscored frames as
boundary-free is exactly what makes a downstream check pass while verifying
nothing.

Buffering alone does not work, which was my first attempt. Channel depth creates
lag only when the consumer is slower, and the face branch runs four orders of
magnitude faster per frame than TransNetV2 (0.01ms vs 400ms), so its channels
drain instantly and no lag accumulates. Measured: 106 of 364 frames outran the
detector. The annotator therefore waits on the watermark explicitly. The
detector signals completion so the tail cannot deadlock, and publishes from
flush_remaining too — without that the final frames arrive with no verdict.

Boundaries are deduped on publish, matching what scenes.json does at write time.
A run of adjacent high-scoring frames is one boundary, not several; leaving them
raw made this view report 357 where the file said 13. Now the two agree exactly.

Frames past the detector's last scored window remain unverified and are counted
as such rather than silently marked boundary-free.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

TRACES: AR-007, AR-010 | SR-002

src/scene_boundaries.hpp

@@ -0,0 +1,126 @@
+#pragma once
+/// TRACES: AR-010 | SR-002
+///
+/// SceneBoundaries — the join point of the decode butterfly.
+///
+/// The topology forks after decode: one branch runs TransNetV2 over dense
+/// frames, the other runs face detection over the sampled cadence. Boundaries
+/// found on the first branch have to reach the second, and they cannot be
+/// carried in the frames themselves because the branches are parallel.
+///
+/// **Why this needs a watermark.** TransNetV2 buffers `kWindow` frames before it
+/// can score any of them, so at any instant the detector has an opinion about
+/// everything up to some time T and nothing after it. Without recording T, a
+/// consumer asking "is there a boundary at t?" cannot distinguish *no* from
+/// *not yet* — and those demand opposite behaviour. Silently treating unscored
+/// frames as boundary-free is exactly the class of failure that makes a
+/// verification pass vacuously.
+///
+/// The consumer is held back by channel depth (see main.cpp) so that by the time
+/// it pulls a frame, the detector has already scored past it. `scored_through()`
+/// is what lets that assumption be *checked* rather than assumed.
+
+#include <algorithm>
+#include <condition_variable>
+#include <mutex>
+#include <vector>
+
+class SceneBoundaries {
+public:
+    /// Peaks closer than this are one boundary. Matches the dedup scenes.json
+    /// applies, so the two views agree.
+    static constexpr double kMergeSec = 0.04;
+
+    /// Called by the scene detector as each window is scored. `through` is the
+    /// timestamp up to which its verdict is now final.
+    void publish(const std::vector<double>& ts, double through) {
+        {
+        std::lock_guard<std::mutex> g(mu_);
+        // Dedup on insert, matching what scenes.json does at write time. A run
+        // of adjacent high-scoring frames is one boundary, not several, and
+        // leaving them raw made this view report 357 where the file said 13 —
+        // the same event counted many times. Harmless for is_boundary(), which
+        // absorbs them in its tolerance, but a count nobody can reconcile with
+        // the output file is a bad diagnostic.
+        bounds_.insert(bounds_.end(), ts.begin(), ts.end());
+        std::sort(bounds_.begin(), bounds_.end());
+        bounds_.erase(std::unique(bounds_.begin(), bounds_.end(),
+                                  [](double a, double b) { return b - a < kMergeSec; }),
+                      bounds_.end());
+        scored_through_ = std::max(scored_through_, through);
+        }
+        cv_.notify_all();
+    }
+
+    /// True if a boundary falls within `tol` of `t`.
+    ///
+    /// `tol` exists because the two branches sample at different rates: a
+    /// boundary found on a dense frame rarely lands exactly on a sampled one.
+    /// Half a sample interval is the natural width — it attributes the boundary
+    /// to the nearest sampled frame and no further.
+    bool is_boundary(double t, double tol) const {
+        std::lock_guard<std::mutex> g(mu_);
+        auto it = std::lower_bound(bounds_.begin(), bounds_.end(), t - tol);
+        return it != bounds_.end() && *it <= t + tol;
+    }
+
+    /// The timestamp through which the detector's verdict is final. A consumer
+    /// past this point is asking about frames nobody has looked at yet.
+    double scored_through() const {
+        std::lock_guard<std::mutex> g(mu_);
+        return scored_through_;
+    }
+
+    /// Block until the detector's verdict covers `t`, or it finishes.
+    ///
+    /// Channel depth alone does NOT create the required lag: it only holds
+    /// frames back when the consumer is slower, and the face branch is roughly
+    /// four orders of magnitude faster per frame than TransNetV2. So the join
+    /// has to wait explicitly.
+    ///
+    /// Returns false if the detector finished without ever covering `t`, which
+    /// happens for the tail frames after its last full window. The caller must
+    /// distinguish that from a genuine "no boundary" rather than assuming.
+    bool wait_until_scored(double t) const {
+        std::unique_lock<std::mutex> lk(mu_);
+        cv_.wait(lk, [&] { return finished_ || scored_through_ >= t; });
+        return scored_through_ >= t;
+    }
+
+    /// Called when the detector will publish nothing further. Without this the
+    /// join would deadlock on the tail: those frames are never covered by a full
+    /// window, so waiting for them would wait forever.
+    void finish() {
+        {
+            std::lock_guard<std::mutex> g(mu_);
+            finished_ = true;
+        }
+        cv_.notify_all();
+    }
+
+    std::size_t count() const {
+        std::lock_guard<std::mutex> g(mu_);
+        return bounds_.size();
+    }
+
+    /// Consumers that outran the detector. Nonzero means the face branch is not
+    /// buffered deeply enough for the detector's window, so some frames were
+    /// annotated from an incomplete verdict — a real misconfiguration, and one
+    /// that would otherwise be invisible.
+    void note_outran() const {
+        std::lock_guard<std::mutex> g(mu_);
+        ++outran_;
+    }
+    std::size_t outran() const {
+        std::lock_guard<std::mutex> g(mu_);
+        return outran_;
+    }
+
+private:
+    mutable std::mutex              mu_;
+    mutable std::condition_variable cv_;
+    bool                            finished_{false};
+    std::vector<double> bounds_;
+    double              scored_through_{-1.0};
+    mutable std::size_t outran_{0};
+};