feat(rumble): host-authoritative self-terminating envelopes (0xCA v2)

Rumble was level-triggered, unbounded state on a lossy channel: a non-zero
level meant "buzz until further notice", healed only by the host re-sending
state every 500 ms, and every client guessed when the host had died with its own
magic timeout (SDL 1.5 s, Apple 1.6 s, Android up to 60 s). A lost stop, a
reordered start, or a dead host could drone the motor for seconds.

Make "stuck rumble" inexpressible on the wire. The 0xCA datagram grows a
length-tolerant tail — [u8 seq][u16 ttl_ms] — so it self-terminates: the host
authorizes a level for at most ttl_ms and renews it (~120 ms) while it holds,
letting an abandoned one lapse client-side. seq is a per-pad wrapping reorder
gate (reusing GamepadSnapshot::seq_newer) so a reordered stale start can't
re-light a stopped motor. Decoders read the first 7 bytes as a plain level and
ignore the tail, so no wire-version bump: an old client renders a new host's
levels, and a new client falls back to its prior staleness heuristic against an
old host (ttl = None). All four generation pairings render correctly.

- core: encode_rumble_datagram_v2 / decode_rumble_envelope (datagram.rs); the
  client demux applies the seq gate then forwards (pad, low, high, Option<ttl>);
  next_rumble is unchanged (drops ttl), next_rumble_ttl keeps it; ABI adds
  punktfunk_connection_next_rumble2 + PUNKTFUNK_RUMBLE_NO_TTL, ABI_VERSION 4->5
  (WIRE_VERSION unchanged — the tail is backward-compatible).
- host (punktfunk1.rs): the flat 500 ms refresh becomes a renewal loop that bumps
  seq + stamps a fresh TTL on active pads and drains a short post-stop zero burst,
  then goes quiet. Hatches: PUNKTFUNK_RUMBLE_ENVELOPE=0 (legacy v1 + flat refresh,
  a bisect switch), PUNKTFUNK_RUMBLE_TTL_MS (clamped [150, 5000]).
- renderers honor the TTL as their playback duration/deadline and keep their old
  heuristic only for a legacy (ttl=None) update: pf-client-core (the Deck haptic
  keep-alive is now deadline-bounded so it can't sustain a host-stopped rumble),
  clients/windows (SDL duration), android (JNI packs the lease out-of-band in bit
  48 so any u16 ttl is unambiguous; Kotlin createOneShot(ttl)), apple
  (RumbleRenderer.envelopeDeadline + nextRumble2; sessionStaleSeconds demoted to
  the legacy fallback).
- tests: codec round-trip + tail tolerance + seq-gate reorder (Rust); the probe
  asserts the v2 tail arrived under PUNKTFUNK_TEST_FEEDBACK; the Apple loopback
  asserts ttlMs round-trips end to end; RumbleTuning lease-decision cases.

The host-side idle-timeout from the previous commit is defense in depth on the
game side; this is the guarantee on the client side. Design:
punktfunk-planning/design/rumble-envelope-plan.md.

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

clients/probe/src/main.rs

@@ -1007,6 +1007,10 @@ async fn session(args: Args) -> Result<()> {
     let audio_bytes = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
     let rumble_pkts = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
     let hidout_pkts = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
+    // Set when a self-terminating v2 rumble envelope (0xCA with the seq+ttl tail) arrives — the
+    // Rust-side contract check for `PUNKTFUNK_TEST_FEEDBACK` (asserted at report time). A legacy v1
+    // datagram leaves it false, so this only ever fails when a v2 tail we EXPECTED went missing.
+    let saw_v2_rumble = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
     // Per-AU host timings (0xCF) → the stream loop, which matches them to received AUs by pts
     // and reports the host/network split. try_send: overflow drops samples, never blocks QUIC.
     let (host_timing_tx, host_timing_rx) =
@@ -1018,6 +1022,7 @@ async fn session(args: Args) -> Result<()> {
             rumble_pkts.clone(),
             hidout_pkts.clone(),
         );
+        let saw_v2 = saw_v2_rumble.clone();
         let ht_tx = host_timing_tx;
         let conn2 = conn.clone();
         // Build a multistream decoder for the host-RESOLVED layout so the probe actually decodes
@@ -1026,6 +1031,7 @@ async fn session(args: Args) -> Result<()> {
         tokio::spawn(async move {
             use std::sync::atomic::Ordering::Relaxed;
             let mut hdr_logged = false;
+            let mut rumble_logged = false;
             let layout = punktfunk_core::audio::layout_for(audio_channels, false);
             let mut audio_dec =
                 opus::MSDecoder::new(48_000, layout.streams, layout.coupled, layout.mapping).ok();
@@ -1051,7 +1057,24 @@ async fn session(args: Args) -> Result<()> {
                             Err(e) => tracing::debug!(error = %e, "probe audio decode"),
                         }
                     }
-                } else if punktfunk_core::quic::decode_rumble_datagram(&d).is_some() {
+                } else if let Some(u) = punktfunk_core::quic::decode_rumble_envelope(&d) {
+                    // Log the first rumble so a loopback test can see the self-terminating v2
+                    // envelope tail (seq + TTL) arrived, not just the level.
+                    if !rumble_logged {
+                        rumble_logged = true;
+                        tracing::info!(
+                            pad = u.pad,
+                            low = u.low,
+                            high = u.high,
+                            envelope = ?u.envelope,
+                            "rumble (0xCA)"
+                        );
+                    }
+                    // Record that a v2 tail was present — the Rust-side seq/ttl contract check for
+                    // PUNKTFUNK_TEST_FEEDBACK (asserted at report time).
+                    if u.envelope.is_some() {
+                        saw_v2.store(true, Relaxed);
+                    }
                     r.fetch_add(1, Relaxed);
                 } else if let Some(meta) = punktfunk_core::quic::decode_hdr_meta_datagram(&d) {
                     // HDR static metadata (0xCE). Log the first receipt so a loopback test can
@@ -1292,6 +1315,23 @@ async fn session(args: Args) -> Result<()> {
         }
     }
 
+    // Rust-side rumble-envelope contract check: when the host was told to script a feedback burst
+    // (PUNKTFUNK_TEST_FEEDBACK, shared by a loopback harness), fail if no self-terminating v2 tail
+    // (seq + TTL) arrived — a regression that reverted the host to v1 level datagrams increments the
+    // rumble counter identically and would otherwise pass silently. Only tightens an otherwise-OK run
+    // (a video failure stays the primary error). The level + hidout planes are asserted end-to-end by
+    // the Apple loopback; this covers the seq/ttl tail on the Rust/Linux path.
+    let result = if std::env::var("PUNKTFUNK_TEST_FEEDBACK").as_deref() == Ok("1")
+        && result.is_ok()
+        && !saw_v2_rumble.load(std::sync::atomic::Ordering::Relaxed)
+    {
+        Err(anyhow::anyhow!(
+            "PUNKTFUNK_TEST_FEEDBACK: expected a v2 rumble envelope (0xCA seq+ttl tail), received none"
+        ))
+    } else {
+        result
+    };
+
     // `--quit` closes with the deliberate-quit code so the host skips the keep-alive linger; a normal
     // exit uses code 0 (an unwanted-disconnect close → the host lingers for a reconnect).
     let close_code = if args.quit {