// Stage-2 presenter orchestrator. GOAL ARCHITECTURE (the result of the 2026-07 pacing saga —
// read this before touching presentation):
//
//   net pump ──► VideoDecoder (VT async) ──► newest-wins 1-slot ring ──► RENDER THREAD ──► CAMetalLayer
//
// • The render thread is woken by FRAME ARRIVAL (the decoder callback signals it), never gated on
//   the display link: on macOS the WindowServer's damage tracking / FramePacing does not count our
//   out-of-band presents, so anything display-link-gated stalls exactly when the rest of the screen
//   goes quiet (adaptive-sync displays idle the link down). A decoded frame is always presented
//   promptly. The display link remains only as (a) a vsync CLOCK (phase + period, for the opt-in
//   V-Sync policy below), (b) a retry tick for a frame that couldn't get a drawable (`putBack`),
//   and (c) the iOS ProMotion rate hint.
// • The layer's own displaySyncEnabled stays FALSE on macOS — synced presents starve the drawable
//   pool outright (see MetalVideoPresenter's init for the post-mortem).
// • Present policy is a USER SETTING (DefaultsKey.vsync; PUNKTFUNK_PRESENT_MODE=immediate|vsync
//   overrides it for A/B), resolved once per session in start():
//     – V-Sync OFF (default): present immediately — lowest latency, the long-proven behavior.
//     – V-Sync ON: present(at: next vsync) predicted from the link's last phase/period, at most one
//       period ahead by construction, falling back to immediate when the link data is stale — a
//       schedule can never sit far in the future holding drawables hostage.
// • Present PACING is the stage-2/3/4 presenter split (`PresentPacing`, chosen per session by
//   SessionPresenter from the presenter setting / PUNKTFUNK_PRESENTER): stage-2 presents on
//   frame arrival; stage-3 additionally gates presents on the on-glass callback (`PresentGate`)
//   so the layer's FIFO image queue can never saturate; stage-4 (iOS/tvOS) presents into
//   CAMetalDisplayLink-vended drawables the moment a frame decodes — deadline pacing, where the
//   queue cannot exist at all — see PresentPacing's doc for the full rationale. Under deadline
//   pacing the render thread below is fed by BOTH the decoder callback and the link's per-refresh
//   updates (which vend the drawable), and the V-Sync policy/vsync clock don't apply.
// • Rendering lives on its own thread so any `nextDrawable()` wait lands off-main (input, SwiftUI).
//
// The render thread also stamps the unified latency stages (end-to-end capture→on-glass + decode and
// display stage terms — design/stats-unification.md). Mirrors StreamPump's lifecycle (one per start;
// cancel is permanent). PUNKTFUNK_PRESENT_DEBUG=1 prints per-second pacing stats (see
// PresentDebugStats).
//
// Threading: the pump runs on its own thread; the decoder callback on a VT thread; the render loop on
// the render thread; `renderTick` + `start`/`stop` on the MAIN thread (the view's CADisplayLink fires
// there). Only the ring (lock-guarded), the vsync clock (lock-guarded), and the decoder/presenter
// (internally locked / staged) cross threads.

#if canImport(Metal) && canImport(QuartzCore)
import AVFoundation
import Foundation
import Metal
import PunktfunkShared
import QuartzCore
import os

/// PUNKTFUNK_PRESENT_DEBUG=1: the render thread prints a once-per-second line with the decode
/// (ring-submit) rate, present rate, failed/empty wakes and the slowest render call — for
/// diagnosing pacing regressions without instruments. Plain print: the unbundled CLI client's
/// stdout is the cheapest reliable capture channel.
let presentDebug = ProcessInfo.processInfo.environment["PUNKTFUNK_PRESENT_DEBUG"] == "1"

/// The pf-present line's os_log mirror (subsystem io.unom.punktfunk, category "present") — the
/// SessionModel "stats" mirror's sibling, so DEADLINE sessions stream their pacing decomposition
/// to Console.app wirelessly with no env var / Xcode attach. Always on for deadline pacing (the
/// stats are a few arrays + one log line per second); other pacings keep the env-gated print.
private let presentLog = ClientLog(category: "present")
/// Pump-side events (loss recovery, format seeding) — the stage-2 sibling of StreamPump's log.
private let pumpLog = ClientLog(category: "pump")

/// Decoded-frame hand-off between the decode half and the render thread. The POLICY is the
/// user's presentation intent (design/apple-presentation-rebuild.md — the 2026-07 rebuild that
/// replaced the visible stage picker):
///
/// - `.newestWins` (Prioritize lowest latency, the default): a 1-slot ring — the decoder
///   overwrites (drops the older undisplayed frame), the render thread takes-and-clears. Zero
///   store by construction: any deeper app-held buffer ahead of a latch-paced display becomes a
///   STANDING queue costing one full refresh per slot, forever (the depth-2 gate post-mortem —
///   see SessionPresenter.gateDepth).
/// - `.fifo(capacity: K)` (Prioritize smoothness): a small deliberate jitter buffer. The
///   decoder appends; overflow drops the OLDEST (bounded added latency — the newest keeps
///   flowing); the render thread pops the oldest ONE per present opportunity, so the cadence is
///   the display's. `take` withholds frames until the buffer has PREROLLED to capacity once —
///   without preroll a steady stream drains every frame on arrival and headroom never builds —
///   and re-arms preroll when it runs dry (an underflow: the previous frame persists on glass,
///   a repeat by omission, while headroom rebuilds). Each buffered frame ≈ one refresh interval
///   of jitter absorbed for one interval of added display latency, which the metrics SHOW —
///   only the OS present floor is shaved from the HUD, never the user's chosen buffer.
///
/// Sendable; lock-guarded — decoder callbacks and the render thread cross here.
public enum FrameStorePolicy: Sendable, Equatable {
    case newestWins
    case fifo(capacity: Int)
}

public final class FrameStore<Frame>: @unchecked Sendable {
    private let lock = NSLock()
    private let capacity: Int // 1 = newest-wins semantics
    private let isFifo: Bool
    private var frames: [Frame] = []
    private var prerolled = false
    /// Submissions since the last `drainSubmitted` — the decode rate for the pf-present line.
    private var submitted = 0
    /// Smoothness accounting for the pf-present line: frames dropped by a full buffer, and
    /// runs-dry that re-armed preroll.
    private var overflowDrops = 0
    private var underflows = 0

    public init(policy: FrameStorePolicy) {
        switch policy {
        case .newestWins:
            capacity = 1
            isFifo = false
        case .fifo(let k):
            capacity = max(1, k)
            isFifo = true
        }
    }

    func submit(_ f: Frame) {
        lock.lock()
        if isFifo {
            frames.append(f)
            if frames.count > capacity {
                frames.removeFirst() // oldest goes — bounded latency, the newest keeps flowing
                overflowDrops += 1
            }
        } else {
            frames = [f] // newest wins; the replaced frame is the intended drop point
        }
        submitted += 1
        lock.unlock()
    }

    func drainSubmitted() -> Int {
        lock.lock()
        defer { lock.unlock() }
        let n = submitted
        submitted = 0
        return n
    }

    /// Take-and-reset the smoothness counters (the pf-present `qDrop`/`qDry` stats).
    func drainSmoothing() -> (overflowDrops: Int, underflows: Int) {
        lock.lock()
        defer { lock.unlock() }
        let out = (overflowDrops, underflows)
        overflowDrops = 0
        underflows = 0
        return out
    }

    func take() -> Frame? {
        lock.lock()
        defer { lock.unlock() }
        if isFifo {
            if !prerolled {
                guard frames.count >= capacity else { return nil } // still building headroom
                prerolled = true
            }
            guard !frames.isEmpty else {
                underflows += 1 // ran dry — repeat by omission, rebuild headroom
                prerolled = false
                return nil
            }
            return frames.removeFirst()
        }
        let f = frames.first
        frames.removeAll(keepingCapacity: true)
        return f
    }

    /// Cadence-driven take: hand back the oldest frame only once its DUE time has arrived, so the
    /// store's job becomes "hold what is not due yet" instead of "release one per present
    /// opportunity" (design/presenter-cadence-rework.md §4.3). `due` projects the frame's due
    /// instant on the same clock as `now`; nil (no cadence estimate for that frame) means due
    /// immediately.
    ///
    /// The preroll gate does not apply here. It exists only to build headroom for a per-slot
    /// drain, and under cadence targeting the cushion IS the headroom — prerolling on top would
    /// stack `capacity − 1` frames of standing latency the user never asked for. `underflows` is
    /// not counted either: an empty store is the normal steady state once frames are held until
    /// due, so the honest starvation signal is `CadenceHealth.late` (the due time had already
    /// passed when the frame became presentable), not a run-dry count.
    func take(dueBy now: CFTimeInterval, due: (Frame) -> CFTimeInterval?) -> Frame? {
        lock.lock()
        defer { lock.unlock() }
        guard let oldest = frames.first else { return nil }
        if let at = due(oldest), at > now { return nil } // held — not due yet
        if isFifo { return frames.removeFirst() }
        frames.removeAll(keepingCapacity: true)
        return oldest
    }

    /// Return a frame the render thread took but could not present (no drawable yet, or a
    /// transient render failure). Newest-wins keeps it only while the slot is still empty — a
    /// newer decoded frame wins; FIFO reinserts it at the FRONT (it is the oldest; a transient
    /// capacity+1 is trimmed by the next submit). Without this, a failed present silently LOSES
    /// the frame, and under the host's infinite GOP a static scene sends no replacement until
    /// the next damage — the stale picture would persist.
    func putBack(_ f: Frame) {
        lock.lock()
        if isFifo {
            frames.insert(f, at: 0)
        } else if frames.isEmpty {
            frames = [f]
        }
        lock.unlock()
    }
}

/// The display's vsync grid as last reported by the display link (target timestamp + period,
/// `CACurrentMediaTime` basis), written on main by `renderTick`, read by the render thread to
/// schedule V-Sync-mode presents. A shared box (like `ReadyRing`) so neither thread captures the
/// pipeline itself. Sendable; lock-guarded.
private final class VsyncClock: @unchecked Sendable {
    private let lock = NSLock()
    private var target: CFTimeInterval = 0
    private var period: CFTimeInterval = 0

    func set(target t: CFTimeInterval, period p: CFTimeInterval) {
        lock.lock(); target = t; period = p; lock.unlock()
    }

    /// The next vsync at or after `now`, extrapolated from the last reported phase/period — by
    /// construction less than one period ahead, so a scheduled present can never sit far in the
    /// future holding its drawable. nil (⇒ present immediately) when the link has reported nothing
    /// yet, its period is nonsense, or its data is STALE (an idle/suspended link on an
    /// adaptive-sync display — exactly the case where scheduling onto its grid stalls the stream).
    func nextVsync(after now: CFTimeInterval) -> CFTimeInterval? {
        lock.lock(); defer { lock.unlock() }
        guard period > 0.0005, target > 0, now - target < 0.25 else { return nil }
        if target >= now { return target }
        return target + ceil((now - target) / period) * period
    }
}

/// Tuning for one cadence loop. Gains are SHIFT COUNTS — the loop is fixed-point Int64
/// throughout, so it runs identically on every client and in the offline harness, and carries no
/// float into a present path.
///
/// ⚠ **These values are provisional, and saying so is part of the design.** The plan asks for
/// constants fitted to recorded `(src_pts, received, decoded)` traces (its spike S2), and S2 was
/// never run. What is here is derived from first principles — a proportional time constant of tens
/// of frames, an integral an order slower, a cushion of a few mean-absolute-deviations — and the
/// first real trace should replace them.
///
/// ⚠ A hand-written port of `punktfunk_core::phase::CadenceTuning`, exactly as `AudioRing` is a
/// port of `JitterPolicy`: this pipeline is Swift and does not link that Rust type. **Every
/// constant and rule here must stay in lockstep with it** — `CadenceClockTests` and the Rust
/// `phase::tests` assert the two against the same synthetic input, and that agreement IS the
/// contract.
struct CadenceTuning: Equatable {
    /// Proportional gain on the offset estimate: `1 >> offsetShift` of the residual per frame.
    var offsetShift: UInt8
    /// Integral gain on the per-frame rate (skew) term: `1 >> skewShift`.
    var skewShift: UInt8
    /// EMA weight for the residual mean-absolute-deviation.
    var jitterShift: UInt8
    /// Per-sample residual clamp — one outlier must not yank the estimate.
    var errorClampNs: Int64
    /// Cushion = `mad * cushionNum / cushionDen`, clamped to
    /// `[cushionFloorNs, frameIntervalNs]`.
    var cushionNum: UInt16
    var cushionDen: UInt16
    var cushionFloorNs: Int64
    /// Source-timestamp gap beyond which the loop re-anchors instead of tracking.
    var reanchorGapNs: Int64

    /// For callers that snap the due time onto a display grid afterwards: the snap-up itself
    /// carries roughly half a refresh of implicit slack, so the cushion can be small. Every Apple
    /// present snaps — onto `VsyncClock.nextVsync` under arrival/glass pacing, onto the
    /// CAMetalDisplayLink's own vend under deadline pacing — so this is the one this client runs.
    static func snapping() -> CadenceTuning {
        CadenceTuning(
            offsetShift: 5, skewShift: 10, jitterShift: 5, errorClampNs: 20_000_000,
            cushionNum: 2, cushionDen: 1, cushionFloorNs: 500_000,
            reanchorGapNs: 500_000_000)
    }

    /// For callers presenting at the due time directly (VRR, direct scanout): no implicit slack,
    /// so the cushion must cover more of the distribution on its own.
    static func freeRunning() -> CadenceTuning {
        var t = snapping()
        t.cushionNum = 3
        t.cushionFloorNs = 2_000_000
        return t
    }
}

/// Loop health for the pf-present line — the numbers that say whether the cushion is doing its
/// job. Mirrors `punktfunk_core::phase::CadenceHealth`.
///
/// Residual PERCENTILES are deliberately absent: this type holds no histogram, and the client
/// stat paths (the latency meters) are where distributions belong.
struct CadenceHealth: Equatable {
    /// Frames folded since the last `reset`.
    var frames: UInt64 = 0
    /// …of which the due time was already past when the frame became presentable. The direct
    /// signal that the cushion is too small.
    var late: UInt64 = 0
    /// Times the loop gave up tracking and re-anchored (gap, regression, or explicit reset).
    var reanchors: UInt64 = 0
    var offsetNs: Int64 = 0
    var skewNs: Int64 = 0
    var jitterNs: Int64 = 0
    var cushionNs: Int64 = 0
}

/// Plays frames out on the SOURCE's cadence instead of on their arrival instant.
///
/// The defect it exists for: every client presents a frame as soon as it is decoded, so the
/// transport's jitter — and, on a host whose compositor delivers raggedly, the compositor's —
/// lands on the glass 1:1. The 2026-08-15 Skynet field log has KWin's screencast arriving
/// 0.11–8.22 ms off its own grid (up to a full 120 Hz period) for 24 minutes on a session with the
/// bitrate pinned and zero loss. The loop estimates the offset between the source clock and the
/// present clock and hands back a due time on the source's own timeline plus a cushion sized to
/// the measured jitter.
///
/// **Type-2 on purpose.** It tracks offset *and* per-frame rate, because two free-running crystals
/// produce a ramp and a proportional-only loop lags a ramp forever.
///
/// **It smooths the offset, never the timestamps.** Due is `srcPts + offset + cushion`, so genuine
/// variation in the source's own cadence — a variable-rate renderer, an irregular capture tick —
/// passes straight through, and only the transport's contribution to `ready − pts` is filtered.
/// Anything that made due times more evenly spaced than the source would be a bug.
///
/// **Domain-agnostic by construction.** A constant offset between clock domains is absorbed by the
/// offset estimator, so a caller feeds `readyNs` and reads the due time in ONE domain with no
/// conversion anywhere in this path. On Apple that domain is `CACurrentMediaTime` — the clock
/// `presentAtMediaTime` consumes — so the decode-output instant is converted ONCE on the way in
/// (`Stage2Pipeline.mediaTimeNs(forRealtimeNs:)`) and the due time comes back needing none.
/// Suspend/resume breaks the constant; the gap re-anchor below is what covers it.
///
/// A late frame's due time is returned in the PAST, unclamped: clamping it to `readyNs` would
/// quietly turn every late frame into a fresh anchor, which is precisely the arrival-driven
/// presentation this exists to stop being.
///
/// Prior art is ordinary and old: MPEG-2 TS PCR recovery and RTP playout scheduling (RFC 3550
/// §6.4.1 carries the jitter estimator this MAD mirrors).
///
/// ⚠ A hand-written port of `punktfunk_core::phase::CadenceClock` — see `CadenceTuning` for the
/// lockstep contract. Sendable; lock-guarded — the decode-completion thread folds frames while the
/// render thread reads health.
final class CadenceClock: @unchecked Sendable {
    private let lock = NSLock()
    private let tuning: CadenceTuning
    /// `ready − srcPts`, smoothed. Absorbs the clock-domain constant.
    private var offsetNs: Int64 = 0
    /// Per-frame drift of that offset — the integral term.
    private var skewNs: Int64 = 0
    /// EMA of |residual|, the cushion's input.
    private var madNs: Int64 = 0
    /// nil until the first sample anchors the loop.
    private var lastPtsNs: UInt64?
    /// Last frame interval seen, so `cushionNs` can apply its ceiling.
    private var frameIntervalNs: Int64 = 0
    private var counters = CadenceHealth()

    init(tuning: CadenceTuning) {
        self.tuning = tuning
    }

    /// Force a re-anchor on the next sample. Call on every discontinuity the client already knows
    /// about: reanchor, codec rebuild, surface recreate, jump-to-live, resume.
    func reset() {
        lock.lock()
        lastPtsNs = nil
        skewNs = 0
        // `madNs` deliberately SURVIVES. It describes the link, not the stream, and a cushion that
        // collapsed to its floor at every rebuild would spend the next few hundred frames
        // presenting late — the exact failure the cushion exists to prevent.
        lock.unlock()
    }

    /// Fold one presentable frame and return when it is due, in the present clock domain.
    ///
    /// `readyNs` is when the frame became presentable; `frameIntervalNs` is the nominal source
    /// interval and the cushion's ceiling.
    ///
    /// The result **may be earlier than `readyNs`** — that is a late frame, and the caller's
    /// contract is "already due ⇒ present at the next opportunity", never "drag the grid back to
    /// now".
    func dueNs(srcPtsNs: UInt64, readyNs: Int64, frameIntervalNs: Int64) -> Int64 {
        lock.lock()
        defer { lock.unlock() }
        self.frameIntervalNs = frameIntervalNs
        counters.frames += 1
        let pts = Int64(bitPattern: srcPtsNs)
        let raw = saturatingSub(readyNs, pts)

        let anchored: Bool
        if let last = lastPtsNs {
            // Source time going BACKWARDS, or a gap so long the estimate cannot be trusted to
            // have tracked across it: re-anchor rather than slew for seconds.
            anchored =
                !(srcPtsNs < last
                    || srcPtsNs - last > UInt64(bitPattern: tuning.reanchorGapNs))
        } else {
            anchored = false
        }
        if anchored {
            // Advance the estimate one frame on the rate term, then correct it by a bounded
            // fraction of what the new sample says.
            offsetNs = saturatingAdd(offsetNs, skewNs)
            let err = min(
                max(saturatingSub(raw, offsetNs), -tuning.errorClampNs), tuning.errorClampNs)
            offsetNs = saturatingAdd(offsetNs, shrTowardZero(err, tuning.offsetShift))
            skewNs = saturatingAdd(skewNs, shrTowardZero(err, tuning.skewShift))
            let dev = abs(err) - madNs
            madNs = saturatingAdd(madNs, shrTowardZero(dev, tuning.jitterShift))
        } else {
            offsetNs = raw
            skewNs = 0
            counters.reanchors += 1
        }
        lastPtsNs = srcPtsNs

        let due = saturatingAdd(saturatingAdd(pts, offsetNs), lockedCushionNs())
        if due < readyNs { counters.late += 1 }
        return due
    }

    /// A frame whose timestamp is not on the source cadence — a repeat the host re-anchored at
    /// submit, a stamp its plausibility gate replaced with "now", or one that reached us with no
    /// usable pts at all. Those samples do not lie on the source's timeline, and folding them in
    /// would drag the offset estimate toward "now" exactly when the stream is idle and the
    /// estimate matters most.
    ///
    /// Returns a due time from the CURRENT estimate, leaving offset, skew and jitter untouched:
    /// the frame is simply due once it is ready, cushioned like any other.
    func noteOffCadence(readyNs: Int64, frameIntervalNs: Int64) -> Int64 {
        lock.lock()
        defer { lock.unlock() }
        self.frameIntervalNs = frameIntervalNs
        return saturatingAdd(readyNs, lockedCushionNs())
    }

    func jitterNs() -> Int64 {
        lock.lock()
        defer { lock.unlock() }
        return madNs
    }

    /// How far past the estimate a frame is held, to absorb the measured jitter.
    ///
    /// The one-frame-interval ceiling is an INVARIANT, not a tunable: a cushion past a whole frame
    /// buys latency for smoothness the source cannot supply, and at that point the honest fix is a
    /// deeper buffer the user asked for, not a loop quietly holding frames.
    func cushionNs() -> Int64 {
        lock.lock()
        defer { lock.unlock() }
        return lockedCushionNs()
    }

    func health() -> CadenceHealth {
        lock.lock()
        defer { lock.unlock() }
        var h = counters
        h.offsetNs = offsetNs
        h.skewNs = skewNs
        h.jitterNs = madNs
        h.cushionNs = lockedCushionNs()
        return h
    }

    private func lockedCushionNs() -> Int64 {
        let den = Int64(max(tuning.cushionDen, 1))
        let want = saturatingMul(madNs, Int64(tuning.cushionNum)) / den
        let ceiling = frameIntervalNs > 0 ? frameIntervalNs : Int64.max
        return min(max(want, min(tuning.cushionFloorNs, ceiling)), ceiling)
    }
}

/// Arithmetic shift that rounds toward ZERO, so a negative residual is damped by exactly as much
/// as its positive twin. A plain `>>` rounds toward −∞, which biases a loop that spends its whole
/// life within a few nanoseconds of zero error.
private func shrTowardZero(_ v: Int64, _ shift: UInt8) -> Int64 {
    v < 0 ? -((-v) >> shift) : v >> shift
}

/// The Rust loop is `saturating_*` throughout — a garbage timestamp must clamp the estimate, never
/// trap the render thread. Swift's operators trap instead of saturating, so the port carries its
/// own; for every reachable input these are plain `+`, `−`, `×`.
private func saturatingAdd(_ a: Int64, _ b: Int64) -> Int64 {
    let (v, overflow) = a.addingReportingOverflow(b)
    return overflow ? (b > 0 ? Int64.max : Int64.min) : v
}

private func saturatingSub(_ a: Int64, _ b: Int64) -> Int64 {
    let (v, overflow) = a.subtractingReportingOverflow(b)
    return overflow ? (b > 0 ? Int64.min : Int64.max) : v
}

private func saturatingMul(_ a: Int64, _ b: Int64) -> Int64 {
    let (v, overflow) = a.multipliedReportingOverflow(by: b)
    guard overflow else { return v }
    return (a > 0) == (b > 0) ? Int64.max : Int64.min
}

/// When a ready frame is pushed to the layer — the stage-2 vs stage-3 presenter split. Same decode
/// half, same newest-wins ring; only the present cadence differs.
///
/// - `arrival` (stage-2, the macOS default): present the moment a frame is decoded. Lowest latency
///   while the layer's image queue is shallow — but that queue is FIFO and consumed at one drawable
///   per refresh (iOS always vsync-latches; the macOS 26 compositor latch-paces our out-of-band
///   presents the same way when composited), so at stream rate ≈ refresh rate its depth is STICKY:
///   one early burst (session start, a Wi-Fi clump) fills it to `maximumDrawableCount` and — with
///   arrivals and latches then running at the same rate — it never drains. Every later frame rides
///   ~2–3 refreshes of queue (the measured 23–30 ms display stage on 120 Hz ProMotion panels), and
///   the full-queue regime is where host↔panel clock drift turns into periodic repeats/drops (the
///   "fixed-interval" jitter reports).
/// - `glass` (stage-3; tvOS's default until the 2026-07 rebuild moved it to stage-4, now an
///   on-device A/B rung): at most a small BOUNDED number of presented-but-
///   undisplayed drawables in flight (`PresentGate`; depth 1 — see `SessionPresenter.gateDepth`
///   for why deeper is a regression). The render thread presents only while a gate slot is free
///   (a drawable's presented handler reopens its slot and re-signals); frames decoded meanwhile
///   coalesce in the newest-wins ring. Freshness is preserved by DROPPING stale frames before
///   present instead of queueing them behind the display — the hidden queue latency becomes
///   explicit, correct frame drops. The residual cost: presents serialize on the on-glass
///   callback, whose own delivery latency pushes each present ~a refresh past the frame's decode
///   (the field-measured 14 ms display stage at 120 Hz vs the ~half-refresh floor).
/// - `deadline` (stage-4, the iOS/iPadOS default; iOS/tvOS only — see
///   `PresenterChoice.explicit`): a CAMetalDisplayLink vends ONE drawable per refresh
///   (`preferredFrameLatency` 1) into a newest-wins hand-off slot, and the render thread pairs
///   it with the newest decoded frame THE MOMENT either half arrives — usually the frame, into
///   an already-vended drawable. The image queue cannot exist (one vended drawable in flight,
///   ever), nothing serializes on the on-glass callback (the link's next vend is the pace), and
///   the present is deadline-timed by the system to latch the upcoming refresh. This is the only
///   pacing whose steady state can reach the sub-refresh display floor on the always-vsync-latch
///   platforms; `arrival`/`glass` remain the on-device A/B rungs.
///
/// macOS PyroWave sessions default to `glass` even though the platform default is stage-2: burst
/// presents into a composited (windowed) layer are the trigger pattern for the macOS DCP
/// "mismatched swapID's" KERNEL PANIC, and the one-in-flight gate removes that pattern — see
/// `SessionPresenter.pacing` for the full rationale.
public enum PresentPacing: Sendable {
    case arrival
    case glass
    case deadline
}

/// Newest-wins 1-slot hand-off box (the generic sibling of `ReadyRing`): deadline pacing's
/// drawable stash — the link thread `put`s each update's vended drawable (replacing an
/// unpresented older one, which just returns to the layer's pool), the render thread `take`s.
/// `putBack` returns a taken value only while the slot is still empty, so a fresher `put` from
/// the other thread is never clobbered by a stale return. Internal (not private) for unit tests.
/// Sendable; lock-guarded.
final class LatestBox<T>: @unchecked Sendable {
    private let lock = NSLock()
    private var value: T?
    func put(_ v: T) { lock.lock(); value = v; lock.unlock() }
    func putBack(_ v: T) {
        lock.lock()
        if value == nil { value = v }
        lock.unlock()
    }
    func take() -> T? {
        lock.lock()
        defer { lock.unlock() }
        let v = value
        value = nil
        return v
    }
}

/// The deadline link's frame-latency ASK and property READBACK, published for the HUD to render.
///
/// ⚠ A readback is NOT a grant. `preferredFrameLatency` is a plain read-write float
/// (CAMetalDisplayLink.h carries no doc contract), so reading it returns whatever we last
/// stored unless the system actively clamps the setter — and the 2026-08-13 field run proved
/// how misleading that is: it read 1.00 while the measured vend lead sat at 1.95 refresh
/// periods. The number that tells the truth about scheduling is the vend lead (the HUD's
/// `os present` floor), never this property. The line still earns its place twice over: a
/// readback that DIFFERS from the ask is the one clamp signal the API can give, and the ask
/// must be visible on screen because **on tvOS no log is reachable** — `log stream --device`
/// is gone from modern macOS, `log collect --device-name` needs root and then fails "Device
/// not configured" because an Apple TV has no USB to fall back to, and the libimobiledevice
/// pairing is a different database from Xcode's. Console.app is a GUI.
///
/// A process-global rather than a sixth parameter threaded through SessionModel → StreamView →
/// controller → SessionPresenter → Stage2Pipeline → delegate: it is write-once-per-session
/// diagnostics, and this file already keeps `presentDebug`/`presentLog` at file scope. Reset by
/// `clear()` at session start so a stale session's answer can never be read as this one's.
public final class PresentLinkInfo: @unchecked Sendable {
    public static let shared = PresentLinkInfo()
    private let lock = NSLock()
    private var ask: Float = 0
    private var latency: Float = 0
    private var rangeMin: Float = 0
    private var rangeMax: Float = 0
    private var drawables: Int = 0
    private var present = false

    private init() {}

    func publish(ask: Float, latency: Float, rangeMin: Float, rangeMax: Float, drawables: Int) {
        lock.lock()
        self.ask = ask
        self.latency = latency
        self.rangeMin = rangeMin
        self.rangeMax = rangeMax
        self.drawables = drawables
        present = true
        lock.unlock()
    }

    /// Session start — a link that never comes up must not leave the previous one's answer up.
    public func clear() {
        lock.lock()
        present = false
        lock.unlock()
    }

    /// `nil` until the link's first update (or on a non-deadline rung, which has no link).
    public func snapshot()
        -> (ask: Float, latency: Float, rangeMin: Float, rangeMax: Float, drawables: Int)?
    {
        lock.lock()
        defer { lock.unlock() }
        return present ? (ask, latency, rangeMin, rangeMax, drawables) : nil
    }
}

/// Deadline pacing's staged frame-rate hint, and — on every pacing — the session's nominal source
/// interval (`sourceIntervalNs`). SessionPresenter pushes the stream rate from the
/// MAIN thread (session start + every layout/Reconfigure); the link's own thread drains and
/// applies it, so the CAMetalDisplayLink is only ever touched from the thread that runs it. The
/// floor is PINNED at the stream rate — no idle ramp-down: with a low floor the link idles toward
/// it on a static scene (infinite GOP ⇒ no frames), and the first damage frame after idle would
/// wait out a slow tick before it could present. Empty wakes at stream rate are near-free; the
/// PANEL still idles via VRR because no presents happen. Sendable; lock-guarded.
private final class FrameRateHint: @unchecked Sendable {
    private let lock = NSLock()
    private var pending: CAFrameRateRange?
    private var streamHz: Float = 0
    private var boosted = false
    func stage(hz: Float) {
        guard hz > 0 else { return }
        lock.lock()
        streamHz = hz
        pending = Self.range(hz: hz, boosted: boosted)
        lock.unlock()
    }
    /// Pen-proximity boost: pin `minimum = preferred` at the range's CEILING instead of the
    /// stream rate. UIKit delivers touch/Pencil events at the PANEL's cadence, and the panel
    /// follows this link's vote — so a 60 fps stream on a 120 Hz iPad halves pencil sampling
    /// unless a boost lifts the panel while the Pencil is in range. Presents still pace at
    /// stream rate (extra link updates just vend into the newest-wins stash), so the cost is
    /// empty wakes, scoped to pen proximity.
    func setBoost(_ on: Bool) {
        lock.lock()
        if boosted != on {
            boosted = on
            if streamHz > 0 { pending = Self.range(hz: streamHz, boosted: on) }
        }
        lock.unlock()
    }
    func drain() -> CAFrameRateRange? {
        lock.lock()
        defer { lock.unlock() }
        let p = pending
        pending = nil
        return p
    }
    /// The nominal SOURCE interval in nanoseconds — the cadence clock's cushion ceiling. Read on
    /// every pacing, not just deadline: this box is where the negotiated stream rate already
    /// lives, staged from main on session start and every Reconfigure, and the decode-completion
    /// thread needs it under a lock. 0 = not known yet, which the clock handles by running its
    /// cushion uncapped (the shared core's own behaviour for a zero interval).
    func sourceIntervalNs() -> Int64 {
        lock.lock()
        defer { lock.unlock() }
        return streamHz > 0 ? Int64(1_000_000_000.0 / Double(streamHz)) : 0
    }
    private static func range(hz: Float, boosted: Bool) -> CAFrameRateRange {
        #if os(tvOS)
        // A TV is a FIXED-rate display: there is no ProMotion panel to lift and no Pencil to
        // sample for, so the `max(hz, 120)` ceiling below asks a 60 Hz Apple TV to accept
        // anything up to 120. A range is a promise about how variable our cadence may be, and a
        // scheduler handed 60…120 on a fixed 60 Hz display has every reason to keep a frame of
        // slack in hand — which is what a two-refresh `targetPresentationTimestamp` IS. Pin all
        // three bounds to the stream rate so the deadline has nothing to hedge against.
        // (Field 2026-08-13, Apple TV 4K / tvOS 27: `os present` stuck at ~2 × 16.67 with
        // `preferredFrameLatency = 1` asked for and re-asserted every update; shrinking the
        // drawable pool to 2 moved it not at all.) `boosted` is deliberately ignored — it exists
        // for pen proximity, which tvOS does not have.
        _ = boosted
        return CAFrameRateRange(minimum: hz, maximum: hz, preferred: hz)
        #else
        let cap = max(hz, 120)
        let preferred = boosted ? cap : hz
        return CAFrameRateRange(minimum: preferred, maximum: cap, preferred: preferred)
        #endif
    }
}

/// The client half of phase-locked capture (design/phase-locked-capture.md): the decode
/// callback deposits per-AU arrival stamps (client CLOCK_REALTIME — the core's reassembly-
/// completion time), the deadline link's thread deposits the latch grid, and ~1 Hz that same
/// thread flushes the circular arrival-phase statistic to the host. The statistic is a
/// verbatim port of `punktfunk_core::phase::circular_latch` — the host's v3 controller
/// (grid-locked submits, coherence-gated engage) was tuned against exactly it, and a
/// period-smeared Wi-Fi link correctly reads coherence ≈ 0 there, so the controller never
/// engages where alignment is physically pointless. Shared box (never captures the pipeline);
/// a session's connection binds/unbinds like DecodeReport/KeyframeRecovery.
final class PhaseReporter: @unchecked Sendable {
    private let lock = NSLock()
    private var connection: PunktfunkConnection?
    /// Arrival stamps since the last flush, client CLOCK_REALTIME. Bounded: ~1 s at 240 fps.
    private var arrivalsNs: [Int64] = []
    /// Smallest update-to-update spacing this window: successive `nextLatch` values sit one
    /// panel period apart except across skipped link updates (2×, 3×, …), so the window
    /// minimum IS the period. Re-learned every flush so VRR/mode switches track both ways.
    private var periodNs: Int64 = 0
    private var prevLatchRealNs: Int64 = 0
    private var lastFlushRealNs: Int64 = 0

    func bind(_ c: PunktfunkConnection?) {
        lock.lock()
        connection = c
        arrivalsNs.removeAll()
        periodNs = 0
        prevLatchRealNs = 0
        lastFlushRealNs = 0
        lock.unlock()
    }

    /// Decode-callback side: one AU's arrival (reassembly-completion) stamp.
    func noteArrival(receivedNs: Int64) {
        lock.lock()
        if connection != nil, arrivalsNs.count < 256 { arrivalsNs.append(receivedNs) }
        lock.unlock()
    }

    /// Link-thread side, once per update: where the NEXT latch sits on the client's realtime
    /// clock (the arrival stamps' domain). Learns the period from update spacing and ~1 Hz
    /// converts the window's arrivals into leads against this grid, then reports — a
    /// fire-and-forget datagram push on the control plane.
    func noteGrid(nextLatchRealNs: Int64) {
        lock.lock()
        if prevLatchRealNs > 0 {
            let delta = nextLatchRealNs - prevLatchRealNs
            // 2–100 ms accepts 10–500 Hz panels, rejects wakeup hiccups and clock jumps.
            if delta > 2_000_000, delta < 100_000_000, periodNs == 0 || delta < periodNs {
                periodNs = delta
            }
        }
        prevLatchRealNs = nextLatchRealNs
        guard let c = connection, periodNs > 0, arrivalsNs.count >= 8,
            nextLatchRealNs - lastFlushRealNs >= 1_000_000_000
        else {
            lock.unlock()
            return
        }
        lastFlushRealNs = nextLatchRealNs
        let period = periodNs
        let leadsUs = arrivalsNs.map { a -> UInt64 in
            let m = (nextLatchRealNs - a) % period
            return UInt64(m < 0 ? m + period : m) / 1000
        }
        arrivalsNs.removeAll(keepingCapacity: true)
        periodNs = 0
        let offsetNs = c.clockOffsetNs
        lock.unlock()
        guard
            let (leadMeanNs, coherence) = Self.circularLatch(
                samplesUs: leadsUs, periodNs: period)
        else { return }
        c.reportPhase(
            nextLatchHostNs: UInt64(max(0, nextLatchRealNs + offsetNs)),
            latchPeriodNs: UInt32(clamping: period),
            uncertaintyNs: 1_000_000, // skew residual — same conservative 1 ms as Android
            arrivalLeadNs: UInt32(clamping: leadMeanNs),
            coherenceMilli: coherence)
    }

    /// Verbatim port of `punktfunk_core::phase::circular_latch` (µs samples against an ns
    /// period; nil under 8 samples). The MEAN is what a phase controller can steer under
    /// jitter — a period-spanning distribution's median is immovable — and the coherence
    /// (resultant length, ‰) says whether any phase exists to steer at all.
    static func circularLatch(samplesUs: [UInt64], periodNs: Int64) -> (UInt64, UInt16)? {
        guard samplesUs.count >= 8, periodNs > 0 else { return nil }
        let periodUs = Double(periodNs) / 1000.0
        var x = 0.0
        var y = 0.0
        for s in samplesUs {
            let theta = Double(s).truncatingRemainder(dividingBy: periodUs) / periodUs * 2 * .pi
            x += cos(theta)
            y += sin(theta)
        }
        let n = Double(samplesUs.count)
        let r = (x * x + y * y).squareRoot() / n
        var meanTheta = atan2(y, x)
        if meanTheta < 0 { meanTheta += 2 * .pi }
        return (UInt64(meanTheta / (2 * .pi) * Double(periodNs)), UInt16(r * 1000.0))
    }
}

/// The CAMetalDisplayLink delegate for deadline pacing: each per-refresh update stashes its
/// vended drawable (newest wins) and nudges the render thread — which also wakes on decoder
/// arrivals, so whichever half completes the (frame, drawable) pair triggers the present. Also
/// applies the staged frame-rate hint from the link's own thread. Retained by the link thread's
/// closure (the link holds it weak); captures only the shared boxes, never the pipeline — the
/// same no-self-capture rule as the pump/render threads.
private final class DeadlineLinkDelegate: NSObject, CAMetalDisplayLinkDelegate {
    private let stash: LatestBox<CAMetalDrawable>
    private let renderSignal: DispatchSemaphore
    private let hint: FrameRateHint
    private let stats: PresentDebugStats?
    /// Phase-locked capture's grid feed — this link IS the latch grid presents pace against.
    private let phase: PhaseReporter?
    /// The OS-floor sampler (design/apple-presentation-rebuild.md): every update's vend→glass
    /// lead is recorded so its p50 becomes the "OS present floor" the HUD subtracts from the
    /// shown display/e2e numbers. Self-adapting: ~1 refresh period is the goal, ~2 means the
    /// compositor is running a frame ahead of us (what a 3-slot drawable pool bought it before
    /// `startDeadlinePresenter` clamped stage-4 to 2). Tracks VRR rate changes.
    private let floorMeter: LatencyMeter?
    /// The pool depth this session vends from (`startDeadlinePresenter` sets it on the layer).
    /// Carried only so the one-shot line below reports the two halves of the depth question
    /// together — a `preferredFrameLatency` of 1 against a 3-slot pool is the configuration that
    /// measured a two-refresh floor in the field, and reading either number alone hides that.
    private let drawableCount: Int
    /// The `preferredFrameLatency` this session asks for — 1 by default, PUNKTFUNK_FRAME_LATENCY
    /// for the on-device ladder (see `startDeadlinePresenter` for the ladder's design).
    private let latencyAsk: Float
    /// One-shot: log the link's preferredFrameLatency READBACK after the first re-assert. A
    /// readback differing from the ask ⇒ the system clamps the property (the one clamp signal
    /// it can give); a readback EQUAL to the ask proves nothing — only vendLeadMs does (see
    /// PresentLinkInfo's doc for the field lesson).
    private var loggedEffective = false

    init(
        stash: LatestBox<CAMetalDrawable>, renderSignal: DispatchSemaphore,
        hint: FrameRateHint, stats: PresentDebugStats?, floorMeter: LatencyMeter?,
        phase: PhaseReporter?, drawableCount: Int, latencyAsk: Float
    ) {
        self.stash = stash
        self.renderSignal = renderSignal
        self.hint = hint
        self.stats = stats
        self.floorMeter = floorMeter
        self.phase = phase
        self.drawableCount = drawableCount
        self.latencyAsk = latencyAsk
    }

    func metalDisplayLink(_ link: CAMetalDisplayLink, needsUpdate update: CAMetalDisplayLink.Update) {
        if let range = hint.drain(), link.preferredFrameRateRange != range {
            link.preferredFrameRateRange = range
        }
        // Re-assert the latency ask every update (cheap compare): it was set once before
        // add(to:), and whether a pre-add set survives scheduling is exactly the kind of
        // thing the vendLeadMs stat exists to catch — belt and braces.
        if link.preferredFrameLatency != latencyAsk { link.preferredFrameLatency = latencyAsk }
        // Publish every update, not just the first: the range is re-applied from the staged hint
        // above (mode switch / rate change), and `preferredFrameLatency` is re-asserted right
        // here — so the readback can change mid-session, and a write-once snapshot would keep
        // showing the answer to a question we have since asked again. Cheap: five stores under
        // an uncontended lock, once per refresh.
        let range = link.preferredFrameRateRange
        PresentLinkInfo.shared.publish(
            ask: latencyAsk, latency: link.preferredFrameLatency, rangeMin: range.minimum,
            rangeMax: range.maximum, drawables: drawableCount)
        if !loggedEffective {
            loggedEffective = true
            let msg = String(
                format: "deadline link up: preferredFrameLatency ask=%.2f readback=%.2f "
                    + "maxDrawables=%d range=%.0f-%.0f preferred=%.0f",
                latencyAsk, link.preferredFrameLatency, drawableCount,
                range.minimum, range.maximum, range.preferred ?? 0)
            presentLog.info("\(msg, privacy: .public)")
        }
        // The link's own pipeline depth, measured: how far ahead of glass this vend runs.
        let leadS = update.targetPresentationTimestamp - CACurrentMediaTime()
        stats?.vendLead(ms: leadS * 1000)
        // Same measurement into the floor meter (as a LatencyMeter sample: end = now, start =
        // now − lead) — its 1 s p50 is the OS present floor SessionModel shaves off.
        if leadS > 0, let floorMeter {
            var ts = timespec()
            clock_gettime(CLOCK_REALTIME, &ts)
            let nowNs = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec)
            floorMeter.record(
                ptsNs: UInt64(nowNs - Int64(leadS * 1_000_000_000)), atNs: nowNs, offsetNs: 0)
        }
        // Phase-locked capture: this update's target present, converted into the arrival
        // stamps' CLOCK_REALTIME domain. Per-update cost is one clock read; the reporter
        // itself flushes ~1 Hz.
        if let phase {
            var ts = timespec()
            clock_gettime(CLOCK_REALTIME, &ts)
            let nowNs = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec)
            phase.noteGrid(nextLatchRealNs: nowNs + Int64(leadS * 1_000_000_000))
        }
        stash.put(update.drawable)
        renderSignal.signal()
    }
}

/// Stage-3's present gate: admits `capacity` in-flight (presented, not yet on glass) drawables.
/// The render thread `tryAcquire`s before taking a frame; the drawable's presented handler
/// `release`s and re-signals the render thread. Depth 1 fully serializes presents on the on-glass
/// callback — which costs a refresh whenever the callback's own latency pushes the next present
/// past a vsync; depth 2 keeps one flip queued behind the one scanning out, so a decoded frame
/// presents immediately and latches the very next vsync while the queue still can't build (see
/// `SessionPresenter.gateDepth` for the per-platform choice). `staleAfter` is insurance against a
/// present whose handler never fires (the macOS "out-of-band presents aren't damage" hazard class
/// — see MetalVideoPresenter's init post-mortem): rather than freezing the stream, a full gate
/// force-opens a slot 100 ms after its oldest present, a visible ~10 fps degradation that
/// PUNKTFUNK_PRESENT_DEBUG's `forced` counter exposes (it reads 0 on healthy systems). Internal
/// (not private) for unit tests. Sendable; lock-guarded — the releaser runs on a Metal callback
/// thread.
final class PresentGate: @unchecked Sendable {
    /// How long one pending present may hold its slot before it's presumed lost.
    static let staleAfter: CFTimeInterval = 0.1

    private let lock = NSLock()
    private let capacity: Int
    /// Arm instants of the in-flight presents, oldest first (≤ `capacity` entries).
    private var armed: [CFTimeInterval] = []
    private var forced = 0

    /// `capacity` = the in-flight present budget (clamped to ≥ 1) — see the type doc.
    init(capacity: Int = 1) {
        self.capacity = max(1, capacity)
    }

    /// Arm the gate for one present. False = the gate is full of live presents (none stale) —
    /// leave the frame in the ring; a presented handler's release/re-signal (or the next
    /// display-link tick) retries with the freshest frame then.
    func tryAcquire(now: CFTimeInterval) -> Bool {
        lock.lock()
        defer { lock.unlock() }
        if armed.count >= capacity {
            // Full: reopen only by presuming the OLDEST in-flight present lost (its handler
            // never fired) rather than stalling the stream.
            guard let oldest = armed.first, now - oldest > Self.staleAfter else { return false }
            armed.removeFirst()
            forced += 1
        }
        armed.append(now)
        return true
    }

    /// One in-flight present reached glass (or was dropped, or its render failed before a present
    /// was registered) — free the oldest slot. A release with nothing in flight is a no-op; a
    /// lost present's handler firing late after its stale force-open can transiently over-admit
    /// one flip, which the next glass callback corrects.
    func release() {
        lock.lock()
        if !armed.isEmpty { armed.removeFirst() }
        lock.unlock()
    }

    /// Take-and-reset the force-open count (PUNKTFUNK_PRESENT_DEBUG's `forced` stat).
    func drainForced() -> Int {
        lock.lock()
        defer { lock.unlock() }
        let n = forced
        forced = 0
        return n
    }
}

/// PUNKTFUNK_PRESENT_DEBUG=1 aggregation: one printed line per second from the render thread with
/// the decode rate, render outcomes, the slowest render call (≈ nextDrawable wait) and the deltas
/// between system-reported on-glass times (vsync-aligned presents show clean refresh-period
/// multiples; immediate flips scatter). Lock-guarded — `presented` lands on a Metal callback thread.
private final class PresentDebugStats: @unchecked Sendable {
    /// The session's cadence loop, for the line's `cadence` segment — `nil` under the latency
    /// intent, and then the line is emitted exactly as it was before source-timestamp playout
    /// existed. `late` is the number WP8 gates on: a due time already past when the frame became
    /// presentable is the direct signal that the cushion is too small.
    private let cadence: CadenceClock?
    private let lock = NSLock()
    private var last = CACurrentMediaTime()
    private var ok = 0, failed = 0, empty = 0, dropped = 0, gated = 0, noDrawable = 0
    private var maxRenderMs = 0.0
    private var lastGlassNs: Int64 = 0
    private var glassDeltasMs: [Double] = []
    /// Present-issue → on-glass delay per frame (system presentedTime minus the render call's
    /// start) — the DIRECT decomposition of the display stage: ring/pairing wait lives upstream
    /// of it, queue + present-pipeline cost inside it. Standing queue reads as ~n×period here;
    /// a healthy latch reads under one period.
    private var latchMs: [Double] = []
    /// Deadline pacing: the link's own pipeline depth — `targetPresentationTimestamp - now` at
    /// each update. ~1 period means preferredFrameLatency=1 is honored (a vended drawable can
    /// reach glass at the NEXT refresh); ~2 periods means the system is running a frame ahead
    /// and one whole refresh of the display stage lives INSIDE the link, not in our pairing.
    private var vendLeadMs: [Double] = []
    /// Presented-but-not-yet-on-glass drawables right now / the window's peak — the direct
    /// measurement of the layer image-queue depth the stage-3 gate exists to bound (stage-2 on a
    /// 120 Hz panel saturates this at ~maximumDrawableCount; stage-3 pegs it at the gate depth).
    private var inFlight = 0
    private var maxInFlight = 0

    init(cadence: CadenceClock?) {
        self.cadence = cadence
    }

    func emptyWake() { lock.lock(); empty += 1; lock.unlock() }

    /// A wake that found the stage-3 gate closed (a present still in flight) — the frame stays in
    /// the ring for the handler's re-signal. Includes display-link ticks while gated; a high count
    /// is normal, it just shows the gate working.
    func gatedWake() { lock.lock(); gated += 1; lock.unlock() }

    /// Deadline pacing: a decoded frame is waiting but the link hasn't vended this interval's
    /// drawable yet — the frame presents on the link's next update. A high count just means
    /// decode outruns the link's phase; the wait is bounded by one refresh.
    func noDrawableWake() { lock.lock(); noDrawable += 1; lock.unlock() }

    /// Deadline pacing, LINK thread: one update's vend-to-target distance (see `vendLeadMs`).
    func vendLead(ms: Double) { lock.lock(); vendLeadMs.append(ms); lock.unlock() }

    func renderReturned(ok rendered: Bool, tookMs: Double) {
        lock.lock()
        if rendered {
            ok += 1
            inFlight += 1
            maxInFlight = max(maxInFlight, inFlight)
        } else {
            failed += 1
        }
        maxRenderMs = max(maxRenderMs, tookMs)
        lock.unlock()
    }

    func presented(atNs: Int64?, issuedNs: Int64) {
        lock.lock()
        inFlight = max(0, inFlight - 1) // clamp: the handler can beat renderReturned's increment
        if let atNs {
            if lastGlassNs > 0 { glassDeltasMs.append(Double(atNs - lastGlassNs) / 1e6) }
            lastGlassNs = atNs
            latchMs.append(Double(atNs - issuedNs) / 1e6)
        } else {
            dropped += 1
        }
        lock.unlock()
    }

    func flushIfDue(ring: FrameStore<ReadyFrame>, gate: PresentGate?) {
        lock.lock()
        let now = CACurrentMediaTime()
        guard now - last >= 1 else { lock.unlock(); return }
        last = now
        let decoded = ring.drainSubmitted()
        let smoothing = ring.drainSmoothing()
        let deltas = glassDeltasMs.sorted()
        let p50 = deltas.isEmpty ? 0 : deltas[deltas.count / 2]
        let dMax = deltas.last ?? 0
        let latches = latchMs.sorted()
        let latchP50 = latches.isEmpty ? 0 : latches[latches.count / 2]
        let latchMax = latches.last ?? 0
        let vends = vendLeadMs.sorted()
        let vendP50 = vends.isEmpty ? 0 : vends[vends.count / 2]
        let vendMax = vends.last ?? 0
        let inflightMax = maxInFlight
        // Loop health, appended only where a loop exists — `late`/`frames` is WP8's cushion
        // criterion and `reanchor` says whether the estimate is tracking at all.
        let loop = cadence?.health()
        let cadenceLine =
            loop.map {
                String(
                    format: " cadence late=%llu/%llu reanchor=%llu jitterUs=%lld cushionUs=%lld "
                        + "skewNs=%lld",
                    $0.late, $0.frames, $0.reanchors, $0.jitterNs / 1000, $0.cushionNs / 1000,
                    $0.skewNs)
            } ?? ""
        let line = String(
            format: "pf-present decoded=%d ok=%d fail=%d empty=%d gated=%d noDrawable=%d "
                + "dropped=%d qDrop=%d qDry=%d maxRenderMs=%.1f inflightMax=%d forced=%d "
                + "glassDeltaMs p50=%.2f max=%.2f n=%d latchMs p50=%.2f max=%.2f "
                + "vendLeadMs p50=%.2f max=%.2f",
            decoded, ok, failed, empty, gated, noDrawable, dropped,
            smoothing.overflowDrops, smoothing.underflows, maxRenderMs, inflightMax,
            gate?.drainForced() ?? 0, p50, dMax, deltas.count, latchP50, latchMax,
            vendP50, vendMax) + cadenceLine
        ok = 0; failed = 0; empty = 0; dropped = 0; gated = 0; noDrawable = 0
        maxRenderMs = 0
        maxInFlight = inFlight // the window peak restarts from the live depth
        glassDeltasMs.removeAll(keepingCapacity: true)
        latchMs.removeAll(keepingCapacity: true)
        vendLeadMs.removeAll(keepingCapacity: true)
        lock.unlock()
        // Console.app first (the on-device readout — see presentLog); stdout only under the env
        // lever (the CLI client's capture channel).
        presentLog.info("\(line, privacy: .public)")
        if presentDebug {
            print(line)
            fflush(stdout) // stdout is a pipe when captured — flush per line or nothing shows
        }
    }
}

/// Bridges the VideoToolbox decode-completion callback to the core Automatic-bitrate controller's
/// decode signal. Created as a pipeline property so the decoder's `onDecoded` callback (built in
/// `init`, before the connection exists) can capture it, then `start` binds the live connection +
/// the arming flag once known — the same "reference captured in init, configured in start" shape as
/// `recovery`/`gate`. `record` runs on VideoToolbox's callback thread; `bind` runs once on the main
/// thread before the pump feeds the first AU, so the plain fields are safe (set-once, then read).
private final class DecodeReport: @unchecked Sendable {
    private weak var connection: PunktfunkConnection?
    private var enabled = false
    func bind(_ connection: PunktfunkConnection) {
        self.connection = connection
        self.enabled = connection.wantsDecodeLatency()
    }
    /// Report received→decoded for one frame, in µs. Both stamps are client `CLOCK_REALTIME`
    /// (no skew). Skips when the controller isn't armed, so it's free to call on every decode.
    func record(receivedNs: Int64, decodedNs: Int64) {
        guard enabled, let c = connection else { return }
        let us = (decodedNs - receivedNs) / 1000
        if us > 0 { c.reportDecodeUs(UInt32(min(us, Int64(UInt32.max)))) }
    }
}

public final class Stage2Pipeline {
    private let ring: FrameStore<ReadyFrame>
    private let presenter: MetalVideoPresenter
    private let decoder: VideoDecoder
    /// Present cadence — `.arrival` (stage-2), `.glass` (stage-3, the present gate) or
    /// `.deadline` (stage-4, the CAMetalDisplayLink engine). Fixed for the pipeline's lifetime;
    /// SessionPresenter resolves it per session (see PresentPacing).
    private let pacing: PresentPacing
    /// The glass gate's in-flight present budget (`PresentGate` capacity) — meaningful only under
    /// `.glass`; SessionPresenter resolves it per platform (see `SessionPresenter.gateDepth`).
    private let gateDepth: Int
    /// macOS smoothness: pace presents onto the vsync grid (`present(at:)` via the VsyncClock),
    /// at most one per vsync, so the FIFO store drains on the display's cadence rather than on
    /// arrival. Ignored under `.deadline` (the link IS the cadence there).
    private let vsyncPaced: Bool
    /// Source-timestamp playout for the SMOOTHNESS intent: every decoded frame is stamped with
    /// when it is due on the host's own cadence, and the present decision aims there instead of at
    /// the moment the frame happened to decode. `nil` under `latency`, whose path keeps no cadence
    /// arithmetic in it at all — see the intent gate in `init`.
    private let cadence: CadenceClock?
    private let endToEndMeter: LatencyMeter?
    private let decodeMeter: LatencyMeter?
    private let displayMeter: LatencyMeter?
    /// The measured OS present floor (deadline pacing only): each link update's vend→glass lead
    /// is recorded here, and its p50 is what SessionModel subtracts from the shown display/e2e
    /// numbers — the pipeline-depth cost no client controls (design/apple-presentation-rebuild.md).
    private let presentFloorMeter: LatencyMeter?
    private let recovery = KeyframeRecovery()
    /// Feeds the core Automatic-bitrate controller's decode signal from the decode callback; `start`
    /// binds the live connection + arming flag (see DecodeReport).
    private let decodeReport = DecodeReport()
    private let phaseReporter = PhaseReporter()
    /// Post-loss freeze-until-reanchor gate (shared core policy via the C ABI). Created here seeded 0;
    /// `start` reseeds it to the live connection's drop count. Captured by the decoder callbacks
    /// (which withhold concealed frames) and driven by the pump (arm on a gap, poll per iteration).
    private let gate = ReanchorGate(framesDropped: 0)
    private var token = StopFlag()
    /// LIVE host↔client clock offset, read AT EACH RECORD — never cached per session. Until
    /// 2026-08-13 this was a `let` snapshot of the connect-time handshake, and on a host whose
    /// wall clock steps (a VM under NTP) the frozen value silently shifted every host-anchored
    /// stat — field evidence: hostnet 17–21 ms one session, a physically impossible 4.4 ms the
    /// next, same wired host. The core re-syncs the estimate mid-stream (60 s + step detection);
    /// each call is an atomic load behind the FFI.
    private var clockOffset: () -> Int64 = { 0 }
    /// Signalled when the pump thread exits, so `stop()` can join it (bounded) before `decoder.reset()`
    /// — otherwise a pump iteration already past its `token.isStopped` check can rebuild a decode session
    /// right after the reset (a brief orphan session). `pumpJoinable` is armed by `start`, consumed by
    /// the first `stop` (so the idempotent second `stop`/deinit doesn't block on an already-drained
    /// semaphore). start/stop are sequential lifecycle calls, so the plain flag is safe.
    private let pumpStopped = DispatchSemaphore(value: 0)
    private var pumpJoinable = false

    /// Render-thread plumbing. `renderSignal` wakes the render thread — signalled by the DECODER
    /// callback on every frame (the primary trigger: presentation must never be gated on the
    /// display link, see the header) and by each display-link tick (the `putBack` retry + the
    /// vsync-clock refresh). Signals coalesce harmlessly (an extra wake finds an empty ring and
    /// goes back to sleep). `vsyncClock` is the link's last phase/period for V-Sync-mode
    /// scheduling. Lock-guarded boxes — the render thread, like the pump thread, must not capture
    /// `self`, or a missed stop() would leak a spinning pipeline. `renderStopped`/`renderJoinable`
    /// mirror the pump's bounded join.
    private let renderSignal = DispatchSemaphore(value: 0)
    private let vsyncClock = VsyncClock()
    private let renderStopped = DispatchSemaphore(value: 0)
    private var renderJoinable = false
    /// Deadline pacing's staged CAMetalDisplayLink frame-rate hint (see `FrameRateHint`).
    /// Created unconditionally (cheap); only the deadline link thread drains it.
    private let frameRateHint = FrameRateHint()

    /// The Metal layer the hosting view installs + sizes.
    public var layer: CAMetalLayer { presenter.layer }

    /// Unified-stats meters (design/stats-unification.md): `endToEndMeter` records the headline
    /// end-to-end (capture→on-glass, skew-corrected); `decodeMeter` the decode stage
    /// (received→decoded); `displayMeter` the display stage (decoded→on-glass, the ring wait +
    /// render + vsync — the tail stage-2 exists to shorten). All optional: metering never gates
    /// the presenter choice. Returns nil if Metal can't be set up (headless / no GPU) — caller
    /// falls back to the stage-1 presenter. `pacing` selects the stage-2 (arrival) vs stage-3
    /// (glass-gated) present cadence — see PresentPacing; `gateDepth` is the glass gate's
    /// in-flight present budget (see `SessionPresenter.gateDepth`).
    public init?(
        endToEndMeter: LatencyMeter?,
        decodeMeter: LatencyMeter? = nil,
        displayMeter: LatencyMeter? = nil,
        presentFloorMeter: LatencyMeter? = nil,
        pacing: PresentPacing = .arrival,
        gateDepth: Int = 1,
        storePolicy: FrameStorePolicy = .newestWins,
        vsyncPaced: Bool = false
    ) {
        guard let presenter = MetalVideoPresenter.make() else { return nil }
        self.presenter = presenter
        self.pacing = pacing
        self.gateDepth = gateDepth
        self.vsyncPaced = vsyncPaced
        self.ring = FrameStore(policy: storePolicy)
        self.endToEndMeter = endToEndMeter
        self.decodeMeter = decodeMeter
        self.displayMeter = displayMeter
        self.presentFloorMeter = presentFloorMeter
        // The intent gate: source-timestamp playout is what `smooth` MEANS now, and `latency` is
        // defined as arrival-driven with no cushion — so the store policy, which is the intent's
        // only other expression (`PresentPriority.storePolicy`: smooth → fifo, latency →
        // newest-wins), is what decides whether a clock exists at all. A latency session runs the
        // same present path it ran before this existed.
        switch storePolicy {
        case .newestWins: self.cadence = nil
        case .fifo: self.cadence = CadenceClock(tuning: .snapping())
        }
        let ring = ring
        let recovery = recovery
        let renderSignal = renderSignal
        let gate = gate
        let decodeReport = decodeReport
        let phaseReporter = phaseReporter
        let cadence = cadence
        let rateHint = frameRateHint
        self.decoder = VideoDecoder(
            onDecoded: { frame in
                // Decode stage = received→decoded, both client CLOCK_REALTIME (offset 0 — no
                // skew applies). Stamped at decode completion, so it covers every decoded frame,
                // including ones the re-anchor gate withholds or the newest-wins ring drops.
                decodeMeter?.record(
                    ptsNs: UInt64(frame.receivedNs), atNs: frame.decodedNs, offsetNs: 0)
                // Same interval, reported to the core bitrate controller so Automatic caps at this
                // device's real decode limit instead of the network link ceiling. Every decoded
                // frame (not just presented ones), so a newest-wins drop can't hide the backlog.
                decodeReport.record(receivedNs: frame.receivedNs, decodedNs: frame.decodedNs)
                // Phase-locked capture's arrival half: the stamp VALUE is reassembly
                // completion, so recording it at decode adds no bias to the 1 Hz aggregate.
                phaseReporter.noteArrival(receivedNs: frame.receivedNs)
                // Freeze-until-reanchor: WITHHOLD a decoder-concealed post-loss frame (the gray/
                // garbage VideoToolbox returns Ok for a reference-missing delta) — don't submit it,
                // so the CAMetalLayer keeps its last good drawable on glass. The gate lifts (returns
                // present) on a proven clean re-anchor (IDR / RFI anchor / 2nd recovery mark) or the
                // bounded backstop. decoderKeyframe=false: VT doesn't flag IDRs, the wire FLAG_SOF does.
                guard gate.onDecoded(flags: frame.flags) else { return }
                // Decoder OUTPUT is where the cadence loop is sampled — the instant the frame
                // becomes presentable. Receipt would not model decode at all and could hand back a
                // due time already past by the moment the frame exists; dequeue would fold the
                // present path's own wait into the estimate and make the loop chase its output.
                ring.submit(Stage2Pipeline.dated(frame, by: cadence, hint: rateHint))
                // FRAME ARRIVAL is the render trigger (never the display link — see the header).
                renderSignal.signal()
            },
            // Async decode failure (a bad P-frame referencing a lost/corrupt IDR): fold it into the
            // gate's no-output streak (which arms the freeze after a short run, matching the desktop),
            // and when that trips ask the host for a fresh IDR now (infinite GOP — it wouldn't
            // otherwise come soon). Throttled in KeyframeRecovery.
            onDecodeError: { _ in if gate.onNoOutput() { recovery.request() } })
    }

    /// Start pulling AUs into the decoder. MAIN THREAD. `onFrame` fires per AU at receipt (the
    /// host+network / capture→received meter, exactly as stage-1); `onSessionEnd` on close.
    /// `clockOffsetNs` (host minus client) makes the end-to-end stamp cross-machine valid.
    public func start(
        connection: PunktfunkConnection,
        onFrame: (@Sendable (AccessUnit) -> Void)?,
        onSessionEnd: (@Sendable () -> Void)?,
        onDecodedSize: (@Sendable (Int, Int) -> Void)? = nil
    ) {
        clockOffset = { connection.clockOffsetNs } // live (re-synced) — see the field doc
        recovery.bind(connection) // arm host-keyframe recovery for this session
        decodeReport.bind(connection) // arm the Automatic-bitrate decode signal for this session
        phaseReporter.bind(connection) // arm phase reports (flushed only by the deadline link)
        gate.reseed(framesDropped: connection.framesDropped()) // baseline the freeze to this session
        // A fresh session is a fresh source clock: re-anchor on its first frame rather than slew
        // for seconds off the previous host's offset. (Mid-session discontinuities — background
        // resume, a stream idle under the infinite GOP — arrive as a source-timestamp gap the loop
        // re-anchors on by itself; this seam covers the one it cannot see.)
        cadence?.reset()
        token = StopFlag() // fresh token per start — a stop is permanent (like StreamPump)

        // Configure the decoder's chroma + the layer's initial colorimetry before the first frame. The
        // chroma subsampling drives only the decode pixel format (orthogonal to HDR/depth); the HDR
        // config is the Welcome's latched value, which a mid-session flip then overrides per-frame.
        decoder.setChroma444(connection.isChroma444)
        decoder.setCodec(connection.videoCodec)
        presenter.configure(hdr: connection.isHDR)

        let token = token
        let decoder = decoder
        let recovery = recovery
        let presenter = presenter
        let pumpStopped = pumpStopped
        let reanchorGate = gate
        // PyroWave rides a different decode half: no CMFormatDescription/VideoToolbox machinery
        // (a wavelet AU has no parameter sets), no keyframe recovery or re-anchor freeze (the
        // stream is all-intra and Phase 4's partial delivery WANTS lossy frames on glass as
        // localized blur, not a freeze). The ready ring, render thread, pacing and meters are
        // shared unchanged.
        let thread: Thread
        if connection.videoCodec == .pyrowave {
            thread = Self.makePyroWavePump(
                connection: connection, token: token, pumpStopped: pumpStopped,
                ring: ring, renderSignal: renderSignal,
                device: presenter.metalDevice, queue: presenter.metalQueue,
                decodeMeter: decodeMeter, cadence: cadence, rateHint: frameRateHint,
                onFrame: onFrame, onSessionEnd: onSessionEnd, onDecodedSize: onDecodedSize)
        } else {
            thread = Thread {
            defer { pumpStopped.signal() } // let stop() join the pump (bounded) before decoder.reset()
            var format: CMVideoFormatDescription?
            // Report coded dims to the resize overlay only on a CHANGE (new-mode IDR), not per
            // loss-recovery IDR at the same size (see StreamPump).
            var lastDecodedDims: CMVideoDimensions?
            var lastFramesDropped = connection.framesDropped()
            // Persistent recovery WANT, not a one-shot edge (see StreamPump for the full rationale):
            // keep asking until an IDR lands so a request swallowed by the throttle is re-sent.
            var awaitingIDR = false
            // 4:4:4 backstop: a run of decode/create failures in a 4:4:4 session means this device can't
            // decode 4:4:4 at the negotiated resolution (the HW probe clears the common case but not a
            // resolution-ceiling miss). End cleanly instead of looping on a black screen.
            var decodeFailRun = 0
            // Every iteration drains its own autorelease pool: this thread has no runloop, so
            // autoreleased VT/CM temporaries would otherwise accumulate until session end.
            // `false` = session over — exit the loop (the closure can't `break` across itself).
            var alive = true
            while alive, !token.isStopped {
                alive = autoreleasepool { () -> Bool in
                do {
                    // Background keep-alive: drain one AU (flow control + host pacing) and discard it
                    // BEFORE any VideoToolbox decode or Metal render — no GPU work off-screen. The
                    // decoder session is left intact; exitBackground requests a fresh IDR and the
                    // re-anchor gate arms on the resumed frame-index gap so concealed frames are
                    // withheld until it lands.
                    if connection.isVideoDropped {
                        _ = try connection.nextAU(timeoutMs: 100)
                        return true
                    }
                    // Loss recovery (the primary path). The reassembler drops unrecoverable AUs and the
                    // decoder conceals the reference-missing deltas — often WITHOUT an error callback —
                    // so key off the drop count climbing, then keep asking (awaitingIDR) until a fresh
                    // IDR re-anchors decode.
                    let dropped = connection.framesDropped()
                    if dropped > lastFramesDropped {
                        lastFramesDropped = dropped
                        awaitingIDR = true
                    }
                    if awaitingIDR { recovery.request() }
                    // Freeze backstop: a drop-count climb arms the gate (in case the frame-index gap
                    // below was itself lost), and an overdue freeze re-asks for the re-anchor.
                    if reanchorGate.poll(framesDropped: dropped) { recovery.request() }
                    // Drain HDR mastering metadata (0xCE) and hand it to the PRESENTER (→ CAEDRMetadata).
                    // Polled UNCONDITIONALLY (not gated on connection.isHDR, the fixed Welcome flag): the
                    // host sends 0xCE only for HDR, INCLUDING a mid-session SDR→HDR transition (a game
                    // entering HDR — the host re-inits its encoder) the Welcome flag would never reflect.
                    // Non-blocking; nil for an SDR stream.
                    if let meta = try? connection.nextHdrMeta(timeoutMs: 0) {
                        presenter.setHdrMeta(meta)
                    }
                    guard let au = try connection.nextAU(timeoutMs: 100) else { return true }
                    // Loss recovery (RFI): a forward frame-index gap fires a throttled reference-
                    // frame-invalidation request so an RFI-capable host (AMD LTR / NVENC) recovers
                    // with a cheap clean P-frame instead of a full IDR. The framesDropped-driven
                    // recovery above stays the backstop for when the recovery frame itself is lost.
                    // The same gap is the earliest, most precise signal to ARM the display freeze —
                    // the following concealed frames are withheld until a clean re-anchor.
                    // Credited arm: the gap width pre-covers the reassembler's ~120 ms-later
                    // framesDropped climb for the same loss, so a fast RFI anchor that heals in
                    // between isn't re-frozen by it (the double-arm race).
                    let gapWidth = connection.noteFrameIndexGapWidth(au.frameIndex)
                    if gapWidth > 0 { reanchorGate.arm(expectingDrops: UInt64(gapWidth)) }
                    onFrame?(au)
                    if let f = connection.videoCodec.formatDescription(fromKeyframe: au.data) {
                        format = f          // refreshed on every IDR (mode changes included)
                        let dims = CMVideoFormatDescriptionGetDimensions(f)
                        if lastDecodedDims?.width != dims.width || lastDecodedDims?.height != dims.height {
                            lastDecodedDims = dims
                            onDecodedSize?(Int(dims.width), Int(dims.height))
                        }
                        awaitingIDR = false // a fresh IDR re-anchored decode — recovery complete
                    }
                    if format == nil {
                        // No decodable format yet: the opening IDR's parameter sets never
                        // arrived (or never parsed), and under the host's infinite GOP nothing
                        // re-delivers them unless we ASK. Without this the guard below drops
                        // every AU silently, forever — the field "black stream, zero recovery
                        // requests" state (2026-08-12): the host streams perfectly, the client
                        // shows nothing and says nothing. awaitingIDR routes through the same
                        // 100 ms-throttled recovery.request() at the top of the loop.
                        if !awaitingIDR {
                            pumpLog.warning(
                                "video: received AUs but no decodable format (missing/unparsed parameter sets) — requesting an IDR until one seeds it"
                            )
                        }
                        awaitingIDR = true
                    }
                    guard let f = format, !token.isStopped else { return true }
                    if decoder.decode(au: au, format: f) {
                        decodeFailRun = 0
                    } else {
                        // Submit/decoder error: drop the session and re-gate on the next IDR's in-band
                        // parameter sets (a delta frame can't recover) and keep asking for that IDR.
                        decoder.reset()
                        awaitingIDR = true
                        decodeFailRun += 1
                        // ~3 s of solid failure in a 4:4:4 session (and only there — a 4:2:0 loss
                        // recovers within a GOP) ⇒ 4:4:4 isn't decodable here; end the session.
                        if connection.isChroma444, decodeFailRun >= 180 {
                            if !token.isStopped { onSessionEnd?() }
                            return false
                        }
                    }
                    return true
                } catch {
                    if !token.isStopped { onSessionEnd?() }
                    return false // session closed
                }
                }
            }
            }
        }
        thread.name = "punktfunk-stage2-pump"
        thread.qualityOfService = .userInteractive
        pumpJoinable = true
        thread.start()

        // The present half. Deadline pacing (stage-4) swaps it wholesale: a CAMetalDisplayLink
        // vends the drawables and its per-refresh updates co-drive the render thread — see
        // startDeadlinePresenter. The V-Sync policy below doesn't apply there (the link deadline-
        // times every present). Deadline sessions ALWAYS carry the stats (their pf-present line
        // streams to Console.app via presentLog — the on-device pacing decomposition).
        let debugStats =
            (presentDebug || pacing == .deadline) ? PresentDebugStats(cadence: cadence) : nil
        if pacing == .deadline {
            startDeadlinePresenter(debugStats: debugStats)
            return
        }

        // The render thread: one present per display-link signal. It owns every layer format/colour/
        // drawable interaction (see MetalVideoPresenter's threading notes); with displaySyncEnabled on,
        // nextDrawable's up-to-a-frame wait lands here instead of on main. The 100 ms timed wait is
        // only the stop-flag poll for a session whose link stopped ticking.
        let ring = ring
        let endToEndMeter = endToEndMeter
        let displayMeter = displayMeter
        let clockOffset = clockOffset
        let renderSignal = renderSignal
        let renderStopped = renderStopped
        // Present policy — the user's V-Sync setting (default OFF = immediate, the long-proven
        // lowest-latency behavior); PUNKTFUNK_PRESENT_MODE=immediate|vsync overrides it for A/B.
        // Resolved once per session.
        let presentMode = ProcessInfo.processInfo.environment["PUNKTFUNK_PRESENT_MODE"]
        // `vsyncPaced` (macOS smoothness) FORCES vsync scheduling — the FIFO store must drain
        // on the display cadence, one frame per vsync, or the buffer degenerates to arrival.
        let vsyncPaced = vsyncPaced
        let vsyncEnabled = vsyncPaced || presentMode == "vsync"
            || (presentMode != "immediate"
                && SessionSettings.current.vsync)
        let vsyncClock = vsyncClock
        // Stage-3's bounded in-flight present gate; nil = stage-2's present-on-arrival. A local
        // (like the ring) so neither the render thread nor the presented handlers capture `self`.
        let gate: PresentGate? = pacing == .glass ? PresentGate(capacity: gateDepth) : nil
        // Cadence targeting turns the store into a holding buffer: a frame comes out once it is
        // DUE, not once a present opportunity exists (§4.3). The latency intent has no clock and
        // keeps the unconditional take, byte for byte.
        let takeReady: () -> ReadyFrame? = cadence == nil
            ? { ring.take() }
            : { ring.take(dueBy: CACurrentMediaTime(), due: { $0.dueMediaTime }) }
        let renderThread = Thread {
            defer { renderStopped.signal() }
            // macOS smoothness: the vsync this thread last presented onto — at most ONE present
            // per vsync so the FIFO drains on the display's cadence. Thread-confined.
            var lastPresentTarget: CFTimeInterval = 0
            // Every iteration drains its own autorelease pool (`return` = the old `continue`):
            // this thread has no runloop, and `nextDrawable()` AUTORELEASES each CAMetalDrawable —
            // without a per-iteration pool every presented frame's drawable object (plus its
            // texture-descriptor/array retinue, ~2 MB/min at 120 fps) piles up until session end.
            while !token.isStopped { autoreleasepool {
                if renderSignal.wait(timeout: .now() + .milliseconds(100)) == .timedOut {
                    debugStats?.flushIfDue(ring: ring, gate: gate)
                    return
                }
                // Smoothness pacing: this vsync's present slot already taken — the frame stays
                // in the store, and the next display-link tick re-signals. (Tolerance well under
                // any refresh period; a stale clock ⇒ nil target ⇒ no dedup, present flows.)
                if vsyncPaced, let t = vsyncClock.nextVsync(after: CACurrentMediaTime()),
                   abs(t - lastPresentTarget) < 0.002 {
                    debugStats?.gatedWake()
                    debugStats?.flushIfDue(ring: ring, gate: gate)
                    return
                }
                // Stage-3: while a present is in flight, don't take from the ring at all — frames
                // keep coalescing there (newest wins, the intended drop point) and the presented
                // handler re-signals the moment the slot frees. Checked BEFORE the take so a gated
                // frame is never bounced through putBack.
                if let gate, !gate.tryAcquire(now: CACurrentMediaTime()) {
                    debugStats?.gatedWake()
                    debugStats?.flushIfDue(ring: ring, gate: gate)
                    return
                }
                guard !token.isStopped, let frame = takeReady() else {
                    gate?.release() // armed but nothing to render — don't hold the gate stale
                    debugStats?.emptyWake()
                    debugStats?.flushIfDue(ring: ring, gate: gate)
                    return
                }
                // V-Sync ON: flip on the next predicted vsync (< one period out, stale link ⇒
                // immediate — see VsyncClock). OFF: flip as soon as the GPU finishes.
                //
                // Under cadence targeting the grid is entered at the frame's DUE time rather than
                // at this instant, so two frames the host emitted one period apart land one period
                // apart on glass however unevenly they arrived. Never before `now`: a due time in
                // the past means the frame is late, not that the grid moves back.
                let now = CACurrentMediaTime()
                let presentAt = vsyncEnabled
                    ? vsyncClock.nextVsync(after: max(now, frame.dueMediaTime ?? now)) : nil
                let renderStarted = CACurrentMediaTime()
                let issuedNs = Stage2Pipeline.realtimeNs(forDisplayLinkTimestamp: renderStarted)
                let onGlass: (Int64?) -> Void = { presentedNs in
                    // Stage-3: the flip reached glass (or was dropped) — free the present slot,
                    // then re-signal so the freshest waiting ring frame goes out immediately.
                    if let gate {
                        gate.release()
                        renderSignal.signal()
                    }
                    // Fallback stamp for a dropped drawable (no system presentedTime): "now" on
                    // the Metal callback, converted to the CLOCK_REALTIME the meters live in.
                    let atNs = presentedNs
                        ?? Stage2Pipeline.realtimeNs(forDisplayLinkTimestamp: CACurrentMediaTime())
                    // End-to-end = capture→on-glass, measured directly (skew-corrected via the
                    // connect-time clock offset) — the HUD headline.
                    endToEndMeter?.record(ptsNs: frame.ptsNs, atNs: atNs, offsetNs: clockOffset())
                    // Display stage = decoded → on-glass. Both instants are client CLOCK_REALTIME,
                    // so no skew offset applies.
                    displayMeter?.record(ptsNs: UInt64(frame.decodedNs), atNs: atNs, offsetNs: 0)
                    debugStats?.presented(atNs: presentedNs, issuedNs: issuedNs)
                }
                // One present tail, two decode sources: the VideoToolbox biplanar buffer or the
                // PyroWave Metal planes — the ring, pacing and meters are agnostic to which.
                let rendered: Bool
                switch frame.image {
                case .video(let pixelBuffer, let isHDR):
                    rendered = presenter.render(
                        pixelBuffer, isHDR: isHDR, presentAtMediaTime: presentAt,
                        onPresented: onGlass)
                case .planar(let planes):
                    rendered = presenter.renderPlanar(
                        planes, presentAtMediaTime: presentAt, onPresented: onGlass)
                }
                debugStats?.renderReturned(
                    ok: rendered, tookMs: (CACurrentMediaTime() - renderStarted) * 1000)
                if !rendered {
                    gate?.release() // no present registered — its handler will never fire
                    ring.putBack(frame)
                } else if vsyncPaced, let presentAt {
                    lastPresentTarget = presentAt // this vsync's slot is now taken
                }
                debugStats?.flushIfDue(ring: ring, gate: gate)
            } }
        }
        renderThread.name = "punktfunk-stage2-render"
        renderThread.qualityOfService = .userInteractive
        renderJoinable = true
        renderThread.start()
    }

    /// Deadline pacing's present half (stage-4 — see `PresentPacing.deadline`): a
    /// CAMetalDisplayLink on its own runloop thread vends ONE drawable per refresh into the
    /// newest-wins stash, and the render thread pairs it with the newest decoded frame the
    /// moment either half completes the pair — the common case is a decoded frame presenting
    /// instantly into an already-vended drawable, which the system then latches at the upcoming
    /// refresh (`preferredFrameLatency` 1). No image queue can form (one vended drawable in
    /// flight, ever) and nothing serializes on the on-glass callback. An unpresented stashed
    /// drawable is simply replaced by the next update (back to the layer's pool), so the stash
    /// is never stale by more than a refresh while the link runs.
    ///
    /// Threading mirrors the arrival/glass half: neither thread captures `self`; the link is
    /// created, driven and invalidated entirely on its own thread (CAMetalDisplayLink is only
    /// ever touched there — the frame-rate hint crosses via `FrameRateHint`); the link thread's
    /// runloop iterations each drain an autorelease pool (a vended CAMetalDrawable is
    /// autoreleased like a `nextDrawable()` one — see the render loop's identical rule); the
    /// 100 ms runloop horizon is the stop-flag poll, so teardown is bounded without a join.
    private func startDeadlinePresenter(debugStats: PresentDebugStats?) {
        let token = token
        let ring = ring
        let renderSignal = renderSignal
        let renderStopped = renderStopped
        let presenter = presenter
        let endToEndMeter = endToEndMeter
        let displayMeter = displayMeter
        let clockOffset = clockOffset
        let hint = frameRateHint
        let layer = presenter.layer
        let stash = LatestBox<CAMetalDrawable>()
        // Cadence targeting under deadline pacing: the link's vend IS the grid snap, so the clock
        // only has to hold a frame back until it is due and the next update presents it — at most
        // one refresh later. Same holding-buffer rule as the arrival/glass loop (§4.3); latency
        // sessions have no clock and take unconditionally.
        let takeReady: () -> ReadyFrame? = cadence == nil
            ? { ring.take() }
            : { ring.take(dueBy: CACurrentMediaTime(), due: { $0.dueMediaTime }) }

        // ⭐ Shrink the drawable pool to 2 for THIS pacing — the measured fix for a present floor
        // stuck at two refreshes (field 2026-08-13, Apple TV 4K / tvOS 27: `os present +32.5` at
        // 60 Hz = 1.95 × 16.67, i.e. the system running a whole frame ahead of us).
        //
        // `maximumDrawableCount` is 3 from MetalVideoPresenter.make(), and its rationale there —
        // "more in-flight drawables before nextDrawable() has to block" — is a STAGE-2 concern.
        // Stage-4 never calls nextDrawable(): every drawable is vended by the link
        // (`update.drawable` → stash → `render(into:)`), so the third slot buys this path nothing
        // and costs it a refresh — a pool of 3 is exactly the room the compositor needs to keep
        // two presents queued ahead of scanout, which is what `preferredFrameLatency = 1` is
        // asking it not to do. Two slots is the shallowest pool that still double-buffers: one
        // vended (stashed or being rendered), one being scanned out.
        //
        // Set HERE, not on the link thread: this runs before either the render thread or the link
        // thread exists, so the layer still has a single writer (the render thread owns
        // drawableSize/format afterwards — see MetalVideoPresenter's threading notes).
        // PUNKTFUNK_DRAWABLE_COUNT=3 restores the old depth for an on-glass A/B without a
        // rebuild; values outside 2...3 are ignored (CAMetalLayer's own accepted range).
        let drawableCount =
            ProcessInfo.processInfo.environment["PUNKTFUNK_DRAWABLE_COUNT"]
                .flatMap(Int.init)
                .flatMap { (2...3).contains($0) ? $0 : nil } ?? 2
        layer.maximumDrawableCount = drawableCount

        // The frame-latency ASK (default 1 — wake as late as fits: latch the NEXT refresh).
        // PUNKTFUNK_FRAME_LATENCY overrides it for the on-device ladder. The property is a
        // FLOAT, so sub-frame asks (0.5) are expressible; whether the scheduler honours them —
        // or reacts to the property at all — is exactly what the ladder measures. Field
        // 2026-08-13 (Apple TV 4K, tvOS 27): ask 1 → vend lead 1.95 refresh periods, and the
        // readback echoed the ask throughout (it is a plain property — see PresentLinkInfo).
        // The discriminating runs, watching `os present` (the vend lead), are:
        //   ask=2   → lead grows to ~3 ⇒ the property WORKS and the tvOS floor is ~ask+1;
        //             lead stays ~2 ⇒ the property is INERT here — stop pulling this lever.
        //   ask=0.5 → any lead below ~1.9 ⇒ a real in-regime win to then tune.
        // Clamped to 0...4: negatives/NaN are meaningless, and beyond 4 asked-for frames of
        // latency nothing is being measured.
        let latencyAsk =
            ProcessInfo.processInfo.environment["PUNKTFUNK_FRAME_LATENCY"]
                .flatMap(Float.init)
                .flatMap { $0.isFinite ? min(max($0, 0), 4) : nil } ?? 1

        let floorMeter = presentFloorMeter
        let phaseReporter = phaseReporter
        // The link starts LAZILY — the render thread triggers this after the FIRST decoded
        // frame's reconcileLayer. Started eagerly it vends into the layer's initial 0×0
        // drawableSize for the whole connect window: every vend fails allocation and the system
        // logs "[CAMetalLayer nextDrawable] returning nil because allocation failed" once per
        // refresh until the first frame arrives. Before that frame there is nothing to present
        // anyway, and the first frame waits at most one refresh for the first vend.
        let startLink: () -> Void = {
            let linkThread = Thread {
                let delegate = DeadlineLinkDelegate(
                    stash: stash, renderSignal: renderSignal, hint: hint, stats: debugStats,
                    floorMeter: floorMeter, phase: phaseReporter,
                    drawableCount: drawableCount, latencyAsk: latencyAsk)
                let link = CAMetalDisplayLink(metalLayer: layer)
                link.preferredFrameLatency = latencyAsk // see the ladder note above
                if let range = hint.drain() { link.preferredFrameRateRange = range }
                link.delegate = delegate // weak — this closure is the strong ref
                link.add(to: RunLoop.current, forMode: .default)
                while !token.isStopped {
                    autoreleasepool {
                        _ = RunLoop.current.run(
                            mode: .default, before: Date(timeIntervalSinceNow: 0.1))
                    }
                }
                link.invalidate()
            }
            linkThread.name = "punktfunk-stage4-link"
            linkThread.qualityOfService = .userInteractive
            linkThread.start()
        }

        let renderThread = Thread {
            defer { renderStopped.signal() }
            // Whether startLink ran — render-thread confined (only this thread triggers it).
            var linkLive = false
            // Per-iteration autorelease pool — same contract as the arrival/glass loop (the
            // vended drawable and its retinue are autoreleased objects on a runloop-less thread).
            while !token.isStopped { autoreleasepool {
                if renderSignal.wait(timeout: .now() + .milliseconds(100)) == .timedOut {
                    debugStats?.flushIfDue(ring: ring, gate: nil)
                    return
                }
                // Present needs the PAIR — frame first. The frame drives the layer reconcile,
                // which must run even when NO drawable is vended yet: the link vends from the
                // layer's CURRENT config, so drawableSize/format have to be right before a vend
                // can succeed at all (see reconcileLayer — the session-start bootstrap, where
                // the layer still has its initial 0×0 size and every vend fails allocation).
                guard !token.isStopped, let frame = takeReady() else {
                    debugStats?.emptyWake()
                    debugStats?.flushIfDue(ring: ring, gate: nil)
                    return
                }
                switch frame.image {
                case .video(let pixelBuffer, let isHDR):
                    presenter.reconcileLayer(
                        decodedSize: CGSize(
                            width: CVPixelBufferGetWidth(pixelBuffer),
                            height: CVPixelBufferGetHeight(pixelBuffer)),
                        isHDR: isHDR)
                case .planar(let planes):
                    presenter.reconcileLayer(
                        decodedSize: CGSize(width: planes.width, height: planes.height),
                        isHDR: planes.pq)
                }
                // First frame: the layer now has a real config — start vending (see startLink).
                if !linkLive {
                    linkLive = true
                    startLink()
                }
                guard let drawable = stash.take() else {
                    // No vend yet (session start: the reconcile above just unblocked the
                    // allocator, the link's next update delivers; steady state: decode beat the
                    // link's phase). putBack keeps newest-wins — a fresher decode replaces this
                    // frame while it waits, and the update's signal retries the pairing.
                    ring.putBack(frame)
                    debugStats?.noDrawableWake()
                    debugStats?.flushIfDue(ring: ring, gate: nil)
                    return
                }
                let renderStarted = CACurrentMediaTime()
                let issuedNs = Stage2Pipeline.realtimeNs(forDisplayLinkTimestamp: renderStarted)
                let onGlass: (Int64?) -> Void = { presentedNs in
                    let atNs = presentedNs
                        ?? Stage2Pipeline.realtimeNs(forDisplayLinkTimestamp: CACurrentMediaTime())
                    endToEndMeter?.record(ptsNs: frame.ptsNs, atNs: atNs, offsetNs: clockOffset())
                    displayMeter?.record(ptsNs: UInt64(frame.decodedNs), atNs: atNs, offsetNs: 0)
                    debugStats?.presented(atNs: presentedNs, issuedNs: issuedNs)
                }
                let rendered: Bool
                switch frame.image {
                case .video(let pixelBuffer, let isHDR):
                    rendered = presenter.render(
                        pixelBuffer, isHDR: isHDR, into: drawable, onPresented: onGlass)
                case .planar(let planes):
                    rendered = presenter.renderPlanar(
                        planes, into: drawable, onPresented: onGlass)
                }
                debugStats?.renderReturned(
                    ok: rendered, tookMs: (CACurrentMediaTime() - renderStarted) * 1000)
                if !rendered {
                    // The vended drawable is spent either way (an unused/mismatched one drops
                    // back to the pool); the frame retries on the link's next vend. A format
                    // mismatch (mid-session HDR flip caught between the layer reconfigure and
                    // the next vend) self-heals the same way — see encodePresent's guard.
                    ring.putBack(frame)
                }
                debugStats?.flushIfDue(ring: ring, gate: nil)
            } }
        }
        renderThread.name = "punktfunk-stage2-render"
        renderThread.qualityOfService = .userInteractive
        renderJoinable = true
        renderThread.start()
    }

    /// MAIN thread, once per display-link tick: refresh the vsync clock (V-Sync-mode scheduling)
    /// and nudge the render thread. The nudge is NOT the presentation trigger — frame arrival is
    /// (see the header) — it only retries a frame a transient `nextDrawable` failure put back into
    /// the ring, which matters under the host's infinite GOP where a static scene sends no
    /// replacement frame. Arrival/glass pacing only — deadline sessions have no CADisplayLink
    /// (their CAMetalDisplayLink's updates are both clock and retry).
    public func renderTick(targetMediaTime: CFTimeInterval, period: CFTimeInterval) {
        vsyncClock.set(target: targetMediaTime, period: period)
        renderSignal.signal()
    }

    /// MAIN thread (SessionPresenter — session start + every layout/Reconfigure): hint the
    /// deadline link with the stream cadence. Staged; the link's own thread applies it (see
    /// `FrameRateHint`). Under arrival/glass pacing no link reads it — the hosting view's
    /// CADisplayLink is the hinted one there — but the stored rate is still the cadence clock's
    /// nominal source interval, and hence its cushion ceiling, on every pacing.
    public func setFrameRateHint(hz: Float) {
        frameRateHint.stage(hz: hz)
    }

    /// Pen-proximity rate boost (drawing workloads): drive the deadline link — and with it the
    /// panel, whose cadence paces UIKit's touch/Pencil event delivery — at the range ceiling
    /// while a Pencil is in range, so a sub-panel-rate stream stops halving pencil sampling.
    /// Staged like the rate hint; no-op under arrival/glass pacing. MAIN thread.
    public func setInteractionBoost(_ on: Bool) {
        frameRateHint.setBoost(on)
        presentLog.info("pen boost \(on ? "engaged" : "released", privacy: .public)")
    }

    /// Forward the layout-derived drawable pixel size to the presenter (MAIN thread — see
    /// `MetalVideoPresenter.setDrawableTarget`).
    public func setDrawableTarget(_ size: CGSize) {
        presenter.setDrawableTarget(size)
    }

    #if os(macOS)
    /// Forward the windowed present mechanism (MAIN thread — see
    /// `MetalVideoPresenter.setWindowedPresent`, the DCP swapID-panic mitigation).
    func setWindowedPresent(_ mode: WindowedPresentMode) {
        presenter.setWindowedPresent(mode)
    }

    /// The windowed `surface` present target the hosting SessionPresenter installs as a sibling
    /// ABOVE `layer` (transparent while unused — see `MetalVideoPresenter.surfaceLayer`).
    var surfaceLayer: CALayer { presenter.surfaceLayer }
    #endif

    /// Forward the display's current EDR headroom to the presenter (MAIN thread — a `UIScreen`
    /// read). tvOS flips HDR presentation between PQ passthrough and the in-shader tone-map on
    /// it; see `MetalVideoPresenter.setDisplayHeadroom`.
    public func setDisplayHeadroom(_ headroom: CGFloat) {
        presenter.setDisplayHeadroom(headroom)
    }

    /// Stop the pump + render thread (≤ one poll timeout each) and drop the decode session. MAIN
    /// THREAD; idempotent. Does not close the connection. A restart needs a fresh Stage2Pipeline
    /// (the stop is permanent).
    public func stop() {
        token.stop()
        // Join the pump (bounded: ≤ one nextAU poll + an in-flight decode) before resetting the decoder,
        // so the pump can't rebuild a session right after the reset. Only the first stop joins; a
        // repeat/deinit stop skips the already-drained semaphore.
        if pumpJoinable {
            pumpJoinable = false
            _ = pumpStopped.wait(timeout: .now() + 0.5)
        }
        // Wake + join the render thread (bounded: it may sit in `nextDrawable` for up to ~a frame; a
        // timed-out join is fine — the loop exits at its next stop-flag check, and a final present on
        // the detached layer is harmless).
        if renderJoinable {
            renderJoinable = false
            renderSignal.signal()
            _ = renderStopped.wait(timeout: .now() + 0.5)
        }
        decoder.reset()
        recovery.bind(nil) // stop requesting keyframes once the session is torn down
        phaseReporter.bind(nil) // and stop phase reports toward the dead connection
    }

    deinit {
        token.stop()
        renderSignal.signal() // wake the render thread so it can observe the stop and exit
    }

    /// The PyroWave pump: AUs go straight into the Metal wavelet decoder (no VideoToolbox, no
    /// format descriptions), decoded planes ride the same ready ring / render thread. All-intra
    /// stream, so none of the VT pump's recovery machinery applies: keyframe/RFI requests are
    /// silenced host-side for this codec, and a lossy (partial-delivery) frame is MEANT to
    /// present as localized blur — never a freeze. Static + capture-by-parameter for the same
    /// reason the VT pump avoids capturing `self` (a missed stop must not leak a live pipeline).
    private static func makePyroWavePump(
        connection: PunktfunkConnection, token: StopFlag, pumpStopped: DispatchSemaphore,
        ring: FrameStore<ReadyFrame>, renderSignal: DispatchSemaphore,
        device: MTLDevice, queue: MTLCommandQueue,
        decodeMeter: LatencyMeter?, cadence: CadenceClock?, rateHint: FrameRateHint,
        onFrame: (@Sendable (AccessUnit) -> Void)?,
        onSessionEnd: (@Sendable () -> Void)?,
        onDecodedSize: (@Sendable (Int, Int) -> Void)?
    ) -> Thread {
        // The chunk-aligned parse window = the session's negotiated shard payload (Welcome);
        // the 64-byte floor mirrors the Rust client's guard against a nonsense value.
        let windowSize = max(64, Int(connection.shardPayload))
        return Thread {
            defer { pumpStopped.signal() }
            // Compiles the two compute kernels on the session's first frames' thread — ~tens of
            // ms, once per session. Failure = this device can't run the negotiated codec (the
            // advertisement probe should have prevented this); end the session cleanly.
            guard let decoder = MetalWaveletDecoder(device: device, queue: queue) else {
                if !token.isStopped { onSessionEnd?() }
                return
            }
            // Newest decoded frame index — a late partial (the reassembler's 30 ms fuse can
            // deliver one behind a newer complete frame) must not travel back in time.
            var newestIndex: UInt32?
            var lastDims: (w: Int, h: Int)?
            var alive = true
            while alive, !token.isStopped {
                alive = autoreleasepool { () -> Bool in
                    do {
                        // Background keep-alive: drain + discard before the Metal wavelet decode
                        // (PyroWave is all-intra, so the resumed frame heals on its own — no IDR
                        // request needed, just no GPU work off-screen).
                        if connection.isVideoDropped {
                            _ = try connection.nextAU(timeoutMs: 100)
                            return true
                        }
                        guard let au = try connection.nextAU(timeoutMs: 100) else { return true }
                        onFrame?(au)
                        if let newest = newestIndex,
                           Int32(bitPattern: au.frameIndex &- newest) <= 0 {
                            return true // stale (or duplicate) frame — skip
                        }
                        guard !token.isStopped else { return true }
                        let chunkAligned =
                            au.flags & PunktfunkConnection.userFlagChunkAligned != 0
                        let ptsNs = au.ptsNs
                        // Decode stage starts at the PULL (matching the VT path's FrameContext —
                        // receipt→pull is the HUD's separate client-queue term, ABI v9 split).
                        let receivedNs = au.pulledNs
                        let flags = au.flags
                        let submitted = decoder.decode(
                            au: au.data, chunkAligned: chunkAligned, windowSize: windowSize
                        ) { planes in
                            // Metal completed-handler thread — stamp + enqueue, don't block
                            // (the exact contract of the VT output callback).
                            guard let planes else { return }
                            var ts = timespec()
                            clock_gettime(CLOCK_REALTIME, &ts)
                            let decodedNs =
                                Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec)
                            decodeMeter?.record(
                                ptsNs: UInt64(receivedNs), atNs: decodedNs, offsetNs: 0)
                            // Same cadence sample as the VideoToolbox half: the wavelet decode's
                            // completion IS this frame's presentable instant.
                            ring.submit(
                                Stage2Pipeline.dated(
                                    ReadyFrame(
                                        ptsNs: ptsNs, receivedNs: receivedNs,
                                        decodedNs: decodedNs, image: .planar(planes),
                                        flags: flags),
                                    by: cadence, hint: rateHint))
                            renderSignal.signal()
                        }
                        if submitted {
                            newestIndex = au.frameIndex
                            // Decoded-size changes come from the SOF dims (this is also how a
                            // mid-stream Reconfigure lands here) — report like the VT pump.
                            if let size = decoder.decodedSize,
                               lastDims?.w != size.width || lastDims?.h != size.height {
                                lastDims = (size.width, size.height)
                                onDecodedSize?(size.width, size.height)
                            }
                        }
                        // A dropped AU (malformed / SOF lost / too few blocks) is just skipped:
                        // every PyroWave frame is independently decodable, the next one heals.
                        return true
                    } catch {
                        if !token.isStopped { onSessionEnd?() }
                        return false // session closed
                    }
                }
            }
        }
    }

    /// Convert a `CADisplayLink.targetTimestamp` (CACurrentMediaTime basis) to a `CLOCK_REALTIME`
    /// nanosecond instant — the present clock the AU pts + skew offset live in. Projects to the target
    /// present time (when the frame is actually on glass), not the moment we drew.
    public static func realtimeNs(forDisplayLinkTimestamp t: CFTimeInterval) -> Int64 {
        let caNow = CACurrentMediaTime()
        var ts = timespec()
        clock_gettime(CLOCK_REALTIME, &ts)
        let realtimeNow = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec)
        return realtimeNow + Int64((t - caNow) * 1_000_000_000)
    }

    /// The exact inverse: a client `CLOCK_REALTIME` nanosecond instant (`ReadyFrame.decodedNs`)
    /// expressed on the `CACurrentMediaTime` timeline the present path schedules against.
    ///
    /// It reads the two clocks in the SAME ORDER as `realtimeNs(forDisplayLinkTimestamp:)` above
    /// and forms the same difference, so the sub-microsecond skew between the two reads is the
    /// same sign in both and cancels on a round trip.
    ///
    /// The cadence loop needs this because its rule is one domain in, SAME domain out: it is fed
    /// the decode-output instant in media time and its due time comes back in media time, with no
    /// second conversion anywhere downstream. (A constant realtime↔media offset would be absorbed
    /// by the loop's own offset estimator and need no conversion at all — but the two clocks
    /// diverge across device sleep, which is exactly why the conversion is done per frame here
    /// rather than once per session.)
    static func mediaTimeNs(forRealtimeNs t: Int64) -> Int64 {
        let caNow = CACurrentMediaTime()
        var ts = timespec()
        clock_gettime(CLOCK_REALTIME, &ts)
        let realtimeNow = Int64(ts.tv_sec) * 1_000_000_000 + Int64(ts.tv_nsec)
        return Int64(caNow * 1_000_000_000) + (t - realtimeNow)
    }

    /// Stamp a decoded frame with when it is DUE on the source's cadence, at the moment it enters
    /// the ready store. Returns the frame untouched when the session has no clock (the latency
    /// intent).
    ///
    /// A frame whose wire pts did not survive (`ptsNs == 0` — the decoder's "unknown" value) is
    /// not on the source's timeline at all, so it is folded through `noteOffCadence`: due as soon
    /// as it is ready, and the estimate left alone. Folding "now" in would drag the offset toward
    /// this instant precisely when the loop has the least evidence.
    private static func dated(
        _ frame: ReadyFrame, by clock: CadenceClock?, hint: FrameRateHint
    ) -> ReadyFrame {
        guard let clock else { return frame }
        let readyNs = mediaTimeNs(forRealtimeNs: frame.decodedNs)
        let interval = hint.sourceIntervalNs()
        let dueNs =
            frame.ptsNs > 0
            ? clock.dueNs(
                srcPtsNs: frame.ptsNs, readyNs: readyNs, frameIntervalNs: interval)
            : clock.noteOffCadence(readyNs: readyNs, frameIntervalNs: interval)
        var dated = frame
        dated.dueMediaTime = Double(dueNs) / 1_000_000_000
        return dated
    }
}
#endif
