feat(core,host,client): PyroWave datagram-aligned packets + partial-frame delivery (Phase 4, §4.4)

PyroWave AUs now packetize on the negotiated shard payload, so a lost datagram
costs a few wavelet blocks of localized blur rather than a whole frame — and the
client can render an aged-out lossy frame instead of freezing until the next one.

Host (opt-in, PyroWave only):
- The encoder packetizes at the shard payload behind a 4-byte window prefix
  (used-len u16 + kind u16). Whole packets pack into WIN_PACKED windows; a packet
  too large for one shard (PyroWave 32x32 blocks are atomic and can exceed a
  shard) rides a WIN_FRAG_FIRST/CONT/LAST chain. `set_wire_chunking()` joins the
  Encoder trait (forwarded through TrackedEncoder — the silent-no-op trap);
  EncodedFrame.chunk_aligned marks the AU.
- virtual_stream tags the AU with USER_FLAG_CHUNK_ALIGNED and re-applies chunking
  after every encoder (re)build, the adaptive-bitrate rebuild included.

Core:
- USER_FLAG_CHUNK_ALIGNED (0x40) wire bit. Reassembler opt-in
  (set_deliver_partial): a chunk-aligned frame that ages out with holes is handed
  over as Frame{complete:false} — received shards at their exact offsets, missing
  ranges zero-filled — instead of being dropped. Partials age out on a tight 30ms
  fuse (PARTIAL_WINDOW_NS) instead of the 120ms loss window: each frame is
  independently decodable, so an ancient partial has no value in a live stream.
  Newest-wins. A partial still counts as dropped for loss reporting.

Client (PyroWave decode):
- The session opts in when codec == PyroWave. The decoder walks the AU
  window-by-window, skipping zero (missing) windows and reassembling FRAG chains,
  then decodes whatever survived. A newest-decoded-index guard drops partials the
  pump has already moved past (no time-travel present).

Also fixes a redundant-closure clippy nit in the PyroWave planar-present path.

Validated on an RTX 5070 Ti under 2% netem loss with FEC pinned off: 60fps
sustained entirely via partials, e2e 43ms p50 (146ms before the fuse) vs 23ms
lossless, no keyframe-recovery chatter. Tests green: core 149, host 310 + the
GPU-gated encoder smoke (framed-window walk + FRAG reassembly + upstream
round-trip), client 26; clippy clean on the pyrowave feature combos.

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

crates/punktfunk-host/src/encode.rs

@@ -29,6 +29,10 @@ pub struct EncodedFrame {
     /// clean re-anchor). Without it the client's freeze can only lift on an IDR — which the host
     /// suppresses after a successful RFI (the cooldown), a ~1 s frozen stall per loss event.
     pub recovery_anchor: bool,
+    /// The AU is shard-aligned self-delimiting chunks (see [`Encoder::set_wire_chunking`]);
+    /// the session stamps [`punktfunk_core::packet::USER_FLAG_CHUNK_ALIGNED`] so the client
+    /// windows its parse and may opt into partial delivery. Only the PyroWave backend sets it.
+    pub chunk_aligned: bool,
 }
 
 /// Codec selection negotiated with the client.
@@ -356,6 +360,13 @@ pub trait Encoder: Send {
     fn reconfigure_bitrate(&mut self, _bps: u64) -> bool {
         false
     }
+    /// Wire-chunk the encoder's AUs at the session's shard payload size (the PyroWave
+    /// datagram-aligned mode, plan §4.4): every `shard_payload` window of the emitted AU
+    /// starts a fresh self-delimiting codec packet, zero-padded to the window — so a lost
+    /// datagram costs a few coefficient blocks, not the frame. AUs produced this way are
+    /// flagged [`EncodedFrame::chunk_aligned`] and the session marks them on the wire.
+    /// Default: no-op (the H.26x backends' bitstreams cannot be cut losslessly).
+    fn set_wire_chunking(&mut self, _shard_payload: usize) {}
     /// Signal end-of-stream. After this, drain the remaining AUs with [`poll`](Self::poll)
     /// until it returns `None` — NVENC buffers frames internally even at `delay=0`.
     fn flush(&mut self) -> Result<()>;
@@ -519,6 +530,12 @@ impl Encoder for TrackedEncoder {
     fn invalidate_ref_frames(&mut self, first_frame: i64, last_frame: i64) -> bool {
         self.inner.invalidate_ref_frames(first_frame, last_frame)
     }
+    // The classic TrackedEncoder trap: a defaulted trait method that isn't forwarded
+    // silently no-ops through the wrapper (bit the direct-NVENC work, then THIS — the
+    // §4.4 chunking probe run hit the default while the plan said Some(1408)).
+    fn set_wire_chunking(&mut self, shard_payload: usize) {
+        self.inner.set_wire_chunking(shard_payload)
+    }
     fn poll(&mut self) -> Result<Option<EncodedFrame>> {
         self.inner.poll()
     }

crates/punktfunk-host/src/encode/linux/mod.rs

@@ -652,6 +652,7 @@ impl Encoder for NvencEncoder {
                     pts_ns,
                     keyframe: pkt.is_key(),
                     recovery_anchor: false,
+                    chunk_aligned: false,
                 }))
             }
             // No packet ready yet (need another input frame).

crates/punktfunk-host/src/encode/linux/nvenc_cuda.rs

@@ -1148,6 +1148,7 @@ impl Encoder for NvencCudaEncoder {
                 pts_ns,
                 keyframe,
                 recovery_anchor: anchor,
+                chunk_aligned: false,
             }))
         }
     }

crates/punktfunk-host/src/encode/linux/pyrowave.rs

@@ -43,6 +43,13 @@ const IMPORT_CACHE_CAP: usize = 16;
 /// Headroom over the per-frame rate budget for the packetized bitstream (block headers + meta;
 /// the rate controller itself never exceeds the budget).
 const BS_SLACK: usize = 256 * 1024;
+/// Chunked-mode window framing (§4.4): 4-byte prefix per shard-sized window.
+const WINDOW_PREFIX: usize = 4;
+/// Window kinds: whole packets / an oversized packet's fragments.
+const WIN_PACKED: u16 = 0;
+const WIN_FRAG_FIRST: u16 = 1;
+const WIN_FRAG_CONT: u16 = 2;
+const WIN_FRAG_LAST: u16 = 3;
 
 /// The DRM modifiers the PyroWave device can import as a SAMPLED image of the capture's
 /// packed-RGB format. The capture advertises these for the pyrowave passthrough instead of
@@ -178,6 +185,10 @@ pub struct PyroWaveEncoder {
     fps: u32,
     /// Per-frame bitstream budget (hard CBR): `bitrate / (8 * fps)`.
     frame_budget: usize,
+    /// Datagram-aligned mode (plan §4.4): packetize at this boundary and pad every codec
+    /// packet to it, so each wire shard carries whole self-delimiting packets. `None` =
+    /// one packet per AU (the dense MVP shape).
+    wire_chunk: Option<usize>,
     bitstream: Vec<u8>,
     pending: VecDeque<EncodedFrame>,
     frame_count: u64,
@@ -532,6 +543,7 @@ impl PyroWaveEncoder {
             height: h,
             fps,
             frame_budget,
+            wire_chunk: None,
             bitstream: Vec::new(),
             pending: VecDeque::new(),
             frame_count: 0,
@@ -865,31 +877,100 @@ impl PyroWaveEncoder {
         dev.wait_for_fences(&[self.fence], true, 5_000_000_000)
             .context("pyrowave encode fence")?;
 
-        // ---- packetize: boundary = whole buffer, so the AU is exactly one pyrowave packet ----
+        // ---- packetize ----
+        // Dense (default): boundary = whole buffer → the AU is exactly one pyrowave packet.
+        // Datagram-aligned (§4.4, `set_wire_chunking`): boundary = the wire shard payload;
+        // each codec packet is zero-padded to the boundary so every shard carries whole
+        // self-delimiting packets — the client windows its parse and a lost shard costs
+        // only those blocks. Padding cost is small: the packetizer fills close to the
+        // boundary by design.
         let cap = self.frame_budget + BS_SLACK;
         self.bitstream.resize(cap, 0);
+        // Chunked mode reserves 4 bytes per window for the framing prefix.
+        let boundary = self.wire_chunk.map(|c| c - WINDOW_PREFIX).unwrap_or(cap);
         let mut n: usize = 0;
         pw_check(
-            pw::pyrowave_encoder_compute_num_packets(self.pw_enc, cap, &mut n),
+            pw::pyrowave_encoder_compute_num_packets(self.pw_enc, boundary, &mut n),
             "compute_num_packets",
         )?;
-        if n != 1 {
-            bail!("pyrowave: expected a single packet at boundary {cap}, got {n}");
+        if n == 0 || (self.wire_chunk.is_none() && n != 1) {
+            bail!("pyrowave: unexpected packet count {n} at boundary {boundary}");
         }
-        let mut packet = pw::pyrowave_packet { offset: 0, size: 0 };
+        let mut packets = vec![pw::pyrowave_packet { offset: 0, size: 0 }; n];
         let mut out_n: usize = 0;
         pw_check(
             pw::pyrowave_encoder_packetize(
                 self.pw_enc,
-                &mut packet,
-                cap,
+                packets.as_mut_ptr(),
+                boundary,
                 &mut out_n,
                 self.bitstream.as_mut_ptr() as *mut std::ffi::c_void,
                 cap,
             ),
             "packetize",
         )?;
-        let au = self.bitstream[packet.offset..packet.offset + packet.size].to_vec();
+        packets.truncate(out_n.max(1));
+        let au = if let Some(chunk) = self.wire_chunk {
+            // Window framing (§4.4): each `chunk`-sized window opens with a 4-byte prefix
+            // (u16 used-length + u16 kind) and carries either WHOLE self-delimiting codec
+            // packets (PACKED — several small ones share a window) or one fragment of an
+            // oversized packet (FRAG chain — pyrowave 32×32 blocks are atomic and may
+            // exceed a shard). A lost shard zeroes its window (used = 0) — the receiver
+            // skips it and drops any fragment chain it interrupts.
+            let payload_max = chunk - WINDOW_PREFIX;
+            let mut au: Vec<u8> = Vec::with_capacity((packets.len() + 1) * chunk);
+            // The currently-open PACKED window: (start offset of its prefix, bytes used).
+            let mut open: Option<(usize, usize)> = None;
+            let close = |au: &mut Vec<u8>, open: &mut Option<(usize, usize)>, chunk: usize| {
+                if let Some((start, used)) = open.take() {
+                    au[start..start + 2].copy_from_slice(&(used as u16).to_le_bytes());
+                    au[start + 2..start + 4].copy_from_slice(&WIN_PACKED.to_le_bytes());
+                    au.resize(start + chunk, 0);
+                }
+            };
+            for p in &packets {
+                let bytes = &self.bitstream[p.offset..p.offset + p.size];
+                if p.size <= payload_max {
+                    let fits = open.is_some_and(|(_, used)| used + p.size <= payload_max);
+                    if !fits {
+                        close(&mut au, &mut open, chunk);
+                        let start = au.len();
+                        au.resize(start + WINDOW_PREFIX, 0);
+                        open = Some((start, 0));
+                    }
+                    au.extend_from_slice(bytes);
+                    if let Some((_, used)) = open.as_mut() {
+                        *used += p.size;
+                    }
+                } else {
+                    // Oversized packet: its own FRAG chain of full windows.
+                    close(&mut au, &mut open, chunk);
+                    let mut off = 0usize;
+                    while off < p.size {
+                        let take = (p.size - off).min(payload_max);
+                        let kind = if off == 0 {
+                            WIN_FRAG_FIRST
+                        } else if off + take == p.size {
+                            WIN_FRAG_LAST
+                        } else {
+                            WIN_FRAG_CONT
+                        };
+                        let start = au.len();
+                        au.resize(start + WINDOW_PREFIX, 0);
+                        au[start..start + 2].copy_from_slice(&(take as u16).to_le_bytes());
+                        au[start + 2..start + 4].copy_from_slice(&kind.to_le_bytes());
+                        au.extend_from_slice(&bytes[off..off + take]);
+                        au.resize(start + chunk, 0);
+                        off += take;
+                    }
+                }
+            }
+            close(&mut au, &mut open, chunk);
+            au
+        } else {
+            let p = &packets[0];
+            self.bitstream[p.offset..p.offset + p.size].to_vec()
+        };
         self.frame_count += 1;
         self.pending.push_back(EncodedFrame {
             data: au,
@@ -898,6 +979,7 @@ impl PyroWaveEncoder {
             // whole recovery story (plan §1.2).
             keyframe: true,
             recovery_anchor: false,
+            chunk_aligned: self.wire_chunk.is_some(),
         });
         Ok(())
     }
@@ -960,6 +1042,17 @@ impl Encoder for PyroWaveEncoder {
         true
     }
 
+    fn set_wire_chunking(&mut self, shard_payload: usize) {
+        // Sanity floor: a boundary below one block header + payload word is meaningless.
+        if shard_payload >= 64 {
+            self.wire_chunk = Some(shard_payload);
+            tracing::info!(
+                shard_payload,
+                "pyrowave: datagram-aligned packetization on (partial-frame loss mode)"
+            );
+        }
+    }
+
     fn flush(&mut self) -> Result<()> {
         // Synchronous per-frame encode: nothing buffered beyond `pending`.
         Ok(())
@@ -1127,6 +1220,83 @@ mod tests {
             );
         }
 
+        // Datagram-aligned mode (§4.4): every emitted AU is a whole number of framed
+        // windows — 4-byte prefix (used-length + kind), whole packets or FRAG chains for
+        // oversized atomic blocks, zero padding after `used`. Walking + reassembling the
+        // fragments must reproduce a decodable packet stream.
+        enc.set_wire_chunking(1408);
+        enc.submit(&cpu_frame(w, h, 500, [90, 60, 30, 255]))
+            .expect("chunked submit");
+        let au = enc.poll().expect("poll").expect("chunked AU");
+        assert!(au.chunk_aligned);
+        assert_eq!(au.data.len() % 1408, 0, "AU is a whole number of windows");
+        // SAFETY: test-only FFI with locally-owned buffers.
+        unsafe {
+            let mut dev: pw::pyrowave_device = std::ptr::null_mut();
+            assert_eq!(
+                pw::pyrowave_create_default_device(&mut dev),
+                pw::pyrowave_result_PYROWAVE_SUCCESS
+            );
+            let dinfo = pw::pyrowave_decoder_create_info {
+                device: dev,
+                width: w as i32,
+                height: h as i32,
+                chroma: pw::pyrowave_chroma_subsampling_PYROWAVE_CHROMA_SUBSAMPLING_420,
+                fragment_path: false,
+            };
+            let mut dec: pw::pyrowave_decoder = std::ptr::null_mut();
+            assert_eq!(
+                pw::pyrowave_decoder_create(&dinfo, &mut dec),
+                pw::pyrowave_result_PYROWAVE_SUCCESS
+            );
+            let mut frag: Vec<u8> = Vec::new();
+            let mut pushed = 0usize;
+            for win in au.data.chunks(1408) {
+                let used = u16::from_le_bytes([win[0], win[1]]) as usize;
+                let kind = u16::from_le_bytes([win[2], win[3]]);
+                assert!(4 + used <= win.len(), "window overrun");
+                assert!(win[4 + used..].iter().all(|&b| b == 0), "non-zero padding");
+                let body = &win[4..4 + used];
+                match kind {
+                    0 => {
+                        assert_eq!(
+                            pw::pyrowave_decoder_push_packet(
+                                dec,
+                                body.as_ptr() as *const _,
+                                body.len()
+                            ),
+                            pw::pyrowave_result_PYROWAVE_SUCCESS
+                        );
+                        pushed += body.len();
+                    }
+                    1 => frag = body.to_vec(),
+                    2 => frag.extend_from_slice(body),
+                    3 => {
+                        frag.extend_from_slice(body);
+                        assert_eq!(
+                            pw::pyrowave_decoder_push_packet(
+                                dec,
+                                frag.as_ptr() as *const _,
+                                frag.len()
+                            ),
+                            pw::pyrowave_result_PYROWAVE_SUCCESS
+                        );
+                        pushed += frag.len();
+                        frag.clear();
+                    }
+                    k => panic!("unknown window kind {k}"),
+                }
+            }
+            assert!(pushed > 0, "chunked AU carries real packets");
+            assert!(
+                pw::pyrowave_decoder_decode_is_ready(dec, false),
+                "chunked AU incomplete after framed walk"
+            );
+            pw::pyrowave_decoder_destroy(dec);
+            pw::pyrowave_device_destroy(dev);
+        }
+        enc.set_wire_chunking(0); // below the floor — back to dense
+
         // In-place rate retarget + encoder rebuild both keep encoding.
         assert!(enc.reconfigure_bitrate(100_000_000));
         assert!(enc.reset());

crates/punktfunk-host/src/encode/linux/vaapi.rs

@@ -297,6 +297,7 @@ fn poll_encoder(enc: &mut encoder::video::Encoder, fps: u32) -> Result<Option<En
                 pts_ns: pts * 1_000_000_000 / fps as u64,
                 keyframe: pkt.is_key(),
                 recovery_anchor: false,
+                chunk_aligned: false,
             }))
         }
         Err(ffmpeg::Error::Other { errno })

crates/punktfunk-host/src/encode/linux/vulkan_video.rs

@@ -1713,6 +1713,7 @@ impl VulkanVideoEncoder {
             pts_ns: f.pts_ns,
             keyframe: f.keyframe,
             recovery_anchor: f.recovery_anchor,
+            chunk_aligned: false,
         })
     }
 

crates/punktfunk-host/src/encode/sw.rs

@@ -216,6 +216,7 @@ impl Encoder for OpenH264Encoder {
                 pts_ns,
                 keyframe,
                 recovery_anchor: false,
+                chunk_aligned: false,
             });
         }
         self.frame_idx += 1;

crates/punktfunk-host/src/encode/windows/amf.rs

@@ -1937,6 +1937,7 @@ unsafe fn drain_one_output(
         pts_ns,
         keyframe: key_prop || forced,
         recovery_anchor,
+        chunk_aligned: false,
     }))
 }
 

crates/punktfunk-host/src/encode/windows/ffmpeg_win.rs

@@ -340,6 +340,7 @@ fn poll_encoder(enc: &mut encoder::video::Encoder, fps: u32) -> Result<PollOutco
                 pts_ns: pts * 1_000_000_000 / fps as u64,
                 keyframe: pkt.is_key(),
                 recovery_anchor: false,
+                chunk_aligned: false,
             }))
         }
         Err(ffmpeg::Error::Other { errno })

crates/punktfunk-host/src/encode/windows/nvenc.rs

@@ -1224,6 +1224,7 @@ impl NvencD3d11Encoder {
                 pts_ns,
                 keyframe,
                 recovery_anchor: anchor,
+                chunk_aligned: false,
             });
         Ok(())
     }
@@ -1649,6 +1650,7 @@ impl Encoder for NvencD3d11Encoder {
                 pts_ns,
                 keyframe,
                 recovery_anchor: anchor,
+                chunk_aligned: false,
             }))
         }
     }

crates/punktfunk-host/src/pipeline.rs

@@ -27,6 +27,7 @@ pub fn pump_once(
         pts_ns,
         keyframe,
         recovery_anchor,
+        chunk_aligned: _,
     }) = encoder.poll()?
     {
         let mut flags = FLAG_PIC as u32;

crates/punktfunk-host/src/punktfunk1.rs

@@ -3978,7 +3978,12 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
     // path now reads this typed `SessionPlan` instead of re-deriving from config at each dispatch site
     // (the latent "capture and encode disagree on the backend" hazard, plan §2.4). `bit_depth` is the
     // only per-session input — capture/topology/encoder are otherwise pure functions of `HostConfig`.
-    let plan = crate::session_plan::SessionPlan::resolve(ctx.bit_depth, ctx.chroma, ctx.codec);
+    let mut plan = crate::session_plan::SessionPlan::resolve(ctx.bit_depth, ctx.chroma, ctx.codec);
+    // PyroWave rides the datagram-aligned wire mode (§4.4): every encoder this session opens
+    // packetizes at the negotiated shard payload, so a lost datagram costs blocks, not frames.
+    if ctx.codec == crate::encode::Codec::PyroWave {
+        plan.wire_chunk = Some(ctx.session.shard_payload());
+    }
     tracing::info!(?plan, "resolved session plan");
     // Single-process path: unpack the context into the locals the loop below uses (names unchanged, so the
     // body is byte-for-byte the same; the receivers are now owned but `try_recv()` is identical).
@@ -4430,12 +4435,15 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
                     bit_depth,
                     plan.chroma,
                 ) {
-                    Ok(new_enc) => {
+                    Ok(mut new_enc) => {
                         tracing::info!(
                             from_kbps = bitrate_kbps,
                             to_kbps = new_kbps,
                             "encoder rebuilt at new bitrate (adaptive bitrate)"
                         );
+                        if let Some(c) = plan.wire_chunk {
+                            new_enc.set_wire_chunking(c);
+                        }
                         enc = new_enc;
                         bitrate_kbps = new_kbps;
                         live_bitrate.store(new_kbps, Ordering::Relaxed);
@@ -4866,6 +4874,11 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
             if au.recovery_anchor {
                 flags |= punktfunk_core::packet::USER_FLAG_RECOVERY_ANCHOR;
             }
+            // Datagram-aligned PyroWave AU (plan §4.4): the client windows its parse at the
+            // shard payload and may opt into partial delivery of lossy frames.
+            if au.chunk_aligned {
+                flags |= punktfunk_core::packet::USER_FLAG_CHUNK_ALIGNED;
+            }
             // Re-send the HDR mastering metadata (0xCE) on each keyframe (a decoder-resync point) and
             // whenever it changed, so a client that dropped the best-effort datagram re-converges.
             if let Some(m) = last_hdr_meta {
@@ -5194,7 +5207,7 @@ fn build_pipeline(
     };
     // `bit_depth` is the handshake-negotiated value (8, or 10 = HEVC Main10 when the client
     // advertised VIDEO_CAP_10BIT and the host opted in). Threaded down from the Welcome.
-    let enc = crate::encode::open_video(
+    let mut enc = crate::encode::open_video(
         plan.codec,
         frame.format,
         frame.width,
@@ -5206,6 +5219,9 @@ fn build_pipeline(
         plan.chroma,
     )
     .context("open video encoder")?;
+    if let Some(c) = plan.wire_chunk {
+        enc.set_wire_chunking(c);
+    }
     // Post-open cross-check: the Welcome already committed `chroma_format` from the pre-open probe, so
     // warn loudly if the encoder actually opened a different chroma than negotiated (the in-band SPS is
     // authoritative for the decoder, but a mismatch means the probe and the live open disagreed).

crates/punktfunk-host/src/session_plan.rs

@@ -97,6 +97,10 @@ pub struct SessionPlan {
     /// Handshake-negotiated video codec the encoder emits — HEVC by default, H.264 for a GPU-less
     /// software host (`resolve_codec` over the client's advertised codecs ∩ the host's capability).
     pub codec: crate::encode::Codec,
+    /// Datagram-aligned wire chunking for the encoder (plan §4.4): `Some(shard_payload)` on a
+    /// PyroWave session — applied to EVERY encoder this plan opens (initial + all rebuilds) so
+    /// AUs stay shard-aligned across mode/bitrate/stall rebuilds. `None` for the H.26x codecs.
+    pub wire_chunk: Option<usize>,
 }
 
 impl SessionPlan {
@@ -115,6 +119,7 @@ impl SessionPlan {
             hdr: bit_depth >= 10,
             chroma,
             codec,
+            wire_chunk: None,
         }
     }