fix(client/net): split the receipt stamp from the pull + bleed standing latency

The two-pair investigation (wired Mac clients stuck at a rock-steady
~18-19 ms "network" that survived the load ending and cleared only on
reconnect) exposed two structural gaps, one of measurement and one of
recovery:

- Receipt was stamped at the hand-off PULL (Swift nextAU, pf-client-core,
  Android decode loops), not at reassembly completion — so any client-side
  standing state between the reassembler and the pull read as NETWORK
  latency, undiagnosable from the HUD. ABI v9: `PunktfunkFrame`/`Frame`
  grow `received_ns`, stamped by `Session::poll_frame` as the AU crosses
  the session boundary. Every embedder now uses the core stamp; the Apple
  client keeps the pull instant as `AccessUnit.pulledNs` and shows the
  receipt→pull wait as its own "client queue" term (detailed HUD tier from
  2 ms + a `queue_p50` stats-log field). Decode stages keep their pull
  anchor on all platforms, so no historical stage shifts meaning.

- The jump-to-live detectors deliberately ignore anything under 6 queued
  frames / 400 ms behind — so a small, constant, loss-free elevation (a
  sub-frame standing backlog, or a stale clock offset after a wall-clock
  step/slew) is carried for the rest of the session. New third detector
  (`StandingLatency`, unit-tested ladder): window-MIN one-way delay
  ≥ 10 ms above the session floor with zero loss for ~4.5 s escalates
  gently — a free clock re-sync first (an applied re-sync re-bases the
  floor), then at most 3 flush+keyframe bleeds sharing the jump-to-live
  cooldown, then a loud disarm naming what it means. Loss windows reset
  the run: congestion belongs to FEC/ABR, not this detector.

Also: mid-stream re-sync apply/discard logs debug→info — they are the
forensic trail for the stale-offset case and were invisible in the field.

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

crates/punktfunk-core/src/abi.rs

@@ -125,6 +125,12 @@ pub struct PunktfunkFrame {
     pub frame_index: u32,
     pub pts_ns: u64,
     pub flags: u32,
+    /// Wall-clock reassembly-completion instant (ns since the Unix epoch, CLOCK_REALTIME — the
+    /// clock `pts_ns` and the skew handshake use). THIS is the receipt stamp for latency math:
+    /// a stamp the embedder takes itself at the poll return additionally contains the
+    /// pre-decode hand-off queue wait, so a client-side standing backlog would masquerade as
+    /// network latency (ABI v9 — the 2026-07 two-pair standing-latency investigation).
+    pub received_ns: u64,
 }
 
 /// Snapshot of session counters.
@@ -391,6 +397,7 @@ pub unsafe extern "C" fn punktfunk_client_poll_frame(
                         frame_index: f.frame_index,
                         pts_ns: f.pts_ns,
                         flags: f.flags,
+                        received_ns: f.received_ns,
                     };
                 }
                 PunktfunkStatus::Ok
@@ -1744,6 +1751,7 @@ pub unsafe extern "C" fn punktfunk_connection_next_au(
                         frame_index: f.frame_index,
                         pts_ns: f.pts_ns,
                         flags: f.flags,
+                        received_ns: f.received_ns,
                     };
                 }
                 PunktfunkStatus::Ok

crates/punktfunk-core/src/client/frame_channel.rs

@@ -73,6 +73,142 @@ pub(crate) const NOOP_CLOCK_FLUSHES_TO_DISARM: u32 = 2;
 /// FIRST no-op clock flush — the moment a step is actually suspected.
 pub(crate) const CLOCK_RESYNC_INTERVAL: Duration = Duration::from_secs(60);
 
+/// Standing-latency bleed (the 2026-07 two-pair investigation): how far above the session's own
+/// one-way-delay floor a report window's MINIMUM must sit to count as a standing elevation. The
+/// jump-to-live detectors above deliberately ignore anything below ~6 frames / 400 ms, so a
+/// small standing state — a sub-frame kernel/reassembly backlog, or a stale clock offset after a
+/// wall-clock step — is carried forever and reads as permanent extra "network" latency. 10 ms
+/// sits above skew-handshake error + normal LAN jitter, and below a single 60 fps frame period,
+/// so the observed one-frame plateau (~17 ms) trips it while a healthy stream cannot.
+pub(crate) const STANDING_LAT_THRESH_NS: i128 = 10_000_000;
+
+/// Consecutive elevated report windows (~750 ms each) before the bleed escalates — ~4.5 s of a
+/// continuously standing, loss-free elevation. Windows with any loss reset the run: loss means
+/// genuine congestion, which the FEC/ABR machinery owns, not this detector.
+pub(crate) const STANDING_LAT_WINDOWS: u32 = 6;
+
+/// Per-session cap on flush+keyframe bleeds. A standing state that survives a clock re-sync AND
+/// this many local flushes is not local and not clock — the path latency itself changed; the
+/// detector disarms with a warning instead of paying a recovery keyframe every few seconds.
+pub(crate) const STANDING_LAT_MAX_BLEEDS: u32 = 3;
+
+/// What the standing-latency detector asks the pump to do this window (see [`StandingLatency`]).
+#[derive(Debug, PartialEq, Eq)]
+pub(crate) enum StandingLatAction {
+    None,
+    /// First escalation: ask for a mid-stream clock re-sync — free, and a stale offset from a
+    /// stepped/slewed wall clock produces exactly this signature (an applied re-sync re-bases
+    /// the floor via the pump's `clock_gen` watch, clearing the elevation if that was the cause).
+    Resync {
+        above_ms: i64,
+    },
+    /// The elevation survived a re-sync attempt: flush the local receive backlog + request a
+    /// keyframe (the jump-to-live action), draining a real sub-threshold standing queue. The
+    /// pump reports execution back via [`StandingLatency::bled`]; an unexecuted action simply
+    /// re-arms next window.
+    Bleed {
+        above_ms: i64,
+    },
+    /// Bleed cap reached and the elevation is back: give up and say so.
+    Disarm {
+        above_ms: i64,
+    },
+}
+
+/// Detector for a small, constant, loss-free one-way-delay elevation — the standing state the
+/// jump-to-live thresholds deliberately tolerate. Tracks the session's OWD floor (minimum of
+/// report-window minimums since start / last re-base) and escalates when windows sit
+/// persistently above it: re-sync first, then a bounded number of flush+keyframe bleeds, then
+/// disarm. Pure state machine (no clocks, no I/O) so the escalation ladder is unit-testable.
+pub(crate) struct StandingLatency {
+    /// Lowest window-minimum OWD seen since session start / last [`rebase`](Self::rebase).
+    floor_ns: Option<i128>,
+    /// Minimum per-frame OWD this report window; `None` = no frames yet.
+    window_min_ns: Option<i128>,
+    /// Consecutive elevated windows.
+    run: u32,
+    /// The current elevation already got its re-sync request — next escalation is a bleed.
+    resync_tried: bool,
+    bleeds: u32,
+    disarmed: bool,
+}
+
+impl StandingLatency {
+    pub(crate) fn new() -> Self {
+        StandingLatency {
+            floor_ns: None,
+            window_min_ns: None,
+            run: 0,
+            resync_tried: false,
+            bleeds: 0,
+            disarmed: false,
+        }
+    }
+
+    /// Feed one frame's skew-corrected OWD (capture→reassembly-complete, ns). Caller gates on a
+    /// live clock offset and plausibility (0 < owd < 10 s), like the ABR OWD signal.
+    pub(crate) fn note_frame(&mut self, owd_ns: i128) {
+        self.window_min_ns = Some(match self.window_min_ns {
+            Some(m) => m.min(owd_ns),
+            None => owd_ns,
+        });
+    }
+
+    /// Close a report window. `loss_free` = the window carried zero loss (loss resets the run —
+    /// congestion is the FEC/ABR machinery's problem, and queues under loss are not "standing").
+    pub(crate) fn on_window(&mut self, loss_free: bool) -> StandingLatAction {
+        let Some(wmin) = self.window_min_ns.take() else {
+            return StandingLatAction::None; // no frames this window — no evidence either way
+        };
+        let floor = *self.floor_ns.get_or_insert(wmin);
+        self.floor_ns = Some(floor.min(wmin));
+        let above_ns = wmin - floor;
+        if self.disarmed {
+            return StandingLatAction::None;
+        }
+        if !loss_free || above_ns < STANDING_LAT_THRESH_NS {
+            self.run = 0;
+            if above_ns < STANDING_LAT_THRESH_NS {
+                self.resync_tried = false; // elevation cleared — a future one re-syncs first again
+            }
+            return StandingLatAction::None;
+        }
+        self.run += 1;
+        if self.run < STANDING_LAT_WINDOWS {
+            return StandingLatAction::None;
+        }
+        self.run = 0; // each escalation gets a fresh observation run
+        let above_ms = (above_ns / 1_000_000) as i64;
+        if !self.resync_tried {
+            self.resync_tried = true;
+            StandingLatAction::Resync { above_ms }
+        } else if self.bleeds < STANDING_LAT_MAX_BLEEDS {
+            StandingLatAction::Bleed { above_ms }
+        } else {
+            self.disarmed = true;
+            StandingLatAction::Disarm { above_ms }
+        }
+    }
+
+    /// The pump executed a [`StandingLatAction::Bleed`] (flush + keyframe). The floor is KEPT: a
+    /// successful bleed brings OWD back down to it (elevation clears naturally); an unsuccessful
+    /// one leaves the elevation visible so the ladder continues toward the cap.
+    pub(crate) fn bled(&mut self) {
+        self.bleeds += 1;
+        self.window_min_ns = None;
+    }
+
+    /// A mid-stream clock re-sync was APPLIED (the pump's `clock_gen` watch): every OWD reading
+    /// shifted, so the floor and any elevation measured under the old offset are meaningless —
+    /// re-learn from scratch. The bleed budget survives (it caps keyframes per session).
+    pub(crate) fn rebase(&mut self) {
+        self.floor_ns = None;
+        self.window_min_ns = None;
+        self.run = 0;
+        self.resync_tried = false;
+    }
+}
+
 /// Client decode-stage latency accumulator for the adaptive-bitrate controller's decode signal.
 /// The embedder adds one sample per decoded frame ([`NativeClient::report_decode_us`], µs from the
 /// AU leaving [`NativeClient::next_frame`] to its decoded output) and the data-plane pump drains a
@@ -191,6 +327,7 @@ mod frame_channel_tests {
             pts_ns: i as u64,
             flags: 0,
             complete: true,
+            received_ns: 0,
         }
     }
 
@@ -258,3 +395,143 @@ mod frame_channel_tests {
         assert_eq!(popped(&ch), Some(total - FRAME_QUEUE_HARD_CAP as u32));
     }
 }
+
+#[cfg(test)]
+mod standing_latency_tests {
+    use super::{
+        StandingLatAction, StandingLatency, STANDING_LAT_MAX_BLEEDS, STANDING_LAT_THRESH_NS,
+        STANDING_LAT_WINDOWS,
+    };
+
+    const FLOOR: i128 = 2_000_000; // a healthy 2 ms LAN OWD
+    const ELEVATED: i128 = FLOOR + STANDING_LAT_THRESH_NS + 7_000_000; // ~one 60fps frame above
+
+    /// Run `n` windows at `owd`, asserting every window but the last returns None; returns the
+    /// last window's action.
+    fn run_windows(d: &mut StandingLatency, owd: i128, n: u32) -> StandingLatAction {
+        for i in 0..n {
+            d.note_frame(owd);
+            let a = d.on_window(true);
+            if i + 1 < n {
+                assert_eq!(a, StandingLatAction::None, "window {i} escalated early");
+            } else {
+                return a;
+            }
+        }
+        unreachable!("n > 0 by construction");
+    }
+
+    /// Learn a clean floor: one window at the healthy OWD.
+    fn learned(d: &mut StandingLatency) {
+        d.note_frame(FLOOR);
+        assert_eq!(d.on_window(true), StandingLatAction::None);
+    }
+
+    #[test]
+    fn healthy_stream_never_escalates() {
+        let mut d = StandingLatency::new();
+        learned(&mut d);
+        // Jitter riding above the floor but under the threshold: never a run.
+        for _ in 0..(STANDING_LAT_WINDOWS * 4) {
+            d.note_frame(FLOOR + STANDING_LAT_THRESH_NS - 1);
+            assert_eq!(d.on_window(true), StandingLatAction::None);
+        }
+    }
+
+    #[test]
+    fn escalation_ladder_resync_then_bleeds_then_disarm() {
+        let mut d = StandingLatency::new();
+        learned(&mut d);
+        // First full elevated run asks for the free fix: a clock re-sync.
+        assert!(matches!(
+            run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
+            StandingLatAction::Resync { .. }
+        ));
+        // Re-sync didn't help (no rebase came) — each further run is a bleed, up to the cap...
+        for _ in 0..STANDING_LAT_MAX_BLEEDS {
+            assert!(matches!(
+                run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
+                StandingLatAction::Bleed { .. }
+            ));
+            d.bled();
+        }
+        // ...then the detector gives up loudly, once, and stays quiet.
+        assert!(matches!(
+            run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
+            StandingLatAction::Disarm { .. }
+        ));
+        d.note_frame(ELEVATED);
+        assert_eq!(d.on_window(true), StandingLatAction::None);
+    }
+
+    #[test]
+    fn loss_windows_reset_the_run() {
+        let mut d = StandingLatency::new();
+        learned(&mut d);
+        for _ in 0..(STANDING_LAT_WINDOWS - 1) {
+            d.note_frame(ELEVATED);
+            assert_eq!(d.on_window(true), StandingLatAction::None);
+        }
+        // A lossy window means congestion, not a standing state: run resets...
+        d.note_frame(ELEVATED);
+        assert_eq!(d.on_window(false), StandingLatAction::None);
+        // ...so the ladder needs the full run again before acting.
+        assert!(matches!(
+            run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
+            StandingLatAction::Resync { .. }
+        ));
+    }
+
+    #[test]
+    fn recovery_resets_the_ladder_to_resync_first() {
+        let mut d = StandingLatency::new();
+        learned(&mut d);
+        assert!(matches!(
+            run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
+            StandingLatAction::Resync { .. }
+        ));
+        // The elevation clears on its own (e.g. the successful bleed case, or transient): the
+        // next episode starts back at the free escalation, not at a bleed.
+        d.note_frame(FLOOR);
+        assert_eq!(d.on_window(true), StandingLatAction::None);
+        assert!(matches!(
+            run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
+            StandingLatAction::Resync { .. }
+        ));
+    }
+
+    #[test]
+    fn applied_resync_rebases_and_clears_a_stale_offset_elevation() {
+        let mut d = StandingLatency::new();
+        learned(&mut d);
+        assert!(matches!(
+            run_windows(&mut d, ELEVATED, STANDING_LAT_WINDOWS),
+            StandingLatAction::Resync { .. }
+        ));
+        // The re-sync APPLIES (pump sees clock_gen move) → rebase. The corrected offset brings
+        // OWD readings back to truth; the floor re-learns and nothing ever escalates to a bleed.
+        d.rebase();
+        for _ in 0..(STANDING_LAT_WINDOWS * 2) {
+            d.note_frame(FLOOR);
+            assert_eq!(d.on_window(true), StandingLatAction::None);
+        }
+    }
+
+    #[test]
+    fn empty_windows_are_no_evidence() {
+        let mut d = StandingLatency::new();
+        learned(&mut d);
+        for _ in 0..(STANDING_LAT_WINDOWS - 1) {
+            d.note_frame(ELEVATED);
+            assert_eq!(d.on_window(true), StandingLatAction::None);
+        }
+        // A frameless window (paused stream) neither advances nor resets the run...
+        assert_eq!(d.on_window(true), StandingLatAction::None);
+        // ...so one more elevated window completes it.
+        d.note_frame(ELEVATED);
+        assert!(matches!(
+            d.on_window(true),
+            StandingLatAction::Resync { .. }
+        ));
+    }
+}

crates/punktfunk-core/src/client/pump.rs

@@ -1,8 +1,9 @@
 //! The client worker: QUIC handshake + control/input/datagram tasks + the blocking data-plane pump.
 
 use super::frame_channel::{
-    CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN, FLUSH_LATENCY,
-    NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW, STANDING_TIME,
+    StandingLatAction, StandingLatency, CLOCK_RESYNC_INTERVAL, FLUSH_AFTER, FLUSH_COOLDOWN,
+    FLUSH_LATENCY, NOOP_CLOCK_FLUSHES_TO_DISARM, NOOP_FLUSH_DATAGRAMS, QUEUE_HIGH, QUEUE_LOW,
+    STANDING_TIME,
 };
 use super::worker::reject_from_close;
 use super::*;
@@ -514,15 +515,19 @@ pub(super) async fn run_pump(args: WorkerArgs) {
                                     // late exactly then) — keep the old estimate and let the next
                                     // periodic batch try again.
                                     if accept_resync(rtt_ns, clock_rtt_ns.unwrap_or(0)) {
-                                        clock_offset.store(offset_ns, Ordering::Relaxed);
-                                        clock_gen.fetch_add(1, Ordering::Relaxed);
-                                        tracing::debug!(
+                                        // info, not debug: ≤1/min, and it is THE forensic
+                                        // trail for a stale-offset (stepped/slewed wall clock)
+                                        // latency plateau — the 2026-07 two-pair investigation
+                                        // had to reconstruct this blind.
+                                        tracing::info!(
                                             offset_ns,
                                             rtt_us = rtt_ns / 1000,
                                             "mid-stream clock re-sync applied"
                                         );
+                                        clock_offset.store(offset_ns, Ordering::Relaxed);
+                                        clock_gen.fetch_add(1, Ordering::Relaxed);
                                     } else {
-                                        tracing::debug!(
+                                        tracing::info!(
                                             rtt_us = rtt_ns / 1000,
                                             "clock re-sync batch discarded — RTT above the \
                                              connect-time baseline (congested window)"
@@ -729,6 +734,12 @@ pub(super) async fn run_pump(args: WorkerArgs) {
         let mut clock_detector_armed = true;
         let mut resync_wanted = false;
         let mut seen_clock_gen = pump_clock_gen.load(Ordering::Relaxed);
+        // Standing-latency bleed (see StandingLatency): the third detector, for the small,
+        // constant, loss-free OWD elevation the two jump-to-live detectors deliberately
+        // tolerate (< QUEUE_HIGH frames, < FLUSH_LATENCY behind) — a sub-frame standing
+        // backlog, or a stale clock offset after a wall-clock step, either of which otherwise
+        // reads as permanent extra "network" latency for the rest of the session.
+        let mut standing_lat = StandingLatency::new();
         while !pump_shutdown.load(Ordering::SeqCst) {
             // The live host↔client offset: re-loaded every iteration so an applied mid-stream
             // re-sync takes effect on the very next frame's latency math.
@@ -740,6 +751,10 @@ pub(super) async fn run_pump(args: WorkerArgs) {
                 seen_clock_gen = gen;
                 stale_since = None;
                 noop_clock_flushes = 0;
+                // Every OWD reading shifted with the offset — the standing-latency floor and
+                // any elevation measured under the old one are meaningless now. If a stale
+                // offset WAS the elevation, this is also the moment it gets fixed.
+                standing_lat.rebase();
                 if !clock_detector_armed {
                     clock_detector_armed = true;
                     tracing::info!(
@@ -843,6 +858,51 @@ pub(super) async fn run_pump(args: WorkerArgs) {
                     window_dropped,
                 );
                 let _ = ctrl_tx.try_send(CtrlRequest::Loss(LossReport { loss_ppm }));
+                // Standing-latency bleed: close the detector's window with this report's loss
+                // verdict and run its escalation ladder — re-sync first (free; a stale offset
+                // from a stepped wall clock produces exactly this signature and the applied
+                // re-sync rebases the floor), then a bounded flush+keyframe (drains a real
+                // sub-threshold standing backlog the jump-to-live thresholds tolerate), then a
+                // loud disarm (the path latency itself changed; nothing local fixes that).
+                match standing_lat.on_window(loss_ppm == 0 && window_dropped == 0) {
+                    StandingLatAction::None => {}
+                    StandingLatAction::Resync { above_ms } => {
+                        tracing::info!(
+                            above_ms,
+                            "standing latency above the session floor with zero loss — \
+                             requesting a clock re-sync first (a stale offset reads exactly \
+                             like this)"
+                        );
+                        let _ = ctrl_tx.try_send(CtrlRequest::ClockResync);
+                    }
+                    StandingLatAction::Bleed { above_ms } => {
+                        // Shares the jump-to-live cooldown: an unexecuted bleed simply re-arms
+                        // over the next windows (the detector's run rebuilds).
+                        if last_flush.is_none_or(|t| t.elapsed() >= FLUSH_COOLDOWN) {
+                            last_flush = Some(Instant::now());
+                            flush_in_window = true;
+                            let flushed = session.flush_backlog().unwrap_or(0);
+                            let dropped = frames.clear();
+                            let _ = ctrl_tx.try_send(CtrlRequest::Keyframe);
+                            standing_lat.bled();
+                            tracing::warn!(
+                                above_ms,
+                                flushed_datagrams = flushed,
+                                dropped_frames = dropped,
+                                "standing latency survived a clock re-sync — bled the local \
+                                 backlog (flush + keyframe)"
+                            );
+                        }
+                    }
+                    StandingLatAction::Disarm { above_ms } => {
+                        tracing::warn!(
+                            above_ms,
+                            "standing latency persists after a re-sync and every bleed — not \
+                             local, not clock; the path latency changed. Leaving it be \
+                             (reconnect re-baselines)"
+                        );
+                    }
+                }
                 // Adaptive bitrate: drain any host ack first (its clamp is authoritative), then
                 // feed the controller this window's congestion signals; a decision becomes a
                 // SetBitrate on the control stream.
@@ -981,6 +1041,13 @@ pub(super) async fn run_pump(args: WorkerArgs) {
                         if clock_offset_ns != 0 && lat_ns > 0 {
                             owd_sum_ns += lat_ns;
                             owd_frames += 1;
+                            // The standing-latency detector rides the same signal, but off the
+                            // window MINIMUM (robust against jitter/burst spikes — a standing
+                            // state elevates the floor itself). Same 10 s plausibility clamp as
+                            // the hn stats use.
+                            if lat_ns < 10_000_000_000 {
+                                standing_lat.note_frame(lat_ns);
+                            }
                         }
                         if clock_detector_armed
                             && clock_offset_ns != 0

crates/punktfunk-core/src/lib.rs

@@ -83,7 +83,12 @@ pub use stats::Stats;
 /// `punktfunk_connection_clipboard_{control,offer,fetch,serve,cancel}` +
 /// `punktfunk_connection_next_clipboard`. Additive; the wire grows only backward-compatible control
 /// messages (0x40-0x44) and a new `Welcome::host_caps` bit, so [`WIRE_VERSION`] is unchanged.
-pub const ABI_VERSION: u32 = 8;
+/// v9: `PunktfunkFrame` grew `received_ns` — the reassembly-completion receipt stamp, so
+/// embedders stop stamping receipt at the hand-off pull (which folds the pre-decode queue wait
+/// into apparent network latency). Struct-size change on the frame poll surface = a hard ABI
+/// break for embedders reading `PunktfunkFrame`; nothing on the wire moved, so [`WIRE_VERSION`]
+/// is unchanged.
+pub const ABI_VERSION: u32 = 9;
 
 /// The punktfunk/1 **wire** version — what `Hello`/`Welcome` carry and hosts equality-check.
 /// Deliberately its own constant: [`ABI_VERSION`] tracks the embeddable **C surface**

crates/punktfunk-core/src/packet/reassemble.rs

@@ -489,6 +489,7 @@ impl Reassembler {
                 pts_ns: done.pts_ns,
                 flags: done.user_flags,
                 complete: true,
+                received_ns: 0, // stamped by Session::poll_frame at the session boundary
             }));
         }
         Ok(None)
@@ -592,6 +593,7 @@ impl ReassemblyWindow {
                                 pts_ns: f.pts_ns,
                                 flags: f.user_flags,
                                 complete: false,
+                                received_ns: 0, // stamped by Session::poll_frame at the session boundary
                             });
                         }
                     }

crates/punktfunk-core/src/session.rs

@@ -30,6 +30,14 @@ pub struct Frame {
     /// ([`crate::packet::USER_FLAG_CHUNK_ALIGNED`]) are ever delivered partial; missing
     /// shard ranges are zero-filled at their exact offsets.
     pub complete: bool,
+    /// Wall-clock instant (ns since the Unix epoch, CLOCK_REALTIME basis — the same clock the
+    /// skew handshake compares and the host stamps `pts_ns` with) at which this AU finished
+    /// reassembly, stamped by [`Session::poll_frame`] as the frame leaves the session. Embedders
+    /// that previously stamped receipt themselves at the hand-off pull should use this instead:
+    /// the pull stamp additionally contains the pre-decode queue wait, silently folding any
+    /// client-side standing backlog into the apparent NETWORK latency. The reassembler itself
+    /// leaves this 0 (it owns no clock — the stamp is the session boundary's job).
+    pub received_ns: u64,
 }
 
 /// One end of a stream. Constructed for a single [`Role`]; calling the other role's
@@ -82,6 +90,18 @@ pub struct Session {
     lane_scratch: Vec<Vec<u8>>,
 }
 
+/// Stamp [`Frame::received_ns`] as the frame crosses the session boundary in
+/// [`Session::poll_frame`] — completed frames return the moment their last shard lands, so
+/// stamping at return IS stamping at reassembly completion (µs apart). CLOCK_REALTIME to match
+/// `pts_ns` / the skew handshake (deliberately not monotonic — cross-machine latency math).
+fn stamp_received(mut f: Frame) -> Frame {
+    f.received_ns = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_nanos() as u64)
+        .unwrap_or(0);
+    f
+}
+
 /// Wire-packet count at which a frame's sealing splits across two lanes (plan Phase 1.5):
 /// below it the channel rendezvous (~µs) isn't worth it; at it the halved AES-GCM span
 /// (≥ ~125 µs of ~1 µs/packet work) dwarfs the hand-off. ≈300 KB of wire, i.e. ≥150 Mbps
@@ -684,7 +704,7 @@ impl Session {
                     // Nothing new on the wire — hand over an aged-out partial if one is
                     // waiting (it can only get staler).
                     if let Some(p) = self.reassembler.take_partial() {
-                        return Ok(p);
+                        return Ok(stamp_received(p));
                     }
                     return Err(PunktfunkError::NoFrame);
                 }
@@ -748,12 +768,12 @@ impl Session {
             }
             if let Some(frame) = pushed {
                 StatsCounters::add(&self.stats.frames_completed, 1);
-                return Ok(frame);
+                return Ok(stamp_received(frame));
             }
             // A push that completed nothing may still have aged a partial out — deliver it
             // ahead of further draining (its successors are already arriving).
             if let Some(p) = self.reassembler.take_partial() {
-                return Ok(p);
+                return Ok(stamp_received(p));
             }
         }
     }