review: the batch-2 gate closed — the assembly layer verified, R-1/R-2 fixed

Review checkpoint closing batch 2 (the assembly layer). The gate
opens: the never-binds posture, seam contracts, door fidelity, the
bind-first graduation, feature-gate hygiene, and POC fidelity all
verify clean; two findings were corrected in the review change.

Findings corrected:

- R-1 (major): the UDP front door's in-band decode created a fresh
  DatagramReader per call and kept only the first frame of a batch —
  POC #2's first-cut bug (reader state must persist across calls;
  batched datagrams were silently dropped, mid-frame EOF was silent).
  Fix: an owning InbandBridge (persistent reader + pending queue +
  fail-loud mid-frame EOF, the AssociateSession shape) threaded
  through the channel->socket pump; pinned by three bridge unit
  tests (which fail against the per-call reader) and an end-to-end
  doubled-echo door-udp test.
- R-2 (convention 2): seven non-test expects -> typed errors —
  DoorBuildError (both door builders' .build() is now Result),
  ProducerDoorBuildError/ProducerDoorError (the bind decision is
  serve's typed error), the composed table invariant as an egress
  error, register_forwarded_serving -> Result (the alkcall
  diagnostic surfaced), TokenClientError (wrap on a spent client).
  New error types re-exported; test call sites updated.
- R-3/R-4 (doc residues): producer.md's local-backend section
  described a channel pipe — the landed ProducerDoor serves
  serve_socks5_over on the accepted socket directly;
  associate_with_token's doc (the TCP door refuses ASSOCIATE, never
  calls it); SessionOwner's doc (the accept uses SyncHalves/
  OwnedWrite).

Verification: cargo test (default) green; cargo test --all-features
green; cargo clippy --all-targets [--all-features] -- -D warnings
clean; cargo fmt --check clean; both wasm checks clean (default +
composed); local-on-wasm fails at mio as designed; cargo doc
--no-deps clean.

Task: tasks/implementation/review-assembly-layer.md (status:
completed; the batch-2 gate opens release polish).

docs/architecture/producer.md

@@ -130,24 +130,31 @@ Contracts (POC-pinned, load-bearing):
 
 The optional shape where the producer serves from a real listener:
 a feature-gated backend binds `socks5://127.0.0.1:port` (or any
-address) and pipes each accepted TCP connection raw onto a channels
+address) and hands each accepted TCP connection to the landed
-`BiStream` (`copy_bidirectional`, zero protocol translation — POC
+substrate-free entry `Socks5Producer::serve_socks5_over` directly —
-#1's curl front door validated the pipe shape). The same door shape
+the RFC conversation runs on the accepted socket (the NoAuth
-in reverse: this is a *front door for the producer side* — vanilla
+in-stream shape: greeting, command, reply, data plane, all on the
-clients reach a locally-served SOCKS5 resource through it. The door
+accepted socket), and the seams dial/egress/accept locally. The same
-surface (NoAuth raw pipe vs RFC 1929 cred-bearing, ADR-005) is the
+door shape in reverse: this is a *front door for the producer side*
-consumer door's code shape (see [consumer.md](consumer.md)).
+— vanilla clients reach a locally-served SOCKS5 resource through it
+(`src/local/producer_door.rs`, the `ProducerDoor`/builder/guard
+surface). A cred-bearing variant (RFC 1929) would wrap
+`accept_password_auth` before `serve_socks5_over` — a documented
+follow-up, not landed.
 
 The door's accept path owns the idle defense the protocol layer does
-not provide (ADR-014): an accepted connection that sends no first
+not provide (ADR-014): an accepted connection that sends no first byte
-byte within a configured window is closed door-side (an accepted
+within a configured window is closed door-side (an accepted socket is
-socket is not a channel — the cap regime does not bound it), plus
+not a channel — the cap regime does not bound it), plus the
-per-credential/per-peer session caps and accept-queue bounds as the
+max-concurrent-sessions cap (an accept beyond the bound is refused
-deployment's policy.
+door-side, no session spawned) — the two constructor knobs.
 
 The producer-side local *backend* for the I/O planes (the
 local-socket providers — TCP dial, UDP egress, BIND listener) is
-ADR-008's variant set, also feature `local`.
+ADR-008's variant set, also feature `local`. The BIND command gate
+on this path is the builder's `bind_policy` (there is no open-op
+identity on an accepted socket — the accepted socket is the whole
+caller; the pump handler's identity-derived gate does not apply).
 
 ## Wiring (Assembly Layer)
 

src/composed/mod.rs

@@ -56,19 +56,17 @@
 //! `alkcall::BiStream` is `Send + Unpin` but NOT `Sync` (its inner is
 //! a `dyn AsyncReadWrite` trait object). 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
+//! needs rides inside the halves — a provider-side owner wrapper")
-//! designed here as [`SessionOwner`] and shared with the composed
+//! takes two shapes here: [`SessionOwner`] (the generic half-wraps-
-//! accept (the sibling task): an alktunnels half wrapped with its
+//! session wrapper — the egress's design point, exported as the
-//! session inside, so the handle payload's `Sync` bound is met by
+//! wrapper-aware deployment's surface) and the accept's own
-//! ownership and the channel entry reaps when the wrapper (and its
+//! [`SyncHalves`]/[`OwnedWrite`] adapter pair (a `std::sync::Mutex`
-//! session) drops. For the egress the wrapper is not the handle
+//! around the half — `Mutex<T: Send>` is `Sync`, no unsafe — with
-//! payload's burden — the egress owns the session in the table entry
+//! the channel session owned inside the payload, [`ForwardedOwner`]:
-//! (the POC's shape); the relay task owns the bare read half. The
+//! dropping the handle drops the session AND the halves together).
-//! accept variant meets the `+ Sync` bound with the [`SyncHalves`]
+//! For the egress the wrapper is not the handle payload's burden —
-//! adapter (a `std::sync::Mutex` around the joined halves — a
+//! the egress owns the session in the table entry (the POC's shape);
-//! `Mutex<T: Send>` is `Sync`, no unsafe) with the channel session
+//! the relay task owns the bare read half.
-//! owned inside the payloads ([`ForwardedOwner`] — dropping the handle
-//! drops the session AND the halves together).
 
 use std::collections::HashMap;
 use std::net::SocketAddr;
@@ -126,9 +124,11 @@ pub fn key_of(addr: &TargetAddr) -> String {
 }
 
 /// The owner wrapper (the batch-1 review's seam contract vs
-/// alktunnels' surface — the shared design point with the composed
+/// alktunnels' surface — the owner-wrapper posture's generic shape;
-/// accept): an alktunnels half wrapped with its [`TunnelSession`]
+/// the composed accept meets the `+ Sync` bound with its own
-/// inside. The landed handle payloads carry `+ Sync` (the
+/// [`SyncHalves`]/[`OwnedWrite`] adapter pair instead): 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
@@ -327,7 +327,15 @@ impl ComposedEgress {
         }
         self.order.retain(|(k, _)| k != &key);
         self.order.push_back((key.clone(), addr.clone()));
-        Ok(self.entries.get_mut(&key).expect("entry inserted above"))
+        // The entry was inserted above (or the key was already
+        // present): an absent entry here is a structural invariant
+        // break, not a runtime condition — surfacing it as an
+        // egress error instead of a panic (convention 2).
+        self.entries.get_mut(&key).ok_or_else(|| {
+            EgressError::General(format!(
+                "tunnel table invariant broken: `{key}` absent after insert"
+            ))
+        })
     }
 }
 
@@ -641,26 +649,25 @@ impl AsyncWrite for OwnedWrite {
 /// `tunnel:open`). The returned registry handle feeds the factories;
 /// keep it alongside the client they ride.
 ///
-/// # Panics
+/// # Errors
 ///
-/// If the underlying registration fails (a duplicate
+/// A duplicate `channels/tunnel/forwarded` op id on the registry (an
-/// `channels/tunnel/forwarded` op id on the registry — an assembly
+/// assembly bug, not a runtime condition) — the underlying
-/// bug, not a runtime condition). `register_openable`'s failure type
+/// registration's `String` diagnostic, surfaced (convention 2: no
-/// is a `String` diagnostic, not a typed error.
+/// panic path; the caller decides).
 pub fn register_forwarded_serving(
     core: &ChannelCore,
     on_registry: &Arc<alkcall::registry::registration::OperationRegistry>,
     auth: AuthContext,
-) -> CorrelationRegistry {
+) -> Result<CorrelationRegistry, String> {
     let registry: CorrelationRegistry = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
     alktunnels::producer::register_tunnel_forwarded_openable(
         core,
         on_registry,
         auth,
         make_forwarded_handler(Arc::clone(&registry)),
-    )
+    )?;
-    .expect("register forwarded openable (the composed accept's serving half)");
+    Ok(registry)
-    registry
 }
 
 /// The composed accept factory (ADR-007 variant 2, producer.md

src/consumer.rs

@@ -440,10 +440,13 @@ impl AssociateSession {
 
     /// The cred-bearing front door's ASSOCIATE re-origination (the
     /// token-carrying twin of [`Self::associate`]): the open rides
-    /// the payload `auth_token` (CF-005 precedence (1)). The TCP door
+    /// the payload `auth_token` (CF-005 precedence (1)). Not called
-    /// uses this only for its RFC-faithful UDP refusal probe (the
+    /// by the landed TCP door — its ASSOCIATE arm refuses
-    /// door's datagram plane is the UDP door's task); the session is
+    /// RFC-faithfully before any session is opened (the datagram
-    /// otherwise identical.
+    /// plane is the UDP door's task, ADR-003); the constructor is
+    /// the session family's completeness (every command carries the
+    /// token twin) and the wrapper-aware deployment's surface. The
+    /// session is otherwise identical.
     pub async fn associate_with_token(
         client: &TokenClient,
         token: Vec<u8>,
@@ -1081,15 +1084,20 @@ impl TokenClient {
     /// Wrap a `ChannelClient`: take its call connection (the open-op
     /// calling surface) and clone the manager. The door's
     /// token-bearing sessions then ride this client.
-    pub async fn wrap(client: &ChannelClient) -> Self {
+    ///
+    /// [`TokenClientError::CallConnectionTaken`] when the client's
+    /// call connection was already taken (a client is a single
+    /// calling surface — the door-only contract; convention 2: no
+    /// panic path on a double wrap).
+    pub async fn wrap(client: &ChannelClient) -> Result<Self, TokenClientError> {
         let call = client
             .take_call_connection()
             .await
-            .expect("the channel client's call connection exists until taken");
+            .ok_or(TokenClientError::CallConnectionTaken)?;
-        Self {
+        Ok(Self {
             call,
             manager: client.manager().clone(),
-        }
+        })
     }
 
     /// The manager (adopt/teardown — the token sessions' prologue uses
@@ -1124,6 +1132,15 @@ impl TokenClient {
     }
 }
 
+/// Why a `TokenClient` refused to construct (the typed construction
+/// error — the client's calling surface was already taken;
+/// convention 2: no panic path).
+#[derive(Debug, thiserror::Error)]
+pub enum TokenClientError {
+    #[error("the channel client's call connection was already taken (a client is one calling surface — the door-only contract)")]
+    CallConnectionTaken,
+}
+
 impl std::fmt::Debug for TokenClient {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("TokenClient").finish_non_exhaustive()

src/door.rs

@@ -248,6 +248,9 @@ impl FrontDoorBuilder {
     /// The local bind address — the door's explicit bind decision
     /// (the never-binds posture: the door is the assembly layer's
     /// spelled-out choice; loopback is the documented posture).
+    /// Required: `.build()` is a `Result` — omitting it is the
+    /// builder's typed construction error (convention 2: no panic
+    /// path).
     pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
         self.bind_addr = Some(addr.into());
         self
@@ -273,16 +276,26 @@ impl FrontDoorBuilder {
         self
     }
 
-    pub fn build(self) -> FrontDoor {
+    pub fn build(self) -> Result<FrontDoor, DoorBuildError> {
-        FrontDoor {
+        let bind_addr = self.bind_addr.ok_or(DoorBuildError::MissingBindAddr)?;
-            bind_addr: self.bind_addr.expect("the door binds an address"),
+        Ok(FrontDoor {
+            bind_addr,
             credentials: self.credentials,
             idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)),
             max_sessions: self.max_sessions.unwrap_or(256),
-        }
+        })
     }
 }
 
+/// Why a door builder refused to construct (the builder's typed
+/// construction error — a missing required field; convention 2: no
+/// panic path on an omitted `.bind`).
+#[derive(Debug, thiserror::Error)]
+pub enum DoorBuildError {
+    #[error("the door requires a bind address (.bind) — the bind decision is never implied")]
+    MissingBindAddr,
+}
+
 impl FrontDoor {
     /// Bind the configured address and run the accept loop
     /// (consumer.md §Wiring: `door.serve(channel_client)`). Per-
@@ -313,10 +326,13 @@ impl FrontDoor {
         // NoAuth door keeps the client's own calling surface
         // (`open_channel` — the raw-pipe shape never carries a
         // token).
-        let token_client = if credentials.is_some() {
+        let token_client = match credentials {
-            Some(Arc::new(TokenClient::wrap(&client).await))
+            Some(_) => Some(Arc::new(
-        } else {
+                TokenClient::wrap(&client)
-            None
+                    .await
+                    .map_err(|err| std::io::Error::other(err.to_string()))?,
+            )),
+            None => None,
         };
 
         loop {
@@ -750,11 +766,12 @@ impl UdpFrontDoorBuilder {
         self
     }
 
-    pub fn build(self) -> UdpFrontDoor {
+    pub fn build(self) -> Result<UdpFrontDoor, DoorBuildError> {
-        UdpFrontDoor {
+        let bind_addr = self.bind_addr.ok_or(DoorBuildError::MissingBindAddr)?;
-            bind_addr: self.bind_addr.expect("the UDP door binds an address"),
+        Ok(UdpFrontDoor {
+            bind_addr,
             idle: self.idle.unwrap_or(Duration::from_secs(30)),
-        }
+        })
     }
 }
 
@@ -850,13 +867,17 @@ impl UdpFrontDoor {
         // encode the RFC reply header (the codec boundary), send the
         // RFC-header'd datagram back to the client's source address.
         // The association's end (the channel EOF) is this pump's clean
-        // exit.
+        // exit. The decode state (the incremental reader + the
+        // batched-datagram pending queue) persists across pump
+        // iterations — the bridge owns it for the association's
+        // lifetime.
         let c2s = {
             let socket = Arc::clone(&socket);
             let client_addr = Arc::clone(&client_addr);
             async move {
+                let mut bridge = InbandBridge::new();
                 loop {
-                    let (addr, payload) = match recv_inband(&mut channel_read).await {
+                    let (addr, payload) = match recv_inband(&mut bridge, &mut channel_read).await {
                         Ok(Some(pair)) => pair,
                         Ok(None) => return UdpDoorExit::AssociationEnded,
                         Err(err) => return UdpDoorExit::Session(err.to_string()),
@@ -904,24 +925,81 @@ async fn send_inband(
     Ok(())
 }
 
-/// The in-band frame read (the read half's shape — the incremental
+/// The in-band frame read (the read half's shape): a fresh-looking
-/// decoder; its state persists across calls, the same `DatagramReader`
+/// async read whose decoder state **persists across calls** — the
-/// the session runs).
+/// bridge's [`InbandBridge`] owns the incremental `DatagramReader`
-async fn recv_inband(
+/// and the batched-datagram pending queue across calls (a datagram
-    read: &mut tokio::io::ReadHalf<&mut AssociationStream>,
+/// split across channel chunks reassembles across reads; frames
-) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError> {
+/// decoded beyond the one returned queue for later reads — the POC
-    let mut chunk = vec![0u8; 16 * 1024];
+/// #2 reader-state finding, the same shape
-    let mut reader = DatagramReader::new();
+/// `AssociateSession::recv_datagram` owns).
-    loop {
+struct InbandBridge {
-        let n = tokio::io::AsyncReadExt::read(read, &mut chunk).await?;
+    reader: DatagramReader,
-        if n == 0 {
+    pending: std::collections::VecDeque<(TargetAddr, Bytes)>,
-            return Ok(None);
+}
-        }
+
-        let dgs = reader.feed(&chunk[..n]);
+impl InbandBridge {
-        if let Some(res) = dgs.into_iter().next() {
+    fn new() -> Self {
-            return res.map(Some).map_err(wire_error);
+        Self {
+            reader: DatagramReader::new(),
+            pending: std::collections::VecDeque::new(),
         }
     }
+
+    /// The next in-band datagram: the pending queue first (a chunk
+    /// may have batched several frames — drain in order), then an
+    /// incremental decode off the read half. `Ok(None)` on channel
+    /// EOF (the association's end); fail-loud mid-frame (never a
+    /// silent partial datagram).
+    async fn recv<R>(
+        &mut self,
+        read: &mut R,
+    ) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError>
+    where
+        R: tokio::io::AsyncRead + Unpin,
+    {
+        if let Some(dg) = self.pending.pop_front() {
+            return Ok(Some(dg));
+        }
+        let mut chunk = vec![0u8; 16 * 1024];
+        loop {
+            let n = tokio::io::AsyncReadExt::read(read, &mut chunk).await?;
+            if n == 0 {
+                if self.reader.is_mid_frame() {
+                    return Err(SocksSessionError::Rfc(SocksError::Io(std::io::Error::new(
+                        std::io::ErrorKind::UnexpectedEof,
+                        "association ended mid-datagram (fail-loud)",
+                    ))));
+                }
+                return Ok(None);
+            }
+            let mut dgs = self.reader.feed(&chunk[..n]).into_iter();
+            if let Some(first) = dgs.next() {
+                for res in dgs {
+                    match res {
+                        Ok(pair) => self.pending.push_back(pair),
+                        Err(err) => return Err(wire_error(err)),
+                    }
+                }
+                match first {
+                    Ok(pair) => return Ok(Some(pair)),
+                    Err(err) => return Err(wire_error(err)),
+                }
+            }
+        }
+    }
+}
+
+/// The channel→socket pump's decode call (the bridge the pump loop
+/// drives — the decode state lives in the bridge, not per call).
+async fn recv_inband<R>(
+    bridge: &mut InbandBridge,
+    read: &mut R,
+) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError>
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    bridge.recv(read).await
 }
 
 /// Map a wire-decode error into the RFC vocabulary (the consumer
@@ -940,3 +1018,118 @@ fn wire_error(err: WireError) -> SocksSessionError {
 /// The association's framed stream type (the split halves' target —
 /// the forked client type over the channel's `BiStream`).
 type AssociationStream = crate::rfc1928::client::Socks5Stream<alkcall::core::types::BiStream>;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::wire::frame as wire_frame;
+    use tokio::io::AsyncWriteExt;
+    use tokio::io::{AsyncRead, DuplexStream};
+
+    /// Drive one pump-shaped decode call: feed `bytes` from the peer
+    /// side, then decode (the recv parks until a frame completes or
+    /// the stream ends — the pump loop's shape).
+    async fn feed_then_recv<R>(
+        bridge: &mut InbandBridge,
+        read: &mut R,
+        writer: &mut DuplexStream,
+        bytes: &[u8],
+    ) -> Result<Option<(TargetAddr, Bytes)>, SocksSessionError>
+    where
+        R: AsyncRead + Unpin,
+    {
+        if !bytes.is_empty() {
+            writer.write_all(bytes).await.expect("feed write");
+            writer.flush().await.expect("feed flush");
+        }
+        bridge.recv(read).await
+    }
+
+    #[tokio::test]
+    async fn batched_frames_queue_not_drop() {
+        // Two frames batched into one feed: the first returns, the
+        // second queues and drains on the next call (the per-call
+        // reader dropped it — review 003 R-1; the pending queue is
+        // the state that persists across the call boundary).
+        let (mut writer, reader) = tokio::io::duplex(1024);
+        let (mut channel_read, _channel_write) = tokio::io::split(reader);
+        let addr = TargetAddr::Ip("10.0.0.2:53".parse().expect("addr"));
+        let mut wire = wire_frame(&addr, b"first").unwrap().to_vec();
+        wire.extend_from_slice(&wire_frame(&addr, b"second").unwrap());
+        let mut bridge = InbandBridge::new();
+
+        let first = feed_then_recv(&mut bridge, &mut channel_read, &mut writer, &wire).await;
+        match first {
+            Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"first"),
+            other => panic!("the first datagram returned: {other:?}"),
+        }
+        let second = bridge.recv(&mut channel_read).await;
+        match second {
+            Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"second"),
+            other => panic!("the batched datagram queued, not lost: {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn eof_mid_frame_fails_loud_not_silent() {
+        // A frame in flight when the stream ends: the decode errors
+        // loud (never a silent partial datagram) — the same shape
+        // AssociateSession::recv_datagram pins.
+        let (mut writer, reader) = tokio::io::duplex(1024);
+        let (mut channel_read, _channel_write) = tokio::io::split(reader);
+        let addr = TargetAddr::Ip("10.0.0.3:53".parse().expect("addr"));
+        let framed = wire_frame(&addr, b"truncated").expect("frame");
+        let mut bridge = InbandBridge::new();
+
+        let feeder = tokio::spawn(async move {
+            writer
+                .write_all(&framed[..3])
+                .await
+                .expect("the partial frame");
+            writer.flush().await.expect("flush");
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            let _ = writer.shutdown().await;
+        });
+
+        let ended = bridge.recv(&mut channel_read).await;
+        assert!(
+            ended.is_err(),
+            "EOF mid-frame is the fail-loud truncated-datagram condition"
+        );
+        feeder.await.expect("feeder");
+    }
+
+    #[tokio::test]
+    async fn partial_frame_state_persists_across_calls() {
+        // One chunk carries frame 1 complete + the first bytes of
+        // frame 2: call 1 returns frame 1 (frame 2's bytes stay in
+        // the bridge's reader), call 2 — after the stream ends —
+        // fails loud on the surviving mid-frame state (a per-call
+        // reader would report a clean Ok(None) EOF here).
+        let (mut writer, reader) = tokio::io::duplex(1024);
+        let (mut channel_read, _channel_write) = tokio::io::split(reader);
+        let addr = TargetAddr::Ip("10.0.0.4:53".parse().expect("addr"));
+        let mut wire = wire_frame(&addr, b"whole").unwrap().to_vec();
+        wire.extend_from_slice(&wire_frame(&addr, b"partial").unwrap()[..3]);
+        let mut bridge = InbandBridge::new();
+
+        let feeder = tokio::spawn(async move {
+            writer.write_all(&wire).await.expect("the mixed chunk");
+            writer.flush().await.expect("flush");
+            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+            let _ = writer.shutdown().await;
+        });
+
+        let first = bridge.recv(&mut channel_read).await;
+        match first {
+            Ok(Some((_src, payload))) => assert_eq!(&payload[..], b"whole"),
+            other => panic!("frame 1 returned: {other:?}"),
+        }
+        let second = bridge.recv(&mut channel_read).await;
+        assert!(
+            second.is_err(),
+            "frame 2's partial bytes survived the call boundary (fail-loud at EOF)"
+        );
+        feeder.await.expect("feeder");
+    }
+}

src/lib.rs

@@ -137,6 +137,8 @@ pub use composed::{
 };
 
 #[cfg(feature = "local")]
-pub use consumer::TokenClient;
+pub use consumer::{TokenClient, TokenClientError};
 #[cfg(feature = "local")]
-pub use door::{CredentialTable, DoorCredential, FrontDoor, UdpFrontDoor};
+pub use door::{CredentialTable, DoorBuildError, DoorCredential, FrontDoor, UdpFrontDoor};
+#[cfg(feature = "local")]
+pub use local::{ProducerDoorBuildError, ProducerDoorError};

src/local/mod.rs

@@ -283,4 +283,6 @@ mod tests;
 
 mod producer_door;
 
-pub use producer_door::{ProducerDoor, ProducerDoorBuilder, ProducerDoorGuard};
+pub use producer_door::{
+    ProducerDoor, ProducerDoorBuildError, ProducerDoorBuilder, ProducerDoorError, ProducerDoorGuard,
+};

src/local/producer_door.rs

@@ -141,16 +141,39 @@ impl ProducerDoorBuilder {
         self
     }
 
-    pub fn build(self) -> ProducerDoor {
+    pub fn build(self) -> Result<ProducerDoor, ProducerDoorBuildError> {
-        ProducerDoor {
+        let producer = self
+            .producer
+            .ok_or(ProducerDoorBuildError::MissingProducer)?;
+        Ok(ProducerDoor {
             bind_addr: self.bind_addr,
-            producer: self.producer.expect("the backend serves a producer"),
+            producer,
             idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)),
             max_sessions: self.max_sessions.unwrap_or(256),
-        }
+        })
     }
 }
 
+/// Why the producer-door builder refused to construct (the builder's
+/// typed construction error — a missing required field; convention 2:
+/// no panic path).
+#[derive(Debug, thiserror::Error)]
+pub enum ProducerDoorBuildError {
+    #[error("the backend requires a producer (.producer) — there is nothing to serve")]
+    MissingProducer,
+}
+
+/// Why the producer door's `serve` refused to start (the typed
+/// construction error — `serve` carries the bind decision, so an
+/// omitted `.bind` is its error, not a panic; convention 2).
+#[derive(Debug, thiserror::Error)]
+pub enum ProducerDoorError {
+    #[error("the backend requires a bind address (.bind) before serve — the bind decision is never implied")]
+    MissingBindAddr,
+    #[error("tcp bind: {0}")]
+    Bind(#[source] std::io::Error),
+}
+
 /// The serving guard: [`ProducerDoor::serve`] returns one; `close()`
 /// ends the accept loop gracefully (the listener is dropped, the
 /// loop's end is awaited — the spawned sessions already running own
@@ -196,15 +219,17 @@ impl ProducerDoor {
     /// (the alktunnels per-accept isolation posture). The listener's
     /// lifetime is the caller's: the returned [`ProducerDoorGuard`]
     /// ends it (`close()` graceful; drop best-effort abort).
-    pub async fn serve(self) -> Result<ProducerDoorGuard, std::io::Error> {
+    pub async fn serve(self) -> Result<ProducerDoorGuard, ProducerDoorError> {
         let ProducerDoor {
             bind_addr,
             producer,
             idle_first_byte,
             max_sessions,
         } = self;
-        let bind_addr = bind_addr.expect("the backend binds an address (builder .bind)");
+        let bind_addr = bind_addr.ok_or(ProducerDoorError::MissingBindAddr)?;
-        let listener = TcpListener::bind(bind_addr).await?;
+        let listener = TcpListener::bind(bind_addr)
+            .await
+            .map_err(ProducerDoorError::Bind)?;
         Ok(serve_on_parts(
             listener,
             idle_first_byte,

tasks/implementation/review-assembly-layer.md

@@ -1,7 +1,7 @@
 ---
 id: review-assembly-layer
 name: Review — the assembly layer (the batch-2 gate)
-status: pending
+status: completed
 depends_on: [implementation/assembly-integration-suite]
 scope: narrow
 risk: low
@@ -62,23 +62,233 @@ Review against the architecture, in this order:
 8. **Docs sync**: the specs match what landed (the integration
    task's reconciliation verified; any residue is a finding).
 
-## Notes
+## Summary
 
-> To be filled by the review agent — see the Summary and the Findings.
+Review performed 2026-09-26 against the eight-criteria order; the gate
+opens (the crate is feature-complete against the Phase 1 spec set —
+release polish remains). One major correctness finding was corrected
+in the review change itself (R-1: the UDP door's in-band bridge lost
+the reader-state contract), plus a convention sweep residue (R-2:
+seven non-test `expect`s → typed errors, the batch-1 R-2 class) and
+two doc residues (R-3, R-4). No seam-contract, never-binds,
+graduation-shape, feature-gate, or POC-fidelity drift was found
+beyond these.
+
+**1. The never-binds posture — clean.** The protocol layer
+(`producer.rs`/`consumer.rs`/`wire.rs`/`params.rs`/`egress.rs`/
+`error.rs`/`rfc1928/`) names no kernel-socket type (`tokio::net`
+appears only in doc prose; zero code references — swept). The default
+dependency graph carries no `tokio/net`/`mio`/`socket2`
+(`cargo tree -e normal -f "{p} [{f}]"`: tokio rides the wasm-clean
+subset `bytes,io-util,macros,rt,sync,time`; `net` appears only under
+`--features local`, which also brings `mio`+`socket2`; `composed`
+adds alktunnels 0.2.0 only and its tokio stays on the clean subset).
+Every kernel bind is an explicit constructor decision in `local`-gated
+code (`local/mod.rs` — the dial's connect, `LocalEgress::bind`'s
+loopback `127.0.0.1:0` (a wildcard bind is the caller's variant, doc
+pinned), `local_bind_accept`'s explicit `bind_host` param, the doors'
+`TcpListener::bind`/`UdpSocket::bind` behind builder `.bind()`) —
+the doors' binds are the doors' own (assembly-layer, ADR-009).
+
+**2. Seam-contract compliance (ADR-004) — clean.** *Readiness*: the
+local egress parks on `sock.readable().await` then `try_recv_from`
+(`WouldBlock` → the spurious `Ok(None)`, never hot); the composed
+egress parks on the shared queue's `recv().await`, and its
+closed-queue condition is a **count-based design** (`live_relays` +
+`ever_opened`) — a real improvement over the POC (the POC held its
+own `inbound_tx` clone, so its queue-`None` arm was unreachable; the
+landed egress actually detects every-relay-ended and returns `Err`,
+pinned by `closed_queue_yields_err_not_a_hot_none`). *Ownership*:
+rides the handle structurally everywhere — the local providers box
+their socket halves into the payloads; the composed accept wraps the
+`Send + Unpin`-only alktunnels halves in `SyncHalves`/`OwnedWrite`
+(`Mutex<T: Send>` is `Sync`, no unsafe) with the `TunnelListener`
+riding INSIDE the write payload (`ForwardedOwner` — the
+disjoint-capture trap is doc-pinned and the whole-struct rebinding is
+in place); the egress keeps the session in the table entry (the POC's
+insert-time-bug lesson, doc-pinned). No `_` bindings of owning values
+(swept). *The accept future owns its resources end-to-end* in both
+variants (the listener lives inside the future; the composed future
+owns the listener + the oneshot receiver — the client-gone select!
+drop tears both down). *Framing-once*: the local egress returns
+`addr + data` inner bytes; the composed egress returns inner bytes
+with its own oversize guard mirroring the wire bound; the tunnel-side
+framing is the tunnel codec's (`frame_datagram`), never the RFC
+frame's.
+
+**3. Door fidelity (ADR-005/ADR-003) — clean, one finding corrected
+(R-1).** The NoAuth door is a literal raw pipe
+(`copy_bidirectional`, zero translation — pinned by
+`door-tcp.rs`'s raw-byte probes). The cred-bearing door terminates
+RFC 1929 locally through the forked `accept_password_auth` surface,
+maps through the static table (the check IS the lookup — one table,
+one truth), and re-originates in-band via the token sessions; the
+never-pipes-raw-post-auth property is pinned by the re-origination
+probe (the vanilla client's command bytes are consumed door-side;
+the in-band conversation is fresh). The token is the mapped
+identity's id riding the open-op `auth_token` payload (CF-005
+precedence (1)) — the password never crosses the channel (pinned);
+the identity never reaches the vanilla client. The RFC UDP-header
+codec (`parse_udp_request`/`new_udp_header`) appears only at the
+door's boundary (the in-band wire stays `{addr, data}` frames —
+pinned end to end through a real echo target); FRAG ≠ 0 is refused by
+dropping the datagram, the association stays healthy (pinned). The
+TCP door's ASSOCIATE arm refuses RFC-faithfully before any session
+opens (no phantom data plane). R-1 is the UDP door's in-band decode
+state — see the Findings.
+
+**4. The graduation's shape (ADR-007 variant 2) — clean.** The
+composed accept advertises the real OS-chosen `bound` (the open-op
+reply's `bound` field via the reply projection; a missing
+`bound`/`listen` FAILS the factory — `Shortage` with an explicit
+no-stand-in message, never an expected-addr stand-in) and the real
+forwarded-op `peer` (a `Path` peer maps to the RFC sentinel — the
+documented addressless case, not a stand-in). The POC's superseded
+workaround (reply#1-first + carried `bind_addr` + sentinel peer) is
+not resurrected — `tests/composed-accept.rs` pins the real addresses
+end to end (`reply#1 advertises the real OS-chosen bound`, `reply#2
+carries the forwarded op's real peer`), and the establishment bound
+covers only the open-op round trip (the listener open is the
+factory's; the accept wait is the teardown-governed quiet-wait phase).
+
+**5. Convention compliance — one residue corrected (R-2).** After
+R-2, no `unwrap`/`expect`/`panic!`/`unreachable!` in non-test library
+code (the seven batch-2 additions were the builder-required-field
+`expect`s and two internal-invariant `expect`s — same class as
+batch-1's R-2; all now typed). `thiserror` surfaces throughout (the
+new builder errors included); naming/module structure hold — the
+feature-gated backend modules (`local/`, `composed/`, `door`) are
+never imported from the protocol modules (swept: zero
+`crate::local`/`crate::door`/`crate::composed` references in the
+protocol layer); producer/consumer vocabulary holds (RFC-facing
+"server"/"client" only inside `src/rfc1928/`).
+
+**6. Feature-gate hygiene (ADR-009) — green, all four targets.**
+`cargo test` (default) green; `cargo test --all-features` green; both
+wasm checks clean for default AND `--features composed` (the
+composed wasm gate verified live); `local` on wasm fails to build
+with exactly the expected shape — `mio` cannot compile on
+`wasm32-unknown-unknown` (the alktty posture's build-error
+guarantee, verified). `local` brings only `tokio/net` (+ `mio`/
+`socket2` transitively); `composed` brings only alktunnels; the
+default graph is unchanged from batch 1 (spot-checked per feature).
+
+**7. POC-fidelity spot-check — clean, with the R-1 exception and one
+documented improvement.**
+- *The composed egress's LRU + closed-queue posture*: the lazy
+  destination→tunnel table, the LRU eviction-before-insert loop, the
+  touch-on-access ordering, the session-inside-the-entry placement,
+  and the relay task's shape are all line-faithful to POC #2's
+  `ComposedEgress` (`/workspace/alksocks-udp-poc/src/egress.rs`).
+  The one deliberate divergence is the closed-queue detection — the
+  POC's `recv().await`-`None` arm was structurally unreachable (it
+  held its own `Sender`); the crate replaces it with the
+  `live_relays`/`ever_opened` count so the association actually ends
+  when every tunnel ends (the POC's stated *intent*, realized;
+  drift-in-the-right-direction, doc-pinned in the struct docs).
+- *The cred-bearing door's re-origination*: matches POC #1's
+  terminate-locally + re-speak shape (the POC's raw-pipe door is the
+  NoAuth variant; the cred-bearing extension is the ADR-005 design,
+  validated by the integration suite's vanilla-client round trips).
+- *The bind-first correlation*: the token→oneshot registry shape
+  with the forwarded handler owning the data-plane pump (R-02 held —
+  the handler's inline `pump_bidi` await tracks the plane) matches
+  the alktunnels 0.2.0 forwarded-op surface; the three-party
+  half-close deadlock note (the bridge completes on EITHER
+  direction's EOF) is the pinned non-obvious-correctness comment
+  done right.
+
+**8. Docs sync — two residues fixed (R-3, R-4).** The integration
+task's reconciliation was verified: overview.md (module map, feature
+gates, dependencies), consumer.md (the front doors, the TokenClient
+door-only contract, the BIND take-guard), producer.md §Wiring, and
+README.md Current State all match the tree. Residues fixed:
+producer.md's local-backend section described the landed shape as a
+channel pipe (`copy_bidirectional` onto a `BiStream`) — the landed
+`ProducerDoor` serves the RFC conversation on the accepted socket
+directly via `serve_socks5_over` (no channel in the path; rewritten,
+including the `bind_policy` gate note the docs omitted);
+`associate_with_token`'s doc claimed "the TCP door uses this" — the
+landed TCP door refuses ASSOCIATE before any session (reworded);
+`SessionOwner`'s docs claimed it is "shared with the composed
+accept" — the accept uses its own `SyncHalves`/`OwnedWrite` pair
+(reworded). No other spec drift.
 
 ## Acceptance Criteria
 
-- [ ] The never-binds posture verified (protocol layer + default
+- [x] The never-binds posture verified (protocol layer + default
       graph)
-- [ ] Seam-contract compliance verified for every provider
+- [x] Seam-contract compliance verified for every provider
-- [ ] Door fidelity verified (both variants + the UDP boundary)
+- [x] Door fidelity verified (both variants + the UDP boundary)
-- [ ] The graduation's shape verified (no stand-ins)
+- [x] The graduation's shape verified (no stand-ins)
-- [ ] Convention compliance swept
+- [x] Convention compliance swept
-- [ ] Feature-gate hygiene verified (all four verification targets)
+- [x] Feature-gate hygiene verified (all four verification targets)
-- [ ] POC-fidelity spot-check of the trickiest assembly behaviors
+- [x] POC-fidelity spot-check of the trickiest assembly behaviors
-- [ ] Docs sync verified; findings recorded; the gate to release
+- [x] Docs sync verified; findings recorded; the gate to release
       polish opens
 
+### Findings
+
+- **R-1 (major → fixed in the review change)**: the UDP front door's
+  in-band decode (`recv_inband`, `src/door.rs`) created a fresh
+  `DatagramReader` per call and returned only the first frame of
+  each feed — a datagram split across channel chunks would have
+  restarted the reassembly from scratch per read, and datagrams
+  batched beyond the first in one chunk were **silently dropped**.
+  This is exactly POC #2's first-cut bug (ADR-003: "reader state
+  persists across calls"; `AssociateSession::recv_datagram` owns the
+  pinned shape) — the door bridge did not carry it, and its own doc
+  comment claimed persistence. Unexercised by the suite (every
+  landed test drives one frame per chunk; the failure needs two
+  frames coalescing into one read — a burst shape real echo targets
+  produce under load). Fix: an owning `InbandBridge` (persistent
+  `DatagramReader` + batched-datagram pending queue, the
+  `AssociateSession` shape) threaded through the channel→socket
+  pump, plus the fail-loud mid-frame EOF (never a silent partial
+  datagram). Pinned by three unit tests on the bridge (batched
+  frames queue-not-drop; partial-frame state persists across calls
+  — the test fails against a per-call reader; EOF mid-frame errors
+  loud) and an end-to-end door-udp test with a doubled-echo egress
+  (`batched_in_band_replies_both_reach_the_vanilla_client`).
+- **R-2 (convention, fixed in the review change)**: seven non-test
+  `expect`s landed with batch 2 (convention 2's rule is the rule;
+  batch-1's R-2 precedent applies): the `FrontDoor`/`UdpFrontDoor`
+  builders' missing-`.bind` panics → `DoorBuildError`
+  (`MissingBindAddr`, `.build()` is now `Result`); the
+  `ProducerDoor` builder's missing-`.producer` panic and `serve`'s
+  missing-`.bind` panic → `ProducerDoorBuildError`/`ProducerDoorError`
+  (the bind decision is `serve`'s typed error, the io error is
+  `#[source]`-wrapped); `ComposedEgress::entry_for`'s
+  inserted-above invariant → an `EgressError::General` structural
+  diagnostic; `register_forwarded_serving`'s registration panic →
+  `Result<CorrelationRegistry, String>` (the underlying alkcall
+  diagnostic surfaced, `# Panics` → `# Errors`);
+  `TokenClient::wrap`'s call-connection-taken panic →
+  `TokenClientError::CallConnectionTaken` (a client is one calling
+  surface — the door-only contract). Test call sites updated
+  (`.expect("the door builds")` etc. — test-side, allowed); the new
+  error types are re-exported (`lib.rs`).
+- **R-3 (doc residue, fixed)**: producer.md's producer-side
+  local-backend section described the landed shape as a raw channel
+  pipe (`copy_bidirectional` onto a `BiStream`, "the consumer door's
+  code shape") — the landed `ProducerDoor` hands each accepted
+  socket to `serve_socks5_over` (the RFC conversation runs on the
+  socket; no channel pipe). Reconciled, including the
+  `bind_policy`-is-this-path's-BIND-gate note and the cred-bearing
+  follow-up posture (not landed).
+- **R-4 (doc residue, fixed)**: `associate_with_token`'s doc claimed
+  the TCP door calls it (it refuses ASSOCIATE before any session);
+  `SessionOwner`'s module + type docs claimed the composed accept
+  shares it (the accept uses `SyncHalves`/`OwnedWrite`). Both
+  reworded to the landed reality.
+- **R-5 (accepted, no fix demanded)**: the end-to-end
+  `batched_in_band_replies...` test does not deterministically
+  exercise the batching (the two echo frames may ride separate
+  channel chunks — timing-dependent); the deterministic pins are the
+  bridge unit tests (which demonstrably fail against the per-call
+  reader). The end-to-end test still guards the regression class
+  through the real stack; the unit tests carry the determinism.
+
 ## References
 
 - docs/architecture/overview.md, wire.md, producer.md, consumer.md
@@ -88,8 +298,4 @@ Review against the architecture, in this order:
   fidelity sources)
 - AGENTS.md (conventions 1–17; the verification commands)
 - The batch-1 review precedent:
-  `tasks/implementation/review-protocol-core.md`
+  `tasks/implementation/review-protocol-core.md`
-
-## Summary
-
-> To be filled on completion

tests/assembly-integration.rs

@@ -389,7 +389,8 @@ async fn wire_composed_suite() -> ComposedSuite {
         ),
         &topo.opener_registry,
         alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
-    );
+    )
+    .expect("register forwarded openable");
 
     // The SOCKS5 producer: the composed accept seam (BIND) + the
     // composed direct-op egress (ASSOCIATE) + a real TCP dial
@@ -504,7 +505,8 @@ async fn full_dash_d_story_one_registration_three_commands_through_the_doors() {
     let door = FrontDoor::builder()
         .credentials(cred_table())
         .bind(door_addr)
-        .build();
+        .build()
+        .expect("the door builds");
     let _tcp_serve = tokio::spawn({
         let client = Arc::clone(&tcp_topo.consumer);
         async move {
@@ -515,7 +517,10 @@ async fn full_dash_d_story_one_registration_three_commands_through_the_doors() {
     });
     let udp_relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
     let udp_relay_addr = udp_relay.local_addr().expect("relay addr");
-    let udp_door = UdpFrontDoor::builder().bind(udp_relay_addr).build();
+    let udp_door = UdpFrontDoor::builder()
+        .bind(udp_relay_addr)
+        .build()
+        .expect("the door builds");
     let _udp_serve = tokio::spawn({
         let client = Arc::clone(&udp_topo.consumer);
         async move {
@@ -725,7 +730,10 @@ async fn door_client_disconnect_tears_the_association_down_no_leaks() {
 
     let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
     let relay_addr = relay.local_addr().expect("relay addr");
-    let door = UdpFrontDoor::builder().bind(relay_addr).build();
+    let door = UdpFrontDoor::builder()
+        .bind(relay_addr)
+        .build()
+        .expect("the door builds");
     let serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move { door.serve_on(relay, client).await.expect("the door serves") }
@@ -780,7 +788,10 @@ async fn door_client_disconnect_tears_the_association_down_no_leaks() {
     // poisoning).
     let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2");
     let relay2_addr = relay2.local_addr().expect("relay addr 2");
-    let door2 = UdpFrontDoor::builder().bind(relay2_addr).build();
+    let door2 = UdpFrontDoor::builder()
+        .bind(relay2_addr)
+        .build()
+        .expect("the door builds");
     let _serve2 = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -963,7 +974,8 @@ async fn mixed_assembly_channels_native_and_door_clients_concurrent() {
     let door = FrontDoor::builder()
         .credentials(cred_table())
         .bind(door_addr)
-        .build();
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&door_topo.consumer);
         async move {
@@ -974,7 +986,10 @@ async fn mixed_assembly_channels_native_and_door_clients_concurrent() {
     });
     let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
     let relay_addr = relay.local_addr().expect("relay addr");
-    let udp_door = UdpFrontDoor::builder().bind(relay_addr).build();
+    let udp_door = UdpFrontDoor::builder()
+        .bind(relay_addr)
+        .build()
+        .expect("the door builds");
     let _udp_serve = tokio::spawn({
         let client = Arc::clone(&udp_door_topo.consumer);
         async move {
@@ -1277,7 +1292,8 @@ async fn producer_door_one_registration_three_commands() {
     let door_addr = listener.local_addr().expect("door addr");
     let door = alksocks::local::ProducerDoor::builder()
         .producer(producer)
-        .build();
+        .build()
+        .expect("the door builds");
     let _guard = door.serve_on(listener).await;
 
     // CONNECT (live).
@@ -1446,7 +1462,10 @@ async fn dash_d_domain_targets_resolve_at_the_producer_seam() {
 
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let door_addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().bind(door_addr).build();
+    let door = FrontDoor::builder()
+        .bind(door_addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {

tests/common/mod.rs

@@ -148,6 +148,57 @@ pub fn echo_egress_factory() -> EgressFactory {
     })
 }
 
+/// A doubled-echo egress factory: every datagram echoes back **twice
+/// back-to-back** (two frames queued as one reply burst — the
+/// producer's frame writer puts both length-prefixed frames on the
+/// channel; the consumer's read may batch them into one chunk). The
+/// reader-state probe for the front doors' bridge (review 003 R-1):
+/// a bridge that re-creates its decoder per read loses the second
+/// datagram; the landed bridge queues it.
+pub fn doubled_echo_egress_factory() -> EgressFactory {
+    alksocks::egress::egress_factory(|| {
+        let (inbound, outbound) = tokio::sync::mpsc::unbounded_channel::<Bytes>();
+        Ok(Box::new(DoubledEchoEgress {
+            inbound,
+            outbound: Mutex::new(outbound),
+        }) as Box<dyn Egress + Send>)
+    })
+}
+
+/// The doubled echo behind [`doubled_echo_egress_factory`] (the
+/// [`QueueEgress`] shape with the echo count two).
+pub struct DoubledEchoEgress {
+    pub inbound: tokio::sync::mpsc::UnboundedSender<Bytes>,
+    pub outbound: Mutex<tokio::sync::mpsc::UnboundedReceiver<Bytes>>,
+}
+
+impl Egress for DoubledEchoEgress {
+    fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> {
+        Box::pin(async move {
+            let mut inner = addr
+                .to_be_bytes()
+                .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
+            inner.extend_from_slice(&payload);
+            let inner = Bytes::from(inner);
+            self.inbound
+                .send(inner.clone())
+                .map_err(|_| EgressError::General("egress closed".into()))?;
+            self.inbound
+                .send(inner)
+                .map_err(|_| EgressError::General("egress closed".into()))
+        })
+    }
+
+    fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
+        Box::pin(async move {
+            match self.outbound.lock().await.recv().await {
+                Some(inner) => Ok(Some(inner)),
+                None => Err(EgressError::General("egress closed".into())),
+            }
+        })
+    }
+}
+
 /// A failing egress factory (the ASSOCIATE-refusal probe): refuses
 /// with the given error, which the producer maps to reply#1's code.
 /// The variant is reconstructed per call (the seam errors are not

tests/composed-accept.rs

@@ -67,7 +67,8 @@ async fn wire_suite_with_identity(identity: alkcall::core::auth::Identity) -> Su
         ),
         &topo.opener_registry,
         alkcall::core::auth::AuthContext::anonymous(b"alk/tunnel"),
-    );
+    )
+    .expect("register forwarded openable");
     // The SOCKS5 openable registers post-hoc (W2's shape) on the
     // producer's registry — with the REAL composed factory, which
     // closes over the opener's client + the correlation registry (both

tests/door-tcp.rs

@@ -222,7 +222,10 @@ async fn noauth_door_raw_pipe_vanilla_connect_round_trip() {
     // the door's own bind).
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().bind(addr).build();
+    let door = FrontDoor::builder()
+        .bind(addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -288,7 +291,11 @@ async fn cred_door_good_creds_connect_round_trip_with_mapped_identity() {
 
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().credentials(table).bind(addr).build();
+    let door = FrontDoor::builder()
+        .credentials(table)
+        .bind(addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -368,7 +375,11 @@ async fn cred_door_bad_creds_rejected_no_channel_opened() {
 
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().credentials(table).bind(addr).build();
+    let door = FrontDoor::builder()
+        .credentials(table)
+        .bind(addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -445,7 +456,11 @@ async fn cred_door_never_pipes_raw_post_auth() {
 
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().credentials(table).bind(addr).build();
+    let door = FrontDoor::builder()
+        .credentials(table)
+        .bind(addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -518,7 +533,11 @@ async fn bind_door_vanilla_conversation_reply_fidelity() {
 
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().credentials(table).bind(addr).build();
+    let door = FrontDoor::builder()
+        .credentials(table)
+        .bind(addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -601,7 +620,8 @@ async fn idle_first_byte_timeout_closes_a_silent_connection() {
     let door = FrontDoor::builder()
         .bind(addr)
         .idle_first_byte(Duration::from_millis(300))
-        .build();
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -658,7 +678,8 @@ async fn session_cap_refuses_beyond_the_bound() {
         .bind(addr)
         .idle_first_byte(Duration::from_secs(30))
         .max_sessions(1)
-        .build();
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -723,7 +744,10 @@ async fn per_connection_failure_isolation() {
 
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().bind(addr).build();
+    let door = FrontDoor::builder()
+        .bind(addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -776,7 +800,10 @@ async fn native_sessions_work_beside_a_noauth_door() {
 
     let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
     let addr = listener.local_addr().expect("door addr");
-    let door = FrontDoor::builder().bind(addr).build();
+    let door = FrontDoor::builder()
+        .bind(addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -822,8 +849,11 @@ async fn bind_session_with_token_round_trip() {
     )
     .await;
 
-    let token_client =
+    let token_client = Arc::new(
-        Arc::new(alksocks::consumer::TokenClient::wrap(topo.consumer.as_ref()).await);
+        alksocks::consumer::TokenClient::wrap(topo.consumer.as_ref())
+            .await
+            .expect("the door's token surface wraps"),
+    );
     let mut session = BindSession::bind_with_token(
         token_client.as_ref(),
         ("bind.token", 0u16).to_target_addr().unwrap(),

tests/door-udp.rs

@@ -100,7 +100,10 @@ async fn udp_door_vanilla_client_round_trip_and_empty_datagram() {
     let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
     let relay_addr = relay.local_addr().expect("relay addr");
     let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
-    let door = UdpFrontDoor::builder().bind(relay_addr).build();
+    let door = UdpFrontDoor::builder()
+        .bind(relay_addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move { door.serve_on(relay, client).await.expect("the door serves") }
@@ -165,6 +168,65 @@ async fn udp_door_vanilla_client_round_trip_and_empty_datagram() {
     );
 }
 
+/// The reader-state pin through the door's bridge (review 003 R-1,
+/// the POC #2 finding carried to the doors): two replies arriving
+/// back-to-back (the doubled echo — both frames ride the producer's
+/// frame writer; the door's channel read may batch them into one
+/// chunk) both reach the vanilla client. A bridge that re-created
+/// its decoder per read lost the second datagram.
+#[tokio::test]
+async fn batched_in_band_replies_both_reach_the_vanilla_client() {
+    let echo_addr = udp_echo_server().await;
+    let topo = harness::wire_consumer_side_with(
+        echo_dial(),
+        Some(harness::doubled_echo_egress_factory()),
+        None,
+        Some(harness::consumer_identity()),
+    )
+    .await;
+
+    let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
+    let relay_addr = relay.local_addr().expect("relay addr");
+    let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
+    let door = UdpFrontDoor::builder()
+        .bind(relay_addr)
+        .build()
+        .expect("the door builds");
+    let _serve = tokio::spawn({
+        let client = Arc::clone(&topo.consumer);
+        async move { door.serve_on(relay, client).await.expect("the door serves") }
+    });
+
+    // One datagram in → two echo datagrams out (both from the relay
+    // socket, both carrying the target as their source address — the
+    // bridge's pending queue drained in order, none lost).
+    client
+        .send_to(&rfc_datagram(echo_addr, b"once"), relay_addr)
+        .await
+        .expect("send");
+    let mut rx = vec![0u8; 65535];
+    for _ in 0..2 {
+        let (n, from) = timeout(BOUND, client.recv_from(&mut rx))
+            .await
+            .expect("reply within bound")
+            .expect("recv");
+        assert_eq!(from, relay_addr, "the reply comes from the relay socket");
+        let (addr, payload) = parse_rfc_datagram(&rx[..n]);
+        assert_eq!(addr, echo_addr);
+        assert_eq!(payload, b"once");
+    }
+
+    // No third reply: exactly the two echoes arrived (the burst was
+    // queued, not duplicated).
+    let mut probe = vec![0u8; 65535];
+    assert!(
+        timeout(Duration::from_millis(300), client.recv_from(&mut probe))
+            .await
+            .is_err(),
+        "exactly two replies arrived"
+    );
+}
+
 /// FRAG ≠ 0 is refused at the door (the v1 posture): the fragment is
 /// dropped — nothing is egressed — and the association stays healthy
 /// (a following normal datagram round-trips).
@@ -182,7 +244,10 @@ async fn frag_nonzero_refused_and_association_stays_healthy() {
     let relay = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay");
     let relay_addr = relay.local_addr().expect("relay addr");
     let client = UdpSocket::bind("127.0.0.1:0").await.expect("client socket");
-    let door = UdpFrontDoor::builder().bind(relay_addr).build();
+    let door = UdpFrontDoor::builder()
+        .bind(relay_addr)
+        .build()
+        .expect("the door builds");
     let _serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move { door.serve_on(relay, client).await.expect("the door serves") }
@@ -253,7 +318,8 @@ async fn idle_bound_closes_an_idle_association_and_a_fresh_one_succeeds() {
     let door = UdpFrontDoor::builder()
         .bind(relay.local_addr().unwrap())
         .idle(Duration::from_millis(300))
-        .build();
+        .build()
+        .expect("the door builds");
     let serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move { door.serve_on(relay, client).await.expect("the door serves") }
@@ -285,7 +351,10 @@ async fn idle_bound_closes_an_idle_association_and_a_fresh_one_succeeds() {
     // is healthy — the POC's strongest observable).
     let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2");
     let relay2_addr = relay2.local_addr().unwrap();
-    let door2 = UdpFrontDoor::builder().bind(relay2_addr).build();
+    let door2 = UdpFrontDoor::builder()
+        .bind(relay2_addr)
+        .build()
+        .expect("the door builds");
     let _serve2 = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {
@@ -332,7 +401,8 @@ async fn fresh_association_succeeds_after_teardown() {
     let door = UdpFrontDoor::builder()
         .bind(relay_addr)
         .idle(Duration::from_millis(200))
-        .build();
+        .build()
+        .expect("the door builds");
     let serve = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move { door.serve_on(relay, client).await.expect("the door serves") }
@@ -358,7 +428,10 @@ async fn fresh_association_succeeds_after_teardown() {
     // A fresh association on the same topology succeeds end to end.
     let relay2 = UdpSocket::bind("127.0.0.1:0").await.expect("bind relay 2");
     let relay2_addr = relay2.local_addr().expect("relay2 addr");
-    let door2 = UdpFrontDoor::builder().bind(relay2_addr).build();
+    let door2 = UdpFrontDoor::builder()
+        .bind(relay2_addr)
+        .build()
+        .expect("the door builds");
     let serve2 = tokio::spawn({
         let client = Arc::clone(&topo.consumer);
         async move {

tests/harness.rs

@@ -77,8 +77,8 @@ pub mod common;
 
 #[allow(unused_imports)]
 pub use common::{
-    datagram_frame as udp_frame, decode_datagram as udp_unframe, echo_dial,
+    datagram_frame as udp_frame, decode_datagram as udp_unframe, doubled_echo_egress_factory,
-    echo_dial_witnessed as witnessing_echo_dial, echo_egress_factory as echo_egress,
+    echo_dial, echo_dial_witnessed as witnessing_echo_dial, echo_egress_factory as echo_egress,
     parked_accept_factory, raw_greet, raw_read_reply, raw_request, read_reply_v4, refusing_dial,
     refusing_egress, write_request, QueueEgress,
 };

tests/local-backend-producer.rs

@@ -80,7 +80,8 @@ async fn serve(
         .producer(producer)
         .idle_first_byte(idle)
         .max_sessions(cap)
-        .build();
+        .build()
+        .expect("the door builds");
     let guard = door.serve_on(listener).await;
     (addr, guard)
 }