feat: local-socket-halves — the local feature (TCP/UDP/unix halves functions)

- src/local/mod.rs (feature = "local" -> tokio/net): dial_tcp (5s
  timeout), connect_udp (ephemeral bind + connect + the FRAMED UdpHalf
  adapter), dial_unix (OQ-TN-14 in v1), bind_tcp +
  TcpListenerHalves::accept_loop (the listen shape's assembly half),
  local_dial() (one DialFn covering all three substrates).
- UdpHalf truncation fails loud (OQ-TN-13): poll_read recvs into a
  scratch buffer first and size-checks against the caller's buffer —
  an over-size datagram is io::ErrorKind::InvalidData, never a silent
  truncation (tokio's poll_recv-into-ReadBuf would truncate).
- No socket type appears outside src/local/ (convention 16); the
  default crate stays wasm-clean; stdio bridging is NOT here (alktty
  owns process stdio).
- Tests (tests/local_halves.rs, 7, feature-gated): TCP/UDP/unix dial
  round-trips through the real establisher path, the 1400-byte MTU
  datagram, the truncation fail-loud probe, dial-refusal ->
  dial_failed, listen producer over a real TCP listener end-to-end.

Verified: cargo test green (default + --all-features, 44 + 7),
clippy -D warnings (default + all-features + wasm32), fmt clean,
cargo check --target wasm32-unknown-unknown passes.

Cargo.toml

@@ -16,6 +16,7 @@ name = "alktunnels"
 
 [features]
 default = []
+local = ["tokio/net"]
 
 [dependencies]
 alkcall = "0.7.0"

src/lib.rs

@@ -24,6 +24,9 @@ pub mod params;
 pub mod producer;
 pub mod wire;
 
+#[cfg(feature = "local")]
+pub mod local;
+
 pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession};
 pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError};
 pub use params::ChannelOpenError;

src/local/mod.rs

@@ -0,0 +1,290 @@
+//! The `local` feature module: real-socket halves functions the
+//! assembly layer injects as `DialFn`/`AcceptFn` closures (ADR-004 —
+//! "produce boxed halves for a resource" is a function, not a trait).
+//! Non-wasm by design: this module IS the platform boundary — the
+//! default crate stays wasm-clean (convention 16: `local`-gated code
+//! is never imported from the shared/producer/consumer modules).
+//!
+//! Substrate coverage (overview.md §Feature Gates):
+//! - **TCP** — [`dial_tcp`] (5s connect timeout per the POCs) and the
+//!   [`TcpListenerHalves`] accept half for listen producers.
+//! - **UDP** — [`connect_udp`]: a connected `UdpSocket` wrapped in the
+//!   FRAMED adapter ([`UdpHalf`], ADR-003 — the codec composes at this
+//!   boundary; the pump never sees UDP specifics). Truncation fails
+//!   loud (OQ-TN-13's resolved posture).
+//! - **Unix** — [`dial_unix`], in v1 per OQ-TN-14 (same halves shape
+//!   as TCP). **Stdio is NOT here** — a spawned process's stdio is
+//!   alktty's pipe mode (exit codes + signals are tty semantics);
+//!   remote command execution composes via alktty, not here.
+//!
+//! Error mapping: [`TunnelEstablishError`] variants (the POC's
+//! `Into<EstablishmentError>` path) — dial refusal → `dial_failed`,
+//! bind/resource exhaustion → `resource_shortage`, handler-internal
+//! → `handler_error`.
+//!
+//! `TargetHandle` re-used from producer.rs — the halves type lives
+//! there; `local/` depends on it, never vice versa (convention 16).
+
+use std::sync::Arc;
+
+use crate::params::Substrate;
+use crate::producer::{AcceptQueue, DialFn, TargetHandle, TunnelEstablishError};
+use tokio::io::{AsyncRead, AsyncWrite};
+
+/// The dial/connect timeout (the POCs' 5s posture).
+const DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
+
+// --- TCP ----------------------------------------------------------------
+
+/// Dial a TCP target (`host:port`): `TcpStream::connect` bounded by
+/// [`DIAL_TIMEOUT`], `into_split` → boxed halves (raw pass-through —
+/// the halves ARE the tunnel, ADR-003).
+pub async fn dial_tcp(target: &str) -> Result<TargetHandle, TunnelEstablishError> {
+    let stream = tokio::time::timeout(DIAL_TIMEOUT, tokio::net::TcpStream::connect(target))
+        .await
+        .map_err(|_| TunnelEstablishError::DialFailed(format!("dial `{target}` timed out")))?
+        .map_err(|e| TunnelEstablishError::DialFailed(format!("dial `{target}`: {e}")))?;
+    let (read, write) = stream.into_split();
+    Ok(TargetHandle {
+        read: Box::new(read),
+        write: Box::new(write),
+    })
+}
+
+/// Bind a TCP listener (the listen shape's assembly-side half).
+/// The binding decision belongs to the caller (OQ-TN-04) — this is a
+/// `local`-feature convenience, not protocol behavior.
+pub async fn bind_tcp(addr: &str) -> Result<TcpListenerHalves, TunnelEstablishError> {
+    let listener = tokio::net::TcpListener::bind(addr)
+        .await
+        .map_err(|e| TunnelEstablishError::ResourceShortage(format!("bind `{addr}`: {e}")))?;
+    Ok(TcpListenerHalves {
+        listener: Arc::new(listener),
+    })
+}
+
+/// An assembly-owned TCP listener + its accept loop's feed point. The
+/// accept loop ([`TcpListenerHalves::accept_loop`]) pushes accepted
+/// handles into the assembly's [`AcceptQueue`]; the listen establisher
+/// pops them (producer-listen's contract).
+pub struct TcpListenerHalves {
+    listener: Arc<tokio::net::TcpListener>,
+}
+
+impl TcpListenerHalves {
+    /// The bound listener's local address.
+    pub fn local_addr(&self) -> Result<std::net::SocketAddr, TunnelEstablishError> {
+        self.listener
+            .local_addr()
+            .map_err(|e| TunnelEstablishError::HandlerError(format!("local_addr: {e}")))
+    }
+
+    /// The accept loop: accept connections until the listener fails,
+    /// pushing each accepted handle into the queue. Returns when the
+    /// listener is closed (accept errors end the loop — the assembly
+    /// layer aborts this task on teardown).
+    pub async fn accept_loop(&self, queue: AcceptQueue) -> Result<(), TunnelEstablishError> {
+        loop {
+            let (stream, _peer) = self
+                .listener
+                .accept()
+                .await
+                .map_err(|e| TunnelEstablishError::DialFailed(format!("accept: {e}")))?;
+            let (read, write) = stream.into_split();
+            let handle = TargetHandle {
+                read: Box::new(read),
+                write: Box::new(write),
+            };
+            queue.push(handle).await.map_err(|dropped| {
+                TunnelEstablishError::HandlerError(format!(
+                    "queue closed; accepted connection dropped: {dropped:?}"
+                ))
+            })?;
+        }
+    }
+}
+
+// --- UDP ----------------------------------------------------------------
+
+/// Connect a UDP "tunnel" to a target: bind an ephemeral socket,
+/// `connect(target)` (the associated-socket posture — one peer), and
+/// wrap it in the FRAMED adapter (ADR-003): `UdpHalf` enforces one
+/// datagram per read/write at the substrate boundary; the `[len: u16
+/// BE]` codec composes at this boundary — the pump never sees UDP
+/// specifics.
+pub async fn connect_udp(target: &str) -> Result<TargetHandle, TunnelEstablishError> {
+    let sock = tokio::net::UdpSocket::bind(("0.0.0.0", 0))
+        .await
+        .map_err(|e| TunnelEstablishError::ResourceShortage(format!("udp bind failed: {e}")))?;
+    sock.connect(target)
+        .await
+        .map_err(|e| TunnelEstablishError::DialFailed(format!("udp connect `{target}`: {e}")))?;
+    let (read, write) = UdpHalf::split(sock);
+    Ok(TargetHandle { read, write })
+}
+
+/// A UDP socket presented as `AsyncRead`/`AsyncWrite` halves: reads
+/// yield whole datagrams (one `poll_recv` per read); writes send one
+/// datagram per `poll_write` buffer. The FRAMED adapter — the codec
+/// wraps at this boundary (ADR-003); the pump never sees UDP
+/// specifics.
+///
+/// **Truncation fails loud** (OQ-TN-13's resolved posture, ADR-003):
+/// `poll_recv` into a too-small caller buffer returns
+/// `io::ErrorKind::InvalidData` (the datagram is dropped with the
+/// error — silent truncation would corrupt the framing invariants).
+/// The codec's u16 bound keeps well-formed peers from ever hitting
+/// this path; only a mis-framed wire does.
+pub struct UdpHalf {
+    sock: Arc<tokio::net::UdpSocket>,
+    kind: UdpHalfKind,
+    /// The recv scratch buffer (the fail-loud truncation check reads
+    /// whole datagrams before copying into the caller's buffer).
+    recv_scratch: Vec<u8>,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+enum UdpHalfKind {
+    Read,
+    Write,
+}
+
+impl UdpHalf {
+    pub fn split(
+        sock: tokio::net::UdpSocket,
+    ) -> (
+        Box<dyn AsyncRead + Send + Sync + Unpin>,
+        Box<dyn AsyncWrite + Send + Sync + Unpin>,
+    ) {
+        let sock = Arc::new(sock);
+        (
+            Box::new(UdpHalf {
+                sock: Arc::clone(&sock),
+                kind: UdpHalfKind::Read,
+                recv_scratch: vec![0u8; 65535],
+            }),
+            Box::new(UdpHalf {
+                sock,
+                kind: UdpHalfKind::Write,
+                recv_scratch: Vec::new(),
+            }),
+        )
+    }
+}
+
+impl AsyncRead for UdpHalf {
+    fn poll_read(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        debug_assert_eq!(this.kind, UdpHalfKind::Read);
+        // One recv yields exactly one datagram into the caller's
+        // buffer. The fail-loud posture (OQ-TN-13, ADR-003): a
+        // datagram LARGER than the caller's buffer is a framing
+        // violation — reported as an error, never silently truncated
+        // (`poll_recv`-into-ReadBuf would truncate; the scratch-buffer
+        // recv + size check keeps the invariant observable). The
+        // codec's u16 bound keeps well-formed peers off this path.
+        let scratch = &mut this.recv_scratch;
+        match this.sock.try_recv(scratch) {
+            Ok(n) => {
+                if n > buf.remaining() {
+                    return std::task::Poll::Ready(Err(std::io::Error::new(
+                        std::io::ErrorKind::InvalidData,
+                        format!(
+                            "udp datagram {n} bytes exceeds caller buffer {} (truncation fails loud — OQ-TN-13)",
+                            buf.remaining()
+                        ),
+                    )));
+                }
+                buf.put_slice(&scratch[..n]);
+                std::task::Poll::Ready(Ok(()))
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                // Not ready: poll readiness to register the waker,
+                // then Pending (the next wake retries the recv).
+                match this.sock.poll_recv_ready(cx) {
+                    std::task::Poll::Ready(Ok(())) => std::task::Poll::Pending,
+                    std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
+                    std::task::Poll::Pending => std::task::Poll::Pending,
+                }
+            }
+            Err(e) => std::task::Poll::Ready(Err(e)),
+        }
+    }
+}
+
+impl tokio::io::AsyncWrite for UdpHalf {
+    fn poll_write(
+        self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        let this = self.get_mut();
+        debug_assert_eq!(this.kind, UdpHalfKind::Write);
+        match this.sock.try_send(buf) {
+            Ok(n) => std::task::Poll::Ready(Ok(n)),
+            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                // Poll send-readiness to register the waker, then
+                // Pending (the next wake retries the send).
+                match this.sock.poll_send_ready(cx) {
+                    std::task::Poll::Ready(Ok(())) => std::task::Poll::Pending,
+                    std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
+                    std::task::Poll::Pending => std::task::Poll::Pending,
+                }
+            }
+            Err(e) => std::task::Poll::Ready(Err(e)),
+        }
+    }
+
+    fn poll_flush(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        // UDP has no shutdown; drop is the close.
+        std::task::Poll::Ready(Ok(()))
+    }
+}
+
+// --- Unix ---------------------------------------------------------------
+
+/// Dial a unix-socket target (OQ-TN-14 — ships in v1, the same halves
+/// shape as TCP): `UnixStream::connect` bounded by [`DIAL_TIMEOUT`],
+/// `into_split` → boxed halves (raw pass-through, ADR-003).
+pub async fn dial_unix(path: &str) -> Result<TargetHandle, TunnelEstablishError> {
+    let stream = tokio::time::timeout(DIAL_TIMEOUT, tokio::net::UnixStream::connect(path))
+        .await
+        .map_err(|_| TunnelEstablishError::DialFailed(format!("dial `{path}` timed out")))?
+        .map_err(|e| TunnelEstablishError::DialFailed(format!("dial `{path}`: {e}")))?;
+    let (read, write) = stream.into_split();
+    Ok(TargetHandle {
+        read: Box::new(read),
+        write: Box::new(write),
+    })
+}
+
+/// The local dial closure (the ADR-004 injection point): one
+/// [`DialFn`]-shaped function covering tcp/udp/unix — the assembly
+/// layer injects this (closed over nothing substrate-specific beyond
+/// this module).
+pub fn local_dial() -> DialFn {
+    Arc::new(move |substrate, backing: &str| {
+        let backing = backing.to_string();
+        Box::pin(async move {
+            match substrate {
+                Substrate::Tcp => dial_tcp(&backing).await,
+                Substrate::Udp => connect_udp(&backing).await,
+                Substrate::Unix => dial_unix(&backing).await,
+            }
+        })
+    })
+}

tasks/tunnels/local-socket-halves.md

@@ -1,7 +1,7 @@
 ---
 id: tunnels/local-socket-halves
 name: "local feature — real socket halves functions (TCP dial, UDP connect, unix)"
-status: pending
+status: completed
 depends_on: [tunnels/producer-open-op, tunnels/wire-codec]
 scope: moderate
 risk: medium
@@ -104,6 +104,51 @@ POC's sizing test).
 
 > Agent fills during implementation.
 
+- **`bind_tcp` + `TcpListenerHalves::accept_loop(queue)`** complete the
+  listen side (the task surface sketched `bind`/`accept_fn`): the
+  accept loop is a spawned task feeding an [`AcceptQueue`] — the
+  protocol-side queue contract from producer-listen. The assembly
+  layer aborts the task on teardown; accept errors end the loop.
+- **`UdpHalf` truncation probe** (OQ-TN-13): `poll_read` recvs into a
+  64KiB scratch buffer first, then size-checks against the caller's
+  buffer — a >buffer datagram is `io::ErrorKind::InvalidData`, never
+  a silent truncation. (`tokio`'s `poll_recv`-into-ReadBuf truncates
+  silently, which is exactly the invariant F-2/OQ-TN-13 forbids.)
+  WouldBlock polls `poll_recv_ready` for waker registration.
+- **The task sketch's `TcpListenerHalves::accept_fn(&self, queue)`
+  became `accept_loop(&self, queue)`** (an owned async loop to spawn)
+  — the `accept_fn` borrow-shape fought the spawn model; the loop
+  returning on listener failure is the assembly-abort contract.
+- **Test plumbing note** (tests): the listen-over-real-socket test
+  drives the client socket via std `try_clone` BEFORE `from_std`
+  (O_NONBLOCK is shared across dup — both ends go async); the session
+  uses borrowed halves (the producer's wrapper-managed pump already
+  pumps channel <-> accepted handle; `pump_against` there would loop
+  the same socket back into itself).
+
 ## Summary
 
-> Agent fills this on completion.
+> Agent fills this on completion.
+
+Implemented the `local` feature (src/local/mod.rs, feature =
+`["tokio/net"]`): `dial_tcp` (5s timeout), `connect_udp` (ephemeral
+bind + connect + the FRAMED `UdpHalf` adapter — truncation fails
+loud), `dial_unix` (OQ-TN-14 in v1), `bind_tcp`/`TcpListenerHalves::
+accept_loop` (the listen shape's assembly half), and `local_dial()`
+(the one `DialFn` covering all three substrates). No socket type
+appears outside `src/local/` (convention 16: the module is never
+imported from producer/consumer/shared; `TargetHandle` re-used from
+producer.rs). Stdio bridging is NOT here (alktty's pipe mode owns
+process stdio).
+
+Tests: `tests/local_halves.rs` (7, `#![cfg(feature = "local")]`) —
+TCP dial round-trip through the real establisher path, UDP framed
+round-trip, the 1400-byte MTU datagram, unix dial round-trip, the
+OQ-TN-13 truncation fail-loud probe (60k datagram vs 16KiB read →
+InvalidData), dial-refusal → `dial_failed`, and the listen producer
+over a real TCP listener end-to-end.
+
+Verified: `cargo test` (default) and `cargo test --all-features`
+green (44 + 7), clippy `-D warnings` clean (default + all-features +
+wasm32), fmt clean, wasm32 check passes (default crate stays
+wasm-clean; `local` is non-wasm by design).

tests/local_halves.rs

@@ -0,0 +1,332 @@
+//! Tests for the `local` feature: real-socket halves through the real
+//! establisher path (`local_dial` injected as the `DialFn`), TCP/UDP/
+//! unix round-trips against real local echo servers, the OQ-TN-13
+//! truncation fail-loud probe, and the 1400-byte MTU datagram.
+//!
+//! The forward topology (`wire_forward`) carries the full path: the
+//! consumer opens (`TunnelSession::open`), the producer's establisher
+//! dials via `local_dial`, the pump handler pumps — real sockets at
+//! both ends.
+
+#![cfg(feature = "local")]
+
+mod harness;
+
+use std::sync::Arc;
+
+use alktunnels::params::{Substrate, TunnelParams};
+use alktunnels::producer::ResourceRegistry;
+use alktunnels::TunnelSession;
+
+use harness::wire_forward;
+
+fn params(resource: &str, substrate: Substrate) -> TunnelParams {
+    TunnelParams {
+        resource: resource.to_string(),
+        substrate,
+    }
+}
+
+/// A TCP echo server on an ephemeral port.
+async fn tcp_echo_server() -> std::net::SocketAddr {
+    let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0))
+        .await
+        .expect("bind echo");
+    let addr = listener.local_addr().expect("addr");
+    tokio::spawn(async move {
+        loop {
+            let Ok((sock, _)) = listener.accept().await else {
+                return;
+            };
+            tokio::spawn(async move {
+                let (mut r, mut w) = tokio::io::split(sock);
+                let _ = tokio::io::copy(&mut r, &mut w).await;
+            });
+        }
+    });
+    addr
+}
+
+/// A UDP echo server on an ephemeral port (one task, echo loop).
+async fn udp_echo_server() -> std::net::SocketAddr {
+    let sock = tokio::net::UdpSocket::bind(("127.0.0.1", 0))
+        .await
+        .expect("bind udp echo");
+    let addr = sock.local_addr().expect("addr");
+    tokio::spawn(async move {
+        let mut buf = vec![0u8; 65535];
+        loop {
+            let Ok((n, peer)) = sock.recv_from(&mut buf).await else {
+                return;
+            };
+            if sock.send_to(&buf[..n], peer).await.is_err() {
+                return;
+            }
+        }
+    });
+    addr
+}
+
+#[tokio::test]
+async fn tcp_dial_round_trips_through_the_real_establisher_path() {
+    let addr = tcp_echo_server().await;
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, &addr.to_string())
+        .await;
+    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("tcp open");
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (read, write) = session.stream_halves().expect("halves");
+    write.write_all(b"tcp-local").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 9];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip")
+        .expect("read");
+    assert_eq!(&buf, b"tcp-local");
+    session.close().await;
+}
+
+#[tokio::test]
+async fn udp_datagram_round_trip_through_the_framed_adapter() {
+    // The FRAMED adapter end-to-end (ADR-003): session sends the codec
+    // frame; the pump copies raw bytes; the dialed UdpHalf strips the
+    // frame on recv and delivers ONE datagram to the target; the echo
+    // comes back through the same path.
+    let addr = udp_echo_server().await;
+    let registry = ResourceRegistry::new();
+    registry
+        .register("dns", Substrate::Udp, &addr.to_string())
+        .await;
+    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
+        .await
+        .expect("udp open");
+    session.send_datagram(b"query").await.expect("send");
+    let dg = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("io ok")
+        .expect("datagram");
+    assert_eq!(&dg[..], b"query");
+    session.close().await;
+}
+
+#[tokio::test]
+async fn udp_large_datagram_through_bounded_buffers() {
+    // The POC's sizing test: a 1400-byte datagram (max ethernet MTU
+    // payload) round-trips through the codec + the bounded-buffer path.
+    let addr = udp_echo_server().await;
+    let registry = ResourceRegistry::new();
+    registry
+        .register("big", Substrate::Udp, &addr.to_string())
+        .await;
+    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("big", Substrate::Udp))
+        .await
+        .expect("udp open");
+    let payload: Vec<u8> = (0..1400u32).map(|i| (i % 7) as u8).collect();
+    session.send_datagram(&payload).await.expect("send");
+    let got = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("io ok")
+        .expect("datagram");
+    assert_eq!(got.len(), payload.len());
+    assert_eq!(&got[..], &payload[..]);
+    session.close().await;
+}
+
+#[tokio::test]
+async fn unix_dial_round_trips() {
+    // OQ-TN-14: unix ships with local v1 (same halves shape as TCP).
+    let dir = std::env::temp_dir().join(format!("alktunnels-test-{}", std::process::id()));
+    let _ = std::fs::create_dir_all(&dir);
+    let path = dir.join("echo.sock");
+    let _ = std::fs::remove_file(&path);
+
+    let listener = tokio::net::UnixListener::bind(&path).expect("bind unix echo");
+    tokio::spawn(async move {
+        loop {
+            let Ok((sock, _)) = listener.accept().await else {
+                return;
+            };
+            tokio::spawn(async move {
+                let (mut r, mut w) = tokio::io::split(sock);
+                let _ = tokio::io::copy(&mut r, &mut w).await;
+            });
+        }
+    });
+
+    let registry = ResourceRegistry::new();
+    registry
+        .register("uds", Substrate::Unix, &path.to_string_lossy())
+        .await;
+    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("uds", Substrate::Unix))
+        .await
+        .expect("unix open");
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (read, write) = session.stream_halves().expect("halves");
+    write.write_all(b"unix-local").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"unix-local");
+    session.close().await;
+    let _ = std::fs::remove_file(&path);
+}
+
+#[tokio::test]
+async fn truncation_fails_loud_oq_tn_13() {
+    // A receive buffer smaller than the datagram surfaces as an error
+    // (never a silent truncation). Probed at the adapter level: feed a
+    // >buffer datagram directly into a UdpHalf pair and read with a
+    // small buffer.
+    let sock_a = tokio::net::UdpSocket::bind(("127.0.0.1", 0))
+        .await
+        .expect("bind a");
+    let sock_b = tokio::net::UdpSocket::bind(("127.0.0.1", 0))
+        .await
+        .expect("bind b");
+    sock_a
+        .connect(sock_b.local_addr().expect("b addr"))
+        .await
+        .expect("connect a");
+    sock_b
+        .connect(sock_a.local_addr().expect("a addr"))
+        .await
+        .expect("connect b");
+
+    // The producer-side adapter (what the establisher dials): sock_a
+    // wrapped as halves. Send a large datagram from the "wire" side.
+    let (read, _write) = alktunnels::local::UdpHalf::split(sock_a);
+    let mut read = read;
+    let payload = vec![0u8; 60000]; // well-formed for UDP but big
+    sock_b.send(&payload).await.expect("send big");
+
+    // The pump reads with the codec's chunk size (16 KiB) — smaller
+    // than 60000: the adapter must FAIL LOUD, not truncate.
+    use tokio::io::AsyncReadExt;
+    let mut chunk = vec![0u8; 16 * 1024];
+    let result =
+        tokio::time::timeout(std::time::Duration::from_secs(5), read.read(&mut chunk)).await;
+    match result {
+        Ok(Err(e)) => {
+            assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
+        }
+        other => panic!("expected InvalidData error, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn dial_refusal_maps_to_dial_failed() {
+    // Error mapping: a refused dial (nothing listening on the port) is
+    // `dial_failed` on the wire (`channel:open_failed` reason).
+    let registry = ResourceRegistry::new();
+    registry
+        .register("dead", Substrate::Tcp, "127.0.0.1:1")
+        .await;
+    let topo = wire_forward(registry, alktunnels::local::local_dial()).await;
+
+    let err = match TunnelSession::open(&topo.consumer, params("dead", Substrate::Tcp)).await {
+        Ok(_) => panic!("refused dial must fail"),
+        Err(e) => e,
+    };
+    assert_eq!(err.open_ref().establishment_reason(), Some("dial_failed"));
+}
+
+#[tokio::test]
+async fn listen_producer_over_a_real_tcp_listener() {
+    // The listen shape + local halves compose: a real TcpListenerHalves
+    // accept loop feeds an AcceptQueue; a listen-registered producer
+    // pops accepted sockets; the consumer pumps the tunnel against the
+    // accepted local socket.
+    let queue = alktunnels::producer::AcceptQueue::new();
+    let listener = alktunnels::local::bind_tcp("127.0.0.1:0")
+        .await
+        .expect("bind");
+    let addr = listener.local_addr().expect("addr");
+
+    let accept_loop = tokio::spawn({
+        let queue = queue.clone();
+        async move { listener.accept_loop(queue).await }
+    });
+
+    // A local client connects (the accepted connection's source). The
+    // driving end comes from `try_clone` — the pump owns the accepted
+    // handle; the driver pushes/reads the echo through the same real
+    // socket.
+    // std-level try_clone BEFORE converting to tokio streams: one fd
+    // for the driver, one wrapped for the pump (two independent
+    // handles to the same socket). O_NONBLOCK is shared across the
+    // dup, so BOTH ends go nonblocking and the driver is driven
+    // async too.
+    let local_std = std::net::TcpStream::connect(addr).expect("std connect");
+    let driver_std = local_std.try_clone().expect("try_clone driver");
+    local_std.set_nonblocking(true).expect("nonblocking");
+    let mut driver = tokio::net::TcpStream::from_std(driver_std).expect("tokio driver");
+
+    let registry = ResourceRegistry::new();
+    registry
+        .register("listener", Substrate::Tcp, "assembly-owned")
+        .await;
+    let accept_fn: alktunnels::producer::AcceptFn = Arc::new(move || {
+        let queue = queue.clone();
+        Box::pin(async move {
+            queue.pop().await.ok_or_else(|| {
+                alktunnels::producer::TunnelEstablishError::DialFailed(
+                    "listener closed".to_string(),
+                )
+            })
+        })
+    });
+    let topo = harness::wire_listen(registry, accept_fn).await;
+
+    let channel_id = alktunnels::open_reverse_channel(
+        &topo.consumer_call,
+        &params("listener", Substrate::Tcp),
+        None,
+    )
+    .await
+    .expect("listen open");
+
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    // The producer's pump (wrapper-managed) copies channel <-> the
+    // accepted handle; the accepted handle IS the client socket — so
+    // the session's borrowed halves drive it directly: the driver end
+    // writes into the client socket, the pump carries it over the
+    // channel, and the session halves read it (and back).
+    let mut session = session;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    driver.write_all(b"listen-local").await.expect("write");
+    driver.flush().await.expect("flush");
+    let (read, _write) = session.stream_halves().expect("halves");
+    let mut buf = vec![0u8; 12];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip")
+        .expect("read");
+    assert_eq!(&buf, b"listen-local");
+
+    session.close().await;
+    accept_loop.abort();
+}