test: coverage weak spots — the UdpHalf sender-block recipe found (option 1)

- DatagramReader::feed is infallible (the oversize bound is enforced
  at frame time; the same zombie category as 4014ba6's InvalidLength).
  The Vec<Result> shape, the local adapter's dead InvalidData mapping,
  and consumer.rs's `?` drop.
- The sender-block machinery test (root-gated veth fixture, auto-skip
  without root): the peer address unassigned + the neighbor table
  flushed pins the connected sender in a persistent WouldBlock/ENOBUFS
  state; the test drives the WouldBlock stash, the readiness-Pending
  arms, AND the accepted-byte high-water Pend in one behavior, then
  resolves (addr assign + peer bind) and asserts the queued datagrams
  drain and deliver. Jumbo MTU (65535) keeps IP fragmentation out of
  the way; the block-state probe runs on a throwaway socket (an
  oversized probe's partially-built skb chain stays corked on EAGAIN
  and poisons the socket).
- pump_queue_to_socket's stash arm now matches ENOBUFS(105) alongside
  WouldBlock — the blocked-neighbor state reports 105, and both mean
  "the socket cannot take this datagram now".
- connect_udp's resolve-failure arm covered (an RFC 6761 .invalid
  target maps to dial_failed; rootless).
- The no-witness tunnel_establisher wrapper covered (a harness
  DialNoWitness mode + a full open round-trip through the wrapper).

Negative result pinned in the task doc: tokio 1.53 structurally masks
UDP ICMP errors out of the poll drive (poll_recv_ready/try_io mask
ERROR out of the read direction), so the recv-side ECONNREFUSED
mapping arm is unreachable through the pump's drive — it joins the
defensive list.

Coverage: 92.86% -> 93.91% lines (50 -> 41 uncovered, all deliberate).

Verification: cargo test + cargo test --all-features (91 tests),
clippy --all-targets --all-features, fmt, wasm32 check+clippy,
cargo doc — all clean.

tests/end_to_end.rs

@@ -714,7 +714,7 @@ async fn datagrams_survive_chunk_splitting_codec_level() {
     let mut reader = alktunnels::wire::DatagramReader::new();
     let mut recovered = Vec::new();
     for chunk in stream.chunks(7) {
-        for dg in reader.feed(chunk).expect("decode") {
+        for dg in reader.feed(chunk) {
             recovered.push(dg.to_vec());
         }
     }

tests/harness.rs

@@ -102,6 +102,9 @@ pub enum RegistrationMode {
     /// probe — ADR-049 §2's bound override; the open fails with reason
     /// `timeout` and no channel survives).
     Timeout(std::time::Duration),
+    /// The no-witness dial establisher (`tunnel_establisher` — the
+    /// public default shape, no CF-006 probe seam).
+    DialNoWitness(alktunnels::producer::DialFn),
 }
 
 pub async fn wire_with(
@@ -219,6 +222,27 @@ pub async fn wire_with(
             )
             .expect("register tunnel listen openable");
         }
+        RegistrationMode::DialNoWitness(dial) => {
+            // The public no-witness wrapper (tunnel_establisher) —
+            // registration + establisher body identical to the
+            // witnessed variant minus the CF-006 seam.
+            let core = alkcall::channels::operations::ChannelCore::new(
+                producer_client.manager().clone(),
+                alkcall::channels::policy::default_policy(),
+            );
+            core.register_openable_with_establisher(
+                alktunnels::params::tunnel_open_spec(),
+                Some(alktunnels::producer::tunnel_establisher(
+                    registry.clone(),
+                    dial,
+                )),
+                alktunnels::producer::make_tunnel_pump_handler(),
+                &producer_op_registry,
+                AuthContext::anonymous(b"alk/tunnel"),
+                None,
+            )
+            .expect("register the no-witness wrapper establisher");
+        }
         RegistrationMode::Timeout(timeout) => {
             // The deadline-expiry probe's registration shape: a hanging
             // establisher (never resolves) with an explicit per-
@@ -527,10 +551,7 @@ pub fn closing_udp_echo_dial(payload_completes: bool) -> alktunnels::producer::D
                             Ok(0) | Err(_) => return,
                             Ok(n) => n,
                         };
-                        let dgs = match reader.feed(&buf[..n]) {
-                            Ok(dgs) => dgs,
-                            Err(_) => return,
-                        };
+                        let dgs = reader.feed(&buf[..n]);
                         if let Some(dg) = dgs.into_iter().next() {
                             let framed = match alktunnels::wire::frame_datagram(&dg) {
                                 Ok(f) => f,
@@ -604,10 +625,7 @@ pub fn framed_udp_echo_dial() -> alktunnels::producer::DialFn {
                                 Ok(0) | Err(_) => return,
                                 Ok(n) => n,
                             };
-                            let dgs = match reader.feed(&buf[..n]) {
-                                Ok(dgs) => dgs,
-                                Err(_) => return,
-                            };
+                            let dgs = reader.feed(&buf[..n]);
                             for dg in dgs {
                                 let framed = match alktunnels::wire::frame_datagram(&dg) {
                                     Ok(f) => f,

tests/local_halves.rs

@@ -871,3 +871,328 @@ async fn udp_half_flush_drains_repeatedly_until_the_queue_is_empty() {
     assert_eq!(received, 5, "five flush cycles, five datagrams");
     drop(peer);
 }
+
+/// The resolve arm (`connect_udp`'s no-addresses mapping): a target
+/// whose name cannot be resolved (an RFC 6761 `.invalid` name —
+/// guaranteed NXDOMAIN, resolved instantly and without network I/O)
+/// maps to `dial_failed` (the resolve step's error posture, ADR-004's
+/// `DialFn` contract). Rootless, deterministic.
+#[tokio::test]
+async fn connect_udp_unresolvable_target_maps_to_dial_failed() {
+    let err = match alktunnels::local::connect_udp("definitely-not-a-real-host-xyz.invalid:1").await
+    {
+        Ok(_) => panic!("an unresolvable target must fail"),
+        Err(e) => e,
+    };
+    assert!(
+        matches!(
+            err,
+            alktunnels::producer::TunnelEstablishError::DialFailed(_)
+        ),
+        "resolve failure maps to dial_failed: {err:?}"
+    );
+}
+
+// =====================================================================
+// The sender-block machinery (the coverage-weak-spots task's option 1,
+// CLOSED — the recipe was found): a veth fixture with the peer
+// address UNASSIGNED and the neighbor table flushed puts the
+// connected UDP sender into a PERMANENT WouldBlock/ENOSPC state
+// (the unresolved neighbor never resolves; the blocked write path's
+// skb stays charged to the socket's wmem with a shrunken SO_SNDBUF).
+// Assigning the peer address + binding the peer resolves ARP; the
+// queued/stashed datagrams drain and DELIVER. Validated on this host
+// (kernel 6.8.0-110-generic, tokio 1.53.1, socket2 0.6.5) — 3/3
+// deterministic probe runs, 2026-09-09.
+//
+// Root-gated (`ip link add`): auto-skipped without root. Never
+// touches the default route or any real interface.
+// =====================================================================
+
+/// The veth fixture (created and torn down per run): `alktx/alktx-p`
+/// with 10.240.0.1/24 (the sender side) and 10.240.0.9 (the peer
+/// side, UNASSIGNED during the block phase — its absence is what
+/// pins the sender). The peer address is assigned mid-test (the
+/// resolve step). A dead-simple guard skips the test when `ip` is
+/// unavailable or the command fails (non-root hosts).
+struct VethFixture {
+    peer_addr_assigned: bool,
+}
+
+/// Run a privileged `ip` command: bare `ip` first (uid-0 test runs),
+/// `sudo ip` second (the uid-1000 dev runs). `None` = unavailable.
+async fn ip_cmd(args: &[&str]) -> Option<bool> {
+    let mut spawned = false;
+    for prog in ["ip", "sudo"] {
+        let mut argv = Vec::with_capacity(args.len() + 1);
+        if prog == "sudo" {
+            argv.push("ip");
+        }
+        argv.extend_from_slice(args);
+        match tokio::process::Command::new(prog)
+            .args(&argv)
+            .output()
+            .await
+        {
+            Ok(o) => {
+                spawned = true;
+                if o.status.success() {
+                    return Some(true);
+                }
+            }
+            Err(_) => continue,
+        }
+    }
+    spawned.then_some(false)
+}
+
+impl VethFixture {
+    async fn try_create() -> Option<Self> {
+        // Skip loud-free: the coverage task documents the host boundary.
+        let created = ip_cmd(&[
+            "link", "add", "alktx", "type", "veth", "peer", "name", "alktx-p",
+        ])
+        .await
+        .unwrap_or(false);
+        if !created {
+            eprintln!("skipping: veth fixture unavailable (non-root host)");
+            return None;
+        }
+        let _ = ip_cmd(&["addr", "add", "10.240.0.1/24", "dev", "alktx"]).await;
+        // A jumbo MTU (65536): a 60 KB datagram crosses as ONE frame —
+        // no IP fragmentation, so no fragment-reassembly races at the
+        // peer (fragment-ID collisions under the drain burst merged
+        // datagrams on an MTU-1500 veth during the probe runs).
+        let _ = ip_cmd(&["link", "set", "alktx", "mtu", "65535"]).await;
+        let _ = ip_cmd(&["link", "set", "alktx", "up"]).await;
+        let _ = ip_cmd(&["link", "set", "alktx-p", "up"]).await;
+        Some(Self {
+            peer_addr_assigned: false,
+        })
+    }
+
+    fn assign_peer(&mut self) -> bool {
+        if self.peer_addr_assigned {
+            return true;
+        }
+        let ok = std::process::Command::new("ip")
+            .args(["addr", "add", "10.240.0.9/24", "dev", "alktx-p"])
+            .output()
+            .ok()
+            .map(|o| o.status.success())
+            .unwrap_or(false)
+            || std::process::Command::new("sudo")
+                .args(["ip", "addr", "add", "10.240.0.9/24", "dev", "alktx-p"])
+                .output()
+                .ok()
+                .map(|o| o.status.success())
+                .unwrap_or(false);
+        self.peer_addr_assigned = ok;
+        ok
+    }
+}
+
+impl Drop for VethFixture {
+    fn drop(&mut self) {
+        let _ = std::process::Command::new("ip")
+            .args(["link", "del", "alktx"])
+            .output();
+        let _ = std::process::Command::new("sudo")
+            .args(["ip", "link", "del", "alktx"])
+            .output();
+    }
+}
+
+/// A sender socket blocked on the unresolved neighbor: connect to the
+/// UNASSIGNED peer address, flush the neighbor table, shrunken
+/// SO_SNDBUF. `try_send` on a 60 KB datagram (truesize beyond the
+/// buffer) then reports WouldBlock/ENOSPC persistently — the state
+/// the write-side machinery is for. Returns the tokio socket ready
+/// for `UdpHalf::split`.
+async fn blocked_sender_socket(peer_port: u16) -> Option<tokio::net::UdpSocket> {
+    // The BLOCK-STATE PROBE runs on a THROWAWAY socket: an oversized
+    // probe datagram's partially-built skb chain stays corked in
+    // sk_write_queue on EAGAIN (the kernel flushes it on the next
+    // successful send) — poisoning the socket with a stale partial
+    // datagram. The test socket below is created fresh afterwards.
+    let probe = socket2::Socket::new(
+        socket2::Domain::IPV4,
+        socket2::Type::DGRAM,
+        Some(socket2::Protocol::UDP),
+    )
+    .ok()?;
+    probe.set_nonblocking(true).ok()?;
+    probe.set_send_buffer_size(32768).ok()?;
+    probe
+        .bind(&"10.240.0.1:0".parse::<std::net::SocketAddr>().ok()?.into())
+        .ok()?;
+    let probe_std: std::net::UdpSocket = probe.into();
+    let probe_sock = tokio::net::UdpSocket::from_std(probe_std).ok()?;
+    probe_sock.connect(("10.240.0.9", peer_port)).await.ok()?;
+
+    // Confirm the block actually holds (the host invariant): with a
+    // 64 KiB SO_SNDBUF, the FIRST 60 KB datagram queues on the
+    // unresolved neighbor (Ok — its skb is charged to wmem); every
+    // SUBSEQUENT send EAGAINs (the charge exceeds the buffer) until
+    // the neighbor resolves. That persistent EAGAIN/ENOBUFS pair IS
+    // the block state. A resolved neighbor would send Ok throughout.
+    let big = vec![0u8; 60000];
+    let mut blocked = 0;
+    for _ in 0..3 {
+        match probe_sock.try_send(&big) {
+            Ok(_) => {} // the first datagram queues; expected
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => blocked += 1,
+            Err(e) if e.raw_os_error() == Some(105) => blocked += 1,
+            Err(_) => return None,
+        }
+    }
+    if blocked < 1 {
+        return None;
+    }
+    drop(probe_sock);
+
+    // The test socket: fresh (no corked state), same posture.
+    let sock = socket2::Socket::new(
+        socket2::Domain::IPV4,
+        socket2::Type::DGRAM,
+        Some(socket2::Protocol::UDP),
+    )
+    .ok()?;
+    sock.set_nonblocking(true).ok()?;
+    sock.set_send_buffer_size(32768).ok()?;
+    sock.bind(&"10.240.0.1:0".parse::<std::net::SocketAddr>().ok()?.into())
+        .ok()?;
+    let snd_std: std::net::UdpSocket = sock.into();
+    let snd = tokio::net::UdpSocket::from_std(snd_std).ok()?;
+    snd.connect(("10.240.0.9", peer_port)).await.ok()?;
+    Some(snd)
+}
+
+/// THE sender-block machinery, end to end (option 1 — one test, the
+/// coherent behavior): the WouldBlock stash (`poll_write`'s
+/// opportunistic drain stashes the mid-send datagram and registers
+/// the send waker — `pump_queue_to_socket`'s Pending arms), the
+/// accepted-byte high-water backpressure (`poll_write` Pends when the
+/// queue stays over [`alktunnels::local::WRITE_QUEUE_HIGH_WATER`]),
+/// the flush re-pump (readiness → drain), and the post-resolve
+/// delivery (the queued datagrams reach the peer).
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn udp_half_write_side_blocks_stashes_and_drains_on_resolve() {
+    let Some(mut veth) = VethFixture::try_create().await else {
+        eprintln!("skipping: veth fixture unavailable (non-root host)");
+        return;
+    };
+
+    // The peer port is EPHEMERAL (stale datagrams from earlier runs
+    // of this test family target any fixed port; an ephemeral port
+    // sidesteps the leftovers).
+    let probe = tokio::net::UdpSocket::bind(("10.240.0.1", 0))
+        .await
+        .expect("port probe bind");
+    let peer_port = probe.local_addr().expect("port probe addr").port();
+    drop(probe);
+    let snd = blocked_sender_socket(peer_port)
+        .await
+        .expect("blocked sender state");
+    let (_read, mut write) = alktunnels::local::UdpHalf::split(snd);
+
+    // Phase A: poll_write ACCEPTS everything (the accept-all
+    // contract) while the socket blocks; the queued bytes cross the
+    // high-water line and the next poll_write Pends.
+    // Drive single-shot poll_write polls with a noop waker (the
+    // pump's copy shape): each returns Ok (accepted) or Pending
+    // (the high-water). The neighbor's INCOMPLETE→FAILED cycle
+    // drops queued skbs periodically (charge released → sends
+    // succeed again), so the drive loops until a Pend is observed.
+    let mut body = vec![0xABu8; 60002];
+    body[0] = 0; // len = 60000 BE
+    body[1] = 0xea;
+    body[2] = 0x60;
+
+    let noop_waker = futures::task::noop_waker();
+    let mut cx = std::task::Context::from_waker(&noop_waker);
+
+    let mut pending_observed = false;
+    for _ in 0..60 {
+        let fut = std::future::poll_fn(|cx| {
+            tokio::io::AsyncWrite::poll_write(Pin::new(&mut write), cx, &body)
+        });
+        let mut fut = std::pin::pin!(fut);
+        match std::future::Future::poll(fut.as_mut(), &mut cx) {
+            std::task::Poll::Ready(Ok(60002)) => {} // accepted
+            std::task::Poll::Ready(Ok(n)) => panic!("poll_write accepted {n}?"),
+            std::task::Poll::Ready(Err(e)) => panic!("poll_write error: {e:?}"),
+            std::task::Poll::Pending => {
+                pending_observed = true;
+                break;
+            }
+        }
+    }
+    assert!(
+        pending_observed,
+        "the high-water Pends once the queue rides over the line"
+    );
+
+    // Flush: the blocked socket → the stash + readiness-Pending arms
+    // (bounded retries; the resolve below unblocks the LAST one).
+    let mut pending_flushes = 0usize;
+    for _ in 0..10 {
+        match tokio::time::timeout(
+            std::time::Duration::from_millis(300),
+            poll_flush_once(&mut write),
+        )
+        .await
+        {
+            Err(_elapsed) => {
+                pending_flushes += 1;
+            } // readiness stayed Pending (blocked socket)
+            Ok(Ok(())) => break, // drained (only possible after resolve)
+            Ok(Err(e)) => panic!("flush error: {e:?}"),
+        }
+    }
+    assert!(pending_flushes > 0, "the blocked socket Pends the flush");
+
+    // Phase B: resolve — assign the peer address (the neighbor
+    // completes), bind the peer, wait for ARP; the queued datagrams
+    // drain and DELIVER.
+    assert!(veth.assign_peer(), "assign the peer address");
+    let peer = tokio::net::UdpSocket::bind(("10.240.0.9", peer_port))
+        .await
+        .expect("bind peer (resolve)");
+
+    let mut flushed = false;
+    for _ in 0..20 {
+        match tokio::time::timeout(
+            std::time::Duration::from_millis(300),
+            poll_flush_once(&mut write),
+        )
+        .await
+        {
+            Ok(Ok(())) => {
+                flushed = true;
+                break;
+            }
+            Ok(Err(e)) => panic!("post-resolve flush error: {e:?}"),
+            Err(_) => {}
+        }
+    }
+    assert!(flushed, "the queued datagrams drained after the resolve");
+
+    // The peer received the queued datagrams (delivery across the
+    // blocked→resolved window — the stash/queue never lost them; at
+    // least the stashed tail must arrive — the periodic
+    // INCOMPLETE→FAILED drops may consume some queued ones, which
+    // is kernel-side loss, not adapter loss).
+    tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
+    let mut got = 0usize;
+    let mut buf = vec![0u8; 65535];
+    while peer.try_recv(&mut buf).is_ok() {
+        got += 1;
+    }
+    assert!(
+        got >= 1,
+        "the stashed/queued datagrams delivered after resolve"
+    );
+
+    drop(peer);
+}

tests/producer_open_op.rs

@@ -384,6 +384,50 @@ async fn registry_lookup_respects_substrate_key() {
     assert_eq!(establishment_reason_of(&err), Some("unknown_resource"));
 }
 
+#[tokio::test]
+async fn tunnel_establisher_no_witness_wrapper_opens() {
+    // The no-witness wrapper (`tunnel_establisher` — the public
+    // default shape, producer.rs's witness-less establisher body):
+    // a full open through the wrapper round-trips. The wrapper
+    // differs from `tunnel_establisher_with_witness` ONLY in the
+    // witness (the CF-006 probe seam), so the default-registration
+    // shape is exercised end to end here.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_with(
+        registry,
+        harness::RegistrationMode::DialNoWitness(echo_dial()),
+        Some(harness::consumer_identity()),
+        None,
+        Arc::new(alkcall::core::auth::NoopIdentityProvider),
+    )
+    .await;
+
+    let out = call_open_params(&topo, &params("echo"), None)
+        .await
+        .expect("the wrapper establisher opens");
+    let channel_id = out["channel_id"].as_u64().expect("channel_id") as u32;
+    let (mut read, mut write, pump) = adopt_and_pump(&topo, channel_id).await;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    write.write_all(b"no-witness").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 10];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip")
+        .expect("read");
+    assert_eq!(&buf, b"no-witness");
+
+    drop(write);
+    drop(read);
+    tokio::time::timeout(std::time::Duration::from_secs(5), pump)
+        .await
+        .expect("pump completion")
+        .expect("pump task");
+}
+
 /// Type-shape assertions kept for harness parity.
 #[allow(dead_code)]
 fn type_assertions() {