fix(desktop): recover frame references and bound relay qualification

crates/rds-desktop/tests/session_v2.rs

@@ -3,8 +3,8 @@
 //! Two `rds_net` endpoints, one running `serve_desktop_with` against a
 //! `SyntheticProducer`, the other a `DesktopSession`. A shared
 //! `SessionClock` makes `FrameHeader` timestamps directly comparable to
-//! client receive times, so viewer-visible latency is measured, not
-//! inferred.
+//! client receive times. This measures complete header arrival, not decoded
+//! pixels or viewer presentation.
 //!
 //! The impairment lane runs on the owned transport (`noq`) with
 //! `impair::ImpairingSocket` wrapping each endpoint's UDP socket:
@@ -392,9 +392,9 @@ async fn view_only_and_failed_injection_never_ack_but_keep_control_alive() {
 }
 
 /// C5 impairment + G5 latency gate: 5% loss + 30 ms jitter on a 50 ms
-/// base — queue stays bounded, stale frames drop, viewer-visible latency
-/// p95 stays in the 150 ms budget and control RTT is unaffected by the
-/// video backlog.
+/// base — queue stays bounded and control RTT is unaffected by the video
+/// backlog. Header-arrival tail bounds below include retransmission; this is
+/// not the clean-link 150 ms gate or an input-to-visible measurement.
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn impaired_link_latency_gate() {
     // 60 fps of ~1 KB frames ≈ one datagram per frame — enough samples
@@ -460,8 +460,8 @@ async fn impaired_link_latency_gate() {
         "too few frames arrived: {}",
         latencies.len()
     );
-    // Protocol queues stay bounded: capture→send wait is the collapse
-    // + channel time only — must stay near zero even under loss.
+    // Protocol queues stay bounded: capture→send wait is the bounded
+    // channel and pacing time — must stay near zero even under loss.
     let queue_p95 = p95(queue_ms);
     let lat_p50 = p50(latencies.clone());
     assert!(
@@ -524,11 +524,11 @@ async fn soak_60fps() {
         p99_age,
         rss_peak
     );
-    // Starvation check: the collapse may legitimately shed frames under
+    // Starvation check: bounded admission may legitimately shed frames under
     // CPU contention, so the floor is sustained flow, not offered rate —
     // a stalled pipeline delivers ~0.
     assert!(ages.len() >= secs as usize * 10, "starved: {}", ages.len());
-    // G5: viewer-visible latency ≤150 ms p95 in-process on a clean link.
+    // G5 synthetic header-arrival age ≤150 ms p95 on a clean loopback link.
     assert!(
         p95_age <= 150,
         "frame age p95 {p95_age}ms exceeds G5 budget"

crates/rds-relay/tests/owned_e2e.rs

@@ -150,36 +150,51 @@ async fn attached_relay_after_silent_direct(dual_stack: bool) {
 
 #[tokio::test]
 async fn relay_forwards_handshake_and_datagrams() {
+    async fn phase<T>(name: &str, work: impl Future<Output = T>) -> T {
+        eprintln!("owned relay fixture: starting {name}");
+        let result = tokio::time::timeout(std::time::Duration::from_secs(5), work)
+            .await
+            .unwrap_or_else(|_| panic!("owned relay fixture timed out during {name}"));
+        eprintln!("owned relay fixture: completed {name}");
+        result
+    }
+
     let _ = tracing_subscriber::fmt()
         .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
         .try_init();
-    let relay = rds_relay::server::serve(
-        EndpointConfig {
-            backend: Backend::Noq,
-            secret_key: Some(key(0)),
-            bind_addrs: vec!["127.0.0.1:0".parse().unwrap()],
-            ..Default::default()
-        },
-        Vec::new(),
+    let relay = phase(
+        "relay startup",
+        rds_relay::server::serve(
+            EndpointConfig {
+                backend: Backend::Noq,
+                secret_key: Some(key(0)),
+                bind_addrs: vec!["127.0.0.1:0".parse().unwrap()],
+                ..Default::default()
+            },
+            Vec::new(),
+        ),
     )
     .await
     .expect("relay up");
 
-    let a = endpoint(1, relay.endpoint_addr()).await;
-    let b = endpoint(2, relay.endpoint_addr()).await;
+    let a = phase("attach A", endpoint(1, relay.endpoint_addr())).await;
+    let b = phase("attach B", endpoint(2, relay.endpoint_addr())).await;
     assert_eq!(relay.endpoints(), 2, "both endpoints attached");
 
     // Exercise the directory boundary too: owned relay locators have their own
     // scheme and public identity, and must survive signed announce/resolve.
-    let directory = rds_discovery::service::serve(
-        "127.0.0.1:0".parse().unwrap(),
-        std::sync::Arc::new(rds_discovery::MemoryStore::default()),
-        rds_discovery::service::ServiceConfig::open_ephemeral(),
+    let directory = phase(
+        "directory startup",
+        rds_discovery::service::serve(
+            "127.0.0.1:0".parse().unwrap(),
+            std::sync::Arc::new(rds_discovery::MemoryStore::default()),
+            rds_discovery::service::ServiceConfig::open_ephemeral(),
+        ),
     )
     .await
     .unwrap();
     let client = rds_discovery::client::Client::new(directory.addr());
-    let _announce = rds_net::announce(
+    let announce = rds_net::announce(
         b.clone(),
         rds_net::AnnounceConfig {
             issuer: rds_discovery::RecordIssuer::memory(ed25519_dalek::SigningKey::from_bytes(
@@ -206,45 +221,56 @@ async fn relay_forwards_handshake_and_datagrams() {
     let target = relay_only(resolved);
     // Accept must be polled while connect is in flight: the server's
     // endpoint only drives the handshake once the incoming is taken.
-    let accept_b = tokio::spawn({
-        let b = b.clone();
-        async move { b.accept().await.expect("incoming").await }
-    });
-    let conn_a = a.connect(target, ALPN).await.expect("connect over relay");
+    let (conn_a, conn_b) = phase("peer handshake", async {
+        tokio::join!(a.connect(target, ALPN), async {
+            b.accept().await.expect("incoming").await
+        })
+    })
+    .await;
+    let conn_a = conn_a.expect("connect over relay");
     assert_eq!(conn_a.remote_id(), b.id());
 
-    let conn_b = accept_b.await.unwrap().expect("accept");
+    let conn_b = conn_b.expect("accept");
     assert_eq!(conn_b.remote_id(), a.id());
 
     // Datagrams both ways — payload is opaque outer-QUIC to the relay.
     conn_a.send_datagram(b"hello".to_vec().into()).unwrap();
-    let got = conn_b.read_datagram().await.unwrap();
+    let got = phase("datagram A to B", conn_b.read_datagram())
+        .await
+        .unwrap();
     assert_eq!(&got[..], b"hello");
     conn_b.send_datagram(b"world".to_vec().into()).unwrap();
-    let got = conn_a.read_datagram().await.unwrap();
+    let got = phase("datagram B to A", conn_a.read_datagram())
+        .await
+        .unwrap();
     assert_eq!(&got[..], b"world");
 
     // Streams ride the same tunnel.
-    let (mut send, mut recv) = conn_a.open_bi().await.unwrap();
-    send.write_all(b"stream-data").await.unwrap();
-    send.finish().unwrap();
-    let (mut bs, mut br) = conn_b.accept_bi().await.unwrap();
-    let mut buf = vec![0u8; 11];
-    br.read_exact(&mut buf).await.unwrap();
-    assert_eq!(&buf, b"stream-data");
-    bs.write_all(b"ack").await.unwrap();
-    bs.finish().unwrap();
-    let ack = recv.read_to_end(usize::MAX).await.unwrap();
-    assert_eq!(&ack, b"ack");
+    phase("reliable stream roundtrip", async {
+        let (mut send, mut recv) = conn_a.open_bi().await.unwrap();
+        send.write_all(b"stream-data").await.unwrap();
+        send.finish().unwrap();
+        let (mut bs, mut br) = conn_b.accept_bi().await.unwrap();
+        let mut buf = vec![0u8; 11];
+        br.read_exact(&mut buf).await.unwrap();
+        assert_eq!(&buf, b"stream-data");
+        bs.write_all(b"ack").await.unwrap();
+        bs.finish().unwrap();
+        let ack = recv.read_to_end(3).await.unwrap();
+        assert_eq!(&ack, b"ack");
+    })
+    .await;
 
     let (forwarded, dropped, bytes) = relay.stats();
     assert!(forwarded > 0, "relay must have forwarded datagrams");
     assert!(bytes > 0);
     assert_eq!(dropped, 0);
 
-    a.close().await;
-    b.close().await;
-    relay.close().await.unwrap();
+    drop(announce);
+    phase("close A", a.close()).await;
+    phase("close B", b.close()).await;
+    phase("close relay", relay.close()).await.unwrap();
+    phase("close directory", directory.close()).await.unwrap();
 }
 
 #[tokio::test]

docs/desktop-frame-delivery.md

@@ -1,11 +1,22 @@
 # Desktop frame sender contract
 
-Scope: partial writes and serving-session cancellation in W6.1/W2.5/W2.6.
-This does not close codec reference-chain or presentation gates.
+Scope: partial writes, reference recovery and serving-session cancellation in
+W6.1/W2.5/W2.6. This does not close real-network codec or presentation gates.
 
 Each encoded frame has one tagged unidirectional stream. An in-flight keyframe
-finishes before a newer frame is selected; a stale delta is abandoned with
-RESET. When the producer closes during a send, its final frame finishes.
+finishes before a newer frame is selected. A delta also finishes when the next
+candidate is another delta, because the successor may depend on it. An
+independent keyframe may replace an in-flight delta with RESET. When the producer
+closes during a send, its final frame finishes.
+
+The producer queue remains two slots. Losing any encoded frame to admission
+backpressure requests an IDR. Queue collapse selects a recent keyframe when
+available and requests recovery for discarded references after that keyframe;
+each drain is bounded to two items. The sender admits deltas only after a
+keyframe and only at the exact next sequence. Gaps, empty encode results and
+sequence exhaustion invalidate the chain. Dependent deltas wait for an IDR,
+without first being sent as undecodable work or requiring a client roundtrip.
+Selection is checked again after pacing, when additional frames may be dropped.
 
 The payload writer retains one pinned `write_all` future across producer
 events. `write_all` may already have accepted a prefix when its other select
@@ -31,16 +42,21 @@ write deadline. No detached worker retains a connection after async teardown.
 Regression coverage uses a 64-byte duplex buffer and a 4096-byte patterned
 payload to force a partial write before each producer event. Both supersession
 and closure previously delivered 4160 bytes; both now preserve exact bytes.
-Other tests retain errors after the producer event, abandon stale deltas without
+Other tests retain errors after the producer event, preserve a partial delta
+before its successor, replace a delta with an independent keyframe without
 waiting for a reader, and observe RESET after aborting an owned sender on real
 iroh/noq connections while a subsequent stream still succeeds.
 
 A separate real-session regression observes the producer being dropped after
 server cancellation with the client connection still alive; this failed before
 task-group ownership was added and passes on both backends.
 
-Remaining work includes reference-aware queue collapsing with real codec continuity tests, per-session
-wire IDs, global encoded/decoded/native allocation accounting, managed viewers,
+Native H.264 coverage checks decode and a pixel across a dropped reference and
+IDR recovery. This is separate from the synthetic QUIC impairment measurements;
+it does not establish real-codec continuity under network loss/reordering.
+
+Remaining work includes real-codec network continuity and overload/IDR-rate
+qualification, per-session wire IDs, global encoded/decoded/native allocation accounting, managed viewers,
 rendering/input release and platform/network acceptance. See the
 [client receive contract](desktop-client-lifecycle.md). No wire format, frame
 priority, dependency, installed binary or published release changes here.

docs/remediation-progress.md

@@ -51,7 +51,7 @@ Neither increment closes these product gaps or any wave.
 | W3.6 | Partial; validated selection and local child failure isolation | Extra paths become eligible on Established; weak policy ownership includes bounded backoff for temporary connection-ID/path-credit exhaustion and candidate-address snapshots. Relay-link route loss and known tunnel closure retire stale relay selection. Mux child send/receive failures are isolated; policy withdraws failed advertisements, excludes failed routes even when last-path close is refused, and closes held connections after all-child loss. Real loopback fixtures preserve open-stream traffic and accept new connections on the surviving child. Policy-observed telemetry exposes sticky event loss and unknown selection. Full path-event resynchronization, lossless retirement metrics, interface/socket recreation, per-service scheduling and physical-network qualification remain open. |
 | W4.4 | Partial; directional service boundaries | Real iroh/noq agents refuse writes with `SyncRead` and reads with `SyncWrite`, permit authorized transfers, and remain usable after refusal/cancellation. A view-only desktop never calls its input sink; failed injection never emits a success ACK. Linux/macOS account isolation, per-path policy, native seat/focus boundaries, consent and concurrent-role qualification remain open. See [receipt](reports/rds-service-scopes-20260926.md). |
 | W5.1/W5.3/W5.5 | Partial; native SSH client and standard PTY | `rds-ssh` uses russh 0.63.3 over pinned managed/direct streams. Explicit host pins, key/agent authentication, PTY/exec acknowledgements, terminal restoration, resize, cancellation and complete exit/output handling have Linux regression coverage and an OpenSSH interop fixture. SSH-specific fixed telemetry names are accepted by Vector; JSON uses a separate private file and terminal console logging pauses during SSH. GDS host/account provisioning, certificates/MFA, native macOS, broker/reattachment and mixed-load/network qualification remain open. See [contract](ssh.md). |
-| W6.1 | Partial; sender partial-write correctness | A pinned payload write preserves progress across supersession/producer closure; write/FIN errors fail, and a frame guard resets on cancellation/error under one 30-second deadline. The serving future owns capture/writer/pacing tasks and resets its control stream on cancellation; normal exit joins async siblings. Failing-before byte/capture regressions and real iroh/noq RESET coverage pass. Reference-aware queue dropping, wire session IDs and native presentation/network gates remain open; see [contract](desktop-frame-delivery.md). |
+| W6.1 | Partial; sender byte correctness and reference recovery | Pinned writes preserve progress; deltas finish before dependent successors and may be replaced by an independent keyframe. Queue loss requests IDR, and a sender sequence guard refuses broken delta chains after selection/pacing. Failure/cancellation resets streams under one deadline; serving owns child tasks. Failing-before byte/lifecycle regressions, native codec recovery and real iroh/noq RESET coverage pass. Real-codec network/overload qualification, wire session IDs and native presentation gates remain open; see [contract](desktop-frame-delivery.md). |
 | W6.2 | Partial; client receive ownership and limits | Four encoded frames per session/eight per process, two blocking decoder calls, owned task groups and cancellation, full handshake deadlines, dimension/sequence checks and rate-limited keyframe recovery are implemented. Real iroh/noq lifecycle and native codec regressions passed locally. Per-session wire IDs, global decoded/native memory accounting, renderer and native platform/network acceptance remain open; see [contract](desktop-client-lifecycle.md). |
 | W6.4 | Partial; ACK semantics and native X11 mapping | ACK follows successful backend acceptance. A bounded session-owned input worker reuses its sink. X11 maps evdev keys/buttons, bounds coordinates and fractional scroll, uses relative motion correctly, selects the explicit screen, checks native errors and releases owned holds on drop. Four failing-before native regressions and two-screen Xvfb checks cover the increment; the Linux CI lane requires actual native execution. Exclusive seat/controller ownership, focus races, custom layouts/IME, dynamic geometry and input-to-visible qualification remain open; see [contract](x11-input.md). |
 | W10.1/W10.2 | Partial; O1 foundation and O2 admin/source increment | Shared bounded Rust telemetry and Vector/OpenObserve pipeline; opt-in authenticated loopback metrics on agent/relay/server, aggregate source observations and old public metrics removal. Durable policy/catalog observations and effective agent revocation revision/lease are implemented. Finer queue/task and upstream adapter coverage, phase/reason correlation, support bundles, private rollout, independent liveness and overhead/platform qualification remain O2–O6; see [contract](observability.md). |

docs/reports/rds-desktop-references-20260926-data.json

@@ -0,0 +1,41 @@
+{
+  "schema_version": 1,
+  "commit": "4221c59a53713de716f4e781a858a3587f991621",
+  "dirty": false,
+  "os": "linux",
+  "architecture": "x86_64",
+  "profile": "test (debug)",
+  "features": "rds-desktop default; dev harness enables transport-noq",
+  "command": "RDS_SOAK_SECS=60 CARGO_BUILD_JOBS=2 cargo test --locked -p rds-desktop --test session_v2 -- --nocapture --test-threads=1",
+  "toolchain": "rustc 1.98.1 (48a229cea 2026-09-01)",
+  "lockfile_sha256": "78723e059a78f49d715542632620884b1a279b0db82acd379259b5344f20acdb",
+  "test_binary_sha256": "79975cd92bf9b3163db368f2c5a09c9adbcf39c7921bfcacdece0419048ab811",
+  "repetitions": 1,
+  "failures": 0,
+  "skips": 0,
+  "tests_passed": 7,
+  "measurement": "synthetic complete frame-header arrival, not decoded/presented pixels",
+  "clean_iroh_loopback": {
+    "duration_seconds": 60,
+    "frames": 3601,
+    "age_p95_ms": 3,
+    "age_p99_ms": 5,
+    "process_rss_peak_kib": 50060
+  },
+  "impaired_noq_loopback": {
+    "frames": 420,
+    "age_p95_ms": 110,
+    "age_p99_ms": 116,
+    "queue_p95_ms": 37,
+    "wire_p95_ms": 80,
+    "control_rtt_p95_ms": 159,
+    "server_forwarded": 1276,
+    "server_dropped": 67,
+    "server_bytes": 355132,
+    "client_forwarded": 1165,
+    "client_dropped": 64,
+    "client_bytes": 57085
+  },
+  "impairment": "Impairment::lossy() at recorded commit, applied under both endpoints UDP sockets",
+  "acceptance_limits": "existing session_v2 thresholds unchanged; no native desktop or WAN acceptance"
+}