feat(core/client): an AU's prefix reaches the decoder while its tail is on the wire

Delivery used to be all-or-nothing: the decoder saw byte 0 only after the
last packet of the AU landed, so the whole transmit time sat in front of
decode. With the slice-streamed wire (previous commit) blocks now arrive
addressable, and a client can opt in (connect's new frame_parts) to
receive each AU's newly-contiguous prefix as Frame::part pieces - offset
tiling, first/last marked, the completing push carrying only the suffix.
A PARTIAL_FRAME-capable decoder then chews slices concurrently with the
remaining network transfer.

The reassembler walks a per-frame cursor over successfully-completed
blocks (failed FEC reconstructs don't advance it), coalesces blocks that
finished out of order into one part, keeps probe filler whole, and stops
short of the final block so the zero-padded tail still trims at
completion. Whole-frame consumers see byte-identical behavior - parts
never flow without the opt-in, and never on PyroWave (its newest-wins
draining assumes whole AUs).

Per-AU accounting keeps its units: OWD/ABR feeds, the inter-arrival
series and the clock-based jump-to-live detector only count completing
deliveries, and FrameChannel::depth() counts AUs so a part-rich queue
can't trip jump-to-live at a fraction of the real backlog. The consumer
contract (gap or orphan part = AU lost: abandon, flush, resync on the
next first) is documented on FramePart; the C ABI keeps parts off until
PunktfunkFrame can express them.

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

clients/android/native/src/session/connect.rs

@@ -225,6 +225,9 @@ pub extern "system" fn Java_io_unom_punktfunk_kit_NativeBridge_nativeConnect<'lo
         // No non-video caps: this client does not render the host cursor locally (no shape/state
         // planes in the jni surface), so advertising CLIENT_CAP_CURSOR would stream cursor-less.
         0,
+        // Slice-progressive delivery: off until the decode loop feeds MediaCodec with
+        // BUFFER_FLAG_PARTIAL_FRAME (P2d flips this behind a FEATURE_PartialFrame probe).
+        false,
         launch, // a store-qualified library id to boot into a game, or None for the desktop
         device_name, // Kotlin's Build.MODEL — the host's approval-list / trust-store label
         pin,    // Some → Crypto on host-fp mismatch

clients/cli/src/main.rs

@@ -798,14 +798,15 @@ from the config directory for a true factory reset."
             },
             punktfunk_core::config::CompositorPref::Auto,
             punktfunk_core::config::GamepadPref::Auto,
-            0,    // bitrate_kbps: the host's default; this connect never presents
-            0,    // video_caps: nothing decodes here
-            2,    // audio_channels
-            0,    // video_codecs: the probe carries no video
-            0,    // preferred_codec
-            None, // display_hdr
-            0,    // client_caps: nothing renders a cursor
-            None, // launch
+            0,     // bitrate_kbps: the host's default; this connect never presents
+            0,     // video_caps: nothing decodes here
+            2,     // audio_channels
+            0,     // video_codecs: the probe carries no video
+            0,     // preferred_codec
+            None,  // display_hdr
+            0,     // client_caps: nothing renders a cursor
+            false, // frame_parts: probe/whole-AU consumer
+            None,  // launch
             Some(punktfunk_core::client::device_name()),
             Some(pin),
             Some(identity),

clients/linux/src/app.rs

@@ -696,9 +696,10 @@ impl AppModel {
                     2,                                // audio_channels: stereo
                     crate::video::decodable_codecs(), // codecs (unused by the probe, but honest)
                     0,                                // preferred_codec: no preference
-                    None, // display_hdr: probe connect, nothing presents
-                    0,    // client_caps: probe connect, nothing renders a cursor
-                    None, // launch: probe connect, no game
+                    None,  // display_hdr: probe connect, nothing presents
+                    0,     // client_caps: probe connect, nothing renders a cursor
+                    false, // frame_parts: probe/whole-AU consumer
+                    None,  // launch: probe connect, no game
                     // Knock under this device's name, not a fingerprint placeholder, when the
                     // probed host doesn't know us yet.
                     Some(pf_client_core::trust::device_name()),

clients/windows/src/probe.rs

@@ -54,10 +54,11 @@ pub fn run_speed_probe(
         0, // video_caps: probe connect, nothing is decoded
         2, // audio_channels: stereo baseline
         decodable_codecs(),
-        0,    // preferred_codec: no preference
-        None, // display_hdr: probe connect, nothing presents
-        0,    // client_caps: probe connect, nothing renders a cursor
-        None, // launch: no game
+        0,     // preferred_codec: no preference
+        None,  // display_hdr: probe connect, nothing presents
+        0,     // client_caps: probe connect, nothing renders a cursor
+        false, // frame_parts: probe/whole-AU consumer
+        None,  // launch: no game
         // Same label a real session sends — a speed test against a host that doesn't know us yet
         // should knock under this device's name, not a fingerprint placeholder.
         Some(punktfunk_core::client::device_name()),

crates/pf-client-core/src/session.rs

@@ -272,6 +272,9 @@ fn pump(
         } else {
             0
         },
+        // Slice-progressive delivery: off — this presenter feeds FFmpeg whole AUs; a partial
+        // avcodec feed path can flip it later.
+        false,
         params.launch.clone(),
         // The host's approval-list / trust-store label for this client. Without it every no-PIN
         // "request access" knock showed up as the fingerprint placeholder "device abcd1234".

crates/punktfunk-core/src/abi.rs

@@ -1871,6 +1871,11 @@ unsafe fn connect_ex_impl(
             // compositing the pointer — only an embedder that renders the cursor planes
             // ([`punktfunk_connection_next_cursor_shape`]/`_state`) may set it. ex7/ex8 pass 0.
             client_caps,
+            // The C ABI cannot carry slice-progressive parts yet — `PunktfunkFrame` has no
+            // part/completeness fields, so a part would be indistinguishable from a whole AU.
+            // An `ex10` variant adds the opt-in together with those fields when an ABI embedder
+            // (Apple) grows a partial-feed decode path.
+            false,
             launch,
             // The C ABI has no device-name parameter (only `punktfunk_pair` takes one), so every
             // embedder gets the OS hostname default — this is what the host's pending-approval

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

@@ -321,9 +321,18 @@ impl FrameChannel {
         self.ready.notify_one();
     }
 
-    /// Pump side: current queued depth — the clock-free standing-queue signal.
+    /// Pump side: current queued depth in ACCESS UNITS — the clock-free standing-queue
+    /// signal. Slice-progressive parts of a still-open AU don't count (the detector's
+    /// thresholds are in frames; counting parts would trip it at a fraction of the real
+    /// backlog), and a queue that stops draining still accumulates completed AUs.
     pub(crate) fn depth(&self) -> usize {
-        self.inner.lock().unwrap().q.len()
+        self.inner
+            .lock()
+            .unwrap()
+            .q
+            .iter()
+            .filter(|f| f.complete)
+            .count()
     }
 
     /// Pump side: discard the whole backlog (the jump-to-live path); returns how many were dropped.
@@ -378,10 +387,34 @@ mod frame_channel_tests {
             pts_ns: i as u64,
             flags: 0,
             complete: true,
+            part: None,
             received_ns: 0,
         }
     }
 
+    /// `depth()` is the standing-queue detector's signal, in AU units: parts of a
+    /// still-open AU must not count (the thresholds are frames — parts would trip
+    /// jump-to-live at a fraction of the real backlog), while completed AUs always do.
+    #[test]
+    fn depth_counts_aus_not_parts() {
+        let ch = FrameChannel::new();
+        let mut p = frame(1);
+        p.complete = false;
+        p.part = Some(crate::session::FramePart {
+            offset: 0,
+            first: true,
+            last: false,
+        });
+        ch.push(p);
+        assert_eq!(
+            ch.depth(),
+            0,
+            "an open AU's prefix part is not a queued frame"
+        );
+        ch.push(frame(2));
+        assert_eq!(ch.depth(), 1);
+    }
+
     fn popped(ch: &FrameChannel) -> Option<u32> {
         match ch.pop(Duration::from_millis(0)) {
             FramePop::Frame(f) => Some(f.frame_index),

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

@@ -354,6 +354,11 @@ impl NativeClient {
         // the video for a session that advertises it, so a non-rendering embedder that sets it
         // streams with NO visible cursor at all. `0` = today's composited behavior.
         client_caps: u8,
+        // Slice-progressive delivery opt-in: AU prefixes arrive as [`Frame`]s with
+        // [`crate::session::Frame::part`]` = Some` while the rest is still on the wire. Set it
+        // ONLY when this embedder's decode path understands parts (e.g. feeds MediaCodec with
+        // BUFFER_FLAG_PARTIAL_FRAME); with it false every AU arrives whole, exactly as before.
+        frame_parts: bool,
         launch: Option<String>,
         // This device's display name, carried in [`crate::quic::Hello::name`]: what the host's
         // pending-approval list shows when an unpaired client knocks, and what its trust store
@@ -438,6 +443,7 @@ impl NativeClient {
                     preferred_codec,
                     display_hdr,
                     client_caps,
+                    frame_parts,
                     launch,
                     name,
                     pin,

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

@@ -520,7 +520,12 @@ impl DataPump {
                     if frame.flags & FLAG_PROBE as u32 != 0 {
                         continue; // speed-test filler, not video — measured via the counters above
                     }
-                    if pump_perf_on {
+                    // A prefix part is not an AU arrival: the inter-arrival series, the OWD
+                    // window and the clock-based staleness detector below all measure per-AU
+                    // signals, so only the delivery that completes an AU feeds them (parts
+                    // would bias OWD low and constantly reset the staleness run).
+                    let is_au = frame.complete;
+                    if pump_perf_on && is_au {
                         let now = Instant::now();
                         if let Some(prev) = last_arrival.replace(now) {
                             // 4096 ≈ 17 s at 240 fps — a stuck window can't grow it unbounded.
@@ -552,7 +557,7 @@ impl DataPump {
                         stale_since = None;
                         standing_since = None;
                     } else {
-                        let lat_ns = if clock_offset_ns != 0 {
+                        let lat_ns = if clock_offset_ns != 0 && is_au {
                             now_realtime_ns() + clock_offset_ns as i128 - frame.pts_ns as i128
                         } else {
                             0
@@ -576,7 +581,7 @@ impl DataPump {
                             && lat_ns > FLUSH_LATENCY.as_nanos() as i128
                         {
                             stale_since.get_or_insert_with(Instant::now);
-                        } else {
+                        } else if is_au {
                             stale_since = None;
                         }
                         let depth = frames.depth();

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

@@ -221,6 +221,12 @@ pub(super) async fn connect_and_handshake(args: &WorkerArgs) -> Result<Handshake
         if welcome.codec == crate::quic::CODEC_PYROWAVE {
             session.set_deliver_partial_frames(true);
         }
+        // Slice-progressive delivery (the embedder's opt-in): AU prefixes hand up as
+        // `Frame::part` pieces while the tail is still on the wire. Never on PyroWave — its
+        // all-intra frame channel drains newest-wins, which assumes whole AUs.
+        if args.frame_parts && welcome.codec != crate::quic::CODEC_PYROWAVE {
+            session.set_deliver_frame_parts(true);
+        }
         Ok::<_, PunktfunkError>((
             session,
             send,

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

@@ -23,6 +23,10 @@ pub(crate) struct WorkerArgs {
     pub(crate) preferred_codec: u8,
     pub(crate) display_hdr: Option<HdrMeta>,
     pub(crate) client_caps: u8,
+    /// Slice-progressive delivery opt-in ([`crate::session::Session::set_deliver_frame_parts`]):
+    /// only an embedder whose decode path understands [`crate::session::Frame::part`] may set
+    /// it. Ignored on all-intra (PyroWave) sessions — their newest-wins draining needs whole AUs.
+    pub(crate) frame_parts: bool,
     pub(crate) launch: Option<String>,
     /// This device's display name, sent in `Hello` (the host's approval list / trust store label).
     pub(crate) name: Option<String>,

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

@@ -5,7 +5,7 @@ use super::*;
 use crate::config::Config;
 use crate::error::Result;
 use crate::fec::ErasureCoder;
-use crate::session::Frame;
+use crate::session::{Frame, FramePart};
 use crate::stats::StatsCounters;
 use std::collections::HashMap;
 use zerocopy::FromBytes;
@@ -85,6 +85,11 @@ struct FrameBuf {
     block_count: usize,
     pts_ns: u64,
     user_flags: u32,
+    /// Slice-progressive delivery cursor ([`Reassembler::deliver_parts`]): count of leading
+    /// blocks already handed up as prefix parts...
+    next_part_block: u16,
+    /// ...and their total data shards — the AU shard offset where the next part starts.
+    delivered_shards: usize,
     /// The whole frame's data region — `total_data_shards × shard_bytes` zeroed bytes. Data
     /// shards are copied to their final offset on arrival; FEC reconstruction writes only the
     /// missing shards' ranges. On completion this Vec IS [`Frame::data`] (truncated to
@@ -187,6 +192,11 @@ pub struct Reassembler {
     deliver_partial: bool,
     /// The newest such partial awaiting pickup (newest-wins: partials are a lossy byproduct).
     pending_partial: Option<Frame>,
+    /// Deliver each video AU's newly-contiguous PREFIX as [`Frame`]s with
+    /// [`Frame::part`]`= Some` while the rest is still in flight (client opt-in — the
+    /// slice-progressive decode path). Works for any multi-block frame — both streamed shapes
+    /// and legacy uniform frames tile their blocks contiguously (`BlockState::base_shard`).
+    deliver_parts: bool,
     /// The video stream's window — its aged-out incomplete frames count into `frames_dropped`
     /// (the client's loss-recovery trigger).
     video: ReassemblyWindow,
@@ -210,6 +220,7 @@ impl Reassembler {
             limits,
             deliver_partial: false,
             pending_partial: None,
+            deliver_parts: false,
             video: ReassemblyWindow::default(),
             probe: ReassemblyWindow::default(),
             recovery_pool: Vec::new(),
@@ -225,6 +236,11 @@ impl Reassembler {
         }
     }
 
+    /// Opt into slice-progressive prefix delivery (see [`Reassembler::deliver_parts`]).
+    pub fn set_deliver_parts(&mut self, on: bool) {
+        self.deliver_parts = on;
+    }
+
     /// Take the newest aged-out partial frame, if one is pending (see `set_deliver_partial`).
     pub fn take_partial(&mut self) -> Option<Frame> {
         self.pending_partial.take()
@@ -260,11 +276,13 @@ impl Reassembler {
             limits,
             deliver_partial,
             pending_partial,
+            deliver_parts,
             video,
             probe,
             recovery_pool,
             in_flight_bytes,
         } = self;
+        let deliver_parts = *deliver_parts;
         let lim = *limits;
         let shard_bytes = hdr.shard_bytes as usize;
         let data_shards = hdr.data_shards as usize;
@@ -454,6 +472,8 @@ impl Reassembler {
                     block_count,
                     pts_ns: hdr.pts_ns,
                     user_flags: hdr.user_flags,
+                    next_part_block: 0,
+                    delivered_shards: 0,
                     buf: vec![0; buf_len],
                     blocks: HashMap::new(),
                     blocks_ok: 0,
@@ -566,8 +586,13 @@ impl Reassembler {
             blocks,
             blocks_ok,
             block_count: frame_block_count,
+            pts_ns: frame_pts_ns,
+            user_flags: frame_user_flags,
+            next_part_block,
+            delivered_shards,
             ..
         } = frame;
+        let (frame_pts_ns, frame_user_flags) = (*frame_pts_ns, *frame_user_flags);
 
         // First packet of a block sizes its state; `data_shards` is already pinned by the
         // derived geometry above, but `recovery_shards` is per-block wire input (adaptive FEC
@@ -713,15 +738,69 @@ impl Reassembler {
             );
             *in_flight_bytes -= done.buf.len();
             done.buf.truncate(done.frame_bytes); // trim trailing-shard zero padding
+                                                 // Slice-progressive consumers already hold the delivered prefix — the completing
+                                                 // packet hands up only the SUFFIX (with `last`), or the degenerate whole-AU part
+                                                 // when nothing was delivered early. Probe filler stays whole either way — the
+                                                 // speed test accounts AUs, not slices.
+            let (data, part) = if deliver_parts && !is_probe {
+                let lo = (done.delivered_shards * shard_bytes).min(done.frame_bytes);
+                let part = FramePart {
+                    offset: lo as u32,
+                    first: lo == 0,
+                    last: true,
+                };
+                if lo == 0 {
+                    (done.buf, Some(part))
+                } else {
+                    (done.buf[lo..].to_vec(), Some(part))
+                }
+            } else {
+                (done.buf, None)
+            };
             return Ok(Some(Frame {
-                data: done.buf,
+                data,
                 frame_index: hdr.frame_index,
                 pts_ns: done.pts_ns,
                 flags: done.user_flags,
                 complete: true,
+                part,
                 received_ns: 0, // stamped by Session::poll_frame at the session boundary
             }));
         }
+        // Slice-progressive delivery: when this packet completed a block that extends the AU's
+        // contiguous prefix (possibly unlocking blocks that finished out of order behind it),
+        // hand the newly-contiguous bytes up as ONE part. Only successfully-completed blocks
+        // count (`done` alone includes failed reconstructs), and the cursor stops short of the
+        // FINAL block — its zero-padded tail is only trimmed at completion above.
+        if deliver_parts && !is_probe {
+            let start = *delivered_shards;
+            while let Some(b) = blocks.get(&*next_part_block) {
+                if !(b.done && (b.reconstructed || b.data_received == b.data_shards)) {
+                    break;
+                }
+                if block_count != 0 && (*next_part_block as usize) + 1 >= block_count {
+                    break;
+                }
+                *delivered_shards = b.base_shard + b.data_shards;
+                *next_part_block += 1;
+            }
+            if *delivered_shards > start {
+                let (lo, hi) = (start * shard_bytes, *delivered_shards * shard_bytes);
+                return Ok(Some(Frame {
+                    data: buf[lo..hi].to_vec(),
+                    frame_index: hdr.frame_index,
+                    pts_ns: frame_pts_ns,
+                    flags: frame_user_flags,
+                    complete: false,
+                    part: Some(FramePart {
+                        offset: lo as u32,
+                        first: start == 0,
+                        last: false,
+                    }),
+                    received_ns: 0, // stamped by Session::poll_frame at the session boundary
+                }));
+            }
+        }
         Ok(None)
     }
 
@@ -830,6 +909,7 @@ impl ReassemblyWindow {
                                 pts_ns: f.pts_ns,
                                 flags: f.user_flags,
                                 complete: false,
+                                part: None,
                                 received_ns: 0, // stamped by Session::poll_frame at the session boundary
                             });
                         }
@@ -894,6 +974,7 @@ mod reset_tests {
             pts_ns: 1,
             flags: 0,
             complete: false,
+            part: None,
             received_ns: 0,
         });
         r.reset();

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

@@ -1296,6 +1296,188 @@ fn slice_streamed_mixed_flag_packet_dropped() {
     assert_eq!(stats.snapshot().frames_dropped, 0, "dropped, not killed");
 }
 
+// ---------------------------------------------------------------------------
+// Slice-progressive prefix delivery (Frame::part — P2c)
+// ---------------------------------------------------------------------------
+
+/// Push packets collecting EVERY delivery (parts and completions), returning them in order.
+fn push_collect(
+    r: &mut Reassembler,
+    coder: &dyn crate::fec::ErasureCoder,
+    stats: &StatsCounters,
+    delivery: &[Vec<u8>],
+) -> Vec<crate::session::Frame> {
+    let mut out = Vec::new();
+    for p in delivery {
+        if let Some(f) = r.push(p, coder, stats).unwrap() {
+            out.push(f);
+        }
+    }
+    out
+}
+
+/// In-order slice delivery streams one part per completed block, offsets tiling exactly, the
+/// final delivery carrying only the suffix with `last` + `complete`.
+#[test]
+fn parts_stream_in_order() {
+    let (pkts, src) = slice_streamed_packets();
+    let cfg = slice_config();
+    let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg));
+    r.set_deliver_parts(true);
+    let coder = coder_for(FecScheme::Gf16);
+    let stats = StatsCounters::default();
+    let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts);
+    assert_eq!(
+        got.len(),
+        4,
+        "three sentinel-block parts + the final suffix"
+    );
+    let mut rebuilt = Vec::new();
+    for (i, f) in got.iter().enumerate() {
+        let part = f
+            .part
+            .expect("parts mode: every delivery carries part meta");
+        assert_eq!(
+            part.offset as usize,
+            rebuilt.len(),
+            "parts tile with no gaps"
+        );
+        assert_eq!(part.first, i == 0);
+        assert_eq!(part.last, i + 1 == got.len());
+        assert_eq!(
+            f.complete, part.last,
+            "complete rides exactly the last part"
+        );
+        rebuilt.extend_from_slice(&f.data);
+    }
+    assert_eq!(rebuilt, src, "concatenated parts must be the byte-exact AU");
+    assert_eq!(
+        stats.snapshot().frames_completed,
+        0,
+        "the reassembler leaves the completion count to the session boundary"
+    );
+}
+
+/// A block completing BEHIND the prefix emits nothing; the block that closes the gap emits
+/// ONE coalesced part spanning everything unlocked.
+#[test]
+fn parts_coalesce_across_reordered_blocks() {
+    let (pkts, src) = slice_streamed_packets();
+    let hdr_of = |p: &Vec<u8>| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap();
+    // Blocks 1 and 2 fully first, then block 0, then the final block.
+    let mut delivery: Vec<Vec<u8>> = Vec::new();
+    for want in [1u16, 2, 0] {
+        delivery.extend(
+            pkts.iter()
+                .filter(|p| {
+                    let h = hdr_of(p);
+                    h.block_count == 0 && h.block_index == want
+                })
+                .cloned(),
+        );
+    }
+    delivery.extend(pkts.iter().filter(|p| hdr_of(p).block_count != 0).cloned());
+
+    let cfg = slice_config();
+    let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg));
+    r.set_deliver_parts(true);
+    let coder = coder_for(FecScheme::Gf16);
+    let stats = StatsCounters::default();
+    let got = push_collect(&mut r, coder.as_ref(), &stats, &delivery);
+    assert_eq!(got.len(), 2, "one coalesced prefix part + the final suffix");
+    let p0 = got[0].part.unwrap();
+    assert_eq!((p0.offset, p0.first, p0.last), (0, true, false));
+    assert_eq!(got[0].data.len(), 720 + 288, "blocks 0-2 in one part");
+    let p1 = got[1].part.unwrap();
+    assert!(p1.last && got[1].complete);
+    let mut rebuilt = got[0].data.clone();
+    rebuilt.extend_from_slice(&got[1].data);
+    assert_eq!(rebuilt, src);
+}
+
+/// Loss inside a block delays its part until FEC reconstructs it — the part then carries the
+/// recovered bytes, still byte-exact.
+#[test]
+fn parts_wait_for_fec_reconstruction() {
+    let (pkts, src) = slice_streamed_packets();
+    let hdr_of = |p: &Vec<u8>| PacketHeader::read_from_bytes(&p[..HEADER_LEN]).unwrap();
+    // Kill two data shards of block 0 — within the 50% parity budget.
+    let delivery: Vec<Vec<u8>> = pkts
+        .iter()
+        .filter(|p| {
+            let h = hdr_of(p);
+            !(h.block_count == 0 && h.block_index == 0 && [1, 7].contains(&h.shard_index))
+        })
+        .cloned()
+        .collect();
+    let cfg = slice_config();
+    let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg));
+    r.set_deliver_parts(true);
+    let coder = coder_for(FecScheme::Gf16);
+    let stats = StatsCounters::default();
+    let got = push_collect(&mut r, coder.as_ref(), &stats, &delivery);
+    let mut rebuilt = Vec::new();
+    for f in &got {
+        rebuilt.extend_from_slice(&f.data);
+    }
+    assert_eq!(
+        rebuilt, src,
+        "reconstructed prefix parts must be byte-exact"
+    );
+    assert!(got.last().unwrap().complete);
+}
+
+/// With parts on, a legacy single-block frame degenerates to ONE whole-AU delivery carrying
+/// `{offset 0, first, last}` — the consumer's feed logic stays uniform.
+#[test]
+fn parts_degenerate_whole_frame() {
+    let cfg = e2e_config(FecScheme::Gf16, 50);
+    let coder = coder_for(FecScheme::Gf16);
+    let mut pk = Packetizer::new(&cfg);
+    let src: Vec<u8> = (0..40).map(|i| i as u8).collect();
+    let pkts = pk.packetize(&src, 7, 0, coder.as_ref()).unwrap();
+    let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg));
+    r.set_deliver_parts(true);
+    let stats = StatsCounters::default();
+    let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts);
+    assert_eq!(got.len(), 1);
+    let f = &got[0];
+    assert_eq!(
+        f.part,
+        Some(crate::session::FramePart {
+            offset: 0,
+            first: true,
+            last: true
+        })
+    );
+    assert!(f.complete);
+    assert_eq!(f.data, src);
+}
+
+/// Parts also flow for the LEGACY streamed shape (uniform full-K sentinels) — the prefix
+/// cursor rides `base_shard`, which both wire shapes maintain.
+#[test]
+fn parts_flow_for_legacy_streamed_frames() {
+    let chunks: Vec<Vec<u8>> = (0..3)
+        .map(|c| (0..50).map(|i| (c * 57 + i * 131 + 7) as u8).collect())
+        .collect();
+    let chunk_refs: Vec<&[u8]> = chunks.iter().map(|c| c.as_slice()).collect();
+    let (pkts, src) = streamed_packets(FecScheme::Gf16, 50, &chunk_refs);
+    let cfg = e2e_config(FecScheme::Gf16, 50);
+    let mut r = Reassembler::new(ReassemblerLimits::from_config(&cfg));
+    r.set_deliver_parts(true);
+    let coder = coder_for(FecScheme::Gf16);
+    let stats = StatsCounters::default();
+    let got = push_collect(&mut r, coder.as_ref(), &stats, &pkts);
+    assert!(got.len() > 1, "sentinel blocks must deliver early parts");
+    let mut rebuilt = Vec::new();
+    for f in &got {
+        rebuilt.extend_from_slice(&f.data);
+    }
+    assert_eq!(rebuilt, src);
+    assert!(got.last().unwrap().complete);
+}
+
 /// A sentinel first-packet commits a MAX-sized frame buffer, so the in-flight budget must
 /// bite after IN_FLIGHT_BUF_FACTOR frames — the amplification bound for one-datagram opens.
 #[test]

crates/punktfunk-core/src/session.rs

@@ -20,6 +20,27 @@ use crate::stats::{Stats, StatsCounters};
 use crate::transport::Transport;
 use zerocopy::IntoBytes;
 
+/// Slice-progressive delivery metadata ([`Session::set_deliver_frame_parts`]): this [`Frame`]
+/// carries one contiguous piece of an access unit, handed up while the rest is still on the
+/// wire so a `PARTIAL_FRAME`-capable decoder can start ahead of the last packet.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct FramePart {
+    /// Byte offset of `data` within the whole AU. The reassembler emits parts in order with
+    /// no gaps — but the pre-decode hand-off may drop entries under memory pressure or a
+    /// jump-to-live clear, so a consumer must still verify `offset` equals its open AU's next
+    /// expected byte and treat a mismatch as the AU lost (abandon + flush the decoder's
+    /// partial input, then discard until the next `first`). The same discard applies to a
+    /// non-`first` part arriving with no AU open.
+    pub offset: u32,
+    /// The AU's first part. A `first` part for a NEW `frame_index` while an earlier AU is
+    /// still open means that AU died mid-flight (aged out or was cleared) — abandon it and
+    /// flush the decoder's partial input; no explicit abort part is ever sent.
+    pub first: bool,
+    /// The AU's final part — the whole AU has now been delivered ([`Frame::complete`] is set
+    /// on exactly this part) and the input may be submitted to the decoder as finished.
+    pub last: bool,
+}
+
 /// A reassembled, FEC-recovered access unit, ready to hand to the platform decoder.
 pub struct Frame {
     pub data: Vec<u8>,
@@ -32,6 +53,10 @@ 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,
+    /// `Some` = slice-progressive delivery is on and this is ONE PIECE of an AU (see
+    /// [`FramePart`]); `data` holds only the part's bytes. `None` = a whole AU (or an
+    /// aged-out chunk-aligned partial — `complete` distinguishes).
+    pub part: Option<FramePart>,
     /// 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
@@ -611,6 +636,16 @@ impl Session {
         self.reassembler.set_deliver_partial(on);
     }
 
+    /// Client opt-in: deliver each AU's newly-contiguous prefix as [`Frame`]s with
+    /// [`Frame::part`]` = Some` while the rest is still on the wire, instead of one whole-AU
+    /// delivery (the slice-progressive decode path — [`crate::packet::USER_FLAG_SLICE_STREAM`]).
+    /// With it on, EVERY video frame delivery carries `part: Some` (a frame with no early
+    /// parts arrives as the degenerate `{offset: 0, first, last}` whole). Do not combine with
+    /// an all-intra (PyroWave) stream: its newest-wins draining assumes whole AUs.
+    pub fn set_deliver_frame_parts(&mut self, on: bool) {
+        self.reassembler.set_deliver_parts(on);
+    }
+
     /// The session's negotiated wire shard payload size (bytes of AU per datagram) —
     /// the window size for chunk-aligned AUs (`USER_FLAG_CHUNK_ALIGNED`).
     pub fn shard_payload(&self) -> usize {
@@ -710,7 +745,11 @@ impl Session {
                 p.packets += 1;
             }
             if let Some(frame) = pushed {
-                StatsCounters::add(&self.stats.frames_completed, 1);
+                // A prefix part is not a completed frame — only the delivery that closes the
+                // AU counts, or parts would multiply the completion rate.
+                if frame.complete {
+                    StatsCounters::add(&self.stats.frames_completed, 1);
+                }
                 return Ok(stamp_received(frame));
             }
             // A push that completed nothing may still have aged a partial out — deliver it

crates/punktfunk-host/src/native.rs

@@ -2282,17 +2282,18 @@ mod tests {
             mode,
             CompositorPref::Auto,
             GamepadPref::Auto,
-            0,    // bitrate_kbps
-            0,    // video_caps
-            2,    // audio_channels
-            0,    // video_codecs (HEVC-only)
-            0,    // preferred_codec
-            None, // display_hdr
-            0,    // client_caps
-            None, // launch
-            None, // name
-            None, // pin (TOFU)
-            None, // identity (host doesn't require pairing)
+            0,     // bitrate_kbps
+            0,     // video_caps
+            2,     // audio_channels
+            0,     // video_codecs (HEVC-only)
+            0,     // preferred_codec
+            None,  // display_hdr
+            0,     // client_caps
+            false, // frame_parts (whole-AU delivery)
+            None,  // launch
+            None,  // name
+            None,  // pin (TOFU)
+            None,  // identity (host doesn't require pairing)
             std::time::Duration::from_secs(10),
         )
         .expect("client connects to synthetic host");
@@ -2455,15 +2456,16 @@ mod tests {
             CompositorPref::Auto,
             GamepadPref::Auto,
             0,
-            0,    // video_caps
-            2,    // audio_channels (stereo)
-            0,    // video_codecs (0 → HEVC-only)
-            0,    // preferred_codec (auto)
-            None, // display_hdr
-            0,    // client_caps
-            None, // launch
-            None, // name: absent on purpose — this test asserts the fingerprint-derived label
-            None, // pin: TOFU — the operator's approval (not a PIN) authorizes this client
+            0,     // video_caps
+            2,     // audio_channels (stereo)
+            0,     // video_codecs (0 → HEVC-only)
+            0,     // preferred_codec (auto)
+            None,  // display_hdr
+            0,     // client_caps
+            false, // frame_parts (whole-AU delivery)
+            None,  // launch
+            None,  // name: absent on purpose — this test asserts the fingerprint-derived label
+            None,  // pin: TOFU — the operator's approval (not a PIN) authorizes this client
             Some((cert, key)),
             std::time::Duration::from_secs(15),
         )
@@ -2524,14 +2526,15 @@ mod tests {
                 CompositorPref::Auto,
                 GamepadPref::Auto,
                 0,
-                0,    // video_caps
-                2,    // audio_channels (stereo)
-                0,    // video_codecs
-                0,    // preferred_codec
-                None, // display_hdr
-                0,    // client_caps
-                None, // launch
-                None, // name
+                0,     // video_caps
+                2,     // audio_channels (stereo)
+                0,     // video_codecs
+                0,     // preferred_codec
+                None,  // display_hdr
+                0,     // client_caps
+                false, // frame_parts (whole-AU delivery)
+                None,  // launch
+                None,  // name
                 None,
                 None,
                 timeout
@@ -2555,14 +2558,15 @@ mod tests {
             CompositorPref::Auto,
             GamepadPref::Auto,
             0,
-            0,    // video_caps
-            2,    // audio_channels (stereo)
-            0,    // video_codecs
-            0,    // preferred_codec
-            None, // display_hdr
-            0,    // client_caps
-            None, // launch
-            None, // name
+            0,     // video_caps
+            2,     // audio_channels (stereo)
+            0,     // video_codecs
+            0,     // preferred_codec
+            None,  // display_hdr
+            0,     // client_caps
+            false, // frame_parts (whole-AU delivery)
+            None,  // launch
+            None,  // name
             Some(host_fp),
             Some((cert.clone(), key.clone())),
             timeout,