feat: consumer-session — TunnelSession (open/adopt, data planes, teardown)

- TunnelSession (src/consumer.rs): open (ChannelClient::open_channel)
  + adopt (ChannelManager::adopt_channel) construction, substrate-
  shaped data planes (raw halves for stream; ADR-003 codec + pending
  frame queue for UDP), pump_against (session-owned pump_bidi handle),
  and teardown ownership per ADR-005: close (abort + ungraceful reap),
  join (await + reap + copy counts, pump-less (0,0,reaped)), Drop
  (abort + sync reap, never leaks). No Clone (compile_fail doc-test).
- open_reverse_channel on the CallConnection (the reverse path's
  two-step: call then adopt — the POC's honest shape).
- Error surfaces (src/error.rs): TunnelOpenError (typed ADR-049 §4
  surface via open_ref/establishment_reason), TunnelIoError (wrong-
  substrate, TruncatedDatagram fail-loud, ChannelTaken), ReverseOpenError.
- Tests: tests/consumer_session.rs (12) over two harness topologies —
  forward (wire_forward/ForwardTopology: consumer connect-side pure
  client, producer serving via adapter) and reverse (the POC shape);
  teardown matrix, W4 half-close, out-of-band close + self-reap (W3),
  F-2 empty datagram, AdoptFailed-on-collision (bogus-id adopt parks
  by design — documented), concurrent sessions. Harness: all_substrate_dial.
- Adopt takes substrate (task-note deviation: the data plane shape
  rode the open call; cannot be inferred from the ID).

Verified: cargo test green (38), clippy -D warnings (native + wasm32),
fmt clean, cargo check --target wasm32-unknown-unknown passes.

Cargo.lock

@@ -58,6 +58,7 @@ dependencies = [
  "thiserror",
  "tokio",
  "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]
@@ -524,6 +525,21 @@ dependencies = [
  "scopeguard",
 ]
 
+[[package]]
+name = "log"
+version = "0.4.34"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
+
+[[package]]
+name = "matchers"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
+dependencies = [
+ "regex-automata",
+]
+
 [[package]]
 name = "memchr"
 version = "2.8.3"
@@ -547,6 +563,15 @@ dependencies = [
  "windows-sys",
 ]
 
+[[package]]
+name = "nu-ansi-term"
+version = "0.50.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
+dependencies = [
+ "windows-sys",
+]
+
 [[package]]
 name = "num"
 version = "0.4.3"
@@ -841,6 +866,15 @@ dependencies = [
  "zmij",
 ]
 
+[[package]]
+name = "sharded-slab"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "signal-hook-registry"
 version = "1.4.8"
@@ -932,6 +966,15 @@ dependencies = [
  "syn 3.0.5",
 ]
 
+[[package]]
+name = "thread_local"
+version = "1.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070"
+dependencies = [
+ "cfg-if",
+]
+
 [[package]]
 name = "tinystr"
 version = "0.8.4"
@@ -999,6 +1042,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a"
 dependencies = [
  "once_cell",
+ "valuable",
+]
+
+[[package]]
+name = "tracing-log"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
+dependencies = [
+ "log",
+ "once_cell",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-subscriber"
+version = "0.3.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319"
+dependencies = [
+ "matchers",
+ "nu-ansi-term",
+ "once_cell",
+ "regex-automata",
+ "sharded-slab",
+ "smallvec",
+ "thread_local",
+ "tracing",
+ "tracing-core",
+ "tracing-log",
 ]
 
 [[package]]
@@ -1050,6 +1123,12 @@ dependencies = [
  "vsimd",
 ]
 
+[[package]]
+name = "valuable"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
+
 [[package]]
 name = "version_check"
 version = "0.9.5"

Cargo.toml

@@ -28,4 +28,5 @@ tracing = "0.1"
 thiserror = "2"
 
 [dev-dependencies]
-tokio = { version = "1", features = ["full", "test-util", "macros"] }
+tokio = { version = "1", features = ["full", "test-util", "macros"] }
+tracing-subscriber = { version = "0.3.23", features = ["env-filter"] }

src/consumer.rs

@@ -1,9 +1,416 @@
 //! The consumer half: `TunnelSession` — the typed client for tunnel
-//! channels (ADR-005). Forward path (`open`), reverse path (`adopt`),
+//! channels (ADR-005). Forward path (`open`), reverse path (`adopt`
-//! substrate-shaped data planes (`stream_halves` / `take_halves` /
+//! and `open_reverse_channel`), substrate-shaped data planes
-//! `send_datagram` / `recv_datagram`), and teardown ownership
+//! (`stream_halves`, `take_halves`, `send_datagram`, `recv_datagram`),
-//! (`close` / `join` / `Drop`).
+//! pump ownership (`pump_against`), and teardown ownership
+//! (`close`, `join`, `Drop`).
 //!
-//! Skeleton module — filled by `tunnels/consumer-session`.
+//! One session = one channel = one tunnel (the channel ID is the flow
+//! key, OQ-TN-02). The session owns the adopted channel entry —
+//! nothing upstream awaits the adopter's pump (the W3 gap); the
+//! session closes it structurally. Role follows the resource, not the
+//! connection (OQ-TN-03): `-L` and `-R` consumers share this one API.
 
-pub use crate::producer::{OP_TUNNEL_OPEN, TUNNEL_ALPN};
+use alkcall::channels::client::{ChannelClient, ChannelOpenError};
+use alkcall::channels::manager::ChannelManager;
+use alkcall::channels::pump::pump_bidi;
+use alkcall::channels::reassembly::{MpscRecvStream, MpscSendStream};
+use alkcall::core::types::BiStream;
+use alkcall::protocol::connection::CallConnection;
+use alkcall::protocol::wire::CallError;
+use bytes::Bytes;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadHalf, WriteHalf};
+
+use crate::error::{ReverseOpenError, TunnelIoError, TunnelOpenError};
+use crate::params::{Substrate, TunnelParams, OP_TUNNEL_OPEN, TUNNEL_ALPN};
+use crate::wire::{frame_datagram, DatagramReader};
+
+/// The typed consumer session (ADR-005): one session = one channel =
+/// one tunnel. Owns the adopted channel entry, the substrate-shaped
+/// data plane, and (when spawned) the session-side pump handle.
+///
+/// The session does **not** implement `Clone` (compile-asserted below):
+/// aliasing a session would alias its teardown; multi-channel
+/// consumers hold a `Vec<TunnelSession>` (or the assembly layer does).
+///
+/// ```compile_fail
+/// use alktunnels::consumer::TunnelSession;
+/// fn assert_clone<T: Clone>() {}
+/// assert_clone::<TunnelSession>();
+/// ```
+///
+/// Fields are `Option` so `take_halves`/`pump_against` can move the
+/// data plane and pump out of the `Drop`-impl type; an emptied session
+/// drops as a no-op (teardown is idempotent — an already-reaped entry
+/// reaps `Err(UnknownChannel)`, ignored).
+pub struct TunnelSession {
+    pub channel_id: u32,
+    manager: Option<ChannelManager>,
+    data: Option<DataPlane>,
+    pump: Option<tokio::task::JoinHandle<(u64, u64)>>,
+}
+
+enum DataPlane {
+    /// Raw pass-through (stream substrates: `tcp`, `unix`) — the
+    /// halves ARE the tunnel (ADR-003); zero tunnel-level framing.
+    Stream(StreamHalves),
+    /// The mandatory `[len: u16 BE]` codec over the halves (`udp`,
+    /// ADR-003): boundary-preserving, `len=0` a legal empty datagram.
+    Datagram(DatagramHalves),
+}
+
+struct StreamHalves {
+    read: ReadHalf<BiStream>,
+    write: WriteHalf<BiStream>,
+}
+
+struct DatagramHalves {
+    read: ReadHalf<BiStream>,
+    write: WriteHalf<BiStream>,
+    reader: DatagramReader,
+    /// Datagrams decoded from a chunk beyond the one returned — a
+    /// single chunk may batch several frames (the codec decodes all
+    /// completions per feed); they queue here for the next
+    /// `recv_datagram`.
+    pending: std::collections::VecDeque<Bytes>,
+}
+
+impl TunnelSession {
+    /// Open a tunnel channel toward a produced resource (the forward
+    /// path, `-L`). Calls the open op
+    /// (`ChannelClient::open_channel(OP_TUNNEL_OPEN, params,
+    /// TUNNEL_ALPN)`), adopts the returned channel ID, splits the
+    /// channel `BiStream`, and presents the substrate-shaped data
+    /// plane. A failed open never yields a session (typed errors —
+    /// ADR-049 §4; no phantom session).
+    pub async fn open(
+        client: &ChannelClient,
+        params: TunnelParams,
+    ) -> Result<Self, TunnelOpenError> {
+        let input = serde_json::to_value(&params).map_err(|e| {
+            TunnelOpenError::Open(ChannelOpenError::CallFailed {
+                error: CallError::internal(format!("params serialize: {e}")),
+            })
+        })?;
+        let (channel_id, send, recv) = client
+            .open_channel(OP_TUNNEL_OPEN, input, TUNNEL_ALPN)
+            .await?;
+        Ok(Self::from_halves(
+            channel_id,
+            params.substrate,
+            client.manager().clone(),
+            send,
+            recv,
+        ))
+    }
+
+    /// Adopt a producer-allocated channel ID (the reverse path, `-R`):
+    /// install the data plane from the adopted halves. Early arrivals
+    /// are parked by the manager and drained on adopt (the adoption
+    /// race — ADR-047 §5). An adopt failure is `AdoptFailed`-class on
+    /// the typed error surface, not an establishment error.
+    ///
+    /// The substrate shapes the data plane (raw halves vs the UDP
+    /// codec) and rode the open call that produced the `channel_id`;
+    /// `adopt` takes it so the session presents the right variant (the
+    /// task sketch omitted it — a datagram session cannot exist without
+    /// it; documented in the task notes).
+    pub async fn adopt(
+        manager: &ChannelManager,
+        channel_id: u32,
+        substrate: Substrate,
+        alpn: impl Into<String>,
+    ) -> Result<Self, TunnelOpenError> {
+        let (send, recv) = manager
+            .adopt_channel(channel_id, alpn, None)
+            .await
+            .map_err(ChannelOpenError::AdoptFailed)?;
+        Ok(Self::from_halves(
+            channel_id,
+            substrate,
+            manager.clone(),
+            send,
+            recv,
+        ))
+    }
+
+    fn from_halves(
+        channel_id: u32,
+        substrate: Substrate,
+        manager: ChannelManager,
+        send: MpscSendStream,
+        recv: MpscRecvStream,
+    ) -> Self {
+        let bidi = BiStream::from_joined(recv, send);
+        let (read, write) = tokio::io::split(bidi);
+        let data = match substrate {
+            Substrate::Tcp | Substrate::Unix => DataPlane::Stream(StreamHalves { read, write }),
+            Substrate::Udp => DataPlane::Datagram(DatagramHalves {
+                read,
+                write,
+                reader: DatagramReader::new(),
+                pending: std::collections::VecDeque::new(),
+            }),
+        };
+        TunnelSession {
+            channel_id,
+            manager: Some(manager),
+            data: Some(data),
+            pump: None,
+        }
+    }
+
+    /// Borrow the raw halves (stream substrates). The halves ARE the
+    /// tunnel — raw pass-through (ADR-003). For owned access, use
+    /// [`TunnelSession::take_halves`]. `None` for datagram sessions
+    /// (drive those with `send_datagram`/`recv_datagram`) or after the
+    /// halves were taken.
+    pub fn stream_halves(
+        &mut self,
+    ) -> Option<(
+        &mut (dyn AsyncRead + Send + Unpin),
+        &mut (dyn AsyncWrite + Send + Unpin),
+    )> {
+        match self.data.as_mut()? {
+            DataPlane::Stream(h) => {
+                let read: &mut (dyn AsyncRead + Send + Unpin) = &mut h.read;
+                let write: &mut (dyn AsyncWrite + Send + Unpin) = &mut h.write;
+                Some((read, write))
+            }
+            DataPlane::Datagram(_) => None,
+        }
+    }
+
+    /// Take the raw halves (either substrate): consumes the session's
+    /// data plane into owned boxed halves. The session stays alive as
+    /// the channel-entry owner — reap with `close`/`join`/`Drop`
+    /// (`join` on a pump-less session completes immediately, reaps
+    /// only, `(0, 0, reaped)`). The halves may be taken once; later
+    /// data-plane calls fail with [`TunnelIoError::ChannelTaken`].
+    /// Datagram halves carry the codec framing (ADR-003) — raw bytes
+    /// on the wire are `[len: u16 BE][payload]` frames.
+    pub fn take_halves(mut self) -> TakenHalves {
+        let manager = self.manager.take();
+        let data = self.data.take();
+        let pump = self.pump.take();
+        let (read, write) = match data {
+            Some(DataPlane::Stream(h)) => (
+                Box::new(h.read) as Box<dyn AsyncRead + Send + Unpin>,
+                Box::new(h.write) as Box<dyn AsyncWrite + Send + Unpin>,
+            ),
+            Some(DataPlane::Datagram(h)) => (
+                Box::new(h.read) as Box<dyn AsyncRead + Send + Unpin>,
+                Box::new(h.write) as Box<dyn AsyncWrite + Send + Unpin>,
+            ),
+            None => unreachable!("take_halves on a spent session"),
+        };
+        TakenHalves {
+            session: TunnelSession {
+                channel_id: self.channel_id,
+                manager,
+                data: None,
+                pump,
+            },
+            read,
+            write,
+        }
+    }
+
+    /// UDP: send one datagram (frame → write → flush; ADR-003's
+    /// mandatory framing). `Oversize` above 65535 (the u16 length
+    /// field would wrap). Empty payloads are legal (`len=0` — the
+    /// F-2 layering).
+    pub async fn send_datagram(&mut self, payload: &[u8]) -> Result<(), TunnelIoError> {
+        match self.data.as_mut() {
+            Some(DataPlane::Datagram(h)) => {
+                let framed = frame_datagram(payload)?;
+                h.write.write_all(&framed).await?;
+                h.write.flush().await?;
+                Ok(())
+            }
+            Some(DataPlane::Stream(_)) => Err(TunnelIoError::WrongSubstrate),
+            None => Err(TunnelIoError::ChannelTaken),
+        }
+    }
+
+    /// UDP: receive the next datagram (boundary-preserving, the
+    /// incremental decode path — the only consumer-visible decode
+    /// path). `Some(bytes)` per datagram — possibly empty (`len=0`
+    /// legal); `None` only on stream EOF (the channels-level sentinel;
+    /// the codec never collides with it, F-2). A stream end
+    /// mid-datagram is [`TunnelIoError::TruncatedDatagram`]
+    /// (fail-loud, OQ-TN-13).
+    pub async fn recv_datagram(&mut self) -> Result<Option<Bytes>, TunnelIoError> {
+        match self.data.as_mut() {
+            Some(DataPlane::Datagram(h)) => {
+                if let Some(dg) = h.pending.pop_front() {
+                    return Ok(Some(dg));
+                }
+                let out = read_one_datagram(&mut h.reader, &mut h.read).await?;
+                match out {
+                    Some((first, rest)) => {
+                        h.pending.extend(rest);
+                        Ok(Some(first))
+                    }
+                    None => Ok(None),
+                }
+            }
+            Some(DataPlane::Stream(_)) => Err(TunnelIoError::WrongSubstrate),
+            None => Err(TunnelIoError::ChannelTaken),
+        }
+    }
+
+    /// Spawn `pump_bidi(channel_halves, accepted_read, accepted_write)`
+    /// against an accepted local connection and hold the returned
+    /// handle (the reverse-flow use; ADR-005). Builder-style: returns
+    /// the session, which now owns the pump — `close` aborts it,
+    /// `join` awaits it, `Drop` aborts it. Consumes the data plane
+    /// (the pump owns the channel halves from here).
+    pub async fn pump_against<S>(mut self, accepted: S) -> Self
+    where
+        S: AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    {
+        let manager = self.manager.take();
+        let data = self.data.take();
+        // The invariant is data-plane XOR pump (a session never holds
+        // both), so this abort is defensive only.
+        if let Some(stale) = self.pump.take() {
+            stale.abort();
+        }
+        let bidi = match data {
+            Some(DataPlane::Stream(h)) => BiStream::from_joined(h.read, h.write),
+            Some(DataPlane::Datagram(h)) => BiStream::from_joined(h.read, h.write),
+            None => unreachable!("pump_against on a spent session"),
+        };
+        let (a_read, a_write) = tokio::io::split(accepted);
+        let pump = tokio::spawn(pump_bidi(bidi, a_read, a_write));
+        TunnelSession {
+            channel_id: self.channel_id,
+            manager,
+            data: None,
+            pump: Some(pump),
+        }
+    }
+
+    /// Tear the tunnel down (the ungraceful path): abort the
+    /// session-owned pump (if any), then reap the adopted channel
+    /// entry (`teardown_channel`). Returns whether the entry existed.
+    pub async fn close(mut self) -> bool {
+        if let Some(pump) = self.pump.take() {
+            pump.abort();
+        }
+        self.manager
+            .take()
+            .map(|m| m.teardown_channel(self.channel_id).is_ok())
+            .unwrap_or(false)
+    }
+
+    /// Await pump completion (both directions finished — half-close
+    /// semantics fall out of `pump_bidi`, W4), THEN reap the adopted
+    /// channel. Returns the `(u64, u64)` copy counts for
+    /// observability plus the reap result. **Pump-less sessions**
+    /// (after `take_halves`, or datagram sessions the caller drives
+    /// directly): completes immediately, reaps only, `(0, 0, reaped)`.
+    pub async fn join(mut self) -> (u64, u64, bool) {
+        let (c2p, p2c) = match self.pump.take() {
+            Some(pump) => match pump.await {
+                Ok(counts) => counts,
+                Err(e) => {
+                    tracing::warn!(
+                        error = %e,
+                        "tunnel session: pump ended abnormally; copy counts unknown"
+                    );
+                    (0, 0)
+                }
+            },
+            None => (0, 0),
+        };
+        let reaped = self
+            .manager
+            .take()
+            .map(|m| m.teardown_channel(self.channel_id).is_ok())
+            .unwrap_or(false);
+        (c2p, p2c, reaped)
+    }
+}
+
+impl Drop for TunnelSession {
+    /// Abort the pump and sync-reap the channel (`teardown_channel` is
+    /// sync — best-effort; `Drop` cannot await). Dropping without
+    /// close/join never leaks the channel entry (ADR-005).
+    fn drop(&mut self) {
+        if let Some(pump) = self.pump.take() {
+            pump.abort();
+        }
+        if let Some(manager) = self.manager.take() {
+            let _ = manager.teardown_channel(self.channel_id);
+        }
+    }
+}
+
+/// The owned halves [`TunnelSession::take_halves`] yields, paired with
+/// the session that still owns the channel entry (reap it via
+/// `TakenHalves::session`).
+pub struct TakenHalves {
+    pub session: TunnelSession,
+    pub read: Box<dyn AsyncRead + Send + Unpin>,
+    pub write: Box<dyn AsyncWrite + Send + Unpin>,
+}
+
+/// Read from the chunk stream until one complete datagram is
+/// assembled (the incremental decode path — ADR-003). `Ok(None)` on
+/// stream EOF; mid-datagram EOF is `TruncatedDatagram` (fail-loud).
+async fn read_one_datagram<R: AsyncRead + Unpin>(
+    reader: &mut DatagramReader,
+    stream: &mut R,
+) -> Result<Option<(Bytes, Vec<Bytes>)>, TunnelIoError> {
+    let mut chunk = vec![0u8; 16 * 1024];
+    loop {
+        let n = stream.read(&mut chunk).await?;
+        if n == 0 {
+            return if reader.is_mid_datagram() {
+                Err(TunnelIoError::TruncatedDatagram)
+            } else {
+                Ok(None)
+            };
+        }
+        let mut dgs = reader.feed(&chunk[..n])?.into_iter();
+        if let Some(first) = dgs.next() {
+            return Ok(Some((first, dgs.collect())));
+        }
+    }
+}
+
+/// Call the producer's open op on channel 0 and extract the allocated
+/// `channel_id` (the reverse path's first step; the POC's shape). The
+/// optional `auth_token` rides the payload — the hub-forwarding path
+/// (precedence: token > `ServingConfig.identity` > transport identity,
+/// CF-005). The assembly layer then constructs the session with
+/// [`TunnelSession::adopt`].
+pub async fn open_reverse_channel(
+    hub_call: &CallConnection,
+    params: &TunnelParams,
+    auth_token: Option<&str>,
+) -> Result<u32, ReverseOpenError> {
+    let substrate_str = match params.substrate {
+        Substrate::Tcp => "tcp",
+        Substrate::Udp => "udp",
+        Substrate::Unix => "unix",
+    };
+    let mut payload = serde_json::json!({
+        "operationId": OP_TUNNEL_OPEN,
+        "input": {
+            "resource": params.resource,
+            "substrate": substrate_str,
+        },
+    });
+    if let Some(token) = auth_token {
+        payload["auth_token"] = serde_json::Value::String(token.to_string());
+    }
+    let response = hub_call.call_with_payload(payload).await;
+    let out = response.result.map_err(ReverseOpenError::Call)?;
+    out.get("channel_id")
+        .and_then(|v| v.as_u64())
+        .map(|id| id as u32)
+        .ok_or(ReverseOpenError::NoChannelId)
+}

src/error.rs

@@ -1,13 +1,13 @@
-//! Tunnel error surfaces: the crate's `TunnelError` plus the typed
+//! Tunnel error surfaces: the crate-level [`TunnelError`], the typed
-//! open-error helpers consumers branch on (ADR-049 §4 surface).
+//! session-open error ([`TunnelOpenError`], ADR-049 §4 surface), and
-//!
+//! the session data-plane error ([`TunnelIoError`]).
-//! Skeleton module — filled by `tunnels/params` (codec + establishment
+
-//! error mapping) and `tunnels/consumer-session` (session errors).
+use alkcall::channels::client::ChannelOpenError;
+use alkcall::protocol::wire::CallError;
+
+use crate::wire::DatagramCodecError;
 
 /// The crate-level error type (thiserror; AGENTS.md convention 2).
-///
-/// Skeleton: the codec and session variants land with
-/// `tunnels/wire-codec` and `tunnels/consumer-session`.
 #[derive(Debug, thiserror::Error)]
 pub enum TunnelError {
     /// The data-plane codec rejected an operation (UDP framing).
@@ -21,3 +21,71 @@ pub enum TunnelError {
     #[error("io: {0}")]
     Io(#[from] std::io::Error),
 }
+
+/// A failed tunnel open or adopt. A failure never yields a session
+/// (no phantom session, mirroring the no-phantom-channel property —
+/// ADR-005). Branch on [`ChannelOpenError::establishment_reason`]
+/// through [`TunnelOpenError::open_ref`] for `channel:open_failed`'s
+/// reason (`dial_failed` / `unknown_resource` / `resource_shortage` /
+/// `handler_error` / `timeout`), or the pre-establishment codes
+/// (`FORBIDDEN`, `channel:too_many_channels`).
+#[derive(Debug, thiserror::Error)]
+pub enum TunnelOpenError {
+    /// The open op failed, or the reply/adopt failed — the typed
+    /// [`ChannelOpenError`] carries the wire [`CallError`] verbatim
+    /// (adopt failures surface as `AdoptFailed`).
+    #[error("tunnel open failed: {0}")]
+    Open(#[from] ChannelOpenError),
+}
+
+impl TunnelOpenError {
+    /// The underlying [`ChannelOpenError`] (the ADR-049 typed surface).
+    pub fn open_ref(&self) -> &ChannelOpenError {
+        match self {
+            TunnelOpenError::Open(e) => e,
+        }
+    }
+}
+
+/// An I/O failure on the session's data plane, or a substrate-shaped
+/// operation attempted on the wrong plane (the POC's shape).
+#[derive(Debug, thiserror::Error)]
+pub enum TunnelIoError {
+    /// The underlying stream failed.
+    #[error("io: {0}")]
+    Io(#[from] std::io::Error),
+    /// The datagram codec rejected the operation (e.g. `Oversize` —
+    /// a >65535-byte send, rejected at frame time per ADR-003).
+    #[error("codec: {0}")]
+    Codec(#[from] DatagramCodecError),
+    /// The stream ended mid-datagram — declared length not satisfied
+    /// (a clean peer never does this; the fail-loud posture, OQ-TN-13
+    /// / ADR-003: never a silent partial datagram).
+    #[error("stream ended mid-datagram (truncated)")]
+    TruncatedDatagram,
+    /// A substrate-shaped operation was attempted on the wrong data
+    /// plane (e.g. `send_datagram` on a stream tunnel).
+    #[error("wrong substrate for this operation")]
+    WrongSubstrate,
+    /// The session's channel halves were already taken — by
+    /// `take_halves` (caller-driven) or `pump_against` (pump-owned);
+    /// no data plane remains on the session.
+    #[error("channel halves already taken from this session")]
+    ChannelTaken,
+}
+
+/// The reverse-path open call failed (`open_reverse_channel` — the
+/// assembly layer calls it before [`crate::consumer::TunnelSession::
+/// adopt`]; the hub's call surface is a `CallConnection`, so the
+/// two-step is the honest API).
+#[derive(Debug, thiserror::Error)]
+pub enum ReverseOpenError {
+    /// The open op failed — the wire [`CallError`]; branch on
+    /// `details.reason` of `channel:open_failed` (ADR-049 §4).
+    #[error("open call failed: {0:?}")]
+    Call(CallError),
+    /// The open op succeeded but the reply carried no `channel_id`
+    /// (malformed responder).
+    #[error("open reply missing channel_id")]
+    NoChannelId,
+}

src/lib.rs

@@ -24,7 +24,8 @@ pub mod params;
 pub mod producer;
 pub mod wire;
 
-pub use error::TunnelError;
+pub use consumer::{open_reverse_channel, TakenHalves, TunnelSession};
+pub use error::{ReverseOpenError, TunnelError, TunnelIoError, TunnelOpenError};
 pub use params::ChannelOpenError;
 pub use params::{establishment_reason, TUNNEL_ALPN, TUNNEL_OPEN_SCOPE};
 pub use wire::MAX_DATAGRAM_LEN;

tasks/tunnels/consumer-session.md

@@ -1,7 +1,7 @@
 ---
 id: tunnels/consumer-session
 name: Consumer half — TunnelSession (open/adopt, data planes, teardown)
-status: pending
+status: completed
 depends_on: [tunnels/params, tunnels/wire-codec, tunnels/producer-open-op]
 scope: broad
 risk: high
@@ -99,6 +99,70 @@ matrix (close/join/Drop paths — no leaked channel entries asserted via
 
 > Agent fills during implementation.
 
+- **`adopt` takes `substrate: Substrate`** (a deviation from the task
+  sketch's `adopt(manager, channel_id, alpn)`): the substrate shapes
+  the data plane (raw halves vs the UDP codec) and cannot be inferred
+  from the channel ID. The ALPN is observability-only (alkcall
+  manager semantics); the substrate rode the open call that produced
+  the ID — passing it to `adopt` keeps the two calls consistent.
+- **`pump_against` takes the whole accepted stream** (`S: AsyncRead +
+  AsyncWrite`) and splits internally — `tokio::io::split` at the
+  pump boundary, matching the POC's `pump_against(accepted)`. Returns
+  the session (builder-style); the pump handle is session-owned and
+  private (close/join/Drop are the only handles needed).
+- **`take_halves(self)` returns `TakenHalves { session, read, write
+  }`** (not bare halves): the session must stay alive as the
+  channel-entry owner (reap via `close`/`join`/`Drop`); returning
+  bare halves would strand the entry. `join` on the held session
+  completes immediately (`(0, 0, reaped)` — pump-less).
+- **`open_reverse_channel`'s hub_call is `&CallConnection`** (not
+  `&Arc<CallConnection>`) — borrowing is the honest shape; callers
+  hold the Arc.
+- **Fields are `Option` + idempotent `Drop`:** a `Drop`-impl type
+  cannot be destructured, so `take_halves`/`pump_against`/`close`/
+  `join` take fields out via `.take()`; an emptied session drops as
+  a no-op. Teardown is idempotent (`teardown_channel` on a reaped
+  entry is `Err(UnknownChannel)`, ignored).
+- **Bogus-ID adopt succeeds by design** (the manager parks early
+  arrivals for any not-yet-seen ID — the adoption-race cover,
+  alkcall ADR-047 §5); the honest `AdoptFailed` probe is the
+  ID-collision path (same ID adopted twice), pinned in the tests.
+- **`recv_datagram` keeps a pending queue**: one chunk may batch
+  several frames; `DatagramReader::feed` decodes all completions per
+  feed, so frames beyond the first queue for subsequent
+  `recv_datagram` calls (dropped frames were a bug the tests caught).
+- **Harness gained `wire_forward`** (`ForwardTopology`): the forward
+  path's consumer is the connect side (`from_connection`, pure
+  consumer) and the producer accepts via the adapter; the accept-side
+  transport-identity posture attaches the dialer's identity in the
+  install hook (modeled; the transport resolves it out-of-band for
+  real). `all_substrate_dial` = stream echo + framed UDP echo.
+
 ## Summary
 
-> Agent fills this on completion.
+> Agent fills this on completion.
+
+Implemented `TunnelSession` (src/consumer.rs, ~415 lines) per
+consumer.md + ADR-005: both construction paths (`open` via
+`ChannelClient::open_channel`, `adopt` via `ChannelManager::
+adopt_channel` + `open_reverse_channel` on the `CallConnection`),
+the substrate-shaped data planes (raw halves for stream; the ADR-003
+codec with a pending frame queue for UDP), pump ownership
+(`pump_against` spawns `pump_bidi` and holds the handle), and the
+full teardown matrix — `close` (abort + ungraceful reap), `join`
+(await + reap + copy counts; pump-less `(0, 0, reaped)`), `Drop`
+(abort + sync reap, never leaks). No `Clone` (compile_fail doc-test).
+
+Tests: `tests/consumer_session.rs` — 12 integration tests over the
+two harness topologies: forward round-trip (session halves), take-
+halves + pump-less join, datagram round-trip incl. the F-2 empty
+datagram, wrong-substrate typed errors, oversize-at-frame-time,
+reverse open+adopt+pump round-trip, W4 half-close, out-of-band close
++ self-reap (the W3 split), Drop-no-leak (pump + pump-less),
+AdoptFailed-on-collision, failed-open-typed (no phantom session),
+concurrent reverse sessions. Harness additions: `wire_forward`
+(`ForwardTopology`) + `all_substrate_dial`/`framed_udp_echo_dial`.
+
+Verified: `cargo test` green (38 total), clippy `-D warnings` clean
+(native + wasm32), fmt clean, wasm32 check passes (default crate
+stays wasm-clean).

tests/consumer_session.rs

@@ -0,0 +1,597 @@
+//! Integration tests for the consumer half — `TunnelSession` per
+//! consumer.md + ADR-005, over the duplex harness. Ported from the
+//! reverse POC's `ReverseTunnel` suite (the W3/W4 probes) plus the
+//! forward POC's session shapes, generalized to the spec's session
+//! type.
+//!
+//! Covered: forward open (session halves drive a duplex round-trip),
+//! reverse open+adopt+pump_against (round-trip, W4 half-close, join
+//! copy counts, out-of-band close + self-reaping, pump-less join),
+//! datagram variant (round-trip incl. the empty datagram — the F-2
+//! layering), wrong-substrate errors, and the teardown matrix
+//! (close/join/Drop — no leaked channel entries via `channel_ids()`).
+
+mod harness;
+
+use alktunnels::params::{Substrate, TunnelParams};
+use alktunnels::producer::ResourceRegistry;
+use alktunnels::{open_reverse_channel, TunnelSession};
+
+use harness::{all_substrate_dial, echo_dial, framed_udp_echo_dial, wire, wire_forward, Topology};
+
+fn params(resource: &str, substrate: Substrate) -> TunnelParams {
+    TunnelParams {
+        resource: resource.to_string(),
+        substrate,
+    }
+}
+
+/// Both managers hold only channel 0 (the phantom-channel property,
+/// asserted from both topologies).
+trait LeakCheck {
+    fn no_data_channels(&self) -> bool;
+}
+
+impl LeakCheck for Topology {
+    fn no_data_channels(&self) -> bool {
+        self.consumer_manager
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0)
+            && self
+                .producer
+                .manager()
+                .channel_ids()
+                .iter()
+                .all(|&id| id == 0)
+    }
+}
+
+impl LeakCheck for harness::ForwardTopology {
+    fn no_data_channels(&self) -> bool {
+        self.consumer
+            .manager()
+            .channel_ids()
+            .iter()
+            .all(|&id| id == 0)
+            && self
+                .producer_manager
+                .channel_ids()
+                .iter()
+                .all(|&id| id == 0)
+    }
+}
+
+fn only_channel_0<T: LeakCheck>(topo: &T) -> bool {
+    topo.no_data_channels()
+}
+
+#[tokio::test]
+async fn forward_open_session_halves_round_trip() {
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_forward(registry, echo_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("open");
+    assert!(topo.producer_manager.has_channel(session.channel_id));
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (read, write) = session.stream_halves().expect("stream session has halves");
+    write.write_all(b"ping").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 4];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip")
+        .expect("read");
+    assert_eq!(&buf, b"ping");
+
+    // Teardown: the session reaps its adopted entry (ADR-005).
+    let channel_id = session.channel_id;
+    let reaped = session.close().await;
+    assert!(reaped, "close reaped the adopted entry");
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(!topo.producer_manager.has_channel(channel_id));
+    assert!(only_channel_0(&topo), "no leaked entries after close");
+}
+
+#[tokio::test]
+async fn forward_open_take_halves_then_pumpless_join() {
+    // consumer.md's pinned semantics: after take_halves the session is
+    // pump-less — join completes immediately, reaps only, (0, 0, reaped).
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_forward(registry, echo_dial()).await;
+
+    let session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("open");
+    let taken = session.take_halves();
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let mut read = taken.read;
+    let mut write = taken.write;
+    write.write_all(b"echo-me").await.expect("write");
+    write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 7];
+    tokio::time::timeout(std::time::Duration::from_secs(5), read.read_exact(&mut buf))
+        .await
+        .expect("round trip through taken halves")
+        .expect("read");
+    assert_eq!(&buf, b"echo-me");
+
+    let session = taken.session;
+    // Drop the halves BEFORE joining: EOF propagates through the mux,
+    // the producer-side wrapper reaps, and the pump-less session reaps
+    // its own adopted entry on join (nothing to await).
+    drop(read);
+    drop(write);
+    let (c2p, p2c, reaped) = session.join().await;
+    assert_eq!((c2p, p2c), (0, 0), "pump-less join: no pump to await");
+    assert!(reaped, "pump-less join reaps the adopted entry");
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn datagram_round_trip_including_empty() {
+    // The datagram variant over the codec (ADR-003): round-trip incl.
+    // the F-2 empty datagram — `len=0` is a legal datagram, never EOF.
+    let registry = ResourceRegistry::new();
+    registry.register("dns", Substrate::Udp, "in-process").await;
+    let topo = wire_forward(registry, framed_udp_echo_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
+        .await
+        .expect("udp open");
+    assert!(
+        session.stream_halves().is_none(),
+        "stream_halves is stream-substrate-only"
+    );
+
+    session
+        .send_datagram(b"query-1")
+        .await
+        .expect("send datagram");
+    session
+        .send_datagram(b"")
+        .await
+        .expect("send empty datagram");
+
+    let dg1 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("no io error")
+        .expect("first datagram");
+    assert_eq!(&dg1[..], b"query-1");
+    let dg2 = tokio::time::timeout(std::time::Duration::from_secs(5), session.recv_datagram())
+        .await
+        .expect("recv timed out")
+        .expect("no io error")
+        .expect("second datagram (empty round-trips — F-2)");
+    assert!(dg2.is_empty(), "the empty datagram round-trips (F-2)");
+
+    let reaped = session.close().await;
+    assert!(reaped);
+    // The producer-side wrapper reap is async (the pump task exits,
+    // the wrapper reaps) — settle before asserting both managers.
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn wrong_substrate_operations_are_typed() {
+    // The POC's WrongSubstrate shape: substrate-shaped operations fail
+    // typed on the wrong plane.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    registry.register("dns", Substrate::Udp, "in-process").await;
+    let topo = wire_forward(registry, all_substrate_dial()).await;
+
+    let mut stream_session = TunnelSession::open(&topo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("tcp open");
+    let err = stream_session
+        .send_datagram(b"nope")
+        .await
+        .expect_err("send_datagram on a stream session");
+    assert!(matches!(err, alktunnels::TunnelIoError::WrongSubstrate));
+    let err = stream_session
+        .recv_datagram()
+        .await
+        .expect_err("recv_datagram on a stream session");
+    assert!(matches!(err, alktunnels::TunnelIoError::WrongSubstrate));
+    stream_session.close().await;
+
+    let mut dg_session =
+        match TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp)).await {
+            Err(_) => panic!("setup: udp dial must succeed in the framed harness"),
+            Ok(mut s) => {
+                assert!(
+                    s.stream_halves().is_none(),
+                    "stream_halves is stream-substrate-only"
+                );
+                s
+            }
+        };
+    dg_session.send_datagram(b"late").await.expect("send works");
+    dg_session.close().await;
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn reverse_open_adopt_pump_round_trip() {
+    // The reverse flow (the POC's ReverseTunnel shape, generalized):
+    // open_reverse_channel → adopt → pump_against → round-trip →
+    // join (copy counts + reap).
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open call");
+
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    assert_eq!(session.channel_id, channel_id);
+
+    let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
+    let session = session.pump_against(accepted_end).await;
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (mut l_read, mut l_write) = tokio::io::split(local_end);
+    l_write.write_all(b"reverse").await.expect("write");
+    l_write.flush().await.expect("flush");
+    let mut buf = vec![0u8; 7];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        l_read.read_exact(&mut buf),
+    )
+    .await
+    .expect("round trip")
+    .expect("read");
+    assert_eq!(&buf, b"reverse");
+
+    // EOF propagates through the two pumps (the two-pump contract).
+    drop(l_write);
+    drop(l_read);
+    let (c2p, p2c, reaped) =
+        tokio::time::timeout(std::time::Duration::from_secs(5), session.join())
+            .await
+            .expect("join timed out");
+    assert!(c2p > 0 && p2c > 0, "both pumps moved bytes: {c2p}, {p2c}");
+    assert!(reaped, "join reaps after pump completion");
+    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+    assert!(
+        !topo.producer.manager().has_channel(channel_id),
+        "producer-side channel reaped (wrapper-managed)"
+    );
+    assert!(only_channel_0(&topo), "no leaked entries after join");
+}
+
+#[tokio::test]
+async fn reverse_half_close_semantics_w4() {
+    // W4's shape: the local side half-closes (shutdown write); the
+    // target sees EOF on its read side while the reverse direction
+    // stays pumpable, and the target's final reply still flows back.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open call");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+
+    let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
+    let session = session.pump_against(accepted_end).await;
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let (mut l_read, mut l_write) = tokio::io::split(local_end);
+    l_write.write_all(b"half").await.expect("write");
+    l_write.flush().await.expect("flush");
+    l_write.shutdown().await.expect("half close");
+
+    let mut buf = vec![0u8; 4];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(5),
+        l_read.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo after half close")
+    .expect("read echo");
+    assert_eq!(&buf, b"half");
+
+    let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.join())
+        .await
+        .expect("join after half close");
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn out_of_band_close_then_session_self_reap() {
+    // The POC's reverse_channel_close_out_of_band shape: a peer-initiated
+    // `channel/close` tears the SERVING side down; the consumer's session
+    // still owns its adopted entry — close/join reaps it (W3 split).
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open call");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    let session = session.pump_against(tokio::io::duplex(1024).0).await;
+
+    let response = topo
+        .consumer_call
+        .call_with_payload(serde_json::json!({
+            "operationId": "channel/close",
+            "input": { "channel_id": channel_id, "reason": "out-of-band" }
+        }))
+        .await;
+    assert!(response.result.is_ok(), "channel/close resolved");
+
+    // The close ran on the PRODUCER side; the consumer's entry remains.
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(
+        !topo.producer.manager().has_channel(channel_id),
+        "producer-side channel reaped by the out-of-band close"
+    );
+    assert!(topo.consumer_manager.has_channel(channel_id));
+    let reaped = session.close().await;
+    assert!(reaped, "session reaped its adopted entry");
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn drop_never_leaks_the_channel_entry() {
+    // ADR-005: Drop = abort + sync reap (best-effort). Dropping a live
+    // session without close/join never leaks the entry.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let ftopo = wire_forward(registry.clone(), echo_dial()).await;
+
+    let session = TunnelSession::open(&ftopo.consumer, params("echo", Substrate::Tcp))
+        .await
+        .expect("open");
+    let channel_id = session.channel_id;
+    assert!(ftopo.consumer.manager().has_channel(channel_id));
+    drop(session);
+    assert!(
+        !ftopo.consumer.manager().has_channel(channel_id),
+        "Drop reaped the adopted entry"
+    );
+
+    // Drop with a session-owned pump: the pump aborts too (the reverse
+    // topology — pump_against is the reverse-flow use).
+    let topo = wire(registry, echo_dial()).await;
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("second reverse open");
+    let session = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("adopt");
+    let session = session.pump_against(tokio::io::duplex(1024).0).await;
+    assert!(topo.consumer_manager.has_channel(channel_id));
+    drop(session);
+    assert!(!topo.consumer_manager.has_channel(channel_id));
+}
+
+#[tokio::test]
+async fn oversize_datagram_rejected_at_frame_time() {
+    // ADR-003: a >65535-byte send is rejected at frame time (`Oversize`),
+    // never a wire overflow.
+    let registry = ResourceRegistry::new();
+    registry.register("dns", Substrate::Udp, "in-process").await;
+    let topo = wire_forward(registry, framed_udp_echo_dial()).await;
+
+    let mut session = TunnelSession::open(&topo.consumer, params("dns", Substrate::Udp))
+        .await
+        .expect("udp open");
+    let err = session
+        .send_datagram(&vec![0u8; alktunnels::MAX_DATAGRAM_LEN + 1])
+        .await
+        .expect_err("oversize must fail");
+    assert!(
+        matches!(
+            err,
+            alktunnels::TunnelIoError::Codec(alktunnels::wire::DatagramCodecError::Oversize(_, _))
+        ),
+        "oversize is a codec error at frame time: {err:?}"
+    );
+    session.close().await;
+}
+
+#[tokio::test]
+async fn failed_open_never_yields_a_session() {
+    // No phantom session: unknown resource → typed error, no channel
+    // anywhere, and `open_ref()` exposes the typed surface.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire_forward(registry, echo_dial()).await;
+
+    let err = match TunnelSession::open(&topo.consumer, params("ghost", Substrate::Tcp)).await {
+        Ok(_) => panic!("unknown resource must not open"),
+        Err(e) => e,
+    };
+    assert_eq!(
+        err.open_ref().establishment_reason(),
+        Some("unknown_resource")
+    );
+    assert!(only_channel_0(&topo));
+
+    // Reverse path on the reverse topology: the call fails typed;
+    // adopt is never reached.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+    let err =
+        match open_reverse_channel(&topo.consumer_call, &params("ghost", Substrate::Tcp), None)
+            .await
+        {
+            Ok(_) => panic!("reverse open of unknown resource must fail"),
+            Err(e) => e,
+        };
+    assert!(matches!(err, alktunnels::ReverseOpenError::Call(_)));
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn adopt_failure_is_typed_not_establishment() {
+    // An adopt failure is `AdoptFailed`-class (post-reply local
+    // failure: ID collision), not an establishment error. A bogus-id
+    // adopt SUCCEEDS by design — the manager parks early arrivals for
+    // any not-yet-seen ID (the adoption race cover, ADR-047 §5) — so
+    // the collision path is the honest probe: adopt the same ID twice.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    let channel_id =
+        open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+            .await
+            .expect("reverse open");
+
+    let first = TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    .expect("first adopt");
+
+    let err = match TunnelSession::adopt(
+        &topo.consumer_manager,
+        channel_id,
+        Substrate::Tcp,
+        alktunnels::TUNNEL_ALPN,
+    )
+    .await
+    {
+        Ok(_) => panic!("adopting a live id twice must fail"),
+        Err(e) => e,
+    };
+    match err.open_ref() {
+        alkcall::channels::client::ChannelOpenError::AdoptFailed(
+            alkcall::channels::manager::ManagerError::ChannelExists(_),
+        ) => {}
+        other => panic!("expected AdoptFailed(ChannelExists), got {other:?}"),
+    }
+
+    first.close().await;
+    tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+    assert!(only_channel_0(&topo));
+}
+
+#[tokio::test]
+async fn concurrent_reverse_sessions_independent() {
+    // Two reverse sessions on one connection: independent channels,
+    // each round-trips its own marker, both reaped.
+    let registry = ResourceRegistry::new();
+    registry
+        .register("echo", Substrate::Tcp, "in-process")
+        .await;
+    let topo = wire(registry, echo_dial()).await;
+
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    let mut sessions = Vec::new();
+    for i in 0..2 {
+        let channel_id =
+            open_reverse_channel(&topo.consumer_call, &params("echo", Substrate::Tcp), None)
+                .await
+                .expect("reverse open");
+        let session = TunnelSession::adopt(
+            &topo.consumer_manager,
+            channel_id,
+            Substrate::Tcp,
+            alktunnels::TUNNEL_ALPN,
+        )
+        .await
+        .expect("adopt");
+        let (accepted_end, local_end) = tokio::io::duplex(64 * 1024);
+        let session = session.pump_against(accepted_end).await;
+        let (mut l_read, mut l_write) = tokio::io::split(local_end);
+        let msg = format!("marker {i}");
+        l_write.write_all(msg.as_bytes()).await.expect("write");
+        l_write.flush().await.expect("flush");
+        let mut buf = vec![0u8; msg.len()];
+        tokio::time::timeout(
+            std::time::Duration::from_secs(5),
+            l_read.read_exact(&mut buf),
+        )
+        .await
+        .expect("round trip")
+        .expect("read");
+        assert_eq!(buf, msg.as_bytes());
+        sessions.push(session);
+    }
+    assert_ne!(sessions[0].channel_id, sessions[1].channel_id);
+
+    for session in sessions {
+        let (_, _, reaped) = session.join().await;
+        assert!(reaped);
+    }
+    assert!(only_channel_0(&topo));
+}
+
+#[allow(dead_code)]
+fn type_assertions() {
+    // The `compile_fail` doc-test in consumer.rs is the no-Clone gate;
+    // this parity stub keeps the type in the test's namespace.
+    let _ = std::mem::size_of::<TunnelSession>();
+}

tests/harness.rs

@@ -1,5 +1,6 @@
 //! Test harness — the producer↔consumer wiring over a
 //! `tokio::io::duplex` carrying the channels 8-byte chunk format.
+#![allow(dead_code)]
 //!
 //! Topology (the reverse POC's shape, mirrored for both directions):
 //! - **producer** (connect side of the channels connection — it runs
@@ -222,6 +223,133 @@ pub async fn wire(registry: ResourceRegistry, dial: alktunnels::producer::DialFn
     .await
 }
 
+/// The forward topology (the `-L` direction): the consumer dials the
+/// transport (connect side — `ChannelClient::from_connection`, a pure
+/// consumer serving nothing) and the producer accepts (the adapter +
+/// an install hook that registers the channel ops + the tunnel
+/// openable on the producer's serving registry). This is the topology
+/// `TunnelSession::open(&channel_client, params)` is normative for:
+/// the consumer's own client calls the open op on the producer's
+/// channel 0.
+///
+/// Identity on the accept-side serving path (the mTLS posture): the
+/// transport authenticated the dialer; the hook attaches the resolved
+/// identity to channel 0 so the serving dispatch sees the caller
+/// (the accept-side analogue of CF-005 (b)'s propagation).
+pub struct ForwardTopology {
+    /// The consumer's own client (connect side, pure consumer). Its
+    /// manager adopted the producer-allocated channel IDs; the
+    /// session reaps via this manager.
+    pub consumer: Arc<ChannelClient>,
+    /// The producer's manager (accept side, serving) — for asserting
+    /// wrapper-managed reaping.
+    pub producer_manager: ChannelManager,
+    /// What the establisher's per-call auth carried (the CF-006 probe).
+    pub identity_witness: Arc<tokio::sync::Mutex<Option<String>>>,
+}
+
+pub async fn wire_forward(
+    registry: ResourceRegistry,
+    dial: alktunnels::producer::DialFn,
+) -> ForwardTopology {
+    let producer_op_registry = Arc::new(OperationRegistry::new());
+    let (producer_manager_tx, mut producer_manager_rx) =
+        tokio::sync::mpsc::channel::<ChannelManager>(1);
+    let identity_witness = Arc::new(tokio::sync::Mutex::new(None::<String>));
+    let dial_witness = Arc::clone(&identity_witness);
+
+    // --- producer (accept side, serving): adapter + install hook ------
+    let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, _auth| {
+        let registry_arc = Arc::clone(&producer_op_registry);
+        let resource_registry = registry.clone();
+        let dial = Arc::clone(&dial);
+        let witness = Arc::clone(&dial_witness);
+        let manager_tx = producer_manager_tx.clone();
+        tokio::spawn(async move {
+            let channel0_bidi = match channel0_conn.accept_bi().await {
+                Ok(s) => s,
+                Err(_) => return,
+            };
+            let (writer, reader) = split_single_stream(channel0_bidi);
+            // The accept-side transport-identity posture: the transport
+            // authenticated the dialer; attach the resolved identity to
+            // channel 0 (the dispatch reads it as the caller identity).
+            let _ = channel0_conn.set_identity(consumer_identity());
+            let call_connection = Arc::new(CallConnection::new_single_stream(
+                channel0_conn,
+                Arc::clone(&writer),
+            ));
+
+            // Register the generic channel ops + the tunnel openable
+            // (the producer.md recipe) before the dispatch loop runs.
+            let core = alkcall::channels::operations::ChannelCore::new(
+                manager.clone(),
+                alkcall::channels::policy::default_policy(),
+            );
+            let ops = ChannelOperations::with_default_policy(manager.clone());
+            ops.register_on(&registry_arc)
+                .expect("register channel ops on producer");
+            register_tunnel_openable(
+                &core,
+                &resource_registry,
+                &registry_arc,
+                AuthContext::anonymous(b"alk/tunnel"),
+                dial,
+                Some(witness),
+            )
+            .expect("register tunnel openable");
+            let _ = manager_tx.send(manager).await;
+
+            let dp = Dispatcher::new(
+                registry_arc,
+                Arc::new(alkcall::core::auth::NoopIdentityProvider),
+            );
+            dp.serve_single_stream(call_connection, reader, writer)
+                .await;
+        })
+    });
+
+    // --- transport ------------------------------------------------------
+    let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
+    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
+    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
+
+    // The dialer's transport identity (the mTLS/QUIC posture) — set
+    // before from_connection. The accept side resolves it out-of-band
+    // (modeled in the hook above).
+    consumer_conn
+        .set_identity(consumer_identity())
+        .expect("transport identity set once");
+
+    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
+    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
+    let _adapter_task = tokio::spawn(async move {
+        let _ = alkcall::core::types::ProtocolHandler::handle(
+            &adapter,
+            producer_conn,
+            &producer_transport_auth,
+        )
+        .await;
+    });
+
+    // --- consumer (connect side): pure client ---------------------------
+    let consumer_client = ChannelClient::from_connection(consumer_conn)
+        .await
+        .expect("consumer channel client init");
+    let consumer_client = Arc::new(consumer_client);
+
+    let producer_manager = producer_manager_rx
+        .recv()
+        .await
+        .expect("producer manager captured");
+
+    ForwardTopology {
+        consumer: consumer_client,
+        producer_manager,
+        identity_witness,
+    }
+}
+
 /// An echo dial closure (the ADR-004 injection point under test): an
 /// in-process pipe per dial — the "target" echoes bytes until EOF.
 /// The consumer side gets split `DuplexStream` halves; a spawned task
@@ -263,6 +391,17 @@ pub fn echo_dial() -> alktunnels::producer::DialFn {
     })
 }
 
+/// A both-substrates dial: stream substrates echo via pipes, UDP
+/// echoes via the framed codec (the datagram tests' single dial).
+pub fn all_substrate_dial() -> alktunnels::producer::DialFn {
+    let echo = echo_dial();
+    let udp = framed_udp_echo_dial();
+    Arc::new(move |substrate: Substrate, backing: &str| match substrate {
+        Substrate::Udp => udp(substrate, backing),
+        Substrate::Tcp | Substrate::Unix => echo(substrate, backing),
+    })
+}
+
 /// A failing dial closure (the `dial_failed` stand-in).
 pub fn failing_dial(message: &'static str) -> alktunnels::producer::DialFn {
     Arc::new(move |_substrate: Substrate, _backing: &str| {
@@ -273,3 +412,60 @@ pub fn failing_dial(message: &'static str) -> alktunnels::producer::DialFn {
         })
     })
 }
+
+/// A framed UDP echo dial (the ADR-003 codec at the target boundary —
+/// the in-process stand-in for the real `local` UDP adapter): the
+/// dial's halves carry the `[len: u16 BE]` wire framing; a spawned
+/// echo task decodes frames from the channel side and re-frames the
+/// echoed payloads back. The producer pump copies raw bytes; the
+/// codec lives at this boundary (the pump never sees UDP specifics).
+pub fn framed_udp_echo_dial() -> alktunnels::producer::DialFn {
+    Arc::new(move |substrate: Substrate, backing: &str| {
+        let backing = backing.to_string();
+        Box::pin(async move {
+            match substrate {
+                Substrate::Udp => {
+                    let (consumer_side, target_side) = tokio::io::duplex(64 * 1024);
+                    tokio::spawn(async move {
+                        use alktunnels::wire::DatagramReader;
+                        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+                        let (mut t_read, mut t_write) = tokio::io::split(target_side);
+                        let mut reader = DatagramReader::new();
+                        let mut buf = vec![0u8; 16 * 1024];
+                        loop {
+                            let n = match t_read.read(&mut buf).await {
+                                Ok(0) | Err(_) => return,
+                                Ok(n) => n,
+                            };
+                            let dgs = match reader.feed(&buf[..n]) {
+                                Ok(dgs) => dgs,
+                                Err(_) => return,
+                            };
+                            for dg in dgs {
+                                let framed = match alktunnels::wire::frame_datagram(&dg) {
+                                    Ok(f) => f,
+                                    Err(_) => return,
+                                };
+                                if t_write.write_all(&framed).await.is_err() {
+                                    return;
+                                }
+                                let _ = t_write.flush().await;
+                            }
+                        }
+                    });
+                    let _ = &backing;
+                    let (c_read, c_write) = tokio::io::split(consumer_side);
+                    Ok(alktunnels::producer::TargetHandle {
+                        read: Box::new(c_read),
+                        write: Box::new(c_write),
+                    })
+                }
+                Substrate::Tcp | Substrate::Unix => {
+                    Err(alktunnels::producer::TunnelEstablishError::DialFailed(
+                        format!("stream dial not wired in the udp harness: {backing}"),
+                    ))
+                }
+            }
+        })
+    })
+}