composed: the direct-op egress — the lazy destination→tunnel table

The `composed` feature's ASSOCIATE egress (ADR-008's composed
variant, ADR-009's gate): `composed::direct_egress` — each
destination dials through an alktunnels 0.2.0 direct-op channel
(`TunnelSession::open_direct`, `Substrate::Udp`), per-target ACL
for free, the POC #2 `ComposedEgress` shape ported to the landed
seam types.

- `Cargo.toml` — `composed = ["dep:alktunnels"]` (alktunnels pinned
  "0.2.0"; rides `composed` alone, `cargo tree --edges normal`
  spot-checked); dev-dep for the composed suite.
- `src/composed/mod.rs` — the lazy destination→tunnel table
  (LRU-capped, `DEFAULT_MAX_TUNNELS` = 256), per-tunnel relay tasks
  draining a shared inbound queue, `SessionOwner` (the owner wrapper
  shared with the composed accept — the batch-1 review's
  Sync-bound wrinkle), the readiness contract (real async park;
  ended relays → association-end `Err`, never a hot `Ok(None)`),
  framing-once, domain pass-through unresolved (ADR-012), the
  backpressure bound documented on the constructor.
- Two deliberate deviations from the POC shape, documented in the
  task's Notes: the closed-queue condition is a live-relay count
  (the egress holds its own `Sender`, so the POC's queue-close arm
  was unreachable through this surface); the tunnel codec is
  alktunnels' (`[len]` + payload), not the RFC-framed reader.
- `tests/composed-egress.rs` — round-trips per destination → one
  tunnel each, table reuse, LRU eviction/re-dial, the readiness
  park, the closed-relay-set `Err`, framing-once through the wire
  layer, domain pass-through — over a `wire_tunnel_direct` harness
  topology (the consumer-side shape with the direct op on the
  producer's serving registry).

Verification: `cargo test` + `--all-features` pass (13 suites);
`cargo check --features composed --target wasm32-unknown-unknown`
clean (the hard gate); clippy `--all-targets --all-features
-- -D warnings` clean; `cargo fmt --check` clean; `cargo doc
--no-deps` clean.

src/composed/mod.rs

@@ -0,0 +1,406 @@
+//! The `composed` feature's egress provider (ADR-008's composed
+//! variants, ADR-009's feature gate): [`direct_egress`] — the UDP
+//! ASSOCIATE egress that dials each destination through an alktunnels
+//! 0.2.0 direct-op channel ([`alktunnels::TunnelSession::open_direct`],
+//! `Substrate::Udp`).
+//!
+//! **Per-target ACL for free** (ADR-008's composed benefit): each
+//! destination rides its own `channels/tunnel/direct`-governed channel
+//! — the producer's `tunnel:direct` scope gate and dial policy apply
+//! per destination, no SOCKS5-layer allowlist needed (ADR-006/ADR-012:
+//! whatever ACL governs the tunnel capability governs everything
+//! reachable through it).
+//!
+//! The shape is POC #2's `ComposedEgress`
+//! (`/workspace/alksocks-udp-poc/src/egress.rs`), ported to the landed
+//! seam types: a lazy destination→tunnel table — each new destination
+//! opens one UDP-tunnel channel, a relay task per tunnel drains
+//! inbound datagrams into a shared queue, the write half stays in the
+//! table (LRU-capped; dropping a write half EOFs the tunnel
+//! producer-side; the entry's session drop reaps the channel —
+//! alktunnels ADR-005's teardown contract).
+//!
+//! Contracts (the landed seam types' normative docs, honored here):
+//!
+//! - **The readiness contract** (load-bearing, `Egress::next_reply`):
+//!   parks on the shared inbound queue (`recv().await` — `Pending`
+//!   until a relay task delivers); a closed queue (all relays ended)
+//!   returns `Err` — never a hot `Ok(None)`. The stage's zero-yield
+//!   defense (review 002 R-1) is the backstop, not the design.
+//! - **Framing-once**: replies carry *inner* bytes (`addr + data` —
+//!   the reply address is the entry's recorded destination, the RFC's
+//!   per-datagram source semantics); the length prefix is the wire
+//!   layer's, applied exactly once downstream. `send` takes the
+//!   payload only; the tunnel-side framing (`[len: u16 BE]`, the
+//!   alktunnels ADR-003 codec) is the tunnel wire's, not the RFC
+//!   wire's.
+//! - **Per-datagram `send` is awaited inline** (the stage
+//!   interleaves sends with reply drains): a slow egress applies
+//!   backpressure to the association, not to other channels — the
+//!   batch-2 composed-egress author inherits this bound knowingly
+//!   (wire.md's note). The table's open cost rides the send path
+//!   (lazy open on first datagram per destination — the POC's
+//!   measured ~0.8 ms).
+//! - **The resource/target mapping**: `open_direct` takes a
+//!   [`SubstrateAddr`] — the destination's [`TargetAddr`] renders to
+//!   it (IP forms direct; domain form passes through unresolved —
+//!   producer-side resolution, the ssh `-D` semantic; ADR-012).
+//!
+//! **The Sync-bound wrinkle** (the batch-1 review's seam contract vs
+//! alktunnels' surface): the landed `TargetHandle`/`BindTargetHandle`
+//! halves are `Box<dyn AsyncRead + Send + Sync + Unpin>`, but
+//! alktunnels' `TakenHalves` yields `Send + Unpin` only. The
+//! owner-wrapper posture (egress.rs's documented contract: "keep-alive
+//! state a substrate needs rides inside the halves — a provider-side
+//! owner wrapper") is designed here as [`SessionOwner`] and shared
+//! with the composed accept (the sibling task): an alktunnels half
+//! wrapped with its session inside, so the handle payload's `Sync`
+//! bound is met by ownership and the channel entry reaps when the
+//! wrapper (and its session) drops. For the egress the wrapper is not
+//! the handle payload's burden — the egress owns the session in the
+//! table entry (the POC's shape); the relay task owns the bare read
+//! half. The accept variant (the sibling task) wraps the accepted
+//! halves in this wrapper.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use alkcall::channels::client::ChannelClient;
+use alktunnels::params::{Substrate, SubstrateAddr};
+use alktunnels::TunnelSession;
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use tokio::io::{AsyncRead, AsyncWrite};
+
+use crate::egress::{Egress, EgressError, EgressFactory};
+pub use crate::rfc1928::util::target_addr::TargetAddr;
+use crate::wire::MAX_INNER_LEN;
+
+/// The default LRU cap (the 256-channel default cap bounds fan-out —
+/// ADR-010's posture; the cap is config, not wire).
+pub const DEFAULT_MAX_TUNNELS: usize = 256;
+
+/// The destination's render into the direct op's [`SubstrateAddr`] (the
+/// resource/target mapping): IP forms direct; domain form passes
+/// through unresolved (producer-side resolution, the ssh `-D`
+/// semantic — ADR-012).
+pub fn substrate_addr_of(addr: &TargetAddr) -> SubstrateAddr {
+    match addr {
+        TargetAddr::Ip(sa) => SubstrateAddr::HostPort {
+            host: sa.ip().to_string(),
+            port: sa.port(),
+        },
+        TargetAddr::Domain(domain, port) => SubstrateAddr::HostPort {
+            host: domain.clone(),
+            port: *port,
+        },
+    }
+}
+
+/// Render the destination to the table's key form (the string the LRU
+/// orders by): `host:port` per [`SubstrateAddr`]'s render (IP literal
+/// or unresolved domain — the domain stays unresolved, so domain and
+/// IP destinations key distinctly, as the client addressed them).
+pub fn key_of(addr: &TargetAddr) -> String {
+    match substrate_addr_of(addr) {
+        SubstrateAddr::HostPort { host, port } => format!("{host}:{port}"),
+        SubstrateAddr::Path { path } => path,
+    }
+}
+
+/// The owner wrapper (the batch-1 review's seam contract vs
+/// alktunnels' surface — the shared design point with the composed
+/// accept): an alktunnels half wrapped with its [`TunnelSession`]
+/// inside. The landed handle payloads carry `+ Sync` (the
+/// `ChannelPlan` payload bound), alktunnels' `TakenHalves` yields
+/// `Send + Unpin` only — the wrapper owns the session beside the half,
+/// so dropping the wrapper drops the half and the session together
+/// (the session's drop reaps the channel entry — alktunnels ADR-005).
+/// Whether the wrapper is `Sync` (the accept's need) is the half's
+/// own `Sync`ness — asserted by the compile-shape test against the
+/// alktunnels 0.2.0 surface; the egress needs only `Send` (the relay
+/// task), which the wrapper always has.
+pub struct SessionOwner<H> {
+    half: H,
+    session: TunnelSession,
+}
+
+impl<H: AsyncRead + Send + Unpin> AsyncRead for SessionOwner<H> {
+    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 = std::pin::Pin::into_inner(self);
+        std::pin::Pin::new(&mut this.half).poll_read(cx, buf)
+    }
+}
+
+impl<H> SessionOwner<H> {
+    /// Wrap the half with its session (the owner).
+    pub fn new(half: H, session: TunnelSession) -> Self {
+        Self { half, session }
+    }
+
+    /// The wrapped half.
+    pub fn half(&self) -> &H {
+        &self.half
+    }
+
+    /// The wrapped half, mutably.
+    pub fn half_mut(&mut self) -> &mut H {
+        &mut self.half
+    }
+
+    /// Unwrap: the half plus the session (the owner), for a caller
+    /// that needs them separated again.
+    pub fn into_parts(self) -> (H, TunnelSession) {
+        (self.half, self.session)
+    }
+}
+
+/// The composed egress: the lazy destination→tunnel table (the
+/// per-association ASSOCIATE seam object). Constructed via
+/// [`direct_egress`] (the `EgressFactory` the producer registration
+/// consumes).
+pub struct ComposedEgress {
+    client: Arc<ChannelClient>,
+    max_tunnels: usize,
+    entries: HashMap<String, TableEntry>,
+    order: std::collections::VecDeque<(String, TargetAddr)>,
+    inbound: tokio::sync::mpsc::Receiver<Inbound>,
+    inbound_tx: tokio::sync::mpsc::Sender<Inbound>,
+    /// Live relay tasks — the readiness contract's closed-queue
+    /// condition: when every relay has ended (each reported via
+    /// [`Inbound::Ended`]) and the queue is drained, `next_reply`
+    /// returns `Err` (the association end), never a hot `Ok(None)`.
+    /// The factory `Sender` alone does not close the queue (the
+    /// egress holds it) — the count is the design.
+    live_relays: usize,
+}
+
+/// The shared inbound queue's message: a tunnel's datagram (keyed by
+/// the table key — the reply address rides the entry) or a relay's
+/// end marker (the tunnel's EOF drained).
+enum Inbound {
+    Datagram(String, Bytes),
+    Ended,
+}
+
+/// One table entry: the tunnel's write half (the udp substrate's
+/// framed half — `[len: u16 BE]` frames on the tunnel wire) + the
+/// reply address to report for datagrams coming back through it + the
+/// session (the channel-entry owner — the egress owns it, not a
+/// handle payload; dropping the entry EOFs the write half and reaps
+/// the channel — alktunnels ADR-005).
+struct TableEntry {
+    _session: TunnelSession,
+    write: Box<dyn AsyncWrite + Send + Unpin>,
+    reply_addr: TargetAddr,
+}
+
+/// The per-tunnel relay task: drain the tunnel's framed read half
+/// into the egress's shared inbound queue. Exits on EOF (the tunnel
+/// ended); the write half stays in the table until eviction or stage
+/// exit. Framing decode is the tunnel codec's (the alktunnels
+/// ADR-003) — the datagrams here are inner payloads (no RFC address
+/// on the tunnel wire; the reply address is the entry's recorded
+/// destination).
+async fn relay_read(
+    mut read: impl AsyncRead + Send + Unpin,
+    tx: tokio::sync::mpsc::Sender<Inbound>,
+    key: String,
+) {
+    use tokio::io::AsyncReadExt;
+    let mut reader = alktunnels::wire::DatagramReader::new();
+    let mut chunk = vec![0u8; 16 * 1024];
+    loop {
+        match read.read(&mut chunk).await {
+            Ok(0) | Err(_) => break,
+            Ok(n) => {
+                for dg in reader.feed(&chunk[..n]) {
+                    if tx.send(Inbound::Datagram(key.clone(), dg)).await.is_err() {
+                        return;
+                    }
+                }
+            }
+        }
+    }
+    let _ = tx.send(Inbound::Ended).await;
+}
+
+/// Assemble the reply's inner bytes (`addr + data` — the entry's
+/// recorded destination + the payload; framing-once: no length prefix
+/// here, the wire layer's `frame_inner` adds it downstream). The
+/// oversize guard mirrors the wire layer's frame bound.
+fn inner_bytes(addr: &TargetAddr, payload: &[u8]) -> Result<Bytes, EgressError> {
+    let mut inner = addr
+        .to_be_bytes()
+        .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
+    inner.extend_from_slice(payload);
+    if inner.len() > MAX_INNER_LEN {
+        return Err(EgressError::General(format!(
+            "reply oversize: {} > {MAX_INNER_LEN}",
+            inner.len()
+        )));
+    }
+    Ok(Bytes::from(inner))
+}
+
+impl ComposedEgress {
+    /// Build the egress (the per-association ASSOCIATE seam object).
+    /// `client` is the channels client the direct ops ride; the
+    /// constructor's config carries the LRU cap (`max_tunnels` — the
+    /// fan-out bound, config not wire; the producer's `tunnel:direct`
+    /// ACL governs which destinations the channels accept — the
+    /// per-target-ACL-for-free benefit).
+    pub fn new(client: Arc<ChannelClient>, max_tunnels: usize) -> Self {
+        let (tx, rx) = tokio::sync::mpsc::channel(64);
+        Self {
+            client,
+            max_tunnels,
+            entries: HashMap::new(),
+            order: std::collections::VecDeque::new(),
+            inbound: rx,
+            inbound_tx: tx,
+            live_relays: 0,
+        }
+    }
+
+    /// The entry for `addr` — open lazily on first datagram (the
+    /// open cost rides the send path, per-datagram awaited inline).
+    async fn entry_for(&mut self, addr: &TargetAddr) -> Result<&mut TableEntry, EgressError> {
+        let key = key_of(addr);
+        if !self.entries.contains_key(&key) {
+            while self.entries.len() >= self.max_tunnels {
+                let Some((evict, _)) = self.order.pop_front() else {
+                    break;
+                };
+                self.entries.remove(&evict);
+            }
+            let target = substrate_addr_of(addr);
+            let session = TunnelSession::open_direct(self.client.as_ref(), target, Substrate::Udp)
+                .await
+                .map_err(|e| EgressError::Unreachable(format!("tunnel open `{key}`: {e}")))?;
+            let taken = session.take_halves();
+            // The session stays INSIDE the entry: dropping it reaps
+            // the channel (ADR-005's teardown contract), which would
+            // EOF the relay task instantly (the POC's insert-time
+            // bug). The relay task owns the bare read half.
+            tokio::spawn(relay_read(taken.read, self.inbound_tx.clone(), key.clone()));
+            self.live_relays += 1;
+            self.entries.insert(
+                key.clone(),
+                TableEntry {
+                    _session: taken.session,
+                    write: taken.write,
+                    reply_addr: addr.clone(),
+                },
+            );
+        }
+        self.order.retain(|(k, _)| k != &key);
+        self.order.push_back((key.clone(), addr.clone()));
+        Ok(self.entries.get_mut(&key).expect("entry inserted above"))
+    }
+}
+
+impl Egress for ComposedEgress {
+    fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> {
+        Box::pin(async move {
+            let entry = self.entry_for(&addr).await?;
+            use tokio::io::AsyncWriteExt;
+            let framed = alktunnels::wire::frame_datagram(&payload)
+                .map_err(|e| EgressError::General(format!("tunnel frame: {e}")))?;
+            let write: &mut (dyn AsyncWrite + Send + Unpin) = &mut *entry.write;
+            write
+                .write_all(&framed)
+                .await
+                .map_err(|e| EgressError::General(format!("tunnel write: {e}")))?;
+            write
+                .flush()
+                .await
+                .map_err(|e| EgressError::General(format!("tunnel flush: {e}")))?;
+            Ok(())
+        })
+    }
+
+    fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
+        Box::pin(async move {
+            loop {
+                // A non-blocking first pass over the shared queue (the
+                // relay tasks may have queued replies already), then a
+                // real async wait on the queue (`Pending` until a
+                // relay task delivers — the readiness contract).
+                loop {
+                    match self.inbound.try_recv() {
+                        Ok(Inbound::Datagram(key, dg)) => {
+                            let addr = match self.entries.get(&key) {
+                                Some(entry) => entry.reply_addr.clone(),
+                                None => continue,
+                            };
+                            return Ok(Some(inner_bytes(&addr, &dg)?));
+                        }
+                        // A relay ended (its tunnel's EOF drained) —
+                        // count it; the closed-queue condition is
+                        // every relay ended.
+                        Ok(Inbound::Ended) => {
+                            self.live_relays = self.live_relays.saturating_sub(1);
+                            continue;
+                        }
+                        Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break,
+                        Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break,
+                    }
+                }
+                if self.live_relays == 0 {
+                    // Every relay ended (the tunnels ended): a closed
+                    // relay set must NOT spin the stage's select (an
+                    // instantly-ready None would starve the channel
+                    // branch) — end the association instead.
+                    return Err(EgressError::General("all tunnel relays ended".into()));
+                }
+                match self.inbound.recv().await {
+                    Some(Inbound::Datagram(key, dg)) => {
+                        let addr = match self.entries.get(&key) {
+                            Some(entry) => entry.reply_addr.clone(),
+                            None => continue,
+                        };
+                        return Ok(Some(inner_bytes(&addr, &dg)?));
+                    }
+                    // A relay ended while parked — loop back: the
+                    // count check above decides whether the
+                    // association ends (never a hot `Ok(None)`).
+                    Some(Inbound::Ended) => {
+                        self.live_relays = self.live_relays.saturating_sub(1);
+                        continue;
+                    }
+                    None => {
+                        return Err(EgressError::General("all tunnel relays ended".into()));
+                    }
+                }
+            }
+        })
+    }
+}
+
+/// The composed egress constructor (ADR-008's composed variant,
+/// producer.md §Wiring): the `EgressFactory` the producer
+/// registration consumes. One per ASSOCIATE; the stage owns the
+/// returned egress; stage exit drops it (write halves dropped = EOF
+/// to the tunnel producers; the entries' sessions reap the channels —
+/// alktunnels ADR-005's teardown contract).
+///
+/// # Backpressure (inherited knowingly)
+///
+/// Per-datagram `send` is awaited inline by the producer stage (it
+/// interleaves sends with reply drains): a slow egress applies
+/// backpressure to the association, not to other channels. The lazy
+/// open's cost (~0.8 ms per new destination in the POC's
+/// measurement) rides the send path.
+pub fn direct_egress(client: Arc<ChannelClient>, max_tunnels: usize) -> EgressFactory {
+    Arc::new(move || {
+        let client = Arc::clone(&client);
+        Box::pin(async move {
+            Ok(Box::new(ComposedEgress::new(client, max_tunnels)) as Box<dyn Egress + Send>)
+        })
+    })
+}

src/lib.rs

@@ -43,10 +43,12 @@
 //!   channel's teardown; no `Clone`.
 //!
 //! `local` (the socket-backed assembly-layer providers —
-//! [`local::tcp_dial`] / [`local::udp_egress`] / [`local::bind_accept`],
-//! ADR-008/009) is feature-gated: the default crate is protocol-only and
-//! wasm-clean by construction, and the composed providers (`composed`)
-//! are still pending.
+//! [`crate::local::tcp_dial`] / [`crate::local::udp_egress`] /
+//! [`crate::local::bind_accept`], ADR-008/009) is feature-gated: the
+//! default crate is protocol-only and wasm-clean by construction, and
+//! the composed providers ride `composed`
+//! ([`crate::composed::direct_egress`] — the alktunnels direct-op
+//! egress, ADR-008/009).
 //!
 //! Architecture: `docs/architecture/overview.md` (spec set:
 //! wire.md, producer.md, consumer.md, rfc1928.md); the cross-cutting
@@ -68,6 +70,9 @@ pub mod wire;
 #[cfg(feature = "local")]
 pub mod local;
 
+#[cfg(feature = "composed")]
+pub mod composed;
+
 #[cfg(feature = "local")]
 pub mod door;
 
@@ -116,6 +121,11 @@ pub use local::{bind_accept, tcp_dial, udp_egress, LocalEgress};
 #[cfg(feature = "local")]
 pub use local::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard};
 
+#[cfg(feature = "composed")]
+pub use composed::{
+    direct_egress, key_of, substrate_addr_of, ComposedEgress, SessionOwner, DEFAULT_MAX_TUNNELS,
+};
+
 #[cfg(feature = "local")]
 pub use consumer::TokenClient;
 #[cfg(feature = "local")]