feat(client): land the -R leg — the acceptor registry + the reverse-listen rows (w4-reverse-listen)

- src/client_accept.rs: the assembly-mutable acceptor registry
  (per-type fixed slots under `tunnel`; refuse-never-replace install;
  payload = the `TunnelChannel` duplex + the open's echo/originator
  fields — the harness `ForwardedAcceptor` vocabulary grown), the
  reverse-listen rows (bound-port surfaced; the listen/cancel forms;
  the runtime-Unix gate at the listen), and the accepted-opens'
  spawned handoff tasks.
- src/client_handler.rs: the `forwarded-tcpip` /
  `forwarded-streamlocal` hooks route through the registry at
  arrival time — registered + admitted → accept + handoff;
  unregistered/unfeatured/cap-exhausted → the clean
  `AdministrativelyProhibited` decline (never accept-and-drop);
  pre-install arrivals decline, never buffered.
- src/client.rs: the dial wires the empty registry into the handler
  + the connection (`SshConnection::acceptors`, the late-install
  accessor) + the listener budget's dedicated counter.
- src/error.rs: `OpError::NoAcceptor` / `ListenRefused` /
  `ListenCapExceeded`; `RequestDenied` → `ListenRefused` (the
  listen/cancel forms' wire decline class; the port-0 mint rides
  alone at this pin — an explicit port's bound *is* the requested
  one, surfaced).
- tests/common/sshd_double.rs: the fake `sshd`'s `-R` half — the
  listen/cancel hooks bind real sockets (the out-param mint is the
  real bound port), the accept loops open the server-initiated
  forwarded channels + pump per accepted connection, cancels match
  by the raw wire strings and tear the listener down.
- tests/client_reverse_listen.rs: the contract rows (listen + data
  leg, the mint, the pre-wire gates, the declines, late-install
  on/off, the cancels, the ceiling, the unix twins).
- tests/tunnel_mapping.rs: the full-circle client ↔ server-direction
  smoke (both `-R` legs through the alkcall fabric double); real-
  `sshd` `-R` stays wave-5 e2e (recorded).
- client.md: the acceptor's landing record (shape, cap relation,
  gates, unix-gate placement, port-0 parity).
- fix(w4-op-rows, test cfg): the test mod's `family_exit_code`
  import gated to the exit-consuming builds (`--features tunnel`
  did not compile its tests before).

Verification: cargo test (default / each feature / --all-features)
green; cargo clippy --all-targets -- -D warnings (none +
--all-features) clean; cargo fmt --check clean; cargo doc --no-deps
clean (default build; all-features carries only the pre-existing
client_ops private-link warnings).

docs/architecture/client.md

@@ -69,6 +69,39 @@ half-capability, which the family forbids (ADR-003 symmetry +
   consumers turn capabilities on/off per connection; the shape is an
   assembly-mutable acceptor registry).
 
+> **The landed acceptor shape (`w4-reverse-listen` — recorded at
+> landing):** the registry is **per-type fixed slots** — exactly the
+> two `-R` data-leg types the client direction can serve
+> (`forwarded-tcpip`, `forwarded-streamlocal`, both under `tunnel`;
+> no map, no generic key), one acceptor trait per type; install is
+> **refuse-never-replace** (the `ListenRouter::register` recorded
+> posture, mirrored client-side); the slots are read **at arrival
+> time** by the dial handler's hooks — a pre-install/unfeatured open
+> is declined at arrival, never buffered. The payload an accepted
+> open hands the assembly: the channel duplex (family-typed — the
+> `-L` rows' `TunnelChannel` byte duplex, one type both legs ride) +
+> the open's echo/originator fields; **the pump is the acceptor's**
+> from the handoff on (the family handler shape) — the crate keeps no
+> accepted-open ledger. Bounded bookkeeping, the cap relation: the
+> registry bounded by construction (two slots); the
+> **reverse-listen count** and the **accepted-open handoff count**
+> are two dedicated counters, each against the same
+> `Limits::max_concurrent_channels` ceiling (the server-side
+> listener/ListenerCap posture mirrored — independent counts, one
+> ceiling; the listens live at the peer, the counter exists so a dial
+> cannot mint an unbounded listener set). The **forward-request rows
+> gate on the registered acceptor** (the issuing-side disposition:
+> a listen whose data leg cannot serve is never asked for —
+> install-first, `OpError::NoAcceptor` pre-wire), plus
+> `OpError::ListenRefused` (the peer's `RequestDenied` — denied
+> listens and unknown-listener cancels, sshd raw-string parity) and
+> `OpError::ListenCapExceeded` (the ceiling, pre-wire). The
+> **runtime-Unix gate** rides the unix *listen* rows only (the bind
+> is the platform-bound op; the `forwarded-streamlocal` ingestion is
+> platform-neutral byte pumping). The `tcpip_forward` reply's mint
+> rides only for `port = 0` requests (russh/sshd parity: an explicit
+> port's bound *is* the requested port, which the row surfaces).
+
 ## Lifecycle primitives (OQ-12 resolution)
 
 - Forward listeners return the **actually-bound port** (russh

src/client.rs

@@ -69,6 +69,10 @@ use std::sync::Arc;
 use russh::Disconnect;
 use russh::keys::PrivateKeyWithHashAlg;
 
+#[cfg(feature = "tunnel")]
+use crate::channel_cap::ChannelCap;
+#[cfg(feature = "tunnel")]
+use crate::client_accept::AcceptorRegistry;
 use crate::client_handler::ClientHandler;
 use crate::config::{DialConfig, Limits};
 use crate::error::DialError;
@@ -81,6 +85,14 @@ pub async fn dial(config: DialConfig) -> Result<SshConnection, DialError> {
 
     let (host, port) = parse_dial_address(&config.address)?;
     let handler = ClientHandler::new(Arc::clone(&config.host_key_policy), host, port);
+    #[cfg(feature = "tunnel")]
+    let (handler, acceptors) = {
+        let acceptors = Arc::new(AcceptorRegistry::new());
+        (
+            handler.with_tunnel_surface(Arc::clone(&acceptors), config.limits),
+            acceptors,
+        )
+    };
 
     let russh_config = Arc::new(build_client_config(&config.limits));
     let mut handle = russh::client::connect(russh_config, config.address.as_str(), handler).await?;
@@ -103,6 +115,10 @@ pub async fn dial(config: DialConfig) -> Result<SshConnection, DialError> {
         limits: config.limits,
         address: config.address,
         username: config.username,
+        #[cfg(feature = "tunnel")]
+        acceptors,
+        #[cfg(feature = "tunnel")]
+        listener_budget: ChannelCap::default(),
     })
 }
 
@@ -167,6 +183,19 @@ pub struct SshConnection {
     limits: Limits,
     address: String,
     username: String,
+    /// The `-R` leg's acceptor registry (the dial constructs it empty;
+    /// the assembly installs/uninstalls across the live connection —
+    /// see [`SshConnection::acceptors`] and
+    /// [`crate::client_accept`]`'s recorded decisions.
+    #[cfg(feature = "tunnel")]
+    acceptors: Arc<AcceptorRegistry>,
+    /// The reverse-listen count's dedicated bound — admitted at the
+    /// listen rows, released at the confirmed cancels and the refused/
+    /// wire-error paths (the cap relation recorded in
+    /// [`crate::client_accept`]; [`crate::channel_cap`]'s mutex makes
+    /// the counts sound under the caller's task).
+    #[cfg(feature = "tunnel")]
+    listener_budget: ChannelCap,
 }
 
 impl SshConnection {
@@ -186,6 +215,25 @@ impl SshConnection {
         &self.username
     }
 
+    /// The `-R` leg's acceptor registry (the
+    /// `w4-reverse-listen` surface, `client_accept`'s module docs
+    /// carry the recorded decisions) — install/uninstall the
+    /// per-type acceptors across the live connection (the
+    /// late-install dynamic). The dial constructs it empty: nothing
+    /// server-initiated is served until a slot holds, fail-closed at
+    /// arrival time.
+    #[cfg(feature = "tunnel")]
+    pub fn acceptors(&self) -> &AcceptorRegistry {
+        &self.acceptors
+    }
+
+    /// The reverse-listen count's shared handle — the op rows
+    /// ([`crate::client_accept`]) read it; private to the crate.
+    #[cfg(feature = "tunnel")]
+    pub(crate) fn listener_budget(&self) -> &ChannelCap {
+        &self.listener_budget
+    }
+
     /// Whether the underlying russh session is closed.
     pub fn is_closed(&self) -> bool {
         self.handle.is_closed()
@@ -223,12 +271,15 @@ impl SshConnection {
 
 impl std::fmt::Debug for SshConnection {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("SshConnection")
+        let mut debug = f.debug_struct("SshConnection");
+        debug
             .field("address", &self.address)
             .field("username", &self.username)
             .field("limits", &self.limits)
-            .field("closed", &self.handle.is_closed())
-            .finish()
+            .field("closed", &self.handle.is_closed());
+        #[cfg(feature = "tunnel")]
+        debug.field("acceptors", &self.acceptors);
+        debug.finish()
     }
 }
 

src/client_accept.rs

@@ -0,0 +1,624 @@
+//! The `-R` leg's client half (`w4-reverse-listen` — `client.md`
+//! §op rows row 4 + §The acceptor): the hub's reverse direction in
+//! family types. Two surfaces, the acceptor registry first:
+//!
+//! 1. **The forward request rows** — [`SshConnection::reverse_listen`]
+//!    / [`SshConnection::reverse_listen_cancel`] / the unix pair
+//!    (`reverse_listen_unix` / `reverse_listen_unix_cancel`): the
+//!    vanilla-server listen requests, the lifecycle primitives
+//!    (`client.md` §Lifecycle — the actually-bound port surfaced,
+//!    every listener explicitly cancellable; the *handles*' surface is
+//!    `w4-lifecycle`'s growth).
+//! 2. **The acceptor registry** ([`AcceptorRegistry`]) — the
+//!    assembly-mutable, per-channel-type registry the dial surface's
+//!    russh handler routes **server-initiated `forwarded-*` opens**
+//!    through. Served only under a registered acceptor for that type
+//!    and under the `tunnel` feature; unregistered/unfeatured accepted
+//!    opens **decline cleanly** — never russh's accept-and-drop
+//!    default (ADR-003 §4, the client-side fail-closed acceptor; the
+//!    reverse-runner fact 1 disposition is ours).
+//!
+//! ## The recorded decisions (the task's decide + record items)
+//!
+//! - **Registry shape: per-type fixed slots, not a typed map** — the
+//!   client direction's registry-servable type set is closed and tiny
+//!   (exactly the `-R` data leg's two: `forwarded-tcpip` /
+//!   `forwarded-streamlocal`, both under `tunnel`); one trait per type
+//!   ([`TcpipForwardedAcceptor`] / [`StreamlocalForwardedAcceptor`])
+//!   makes *which acceptor serves which type* a compile-time fact and
+//!   the fail-closed gate a one-slot read. A generic typed map would
+//!   add runtime dispatch (a typed key + an enum payload) for a set
+//!   that cannot grow without an ADR (any further server-initiated
+//!   type is a mapping, `client.md` §op rows). The harness's
+//!   `ForwardedAcceptor` vocabulary is grown, not forked: the payload
+//!   fields are the same shapes (the open's originator fields + the
+//!   channel duplex), the duplex now family-typed
+//!   ([`TunnelChannel`](crate::TunnelChannel) — the `-L` rows' byte
+//!   duplex, one type both legs ride).
+//! - **Install is refuse-never-replace** ([`AcceptorRegistry::
+//!   install_tcpip`]'s held-slot error): the same posture recorded
+//!   server-side on the `ListenRouter` at `w4-router-collision` —
+//!   a same-slot second install is an assembly program error, declined
+//!   never a quiet displacement of a live acceptor. Uninstall is
+//!   explicit, 1:1 with the install.
+//! - **Late-install across a live connection** (the alkminer
+//!   `install_forward_sink` dynamic — reverse-runner fact 2): the
+//!   registry rides the connected [`SshConnection`]
+//!   ([`SshConnection::acceptors`]) — the dial constructs it empty,
+//!   the assembly installs/uninstalls whenever it decides. The open
+//!   hooks read the slot **at arrival time**, fail-closed: an open
+//!   arriving before its acceptor installs (or after an uninstall)
+//!   is **declined at arrival, never buffered** — no unclaimed-open
+//!   queue exists anywhere in this module's posture (an unclaimed
+//!   open's bytes would have to be buffered unboundedly to accept-
+//!   then-route; the decline is the clean, bounded answer).
+//! - **The listen rows gate on the registered acceptor** — the
+//!   *issuing*-side disposition: a listen whose data leg cannot serve
+//!   is a half-capability (the peer stays bound, every arrival
+//!   declines), so a `reverse_listen` with no registered acceptor for
+//!   the type answers
+//!   [`OpError::NoAcceptor`](crate::OpError) **before any wire op**
+//!   (install-first is the recorded order; the rows never buffer an
+//!   outstanding listen against a later install — re-issuing is the
+//!   assembly's retry, the listen rows are stateless global requests).
+//! - **Bounded bookkeeping (AGENTS.md convention 9) — the cap
+//!   relation** (the server-side listener-counter posture as the
+//!   template): the registry is bounded by construction (two fixed
+//!   slots). The per-connection **listener count** is a dedicated
+//!   in-crate counter admitting at the listen rows and releasing at
+//!   the confirmed cancels and the failed/wire-error paths — bounded
+//!   against the *same* `Limits::max_concurrent_channels` ceiling the
+//!   server direction's listener counter uses (independent counts, one
+//!   ceiling — never a map: the outstanding listens live at the peer;
+//!   the counter exists so a dial cannot mint an unbounded listener
+//!   set the registry would have to serve). The per-connection
+//!   **accepted-open handoff count** is the second dedicated counter
+//!   against the same ceiling (admitted in the dial handler's open
+//!   hook, accepted-with-the-count-held, released when the acceptor's
+//!   `accept` call completes — no accepted-open ledger anywhere
+//!   else: the pump is the acceptor's from the handoff on, the
+//!   channel's transport bounds are russh's windows, and the slot read
+//!   is the arrival admission).
+//! - **The runtime-Unix gate (OQ-11 symmetry)**: the streamlocal
+//!   *listen* rows (`reverse_listen_unix` / `reverse_listen_unix_cancel`)
+//!   compile everywhere and
+//!   decline off-Unix before any wire op (the bind is the
+//!   platform-bound operation — the door puts nothing on the wire it
+//!   cannot track). The `forwarded-streamlocal` *open* hook is
+//!   universal: ingestion is platform-neutral byte pumping — the
+//!   unix-socket leg's socket is the peer-side's.
+//! - **`OpError::ListenRefused`** — the peer's `RequestDenied` on
+//!   every form's decline (a denied listen, a cancel of an unknown
+//!   listener) — channel-scoped honesty: the connection stays usable
+//!   (the `From` mapping's recorded producer, `error.rs`).
+//!
+//! ## The data leg
+//!
+//! Per accepted open: the hook accepts the channel and hands the
+//! assembly the open — the [`TunnelChannel`](crate::TunnelChannel)
+//! duplex + the open's echo/originator fields (the payload types
+//! below) — on a spawned task (the spawn-and-return rule's client
+//! mirror: the hook runs in the packet loop and returns fast). The
+//! pump is the acceptor's from the handoff — the family handler shape
+//! (alktunnels' `ForwardedHandler` recipe, POC-3 §C.4); dropping the
+//! duplex is wire-clean (russh closes the channel).
+//!
+//! ## Test surface
+//!
+//! The wire vehicle is the fake `sshd` double grown with the
+//! server-initiated-forward surfaces (`tests/common/sshd_double.rs`) —
+//! the double binds **real** sockets for its script accepts, opens the
+//! forwarded channels toward the client per accepted connection, and
+//! pumps them (a real `ssh -R`-shaped round trip in-process, the task's
+//! recorded smoke posture). The full-circle alkssh-client ↔
+//! alkssh-server-direction smoke (both `-R` legs through the alkcall
+//! fabric, the composition the assembly makes) rides the fabric
+//! doubles in `tests/tunnel_mapping.rs` — the same doubles the
+//! server-side `-R` rows use; real-`sshd` `-R` is wave-5 e2e
+//! territory, never silently dropped.
+//!
+//! ## Smoke posture (recorded)
+//!
+//! The in-process smoke shape is a real `ssh -R`-shaped round trip: a
+//! TCP connection to the forward's actually-bound port round-trips
+//! through the registered acceptor's duplex (the reverse-runner's
+//! hub-side proxy shape). The full-circle variant against the alkssh
+//! server direction pins the client registry against the server
+//! direction's real arrival pump (the echo/originator/mint shapes
+//! `w3-tunnel-listen` landed); real-`sshd` `-R` is wave-5 e2e.
+
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use parking_lot::Mutex;
+use russh::client::ChannelOpenHandle;
+
+use crate::TunnelChannel;
+use crate::channel_cap::ChannelCap;
+use crate::client::SshConnection;
+use crate::error::OpError;
+use crate::observe;
+
+/// An accepted server-initiated `forwarded-tcpip` open — the acceptor
+/// registry's payload for the
+/// [`TcpipForwardedAcceptor::accept`] call: the open's echo fields
+/// (the listen endpoint the arrival belongs to — the assembly's
+/// routing key; the OpenSSH matcher parity verbatim) + the originator
+/// (the arrival's peer, as the wire carried it) + the channel duplex.
+pub struct ForwardedTcpip {
+    /// The connected address the open echoed — the `reverse_listen`
+    /// wire string (sshd parity).
+    pub connected_address: String,
+    /// The actually-bound port of the minted listener — for a
+    /// `port = 0` mint this is the server-picked number, never `0`.
+    pub connected_port: u32,
+    /// The arrival's peer address (the wire originator).
+    pub originator_address: String,
+    /// The arrival's peer port (the wire originator).
+    pub originator_port: u32,
+    /// The accepted channel's byte duplex — the acceptor owns the
+    /// pump from the handoff on (dropping it is wire-clean).
+    pub stream: TunnelChannel,
+}
+
+impl std::fmt::Debug for ForwardedTcpip {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ForwardedTcpip")
+            .field("connected_address", &self.connected_address)
+            .field("connected_port", &self.connected_port)
+            .field("originator_address", &self.originator_address)
+            .field("originator_port", &self.originator_port)
+            .finish_non_exhaustive()
+    }
+}
+
+/// An accepted server-initiated `forwarded-streamlocal` open — the
+/// streamlocal acceptor's payload (the echoed socket path + the
+/// channel duplex).
+pub struct ForwardedStreamlocal {
+    /// The socket path the open echoed — the `reverse_listen_unix`
+    /// wire string (sshd parity).
+    pub socket_path: String,
+    /// The accepted channel's byte duplex — the acceptor owns the
+    /// pump from the handoff on (dropping it is wire-clean).
+    pub stream: TunnelChannel,
+}
+
+impl std::fmt::Debug for ForwardedStreamlocal {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ForwardedStreamlocal")
+            .field("socket_path", &self.socket_path)
+            .finish_non_exhaustive()
+    }
+}
+
+/// The `forwarded-tcpip` acceptor — the registry's per-type slot
+/// trait. `accept` runs on the spawned handoff task (never the
+/// packet loop): the assembly's pump/home of the open lives inside
+/// it; the crate releases its handoff counter when the call returns,
+/// so the family shape (hand off to a queue, return) keeps the
+/// counter's meaning "open accepted, handoff in flight".
+#[async_trait]
+pub trait TcpipForwardedAcceptor: Send + Sync {
+    /// Serves one accepted open. Never invoked for an unregistered
+    /// type (the registry gate is first); the open is already accepted
+    /// on the wire — refusing one from here is a *data-plane* choice
+    /// (drop the duplex; the channel closes on it), never a wire
+    /// open-decline.
+    async fn accept(&self, open: ForwardedTcpip);
+}
+
+/// The `forwarded-streamlocal` acceptor — the unix slot's trait
+/// (see [`TcpipForwardedAcceptor`] for the call-shape contract).
+#[async_trait]
+pub trait StreamlocalForwardedAcceptor: Send + Sync {
+    /// Serves one accepted open (platform-neutral pumping — see the
+    /// module docs' unix-gate record).
+    async fn accept(&self, open: ForwardedStreamlocal);
+}
+
+/// The held-slot refusal (`AcceptorRegistry::install_*`) — the
+/// refuse-never-replace posture (the server-side `ListenRouter`'s
+/// recorded sibling): the slot already holds a live acceptor; a second
+/// install is the assembly's program error, declined, never a quiet
+/// displacement. An explicit `uninstall_*` frees the slot.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
+#[error("an acceptor is already registered for this channel type")]
+pub struct AcceptorHeld;
+
+/// The assembly-mutable, per-channel-type acceptor registry — the
+/// `-R` data leg's fail-closed gate (see the module docs for the
+/// recorded decisions). Constructed empty at the dial
+/// ([`SshConnection::acceptors`], the live-connection accessor);
+/// install/uninstall is the assembly's to drive whenever it decides.
+pub struct AcceptorRegistry {
+    tcpip: Mutex<Option<Arc<dyn TcpipForwardedAcceptor>>>,
+    streamlocal: Mutex<Option<Arc<dyn StreamlocalForwardedAcceptor>>>,
+}
+
+impl AcceptorRegistry {
+    /// The empty registry — the dial's construct (nothing is served
+    /// until an acceptor installs; ADR-003 §4's fail-closed default).
+    pub fn new() -> Self {
+        Self {
+            tcpip: Mutex::new(None),
+            streamlocal: Mutex::new(None),
+        }
+    }
+
+    /// Installs the `forwarded-tcpip` acceptor. Refuse-never-replace:
+    /// a held slot declines ([`AcceptorHeld`]) — uninstall first if a
+    /// replacement is meant (the late-install toggle's other half).
+    pub fn install_tcpip(
+        &self,
+        acceptor: Arc<dyn TcpipForwardedAcceptor>,
+    ) -> Result<(), AcceptorHeld> {
+        let mut slot = self.tcpip.lock();
+        if slot.is_some() {
+            return Err(AcceptorHeld);
+        }
+        *slot = Some(acceptor);
+        Ok(())
+    }
+
+    /// Uninstalls the `forwarded-tcpip` acceptor — `true` when one was
+    /// held (the toggle's honest report). In-flight opens already
+    /// accepted keep pumping (their handoff happened); arrivals after
+    /// this decline at arrival time (fail-closed, no buffering).
+    pub fn uninstall_tcpip(&self) -> bool {
+        self.tcpip.lock().take().is_some()
+    }
+
+    /// Installs the `forwarded-streamlocal` acceptor (see
+    /// [`Self::install_tcpip`] for the refuse-never-replace posture).
+    pub fn install_streamlocal(
+        &self,
+        acceptor: Arc<dyn StreamlocalForwardedAcceptor>,
+    ) -> Result<(), AcceptorHeld> {
+        let mut slot = self.streamlocal.lock();
+        if slot.is_some() {
+            return Err(AcceptorHeld);
+        }
+        *slot = Some(acceptor);
+        Ok(())
+    }
+
+    /// Uninstalls the `forwarded-streamlocal` acceptor (see
+    /// [`Self::uninstall_tcpip`] for the honest-report posture).
+    pub fn uninstall_streamlocal(&self) -> bool {
+        self.streamlocal.lock().take().is_some()
+    }
+
+    /// The arrival-time slot reads (the open hooks' gates) — cloned
+    /// out so the spawned handoff task owns its acceptor handle.
+    pub(crate) fn tcpip_acceptor(&self) -> Option<Arc<dyn TcpipForwardedAcceptor>> {
+        self.tcpip.lock().clone()
+    }
+
+    pub(crate) fn streamlocal_acceptor(&self) -> Option<Arc<dyn StreamlocalForwardedAcceptor>> {
+        self.streamlocal.lock().clone()
+    }
+}
+
+impl Default for AcceptorRegistry {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl std::fmt::Debug for AcceptorRegistry {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AcceptorRegistry")
+            .field("tcpip", &self.tcpip.lock().is_some())
+            .field("streamlocal", &self.streamlocal.lock().is_some())
+            .finish()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// The server-initiated opens' handoff (the dial handler's routed paths)
+// ---------------------------------------------------------------------------
+
+/// The accepted `forwarded-tcpip` open's handoff task — the
+/// spawn-and-return rule's client mirror (ADR-006 decision 4; the
+/// hook runs in the packet loop and returns fast): accepts the open on
+/// the wire, converts the channel to the family duplex, delivers the
+/// payload to the assembly's acceptor, and releases the handoff
+/// counter when the acceptor's call completes (the recorded release
+/// site — the count's meaning is "accepted, handoff in flight").
+#[cfg_attr(feature = "tunnel", allow(clippy::too_many_arguments))]
+pub(crate) async fn handoff_forwarded_tcpip(
+    acceptor: Arc<dyn TcpipForwardedAcceptor>,
+    open_cap: Arc<ChannelCap>,
+    channel: russh::Channel<russh::client::Msg>,
+    reply: ChannelOpenHandle,
+    connected_address: String,
+    connected_port: u32,
+    originator_address: String,
+    originator_port: u32,
+) {
+    let channel_number = channel.id().number();
+    observe::serving_path_started("forwarded-tcpip", channel_number, None);
+    reply.accept().await;
+    let open = ForwardedTcpip {
+        connected_address,
+        connected_port,
+        originator_address,
+        originator_port,
+        stream: TunnelChannel {
+            stream: channel.into_stream(),
+        },
+    };
+    acceptor.accept(open).await;
+    open_cap.release();
+    observe::serving_path_ended("forwarded-tcpip", channel_number, None);
+}
+
+/// The accepted `forwarded-streamlocal` open's handoff task — the
+/// unix slot's mirror of [`handoff_forwarded_tcpip`].
+pub(crate) async fn handoff_forwarded_streamlocal(
+    acceptor: Arc<dyn StreamlocalForwardedAcceptor>,
+    open_cap: Arc<ChannelCap>,
+    channel: russh::Channel<russh::client::Msg>,
+    reply: ChannelOpenHandle,
+    socket_path: String,
+) {
+    let channel_number = channel.id().number();
+    observe::serving_path_started("forwarded-streamlocal", channel_number, None);
+    reply.accept().await;
+    let open = ForwardedStreamlocal {
+        socket_path,
+        stream: TunnelChannel {
+            stream: channel.into_stream(),
+        },
+    };
+    acceptor.accept(open).await;
+    open_cap.release();
+    observe::serving_path_ended("forwarded-streamlocal", channel_number, None);
+}
+
+// ---------------------------------------------------------------------------
+// The reverse-listen rows (under `tunnel`)
+// ---------------------------------------------------------------------------
+
+impl SshConnection {
+    /// The reverse-listen row (`-R`): asks the vanilla server to listen
+    /// on (address, port) and returns the **actually-bound port** —
+    /// for `port = 0` this is the peer-picked number (`client.md`
+    /// §Lifecycle: the bound-port primitive, the port-0 mint); for an
+    /// explicit port the wire's success reply carries no number (russh/
+    /// sshd parity — the reply's data only exists for mints) and the
+    /// bound port *is* the requested one, which the row surfaces.
+    ///
+    /// Gates, both **before any wire op**: the `forwarded-tcpip`
+    /// acceptor must be registered
+    /// ([`OpError::NoAcceptor`](crate::OpError) — a listen whose data
+    /// leg cannot serve is a half-capability) and the connection's
+    /// listener count must have room
+    /// ([`OpError::ListenCapExceeded`](crate::OpError) — the
+    /// dedicated counter sharing
+    /// [`Limits::max_concurrent_channels`](crate::Limits); see the
+    /// module docs' cap relation). The peer's refusal is
+    /// [`OpError::ListenRefused`](crate::OpError) (the wire's
+    /// `RequestDenied`) and releases the count; the connection stays
+    /// usable after every refusal class. The listener is explicitly
+    /// cancellable ([`SshConnection::reverse_listen_cancel`]; it is
+    /// not implicitly unlistened by anything but the connection's
+    /// end).
+    pub async fn reverse_listen(
+        &self,
+        address: impl Into<String>,
+        port: u32,
+    ) -> Result<u32, OpError> {
+        if self.acceptors().tcpip_acceptor().is_none() {
+            return Err(OpError::NoAcceptor);
+        }
+        if !self
+            .listener_budget()
+            .admit(self.limits().max_concurrent_channels)
+        {
+            return Err(OpError::ListenCapExceeded);
+        }
+        let bound = match self.russh_handle().tcpip_forward(address, port).await {
+            Ok(minted) if port == 0 => minted,
+            Ok(_) => port,
+            Err(error) => {
+                self.listener_budget().release();
+                return Err(OpError::from(error));
+            }
+        };
+        Ok(bound)
+    }
+
+    /// The reverse-listen cancel row (`-R` teardown): the explicit
+    /// cancellation of one bound listener (the lifecycle primitive:
+    /// every listener explicitly cancellable, the wire-honest path —
+    /// the peer tears its listen down, the bound port frees). Matches
+    /// by the raw wire strings (sshd parity). The peer's refusal
+    /// (unknown listener) is [`OpError::ListenRefused`](crate::OpError);
+    /// a confirmed cancel releases the listener count.
+    pub async fn reverse_listen_cancel(
+        &self,
+        address: impl Into<String>,
+        port: u32,
+    ) -> Result<(), OpError> {
+        match self
+            .russh_handle()
+            .cancel_tcpip_forward(address, port)
+            .await
+        {
+            Ok(()) => {
+                self.listener_budget().release();
+                Ok(())
+            }
+            Err(error) => Err(OpError::from(error)),
+        }
+    }
+
+    /// The streamlocal reverse-listen row (the unix `-R`): asks the
+    /// vanilla server to listen on the unix socket at `socket_path`,
+    /// minting no port (the global request's success is the outcome;
+    /// the accepted opens' echo carries the path). Same gates as
+    /// [`SshConnection::reverse_listen`] for the
+    /// `forwarded-streamlocal` slot, plus the **runtime-Unix gate**
+    /// (OQ-11: off-Unix, [`OpError::UnixUnavailable`](crate::OpError)
+    /// before any wire op — the bind is the platform-bound operation;
+    /// the channel ingestion is platform-neutral).
+    #[cfg_attr(not(unix), allow(unused_variables))]
+    pub async fn reverse_listen_unix(&self, socket_path: impl Into<String>) -> Result<(), OpError> {
+        #[cfg(not(unix))]
+        {
+            return Err(OpError::UnixUnavailable);
+        }
+        #[cfg(unix)]
+        {
+            if self.acceptors().streamlocal_acceptor().is_none() {
+                return Err(OpError::NoAcceptor);
+            }
+            if !self
+                .listener_budget()
+                .admit(self.limits().max_concurrent_channels)
+            {
+                return Err(OpError::ListenCapExceeded);
+            }
+            match self.russh_handle().streamlocal_forward(socket_path).await {
+                Ok(()) => Ok(()),
+                Err(error) => {
+                    self.listener_budget().release();
+                    Err(OpError::from(error))
+                }
+            }
+        }
+    }
+
+    /// The streamlocal cancel row (matches by the socket path, sshd
+    /// parity — see [`SshConnection::reverse_listen_cancel`]).
+    #[cfg_attr(not(unix), allow(unused_variables))]
+    pub async fn reverse_listen_unix_cancel(
+        &self,
+        socket_path: impl Into<String>,
+    ) -> Result<(), OpError> {
+        #[cfg(not(unix))]
+        {
+            return Err(OpError::UnixUnavailable);
+        }
+        #[cfg(unix)]
+        {
+            match self
+                .russh_handle()
+                .cancel_streamlocal_forward(socket_path)
+                .await
+            {
+                Ok(()) => {
+                    self.listener_budget().release();
+                    Ok(())
+                }
+                Err(error) => Err(OpError::from(error)),
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::AcceptorHeld;
+    use crate::client_accept::{
+        AcceptorRegistry, ForwardedStreamlocal, ForwardedTcpip, StreamlocalForwardedAcceptor,
+        TcpipForwardedAcceptor,
+    };
+    use std::sync::Arc;
+
+    struct StubTcpip;
+
+    #[async_trait::async_trait]
+    impl TcpipForwardedAcceptor for StubTcpip {
+        async fn accept(&self, _open: ForwardedTcpip) {}
+    }
+
+    struct StubStreamlocal;
+
+    #[async_trait::async_trait]
+    impl StreamlocalForwardedAcceptor for StubStreamlocal {
+        async fn accept(&self, _open: ForwardedStreamlocal) {}
+    }
+
+    /// - *Contract stamp:* the recorded refuse-never-replace posture —
+    ///   a held slot declines the second install (never a quiet
+    ///   displacement of a live acceptor); uninstall frees it.
+    #[test]
+    fn install_refuses_a_held_slot_and_uninstall_frees_it() {
+        let registry = AcceptorRegistry::new();
+        let first: Arc<dyn TcpipForwardedAcceptor> = Arc::new(StubTcpip);
+        assert!(registry.install_tcpip(Arc::clone(&first)).is_ok());
+        assert!(
+            registry.install_tcpip(Arc::new(StubTcpip)).is_err(),
+            "a held slot is refused, never replaced"
+        );
+        let held = registry.tcpip_acceptor().expect("the first stays held");
+        assert!(Arc::ptr_eq(&held, &first));
+
+        assert!(registry.uninstall_tcpip());
+        assert!(
+            !registry.uninstall_tcpip(),
+            "the second uninstall is the honest `false`"
+        );
+        assert!(registry.tcpip_acceptor().is_none());
+        assert!(
+            registry.install_tcpip(first).is_ok(),
+            "an uninstalled slot installs again (the late-install toggle)"
+        );
+    }
+
+    /// - *Contract stamp:* the two slots are independent (the
+    ///   per-type-registry decision) — the empty streamlocal slot
+    ///   stays unfeatured while the tcpip slot serves.
+    #[test]
+    fn the_slots_are_independent() {
+        let registry = AcceptorRegistry::new();
+        assert!(registry.tcpip_acceptor().is_none());
+        assert!(registry.streamlocal_acceptor().is_none());
+
+        registry
+            .install_tcpip(Arc::new(StubTcpip))
+            .expect("the tcp slot installs");
+        assert!(registry.tcpip_acceptor().is_some());
+        assert!(
+            registry.streamlocal_acceptor().is_none(),
+            "the streamlocal slot is untouched"
+        );
+
+        registry
+            .install_streamlocal(Arc::new(StubStreamlocal))
+            .expect("the streamlocal slot installs independently");
+        assert!(registry.streamlocal_acceptor().is_some());
+    }
+
+    /// - *Contract stamp:* the Debug shape (the installed-slot fact,
+    ///   never internals).
+    #[test]
+    fn the_registry_debug_shapes_the_slots() {
+        let registry = AcceptorRegistry::new();
+        let debug = format!("{registry:?}");
+        assert!(
+            debug.contains("tcpip: false") && debug.contains("streamlocal: false"),
+            "{debug}"
+        );
+
+        registry
+            .install_tcpip(Arc::new(StubTcpip))
+            .expect("installs");
+        let debug = format!("{registry:?}");
+        assert!(debug.contains("tcpip: true"), "{debug}");
+    }
+
+    /// - *Contract stamp:* the held-slot error's honest shape (no
+    ///   `Other`, no catch-all).
+    #[test]
+    fn acceptor_held_error_shapes() {
+        let held = AcceptorHeld;
+        assert_eq!(
+            held.to_string(),
+            "an acceptor is already registered for this channel type"
+        );
+    }
+}

src/client_handler.rs

@@ -20,25 +20,39 @@
 //!    the client-side symmetry rule): russh's default posture for every
 //!    `server_channel_open_*` hook is accept-and-drop — a
 //!    half-capability the family forbids. Every hook is overridden
-//!    explicit-over-inherited and answers
+//!    explicit-over-inherited. The `forwarded-tcpip` /
+//!    `forwarded-streamlocal` hooks (the `-R` data leg's types) route
+//!    through the **acceptor registry** ([`crate::client_accept`] —
+//!    the `w4-reverse-listen` surface): served only under a registered
+//!    acceptor for that type and under the `tunnel` feature (the
+//!    [`with_tunnel_surface`](ClientHandler::with_tunnel_surface)
+//!    wiring; a build without `tunnel` has no registry — every hook
+//!    declines); unregistered/unfeatured opens answer
 //!    [`ChannelOpenFailure::AdministrativelyProhibited`] — a clean
 //!    wire-visible refusal the peer observes as an
 //!    unconfirmed open, never an accepted-then-dropped channel. The
-//!    unknown-type gate
+//!    accepted open's handoff spawns (the spawn-and-return rule's
+//!    client mirror): the hook accepts and returns fast, the duplex +
+//!    the open's fields ride the payload to the assembly's acceptor
+//!    call on its own task. Every other type has no registry slot —
+//!    the unknown-type gate
 //!    ([`Handler::should_accept_unknown_server_channel`]) answers
 //!    `false` (the hook's reject-before-open form; its `_unknown`
 //!    sibling stays a defensive reject — wire-reachable only if that
-//!    gate ever changed, also fail-closed). The acceptor registry
-//!    that serves these opens under its feature is
-//!    `w4-reverse-listen`'s surface; this skeleton pins the decline-all
-//!    default every unregistered open gets.
+//!    gate ever changed, also fail-closed), and the never-feature rows
+//!    (x11, agent) decline at any configuration.
 //!
-//! Data-plane hooks stay russh's defaults: every server-initiated
-//! channel is declined, so no data for one can ever arrive. The
-//! `w4-op-rows` / `w4-reverse-listen` tasks grow this skeleton in
-//! place (the wave's sequential client chain); the decline events ride
-//! the crate's [`crate::observe`] vocabulary (`decline_open`,
-//! reason `no-acceptor`).
+//! The accepted-open handoff carries a **bounded bookkeeping**
+//! counter (`AGENTS.md` convention 9, the cap relation recorded in
+//! [`crate::client_accept`]): a dedicated per-connection counter —
+//! sharing the dial's `Limits::max_concurrent_channels` ceiling the
+//! server direction's per-session cap uses — admitted in the hook
+//! before the accept, released when the acceptor's call completes in
+//! the spawned task; a full counter declines the arrival cleanly
+//! (`cap-exceeded`).
+
+#[cfg(feature = "tunnel")]
+use std::sync::Arc;
 
 use russh::ChannelOpenFailure;
 use russh::client::ChannelOpenHandle;
@@ -46,10 +60,30 @@ use russh::client::Msg;
 use russh::keys::PublicKeyOrCertificate;
 use russh::{Channel, ChannelId};
 
+#[cfg(feature = "tunnel")]
+use crate::channel_cap::ChannelCap;
+#[cfg(feature = "tunnel")]
+use crate::client_accept::{
+    AcceptorRegistry, handoff_forwarded_streamlocal, handoff_forwarded_tcpip,
+};
+#[cfg(feature = "tunnel")]
+use crate::config::Limits;
 use crate::host_key::{HostKeyPolicy, PresentedHostKey};
 use crate::observe;
 use crate::{DialError, fingerprint};
 
+/// The `-R` leg's per-connection client-handler state (the
+/// `w4-reverse-listen` surface, under `tunnel`): the acceptor registry
+/// the two forwarded hooks route through, plus the accepted-open
+/// handoff counter and its ceiling (the dial's
+/// `Limits::max_concurrent_channels`).
+#[cfg(feature = "tunnel")]
+pub(crate) struct TunnelSurface {
+    acceptors: Arc<AcceptorRegistry>,
+    limits: Limits,
+    open_cap: std::sync::Arc<ChannelCap>,
+}
+
 /// The per-connection client handler the dial surface constructs —
 /// see the module docs. Test surface
 /// (`#[doc(hidden)]` re-export from the crate root): the contract
@@ -59,6 +93,11 @@ pub struct ClientHandler {
     policy: std::sync::Arc<dyn HostKeyPolicy>,
     host: String,
     port: u16,
+    /// The `-R` surface under `tunnel` (`Some` — the dial wires it;
+    /// the bare-hook tests construct without it to pin the decline-all
+    /// default posture in the featured builds too).
+    #[cfg(feature = "tunnel")]
+    tunnel: Option<TunnelSurface>,
 }
 
 impl ClientHandler {
@@ -74,6 +113,24 @@ impl ClientHandler {
             policy,
             host: host.into(),
             port,
+            #[cfg(feature = "tunnel")]
+            tunnel: None,
+        }
+    }
+
+    /// Wires the `-R` surface (the dial's step under `tunnel`): the
+    /// acceptor registry (constructed empty at the dial; the assembly
+    /// installs through the connection's accessor) + the
+    /// accepted-open counter against the limits' channel ceiling.
+    #[cfg(feature = "tunnel")]
+    pub fn with_tunnel_surface(self, acceptors: Arc<AcceptorRegistry>, limits: Limits) -> Self {
+        Self {
+            tunnel: Some(TunnelSurface {
+                acceptors,
+                limits,
+                open_cap: Arc::new(ChannelCap::default()),
+            }),
+            ..self
         }
     }
 }
@@ -125,39 +182,138 @@ impl russh::client::Handler for ClientHandler {
         Ok(())
     }
 
+    /// The `forwarded-tcpip` hook (the `-R` data leg's gate): the
+    /// registry's slot read at arrival time — registered + room →
+    /// accept and hand the assembly the open on a spawned task; no
+    /// acceptor / no room → the clean decline, never accept-and-drop.
     async fn server_channel_open_forwarded_tcpip(
         &mut self,
         channel: Channel<Msg>,
-        _connected_address: &str,
-        _connected_port: u32,
-        _originator_address: &str,
-        _originator_port: u32,
+        #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] connected_address: &str,
+        #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] connected_port: u32,
+        #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] originator_address: &str,
+        #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] originator_port: u32,
         reply: ChannelOpenHandle,
         _session: &mut russh::client::Session,
     ) -> Result<(), Self::Error> {
-        observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number());
-        reply
-            .reject(ChannelOpenFailure::AdministrativelyProhibited)
-            .await;
-        Ok(())
+        #[cfg(not(feature = "tunnel"))]
+        {
+            observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number());
+            reply
+                .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                .await;
+            Ok(())
+        }
+        #[cfg(feature = "tunnel")]
+        {
+            let Some(surface) = self.tunnel.as_ref() else {
+                observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number());
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+                return Ok(());
+            };
+            let Some(acceptor) = surface.acceptors.tcpip_acceptor() else {
+                observe::decline_open("forwarded-tcpip", "no-acceptor", channel.id().number());
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+                return Ok(());
+            };
+            if !surface
+                .open_cap
+                .admit(surface.limits.max_concurrent_channels)
+            {
+                observe::decline_open("forwarded-tcpip", "cap-exceeded", channel.id().number());
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+                return Ok(());
+            }
+            tokio::spawn(handoff_forwarded_tcpip(
+                acceptor,
+                Arc::clone(&surface.open_cap),
+                channel,
+                reply,
+                connected_address.to_owned(),
+                connected_port,
+                originator_address.to_owned(),
+                originator_port,
+            ));
+            Ok(())
+        }
     }
 
+    /// The `forwarded-streamlocal` hook — the unix-slot's mirror of
+    /// the tcpip gate (the ingestion is platform-neutral: no runtime
+    /// gate here; the listen rows carry the OQ-11 gate on the bind —
+    /// the recorded asymmetry, see [`crate::client_accept`]).
     async fn server_channel_open_forwarded_streamlocal(
         &mut self,
         channel: Channel<Msg>,
-        _socket_path: &str,
+        #[cfg_attr(not(feature = "tunnel"), allow(unused_variables))] socket_path: &str,
         reply: ChannelOpenHandle,
         _session: &mut russh::client::Session,
     ) -> Result<(), Self::Error> {
-        observe::decline_open(
-            "forwarded-streamlocal",
-            "no-acceptor",
-            channel.id().number(),
-        );
-        reply
-            .reject(ChannelOpenFailure::AdministrativelyProhibited)
-            .await;
-        Ok(())
+        #[cfg(not(feature = "tunnel"))]
+        {
+            observe::decline_open(
+                "forwarded-streamlocal",
+                "no-acceptor",
+                channel.id().number(),
+            );
+            reply
+                .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                .await;
+            Ok(())
+        }
+        #[cfg(feature = "tunnel")]
+        {
+            let Some(surface) = self.tunnel.as_ref() else {
+                observe::decline_open(
+                    "forwarded-streamlocal",
+                    "no-acceptor",
+                    channel.id().number(),
+                );
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+                return Ok(());
+            };
+            let Some(acceptor) = surface.acceptors.streamlocal_acceptor() else {
+                observe::decline_open(
+                    "forwarded-streamlocal",
+                    "no-acceptor",
+                    channel.id().number(),
+                );
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+                return Ok(());
+            };
+            if !surface
+                .open_cap
+                .admit(surface.limits.max_concurrent_channels)
+            {
+                observe::decline_open(
+                    "forwarded-streamlocal",
+                    "cap-exceeded",
+                    channel.id().number(),
+                );
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+                return Ok(());
+            }
+            tokio::spawn(handoff_forwarded_streamlocal(
+                acceptor,
+                Arc::clone(&surface.open_cap),
+                channel,
+                reply,
+                socket_path.to_owned(),
+            ));
+            Ok(())
+        }
     }
 
     async fn server_channel_open_direct_tcpip(
@@ -247,10 +403,10 @@ impl russh::client::Handler for ClientHandler {
 /// the hooks directly where the wire cannot reach a hook on its own.
 #[cfg(test)]
 mod tests {
-    use std::sync::Arc;
-
     use russh::client::Handler as _;
 
+    use std::sync::Arc;
+
     use super::ClientHandler;
     use crate::host_key::HostKeyPolicy;
 

src/client_ops.rs

@@ -541,7 +541,11 @@ async fn write_exit_chunk(
 /// surface — the duplex's `poll_shutdown` carries the wire EOF.
 #[cfg(feature = "tunnel")]
 pub struct TunnelChannel {
-    stream: russh::ChannelStream<russh::client::Msg>,
+    /// The russh channel's byte duplex (`pub(crate)` — the `-R` leg's
+    /// acceptor payloads construct the type over a server-initiated
+    /// open's `into_stream()` conversion; downstream crates see the
+    /// type opaque as ever).
+    pub(crate) stream: russh::ChannelStream<russh::client::Msg>,
 }
 
 #[cfg(feature = "tunnel")]
@@ -836,7 +840,9 @@ async fn git_output_pump(
 
 #[cfg(all(test, any(feature = "session", feature = "git", feature = "tunnel")))]
 mod tests {
-    use super::{family_exit_code, open_error};
+    #[cfg(any(feature = "session", feature = "git"))]
+    use super::family_exit_code;
+    use super::open_error;
     use crate::error::OpError;
     use russh::ChannelOpenFailure;
 
@@ -865,6 +871,7 @@ mod tests {
     /// - *Contract stamp:* the family exit-currency clamp (§the
     ///   family plane) — ordinary codes ride identity, beyond-range
     ///   ones narrow (never invent).
+    #[cfg(any(feature = "session", feature = "git"))]
     #[test]
     fn family_exit_code_narrows_beyond_the_i32_range() {
         assert_eq!(family_exit_code(0), 0);

src/error.rs

@@ -338,6 +338,36 @@ pub enum OpError {
     #[error("unix socket forwarding is unavailable on this platform")]
     UnixUnavailable,
 
+    /// The reverse-forward op's channel type has **no acceptor
+    /// registered** (ADR-003 §4's client-side gate surfaced to the
+    /// *issuing* side: a reverse listen whose data leg could not serve
+    /// arrivals is issued never — the half-capability posture,
+    /// answered before any wire op). Featured-but-unregistered is the
+    /// exact class: the unset-declines convention mirrored (the
+    /// acceptor slots are assembly-mutable, install-first is the
+    /// recorded order).
+    #[error("no acceptor is registered for the op's forwarded channel type")]
+    NoAcceptor,
+
+    /// The peer refused the op's reverse-forward global request (the
+    /// `RequestDenied` wire decline class) — a listen denial or the
+    /// cancel forms' unknown-listener answer (sshd parity: match by
+    /// the raw wire strings). Channel-scoped honesty: the connection
+    /// stays usable (row-pinned), and the wire's own reason classes
+    /// are never invented here — the closed family class mirrors the
+    /// wire shape, `RequestRefused`'s global-request counterpart.
+    #[error("the peer refused the reverse-forward request")]
+    ListenRefused,
+
+    /// The op's reverse-listen admission hit the connection's
+    /// dedicated listener ceiling (the listener counter sharing
+    /// `Limits::max_concurrent_channels` — the server-side
+    /// listener-counter posture mirrored; the cap relation recorded in
+    /// `client_accept.rs`) — answered **before any wire op**: a
+    /// request the budget cannot track is issued never.
+    #[error("the connection's reverse-listen count is at its ceiling")]
+    ListenCapExceeded,
+
     /// The sftp route is the OQ-13 decline placeholder (the feature
     /// exists, the route does not until alkfs lands, on either
     /// direction) — answered before any wire op, never a
@@ -352,14 +382,18 @@ pub enum OpError {
 }
 
 impl From<russh::Error> for OpError {
-    /// The one mirror point: a typed open's rejection is the wire
-    /// shape `russh::Error::ChannelOpenFailure` and maps to
+    /// The mirror points: a typed open's rejection is the wire shape
+    /// `russh::Error::ChannelOpenFailure` and maps to
     /// [`OpError::OpenDeclined`] (family-mirrored
-    /// [`OpenDeclineReason`]); every other russh error stays the
+    /// [`OpenDeclineReason`]); a denied reverse-forward *global*
+    /// request (`RequestDenied` — the listen/cancel forms' only wire
+    /// decline shape at this pin, the rows' only producer) maps to
+    /// [`OpError::ListenRefused`]; every other russh error stays the
     /// transparent [`OpError::Wire`].
     fn from(error: russh::Error) -> Self {
         match error {
             russh::Error::ChannelOpenFailure(failure) => Self::OpenDeclined(failure.into()),
+            russh::Error::RequestDenied => Self::ListenRefused,
             other => Self::Wire(other),
         }
     }
@@ -586,6 +620,9 @@ mod tests {
             OpError::CommandTooLong,
             OpError::InvalidRepoId,
             OpError::UnixUnavailable,
+            OpError::NoAcceptor,
+            OpError::ListenRefused,
+            OpError::ListenCapExceeded,
             OpError::SftpNotServed,
             OpError::Wire(russh::Error::Disconnect),
         ]);
@@ -601,6 +638,9 @@ mod tests {
             OpError::CommandTooLong => "command-too-long",
             OpError::InvalidRepoId => "invalid-repo-id",
             OpError::UnixUnavailable => "unix-unavailable",
+            OpError::NoAcceptor => "no-acceptor",
+            OpError::ListenRefused => "listen-refused",
+            OpError::ListenCapExceeded => "listen-cap-exceeded",
             OpError::SftpNotServed => "sftp-not-served",
             OpError::Wire(_) => "wire",
         };
@@ -609,6 +649,16 @@ mod tests {
         }
     }
 
+    /// - *Contract stamp:* the reverse-forward global request's wire
+    ///   decline class (`RequestDenied` — the listen/cancel forms'
+    ///   only producer shape at this pin): `ListenRefused`, never the
+    ///   transparent wire class; the typed-open mapping unchanged.
+    #[test]
+    fn op_error_request_denied_maps_to_listen_refused() {
+        let mapped: OpError = russh::Error::RequestDenied.into();
+        assert!(matches!(mapped, OpError::ListenRefused), "got {mapped:?}");
+    }
+
     /// - *Contract stamp:* POC-1 §6 — a rejected channel open
     ///   surfaces on the awaiting typed call as
     ///   `russh::Error::ChannelOpenFailure(..)`; the op rows map it

src/lib.rs

@@ -20,14 +20,16 @@
 //!
 //! Architecture: `docs/architecture/` (phase 1 — ADRs 001–006 accepted;
 //! the foundations, the server stack, the server-direction mappings, and
-//! the client dial + op rows are landed through wave 4; the `-R` leg and
-//! lifecycle primitives, the validation suite, and the release are later
-//! implementation waves). See also `docs/research/`,
+//! the client dial + op rows + the `-R` acceptor leg are landed through
+//! wave 4; the lifecycle primitives, the validation suite, and the
+//! release are later implementation waves). See also `docs/research/`,
 //! `docs/plans/implementation.md`, and family ADR-003.
 
 mod adapter;
 mod channel_cap;
 mod client;
+#[cfg(feature = "tunnel")]
+mod client_accept;
 mod client_handler;
 mod client_ops;
 mod config;
@@ -49,13 +51,22 @@ mod tty_map;
 mod tunnel_map;
 
 pub use adapter::SshAdapter;
+pub use client::SshConnection;
+pub use client::dial;
 /// Contract-test surface (`#[doc(hidden)]`): the dial surface's
 /// config-mapping and handler-skeleton mechanics (the `serve.rs`
 /// §Test-surface mirror; see `client.rs` §Test surface). Not
 /// documented API.
 #[doc(hidden)]
 pub use client::russh_client_config;
-pub use client::{SshConnection, dial};
+/// The `-R` leg's acceptor registry + the reverse-listen rows'
+/// payload/acceptor types (`w4-reverse-listen` — under `tunnel`; the
+/// recorded decisions live in the `client_accept` module docs).
+#[cfg(feature = "tunnel")]
+pub use client_accept::{
+    AcceptorHeld, AcceptorRegistry, ForwardedStreamlocal, ForwardedTcpip,
+    StreamlocalForwardedAcceptor, TcpipForwardedAcceptor,
+};
 #[doc(hidden)]
 pub use client_handler::ClientHandler;
 /// The client op rows' surfaces (`w4-op-rows`):

tasks/implementation/w4-reverse-listen.md

@@ -1,7 +1,7 @@
 ---
 id: w4-reverse-listen
 name: Wave 4 — the client -R leg (forward requests + the acceptor registry)
-status: pending
+status: completed
 depends_on: [w4-op-rows]
 scope: moderate
 risk: high
@@ -65,21 +65,21 @@ registry that serves:
 
 ## Acceptance Criteria
 
-- [ ] `tcpip_forward` / `streamlocal_forward` (+ cancel forms) land
+- [x] `tcpip_forward` / `streamlocal_forward` (+ cancel forms) land
       with the actually-bound port surfaced (the port-0 mint row);
       the runtime-Unix gate compiles everywhere, declines cleanly
       off-Unix
-- [ ] The acceptor registry lands (shape decided + recorded): served
+- [x] The acceptor registry lands (shape decided + recorded): served
       only under a registered acceptor under `tunnel`;
       unregistered/unfeatured server-initiated opens declined cleanly
       (never accept-and-drop) — wire rows per build
-- [ ] Late-install works across a live connection; a pre-install
+- [x] Late-install works across a live connection; a pre-install
       arrival is declined (decided + recorded, no unclaimed buffering)
-- [ ] The data leg hands accepted opens to the assembly handler;
+- [x] The data leg hands accepted opens to the assembly handler;
       bookkeeping bounded (the cap relation decided + recorded)
-- [ ] Per-build contract rows green through the server double; the
+- [x] Per-build contract rows green through the server double; the
       smoke posture decided + recorded
-- [ ] `cargo test` (+ per-feature + `--all-features`), clippy ×2
+- [x] `cargo test` (+ per-feature + `--all-features`), clippy ×2
       (`-D warnings`), fmt, `cargo doc --no-deps` pass
 
 ## References
@@ -104,6 +104,113 @@ registry that serves:
 
 > To be filled by implementation agent
 
+Implementation decisions (also recorded in `src/client_accept.rs`'s
+module docs and `client.md` §The acceptor's landing block):
+
+- **Registry shape — per-type fixed slots, not a typed map**: the
+  client-direction registry-servable set is closed (exactly the two
+  `-R` data-leg types under `tunnel`); one acceptor trait per type
+  (`TcpipForwardedAcceptor` / `StreamlocalForwardedAcceptor`), the
+  payload an accepted open hands the assembly: the channel duplex
+  (family-typed — the `-L` rows' `TunnelChannel`, one type both legs
+  ride) + the open's echo/originator fields — the harness
+  `ForwardedAcceptor` vocabulary grown, not forked. Install is
+  **refuse-never-replace** (`AcceptorHeld` — the server-side
+  `ListenRouter::register` posture mirrored); uninstall is explicit
+  and reports honestly. A new module `src/client_accept.rs` (one
+  module per file; the rows' method bodies live beside the registry).
+- **The listen rows gate on the registered acceptor** (the
+  issuing-side disposition — a listen whose data leg cannot serve is
+  never asked for; install-first recorded order;
+  `OpError::NoAcceptor` pre-wire). The cancel forms do *not* gate on
+  the acceptor (teardown must work after an uninstall).
+- **Bound-port mapping** (recorded wire fact at this pin): russh's
+  client `tcpip_forward` reply carries a port **only for
+  `port = 0` requests** (sshd parity — an explicit request's bound
+  port *is* the requested port); `reverse_listen` maps both to the
+  actually-bound number it returns.
+- **Bounded bookkeeping — the cap relation**: two dedicated
+  `ChannelCap` counters, each against the same
+  `Limits::max_concurrent_channels` ceiling (the server-side
+  listener/ListenerCap posture mirrored): (a) the per-connection
+  **reverse-listen count** — admitted at the listen rows, released at
+  confirmed cancels and the refused/wire-error paths (never a map —
+  the outstanding listens live at the peer); (b) the **accepted-open
+  handoff count** — admitted in the dial handler's open hook before
+  the accept, released when the acceptor's `accept` call completes in
+  the spawned handoff task; a full counter declines the arrival
+  (`cap-exceeded`, event vocabulary). No accepted-open ledger: the
+  pump is the acceptor's from the handoff on (the family handler
+  shape).
+- **The runtime-Unix gate rides the unix *listen* rows only** (the
+  bind is the platform-bound op; `forwarded-streamlocal` *ingestion*
+  is platform-neutral byte pumping, hook universal compile).
+- **Pre-existing wart fixed**: `client_ops.rs`'s test mod gated
+  `family_exit_code` under a `tunnel`-inclusive cfg — the
+  `--features tunnel` build did not compile its tests before this
+  task; the import/test are now gated to the exit-consuming builds.
+
+Surprises: russh's client reply shapes (above, the port-0-only mint)
+and the `JoinHandle::drop`-is-detach-not-cancel behavior in the
+double's cancel teardown (first draft removed the entry without
+aborting the accept loop — the real listener stayed bound; the double
+aborts explicitly now).
+
 ## Summary
 
-> To be filled on completion
+> To be filled on completion
+
+Landed (`feat(client)`: w4-reverse-listen):
+
+- **`src/client_accept.rs`** — the acceptor registry
+  (`AcceptorRegistry` + the two per-type acceptor traits + the
+  `ForwardedTcpip`/`ForwardedStreamlocal` payloads + `AcceptorHeld`),
+  the reverse-listen rows (`reverse_listen` / `reverse_listen_cancel`
+  / the unix pair, with the gates + dedicated counters), and the
+  accepted-opens' handoff tasks; exports from the crate root under
+  `tunnel`.
+- **`src/client.rs`** — the dial wires the (empty) registry +
+  tunnel surface into the handler and the connection
+  (`SshConnection::acceptors` — the live-connection accessor;
+  the listener budget's private field).
+- **`src/client_handler.rs`** — the two forwarded hooks route through
+  the registry at arrival time (registered + room → accept + spawned
+  handoff; unregistered/cap-exhausted → clean `Administratively-
+  Prohibited` decline with the event vocabulary); every other type
+  keeps the decline-all skeleton; the never-feature rows unchanged.
+- **`src/error.rs`** — `OpError::NoAcceptor` / `ListenRefused` /
+  `ListenCapExceeded` + the `RequestDenied → ListenRefused` mapping
+  (the listen/cancel forms' only wire decline shape at this pin).
+- **`tests/common/sshd_double.rs`** — the fake `sshd`'s `-R` server
+  half: the four listen/cancel hooks (scripted accept/deny via
+  `DblListen`; an accept binds a **real** socket, mints the real
+  bound port for `port = 0` into the out-param, spawns the accept
+  loop — per accepted connection a server-initiated `forwarded-*`
+  open echoing the listen's args + the peer, then the bytes pumped;
+  cancels match by the raw wire strings, tearing the real listener
+  down) + the listen/cancel observations.
+- **`tests/client_reverse_listen.rs`** (`tunnel` build) — the
+  contract rows: listen + data-leg pump round trip (real
+  `ssh -R`-shaped), the port-0 mint, the no-acceptor pre-wire
+  declines (listen + open, with the event rows), the late-install
+  rows (pre-install declined → serve-after; uninstall takes it off),
+  the cancel rows (teardown observable — reconnect refused; unknown
+  → `ListenRefused`), the listener-ceiling row (exceeded pre-wire;
+  cancel recovers), the registry contract through the accessor, and
+  the streamlocal unix rows (+ the compiled off-Unix gate row).
+  Per-build matrix: the *unfeatured* builds' wire rows run in every
+  build through `tests/client_dial.rs`'s decline-all row (the
+  surface itself is the compile-time mirror).
+- **`tests/tunnel_mapping.rs`** — the `client_smoke` module: the
+  full-circle alkssh-client ↔ alkssh-server-direction smoke (both
+  `-R` legs through the alkcall fabric double over a real TCP
+  mount) — the port-0 mint through the server direction's fabric
+  listen + arrival pump into the registered acceptor's duplex, bytes
+  round-tripped end to end; real-`sshd` `-R` remains wave-5 e2e
+  (recorded, never silently dropped).
+
+Verification: `cargo test` (default) / `--features <each>` /
+`--all-features` all green; `cargo clippy --all-targets -- -D
+warnings` × builds (none / all-features) clean; `cargo fmt --check`
+clean; `cargo doc --no-deps` (+ `--all-features`; only the
+pre-existing `client_ops.rs` private-link warnings remain).

tests/client_reverse_listen.rs

@@ -0,0 +1,795 @@
+//! The `-R` leg's contract rows (`w4-reverse-listen`) — driven through
+//! the in-process fake `sshd` grown with the reverse-forward surfaces
+//! ([`sshd_double`](common::sshd_double): the `tcpip-forward` /
+//! `streamlocal-forward` hooks that bind **real** sockets and open the
+//! `forwarded-*` channels toward the client per accepted connection):
+//! the listen rows (featured + registered → served, the port-0 mint,
+//! the cancels + the listener-cap ceiling), the acceptor-registry rows
+//! (featured + unregistered → clean decline; late-install across a
+//! live connection — decline-before-install, serve-after; uninstall;
+//! the data-leg pump round trip), the off-Unix streamlocal rows
+//! (OQ-11), and the connection-stays-usable row after every decline
+//! class. The row/stamp discipline (`tests/common/mod.rs`'s doc)
+//! applies verbatim.
+//!
+//! **Per-build matrix note (recorded):** this file is the
+//! `tunnel`-featured build's rows (the only build whose surface can
+//! issue the reverse rows at all — the unfeatured build's compile-time
+//! mirror); the wire decline rows for the *unfeatured* builds run in
+//! every build through `tests/client_dial.rs`'s
+//! `server_initiated_opens_decline_cleanly_on_the_wire` (the
+//! decline-all skeleton the dial carries in a build without `tunnel`
+//! — no registry exists there, every hook declines).
+#![cfg(feature = "tunnel")]
+
+mod common;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use alkssh::{
+    AcceptorHeld, AcceptorRegistry, DialConfig, ForwardedStreamlocal, ForwardedTcpip, OpError,
+    SshConnection, StreamlocalForwardedAcceptor, TcpipForwardedAcceptor, dial,
+};
+use async_trait::async_trait;
+use common::sshd_double::{DblListen, SshdDouble, SshdObserved, SshdScript, spawn_scripted};
+use common::{MemHostKeyPolicy, TestVault, fixture};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+/// A vault-backed key source over the alpha fixture — the dial
+/// posture (the fixture PEM sealed, decrypt-at-dial).
+fn alpha_key_source() -> Arc<dyn alkssh::KeySource> {
+    let vault = TestVault::unlocked();
+    let blob = vault.seal(fixture("alpha"));
+    Arc::new(alkssh::VaultKeySource::new(vault.handle(), Some(blob)))
+}
+
+/// Dials a fresh double under the given script — one dial per row.
+async fn dial_scripted_limits(
+    script: SshdScript,
+    limits: alkssh::Limits,
+) -> (SshConnection, SshdDouble, SshdObserved) {
+    let (double, accepts, observed) =
+        spawn_scripted(vec![fixture("bravo").private_key()], true, script).await;
+    accepts.register(fixture("alpha").fingerprint);
+
+    let policy = Arc::new(MemHostKeyPolicy::with_default(true));
+    let mut config = DialConfig::new(
+        double.address.clone(),
+        "runner",
+        alpha_key_source(),
+        Arc::clone(&policy) as Arc<dyn alkssh::HostKeyPolicy>,
+    );
+    config.limits = limits;
+    let connection = dial(config).await.expect("the dial succeeds");
+    (connection, double, observed)
+}
+
+async fn dial_scripted(script: SshdScript) -> (SshConnection, SshdDouble, SshdObserved) {
+    dial_scripted_limits(script, alkssh::Limits::default()).await
+}
+
+/// The clean-close observable: the explicit close leaves the double's
+/// session `Ok` — the "the connection stays usable" final act.
+async fn expect_clean_close(connection: SshConnection, mut double: SshdDouble) {
+    assert!(!connection.is_closed());
+    connection.close().await;
+    let outcome = double.next_session_outcome().await;
+    assert!(
+        matches!(outcome, Some(Ok(()))),
+        "the clean close is the double's clean session end, got {outcome:?}"
+    );
+    double.shutdown();
+}
+
+/// A free loopback port the row's listen asks for exactly (the probe
+/// binds, then frees — a tiny race, as everywhere a fixed port is
+/// picked; the rows mostly mint dynamic ports instead).
+async fn pick_port() -> u16 {
+    let probe = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("the probe binds");
+    probe.local_addr().expect("addr").port()
+}
+
+/// The byte round trip the acceptor double's echo pump runs — the
+/// data leg's plain duplex (the bytes back through the accepted
+/// channel; the pump is the assembly's).
+async fn echo_pump(mut stream: alkssh::TunnelChannel) {
+    let mut buf = vec![0u8; 4096];
+    loop {
+        match stream.read(&mut buf).await {
+            Ok(0) | Err(_) => return,
+            Ok(count) => {
+                if stream.write_all(&buf[..count]).await.is_err() {
+                    return;
+                }
+            }
+        }
+    }
+}
+
+/// What the acceptor double records per accepted open — the payload's
+/// echo/originator fields (the duplex rides the type, never recorded).
+#[derive(Clone, Debug, PartialEq)]
+struct OpenRecord {
+    connected_address: String,
+    connected_port: u32,
+    originator_address: String,
+    originator_port: u32,
+}
+
+/// The acceptor double — records the payload's fields and, when the
+/// row asks, pumps the duplex as an echo (the family handler shape:
+/// the assembly owns the pump).
+#[derive(Clone)]
+struct AcceptorDouble {
+    opens: tokio::sync::mpsc::UnboundedSender<OpenRecord>,
+    echo: bool,
+}
+
+#[async_trait]
+impl TcpipForwardedAcceptor for AcceptorDouble {
+    async fn accept(&self, open: ForwardedTcpip) {
+        let record = OpenRecord {
+            connected_address: open.connected_address.clone(),
+            connected_port: open.connected_port,
+            originator_address: open.originator_address.clone(),
+            originator_port: open.originator_port,
+        };
+        if self.echo {
+            tokio::spawn(echo_pump(open.stream));
+        } else {
+            drop(open.stream);
+        }
+        let _ = self.opens.send(record);
+    }
+}
+
+/// The streamlocal acceptor double — the unix slot's shape.
+#[derive(Clone)]
+struct UnixAcceptorDouble {
+    opens: tokio::sync::mpsc::UnboundedSender<String>,
+    echo: bool,
+}
+
+#[async_trait]
+impl StreamlocalForwardedAcceptor for UnixAcceptorDouble {
+    async fn accept(&self, open: ForwardedStreamlocal) {
+        if self.echo {
+            tokio::spawn(echo_pump(open.stream));
+        } else {
+            drop(open.stream);
+        }
+        let _ = self.opens.send(open.socket_path);
+    }
+}
+
+/// The next accepted open's fields — the row's payload observation
+/// (bounded await: a wrong row must not hang).
+async fn next_open(
+    opens: &mut tokio::sync::mpsc::UnboundedReceiver<OpenRecord>,
+    context: &str,
+) -> OpenRecord {
+    tokio::time::timeout(Duration::from_secs(15), opens.recv())
+        .await
+        .unwrap_or_else(|_| panic!("{context}: the accepted open hands off in time"))
+        .expect("{context}: the acceptor sender lives")
+}
+
+/// - *Contract stamp:* `client.md` §op rows row 4 (the `-R` leg's
+///   listen + ingest row) + the module docs' listen-gate record —
+///   featured + registered → served: the
+///   `reverse_listen` request rides the wire (the double's
+///   `tcpip-forward` hook observes it verbatim), the actually-bound
+///   port returns, and the **data leg** round-trips: a connection to
+///   the bound port's bytes flow through the double's accept loop →
+///   the server-initiated `forwarded-tcpip` open → the registered
+///   acceptor's duplex → the echo pump → back, verbatim. The payload
+///   fields are the wire's echo shapes (the connected address/port ==
+///   the listen request's, the originator the arrival's peer).
+#[tokio::test]
+async fn reverse_listen_serves_and_the_data_leg_pumps() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, double, observed) = dial_scripted(scripted).await;
+
+    let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel();
+    let registry: &AcceptorRegistry = connection.acceptors();
+    registry
+        .install_tcpip(Arc::new(AcceptorDouble {
+            opens: opens_tx,
+            echo: true,
+        }))
+        .expect("the slot installs");
+
+    let port = u32::from(pick_port().await);
+    let bound = connection
+        .reverse_listen("127.0.0.1", port)
+        .await
+        .expect("the featured + registered listen serves");
+    assert_eq!(bound, port, "an explicit port binds exactly that port");
+
+    let mut conn =
+        tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range")))
+            .await
+            .expect("the double's real listener accepts");
+    conn.write_all(b"leg-data")
+        .await
+        .expect("the forward's ingress carries the bytes");
+    let mut echoed = [0u8; 8];
+    tokio::time::timeout(Duration::from_secs(15), conn.read_exact(&mut echoed))
+        .await
+        .expect("the round trip completes in time")
+        .expect("the echo reads");
+    assert_eq!(&echoed, b"leg-data", "the data leg round-trips verbatim");
+
+    let record = next_open(&mut opens_rx, "the data-leg row").await;
+    assert_eq!(record.connected_address, "127.0.0.1".to_string());
+    assert_eq!(record.connected_port, port);
+    assert_eq!(record.originator_address, "127.0.0.1".to_string());
+    assert!(
+        record.originator_port > 0,
+        "the originator is the arrival's ephemeral port"
+    );
+
+    assert_eq!(
+        observed.listens(),
+        vec![("tcpip-forward", "127.0.0.1".to_string(), port)],
+        "the listen request rode the wire verbatim"
+    );
+
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* `client.md` §Lifecycle primitives (the
+///   bound-port spec) — `port = 0` mint: the peer picks and the
+///   surfaced number is the actually-bound one (the row connects to
+///   it; the double's listener is the real socket, so the mint is the
+///   real shape sshd's `get_local_port` parity rides).
+#[tokio::test]
+async fn port_zero_mint_surfaces_the_actually_bound_port() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, _double, _observed) = dial_scripted(scripted).await;
+
+    let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel();
+    connection
+        .acceptors()
+        .install_tcpip(Arc::new(AcceptorDouble {
+            opens: opens_tx,
+            echo: false,
+        }))
+        .expect("the slot installs");
+
+    let bound = connection
+        .reverse_listen("127.0.0.1", 0)
+        .await
+        .expect("the mint serves");
+    assert!(
+        bound > 0,
+        "the served mint is the actually-bound port, never the requested 0"
+    );
+
+    drop(
+        tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(bound).expect("in-range")))
+            .await
+            .expect("the surfaced number is a really listening port"),
+    );
+
+    let record = next_open(&mut opens_rx, "the mint row").await;
+    assert_eq!(record.connected_address, "127.0.0.1");
+    assert_eq!(
+        record.connected_port, bound,
+        "the open echoes the bound port"
+    );
+
+    let _ = connection.close().await;
+}
+
+/// - *Contract stamp:* ADR-003 §4's issuing-side mirror + the module
+///   docs' listen-gate record — a reverse listen with no registered
+///   acceptor declines **before any wire op** (`NoAcceptor`; the
+///   half-capability posture: a bound-but-unserveable listener is
+///   never asked for), and the connection stays usable.
+#[tokio::test]
+async fn reverse_listen_without_an_acceptor_declines_before_any_wire_op() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, double, observed) = dial_scripted(scripted).await;
+
+    let result = connection
+        .reverse_listen("127.0.0.1", u32::from(pick_port().await))
+        .await;
+    assert!(
+        matches!(result, Err(OpError::NoAcceptor)),
+        "the unregistered class answers the listen, got {result:?}"
+    );
+    assert!(
+        observed.listens().is_empty(),
+        "nothing rode the wire — the gate answered before the request"
+    );
+
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* ADR-003 §4 — featured + unregistered → clean
+///   decline: a server-initiated `forwarded-tcpip` open on a
+///   live connection with an empty registry answers
+///   `AdministrativelyProhibited` (never russh's accept-and-drop
+///   default, never a buffered open), the decline event rides the
+///   crate vocabulary, and the connection stays usable.
+#[tokio::test]
+async fn unregistered_forwarded_open_declines_cleanly() {
+    common::event_capture::install();
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, mut double, _observed) = dial_scripted(scripted).await;
+
+    let server = double.session_handle().await;
+    let outcome = tokio::time::timeout(
+        Duration::from_secs(15),
+        server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0),
+    )
+    .await
+    .expect("the open resolves in time");
+    assert!(
+        matches!(
+            outcome,
+            Err(russh::Error::ChannelOpenFailure(
+                russh::ChannelOpenFailure::AdministrativelyProhibited
+            ))
+        ),
+        "the unregistered open declines cleanly, got {outcome:?}"
+    );
+
+    let decline = common::event_capture::await_event(&[
+        ("row", "forwarded-tcpip"),
+        ("decline_class", "open"),
+        ("reason", "no-acceptor"),
+    ])
+    .await;
+    assert_eq!(decline.level, "DEBUG");
+    assert_eq!(
+        decline.field("feature_build"),
+        Some(common::event_capture::expected_build())
+    );
+
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* the late-install dynamic (reverse-runner fact
+///   2 — consumers turn capabilities on/off per connection) + the
+///   arrival-time gate — the same live connection: the pre-install
+///   arrival declines, the post-install arrival serves (never
+///   buffered, never half-served), and the acceptor double's late
+///   install is the registry's whole surface.
+#[tokio::test]
+async fn late_install_serves_after_a_pre_install_arrival_declined() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, mut double, _observed) = dial_scripted(scripted).await;
+
+    let server = double.session_handle().await;
+    let pre = tokio::time::timeout(
+        Duration::from_secs(15),
+        server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0),
+    )
+    .await
+    .expect("the pre-install open resolves in time");
+    assert!(
+        matches!(
+            pre,
+            Err(russh::Error::ChannelOpenFailure(
+                russh::ChannelOpenFailure::AdministrativelyProhibited
+            ))
+        ),
+        "the pre-install arrival declines at arrival time, got {pre:?}"
+    );
+
+    let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel();
+    connection
+        .acceptors()
+        .install_tcpip(Arc::new(AcceptorDouble {
+            opens: opens_tx,
+            echo: false,
+        }))
+        .expect("the late install takes");
+
+    let post = tokio::time::timeout(
+        Duration::from_secs(15),
+        server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0),
+    )
+    .await
+    .expect("the post-install open resolves in time");
+    assert!(
+        post.is_ok(),
+        "the post-install arrival serves, got {post:?}"
+    );
+    let record = next_open(&mut opens_rx, "the late-install row").await;
+    assert_eq!(
+        record,
+        OpenRecord {
+            connected_address: "hub.example".into(),
+            connected_port: 4096,
+            originator_address: "originator".into(),
+            originator_port: 0,
+        }
+    );
+
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* the late-install toggle's off half — the
+///   uninstall takes the capability off mid-connection: the
+///   post-uninstall arrival declines at arrival time (fail-closed;
+///   the already-accepted open's channel is the acceptor's business).
+#[tokio::test]
+async fn uninstall_takes_the_capability_off_before_a_later_arrival() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, mut double, _observed) = dial_scripted(scripted).await;
+
+    let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel();
+    connection
+        .acceptors()
+        .install_tcpip(Arc::new(AcceptorDouble {
+            opens: opens_tx,
+            echo: false,
+        }))
+        .expect("the slot installs");
+
+    let server = double.session_handle().await;
+    let served = tokio::time::timeout(
+        Duration::from_secs(15),
+        server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0),
+    )
+    .await
+    .expect("the served open resolves in time");
+    assert!(served.is_ok(), "the installed slot serves, got {served:?}");
+    let record = next_open(&mut opens_rx, "the uninstall row").await;
+    assert_eq!(record.connected_address, "hub.example");
+
+    assert!(
+        connection.acceptors().uninstall_tcpip(),
+        "the uninstall reports the held slot"
+    );
+
+    let post = tokio::time::timeout(
+        Duration::from_secs(15),
+        server.channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0),
+    )
+    .await
+    .expect("the post-uninstall open resolves in time");
+    assert!(
+        matches!(
+            post,
+            Err(russh::Error::ChannelOpenFailure(
+                russh::ChannelOpenFailure::AdministrativelyProhibited
+            ))
+        ),
+        "the post-uninstall arrival declines, got {post:?}"
+    );
+
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* `client.md` §Lifecycle primitives (every
+///   listener explicitly cancellable) + the double's sshd parity —
+///   the cancel tears the real listener down (the bound port
+///   reconnects as refused), a cancel of an unknown listener is the
+///   peer's refusal (`ListenRefused`), and the connection stays
+///   usable after every cancel class.
+#[tokio::test]
+async fn cancel_tears_the_forward_down_and_unknown_cancels_refuse() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, double, observed) = dial_scripted(scripted).await;
+
+    let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel();
+    connection
+        .acceptors()
+        .install_tcpip(Arc::new(AcceptorDouble {
+            opens: opens_tx,
+            echo: false,
+        }))
+        .expect("the slot installs");
+
+    let port = u32::from(pick_port().await);
+    connection
+        .reverse_listen("127.0.0.1", port)
+        .await
+        .expect("the listen binds");
+
+    connection
+        .reverse_listen_cancel("127.0.0.1", port)
+        .await
+        .expect("the confirmed cancel");
+    assert_eq!(
+        observed.cancels(),
+        vec![("cancel-tcpip-forward", "127.0.0.1".to_string(), port)],
+        "the cancel rode the wire verbatim"
+    );
+
+    let mut refused = false;
+    for _ in 0..200 {
+        let reconnect =
+            tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(port).expect("in-range")))
+                .await;
+        match reconnect {
+            Ok(_) => {
+                tokio::time::sleep(Duration::from_millis(25)).await;
+            }
+            Err(err) if err.kind() == std::io::ErrorKind::ConnectionRefused => {
+                refused = true;
+                break;
+            }
+            Err(err) => panic!("the bound port errors, not refuses, after the cancel: {err:?}"),
+        }
+    }
+    assert!(
+        refused,
+        "the forward is torn down — the bound port refuses a reconnect"
+    );
+
+    let again = connection.reverse_listen_cancel("127.0.0.1", port).await;
+    assert!(
+        matches!(again, Err(OpError::ListenRefused)),
+        "the unknown listener's cancel refuses, got {again:?}"
+    );
+    assert!(
+        matches!(
+            connection
+                .reverse_listen_cancel("never-listened", port)
+                .await,
+            Err(OpError::ListenRefused)
+        ),
+        "no match by endpoint either — the raw-string parity holds"
+    );
+
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* the module docs' recorded cap relation (the
+///   server-side listener-counter posture mirrored) — the dedicated
+///   listener count against the `max_concurrent_channels` ceiling:
+///   beyond it the listen declines **before any wire op**
+///   (`ListenCapExceeded`), and a confirmed cancel recovers the
+///   budget.
+#[tokio::test]
+async fn the_listener_ceiling_bounds_the_listen_map_and_recovers_on_cancel() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let limits = alkssh::Limits {
+        max_concurrent_channels: 1,
+        ..alkssh::Limits::default()
+    };
+    let (connection, double, observed) = dial_scripted_limits(scripted, limits).await;
+
+    let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel();
+    connection
+        .acceptors()
+        .install_tcpip(Arc::new(AcceptorDouble {
+            opens: opens_tx,
+            echo: false,
+        }))
+        .expect("the slot installs");
+
+    let first = u32::from(pick_port().await);
+    connection
+        .reverse_listen("127.0.0.1", first)
+        .await
+        .expect("the first listen admits");
+
+    let second = connection
+        .reverse_listen("127.0.0.1", u32::from(pick_port().await))
+        .await;
+    assert!(
+        matches!(second, Err(OpError::ListenCapExceeded)),
+        "beyond the ceiling the listen refuses pre-wire, got {second:?}"
+    );
+    assert_eq!(
+        observed.listens().len(),
+        1,
+        "the ceiling refusal consumed nothing on the wire"
+    );
+
+    connection
+        .reverse_listen_cancel("127.0.0.1", first)
+        .await
+        .expect("the cancel recovers the budget");
+
+    let third = u32::from(pick_port().await);
+    connection
+        .reverse_listen("127.0.0.1", third)
+        .await
+        .expect("the count admits again after the cancel");
+    assert_eq!(
+        observed.listens().len(),
+        2,
+        "the recovered budget's listen rode the wire"
+    );
+
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* the module docs' refuse-never-replace posture
+///   through the connection accessor — the second install is declined
+///   (`AcceptorHeld`), the first acceptor stays served, and the
+///   uninstall/uninstall-again cycle reports honestly.
+#[tokio::test]
+async fn the_connection_accessor_serves_the_registry_contract() {
+    let scripted = SshdScript {
+        tcp_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, _double, _observed) = dial_scripted(scripted).await;
+
+    let (opens_tx, _opens_rx) = tokio::sync::mpsc::unbounded_channel();
+    let registry = connection.acceptors();
+    registry
+        .install_tcpip(Arc::new(AcceptorDouble {
+            opens: opens_tx,
+            echo: false,
+        }))
+        .expect("the first install takes");
+    assert!(
+        matches!(
+            registry.install_tcpip(Arc::new(AcceptorDouble {
+                opens: tokio::sync::mpsc::unbounded_channel().0,
+                echo: false,
+            })),
+            Err(AcceptorHeld)
+        ),
+        "the held slot refuses the second install"
+    );
+    assert!(registry.uninstall_tcpip());
+    assert!(!registry.uninstall_tcpip(), "the honest `false` report");
+
+    let _ = connection.close().await;
+}
+
+/// The unix `-R` rows (the runtime-gate's on-Unix leg's shape): the
+/// listen + the data-leg round trip + the payload's echoed path + the
+/// cancel's wire-honest teardown + the unknown cancel's refusal.
+///
+/// - *Contract stamp:* OQ-11's client mirror (the
+///   `streamlocal_forward` rows compile under `tunnel`, serve on
+///   unix) + `mapping.md` §Global-request rows (the unix listen legs)
+///   + the acceptor registry's streamlocal slot (featured +
+///   registered → served).
+#[cfg(unix)]
+#[tokio::test]
+async fn streamlocal_listen_serves_arrivals_and_cancels_through_the_unix_gate() {
+    let scripted = SshdScript {
+        unix_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, double, observed) = dial_scripted(scripted).await;
+
+    let (ops_tx, mut ops_rx) = tokio::sync::mpsc::unbounded_channel();
+    let registry = connection.acceptors();
+    registry
+        .install_streamlocal(Arc::new(UnixAcceptorDouble {
+            opens: ops_tx,
+            echo: true,
+        }))
+        .expect("the streamlocal slot installs");
+
+    let socket_path = std::env::temp_dir()
+        .join(format!("alkssh-w4-rl-{}", std::process::id()))
+        .to_path_buf();
+    let socket_path = socket_path.to_string_lossy().to_string();
+    let _ = std::fs::remove_file(&socket_path);
+
+    connection
+        .reverse_listen_unix(&socket_path)
+        .await
+        .expect("the unix listen serves");
+    assert_eq!(
+        observed.listens(),
+        vec![("streamlocal-forward", socket_path.clone(), 0)],
+        "the unix listen rode the wire verbatim"
+    );
+
+    let mut conn = tokio::net::UnixStream::connect(&socket_path)
+        .await
+        .expect("the double's real listener accepts");
+    conn.write_all(b"unix-data")
+        .await
+        .expect("the unix forward's ingress carries the bytes");
+    let mut echoed = [0u8; 9];
+    tokio::time::timeout(Duration::from_secs(15), conn.read_exact(&mut echoed))
+        .await
+        .expect("the unix round trip completes in time")
+        .expect("the echo reads");
+    assert_eq!(
+        &echoed, b"unix-data",
+        "the unix data leg round-trips verbatim"
+    );
+
+    let path = tokio::time::timeout(Duration::from_secs(15), ops_rx.recv())
+        .await
+        .expect("the streamlocal handoff arrives in time")
+        .expect("the acceptor sender lives");
+    assert_eq!(
+        path, socket_path,
+        "the open echoes the listen's socket path verbatim"
+    );
+
+    connection
+        .reverse_listen_unix_cancel(&socket_path)
+        .await
+        .expect("the unix cancel confirms");
+    assert_eq!(
+        observed.cancels(),
+        vec![("cancel-streamlocal-forward", socket_path.clone(), 0)],
+        "the unix cancel rode the wire verbatim"
+    );
+
+    let reconnect = tokio::net::UnixStream::connect(&socket_path).await;
+    assert!(
+        reconnect.is_err(),
+        "the unix forward is torn down — the socket path refuses a reconnect, got {reconnect:?}"
+    );
+    assert!(
+        matches!(
+            connection.reverse_listen_unix_cancel(&socket_path).await,
+            Err(OpError::ListenRefused)
+        ),
+        "the unknown unix listener's cancel refuses"
+    );
+
+    let _ = std::fs::remove_file(&socket_path);
+    expect_clean_close(connection, double).await;
+}
+
+/// - *Contract stamp:* OQ-11 (the runtime-Unix gate; universal
+///   compile, clean off-Unix decline) + the module docs' gate record —
+///   off-Unix the unix listen declines **before any wire op**
+///   (`UnixUnavailable`; the platform-refused request consumes
+///   nothing).
+#[cfg(not(unix))]
+#[tokio::test]
+async fn streamlocal_declines_off_unix_before_any_wire_op() {
+    let scripted = SshdScript {
+        unix_listen: DblListen::Accept,
+        ..SshdScript::default()
+    };
+    let (connection, double, observed) = dial_scripted(scripted).await;
+
+    connection
+        .acceptors()
+        .install_streamlocal(Arc::new(UnixAcceptorDouble {
+            opens: tokio::sync::mpsc::unbounded_channel().0,
+            echo: false,
+        }))
+        .expect("the slot installs");
+    let result = connection.reverse_listen_unix("alkssh-row.sock").await;
+    assert!(
+        matches!(result, Err(OpError::UnixUnavailable)),
+        "the off-Unix gate answers the unix listen, got {result:?}"
+    );
+    assert!(
+        observed.listens().is_empty(),
+        "nothing rode the wire — the gate answered before the request"
+    );
+
+    expect_clean_close(connection, double).await;
+}

tests/common/sshd_double.rs

@@ -56,6 +56,24 @@
 //!   changes/signals the client rows assert the wire verbatim
 //!   against). The double stays vanilla-parity: a hand-rolled
 //!   responder standing in for `sshd`, nothing alkssh-typed inside.
+//!
+//! - **The `-R` server half** (`w4-reverse-listen`): the double grows
+//!   the reverse-forward surfaces a vanilla `sshd` really serves — the
+//!   `tcpip-forward` / `streamlocal-forward` global-request hooks
+//!   (scripted accept/deny; an accept **binds a real socket** in
+//!   sshd's role, so a `port = 0` request's minted bound port is the
+//!   socket's real one and the data leg is a genuine
+//!   `ssh -R`-shaped round trip), the accept loop per bound listener
+//!   (per accepted connection: a server-initiated `forwarded-*` open
+//!   toward the client echoing the listen's own args + the
+//!   connection's peer as the originator, then the bytes pumped
+//!   bidirectionally; a client-declined open closes the accepted
+//!   connection — vanilla parity), and the cancel forms removing the
+//!   matching entry (tearing the real listener down; an unknown
+//!   listener denies, sshd parity). The bookkeeping is per-connection
+//!   handler state and bounded by the scripted listens alone — the
+//!   double is a scripted stand-in, not a real server: at most one
+//!   forward per scripted accept is what the rows drive.
 
 #![allow(dead_code)]
 
@@ -89,6 +107,27 @@ pub enum DblDirect {
     Echo,
 }
 
+/// The double's scripted behavior for the client's reverse-forward
+/// listen requests (`tcpip-forward` / `streamlocal-forward@…) — the
+/// `-R` rows' scripted class (`w4-reverse-listen`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum DblListen {
+    /// Deny the listen — the vanilla sshd's refusal (the scripted
+    /// server-refusal class; the wire `RequestFailure` the client's
+    /// op row mirrors). The script's default (fail-closed double).
+    #[default]
+    Deny,
+    /// Accept the listen: the double binds a **real socket** in
+    /// sshd's role (the actually-bound port rides the reply's mint —
+    /// the port-0 row's real shape), records it for the cancel forms,
+    /// and spawns the accept loop (per accepted connection: a
+    /// server-initiated `forwarded-*` open toward the client echoing
+    /// the listen's own args + the connection's peer, then the bytes
+    /// pumped bidirectionally; a client-declined open closes that
+    /// accepted connection — vanilla parity).
+    Accept,
+}
+
 /// The fake `sshd`'s op-row script (the `w4-op-rows` request
 /// surfaces) — what the double answers and, on a success, what the
 /// scripted "program" produces.
@@ -110,6 +149,11 @@ pub struct SshdScript {
     pub direct: DblDirect,
     /// The `direct-streamlocal` opens' behavior.
     pub streamlocal: DblDirect,
+    /// The `tcpip-forward` listen form's behavior (`tcp -R` rows).
+    pub tcp_listen: DblListen,
+    /// The `streamlocal-forward` listen form's behavior (the unix
+    /// `-R` rows).
+    pub unix_listen: DblListen,
     /// The counted refusal — the double declines the *first N* client
     /// session opens (per connection, armed by
     /// `accept_session_channels`): the stays-usable-after-a-decline
@@ -121,7 +165,8 @@ pub struct SshdScript {
 impl Default for SshdScript {
     /// The plain dial rows' posture: session opens accepted (the
     /// `accept_session_channels` flag governs), requests succeed with
-    /// no outputs and no exit, direct opens rejected.
+    /// no outputs and no exit, direct opens rejected, listen forms
+    /// denied (the scripted fail-closed default).
     fn default() -> Self {
         Self {
             request_success: true,
@@ -130,11 +175,18 @@ impl Default for SshdScript {
             exit: None,
             direct: DblDirect::Reject,
             streamlocal: DblDirect::Reject,
+            tcp_listen: DblListen::Deny,
+            unix_listen: DblListen::Deny,
             reject_first_session_opens: 0,
         }
     }
 }
 
+/// One recorded global-request listen/cancel form: (the request
+/// name, the wire endpoint string, the wire port) — the reverse-listen
+/// rows assert the listens the client ops put on the wire.
+pub type ListenRecord = (&'static str, String, u32);
+
 /// The double's per-connection observations — the wire rows assert
 /// the verbatim translations against these (what the client's op
 /// rows put on the wire).
@@ -145,6 +197,8 @@ pub struct SshdObserved {
     shells: Arc<Mutex<Vec<()>>>,
     windows: Arc<Mutex<Vec<WindowRecord>>>,
     signals: Arc<Mutex<Vec<String>>>,
+    listens: Arc<Mutex<Vec<ListenRecord>>>,
+    cancels: Arc<Mutex<Vec<ListenRecord>>>,
 }
 
 /// One recorded pty request's translated shapes — (term, cols, rows,
@@ -185,6 +239,17 @@ impl SshdObserved {
     pub fn signals(&self) -> Vec<String> {
         self.signals.lock().clone()
     }
+
+    /// The global-request listen forms seen (name, endpoint, port —
+    /// the requested port, never the mint).
+    pub fn listens(&self) -> Vec<ListenRecord> {
+        self.listens.lock().clone()
+    }
+
+    /// The global-request cancel forms seen (name, endpoint, port).
+    pub fn cancels(&self) -> Vec<ListenRecord> {
+        self.cancels.lock().clone()
+    }
 }
 
 /// The double's scripted program serve — after a request's success
@@ -244,6 +309,81 @@ async fn direct_echo(mut stream: russh::ChannelStream<Msg>) {
     }
 }
 
+/// The `-R` accept loop per bound listener (the double's sshd role):
+/// per accepted connection a server-initiated `forwarded-*` open
+/// toward the client — the connected address echoes the listen
+/// request's wire string verbatim, the port the actually-bound one,
+/// the originator the accepted connection's peer — then the bytes
+/// pumped bidirectionally between the SSH channel and the accepted
+/// socket. A client-declined open closes that accepted connection
+/// (vanilla parity: the listener stays, the data has nowhere to go).
+/// The loop ends with the listener itself (cancel tears it down; the
+/// session's end fails the opens' sends).
+async fn accept_forward_loop(
+    handle: Handle,
+    listener: tokio::net::TcpListener,
+    connected_address: String,
+    connected_port: u32,
+) {
+    loop {
+        let Ok((conn, peer)) = listener.accept().await else {
+            return;
+        };
+        let opened = handle
+            .channel_open_forwarded_tcpip(
+                connected_address.clone(),
+                connected_port,
+                peer.ip().to_string(),
+                u32::from(peer.port()),
+            )
+            .await;
+        match opened {
+            Ok(channel) => {
+                tokio::spawn(forward_connection(channel, conn));
+            }
+            Err(_) => drop(conn),
+        }
+    }
+}
+
+/// The accepted forward connection's pump (its SSH channel ↔ its
+/// socket) — the data leg's byte plane.
+async fn forward_connection(channel: Channel<Msg>, mut conn: tokio::net::TcpStream) {
+    let mut stream = channel.into_stream();
+    let _ = tokio::io::copy_bidirectional(&mut stream, &mut conn).await;
+}
+
+/// The unix accept loop's twin (the unix `-R` rows' shape): per
+/// accepted socket a `forwarded-streamlocal` open toward the client
+/// echoing the listen path.
+#[cfg(unix)]
+async fn accept_unix_forward_loop(
+    handle: Handle,
+    listener: tokio::net::UnixListener,
+    socket_path: String,
+) {
+    loop {
+        let Ok((conn, _)) = listener.accept().await else {
+            return;
+        };
+        let opened = handle
+            .channel_open_forwarded_streamlocal(socket_path.clone())
+            .await;
+        match opened {
+            Ok(channel) => {
+                tokio::spawn(unix_forward_connection(channel, conn));
+            }
+            Err(_) => drop(conn),
+        }
+    }
+}
+
+#[cfg(unix)]
+async fn unix_forward_connection(channel: Channel<Msg>, mut conn: tokio::net::UnixStream) {
+    let mut stream = channel.into_stream();
+    let _ = tokio::io::copy_bidirectional(&mut stream, &mut conn).await;
+}
+
 /// The fake `sshd`'s per-connection handler — accept-set identity
 /// resolution, explicit rejects everywhere, the `auth_succeeded`
 /// handle carrier and the session-open acceptor stub.
@@ -255,6 +395,13 @@ struct SshdDoubleHandler {
     script: Arc<SshdScript>,
     observed: SshdObserved,
     remaining_declines: std::sync::Arc<std::sync::atomic::AtomicUsize>,
+    /// The `-R` bookkeeping (per-connection, sshd's listener table):
+    /// the key is the cancel-form's match key (the wire endpoint + the
+    /// bound port — a `port = 0` mint's cancel matches the bound one),
+    /// the value the live accept loop (cancel aborts it, dropping the
+    /// real listener with it).
+    tcp_forwards: std::collections::HashMap<(String, u32), tokio::task::JoinHandle<()>>,
+    unix_forwards: std::collections::HashMap<String, tokio::task::JoinHandle<()>>,
 }
 
 impl russh::server::Handler for SshdDoubleHandler {
@@ -485,6 +632,123 @@ impl russh::server::Handler for SshdDoubleHandler {
         self.observed.signals.lock().push(format!("{signal:?}"));
         Ok(())
     }
+
+    /// The `tcpip-forward` listen form: recorded, scripted — an accept
+    /// binds a real `127.0.0.1` listener in sshd's role (a `port = 0`
+    /// request's mint is the socket's real bound port, the out-param
+    /// parity), keys the listener by (wire address, bound port), and
+    /// spawns the accept loop ([`accept_forward_loop`]).
+    async fn tcpip_forward(
+        &mut self,
+        address: &str,
+        port: &mut u32,
+        session: &mut Session,
+    ) -> Result<bool, Self::Error> {
+        self.observed
+            .listens
+            .lock()
+            .push(("tcpip-forward", address.to_string(), *port));
+        if self.script.tcp_listen == DblListen::Deny {
+            return Ok(false);
+        }
+        let bind_target = format!("127.0.0.1:{}", *port);
+        let Ok(listener) = tokio::net::TcpListener::bind(&bind_target).await else {
+            return Ok(false);
+        };
+        let bound = listener
+            .local_addr()
+            .expect("the bound listener has an address")
+            .port();
+        let bound = u32::from(bound);
+        if *port == 0 {
+            *port = bound;
+        }
+        let task = tokio::spawn(accept_forward_loop(
+            session.handle(),
+            listener,
+            address.to_string(),
+            bound,
+        ));
+        self.tcp_forwards.insert((address.to_string(), bound), task);
+        Ok(true)
+    }
+
+    /// The `cancel-tcpip-forward` form: matches by the raw wire
+    /// strings (the bound port for a mint); a hit removes the
+    /// listener (tearing the real socket down — the cancel row's
+    /// observable), an unknown listener denies (sshd parity).
+    async fn cancel_tcpip_forward(
+        &mut self,
+        address: &str,
+        port: u32,
+        _session: &mut Session,
+    ) -> Result<bool, Self::Error> {
+        self.observed
+            .cancels
+            .lock()
+            .push(("cancel-tcpip-forward", address.to_string(), port));
+        match self.tcp_forwards.remove(&(address.to_string(), port)) {
+            Some(listener_task) => {
+                listener_task.abort();
+                Ok(true)
+            }
+            None => Ok(false),
+        }
+    }
+
+    /// The `streamlocal-forward` listen form (the unix `-R`): records,
+    /// scripts, binds the real unix socket (stale file unlinked
+    /// first), and spawns the unix accept loop. Off-Unix the hook is
+    /// not implemented and falls to the trait's deny default (the
+    /// vanilla Windows sshd parity) — the crate's own runtime-Unix
+    /// gate mirrors the refusal client-side before the wire.
+    #[cfg(unix)]
+    async fn streamlocal_forward(
+        &mut self,
+        socket_path: &str,
+        session: &mut Session,
+    ) -> Result<bool, Self::Error> {
+        self.observed
+            .listens
+            .lock()
+            .push(("streamlocal-forward", socket_path.to_string(), 0));
+        if self.script.unix_listen == DblListen::Deny {
+            return Ok(false);
+        }
+        let _ = std::fs::remove_file(socket_path);
+        let Ok(listener) = tokio::net::UnixListener::bind(socket_path) else {
+            return Ok(false);
+        };
+        let task = tokio::spawn(accept_unix_forward_loop(
+            session.handle(),
+            listener,
+            socket_path.to_string(),
+        ));
+        self.unix_forwards.insert(socket_path.to_string(), task);
+        Ok(true)
+    }
+
+    /// The `cancel-streamlocal-forward` form (matches by the path
+    /// alone — the unix cancel's sshd parity).
+    #[cfg(unix)]
+    async fn cancel_streamlocal_forward(
+        &mut self,
+        socket_path: &str,
+        _session: &mut Session,
+    ) -> Result<bool, Self::Error> {
+        self.observed.cancels.lock().push((
+            "cancel-streamlocal-forward",
+            socket_path.to_string(),
+            0,
+        ));
+        match self.unix_forwards.remove(socket_path) {
+            Some(listener_task) => {
+                listener_task.abort();
+                Ok(true)
+            }
+            None => Ok(false),
+        }
+    }
 }
 
 impl SshdDoubleHandler {
@@ -589,6 +853,8 @@ pub async fn spawn_scripted(
                 remaining_declines: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(
                     script_declines,
                 )),
+                tcp_forwards: std::collections::HashMap::new(),
+                unix_forwards: std::collections::HashMap::new(),
             };
             let outcomes = outcomes_tx.clone();
             tokio::spawn(async move {

tests/tunnel_mapping.rs

@@ -1894,6 +1894,166 @@ mod serving {
     }
 }
 
+#[cfg(feature = "tunnel")]
+mod client_smoke {
+    //! The full-circle smoke (`w4-reverse-listen`'s recorded posture):
+    //! the **alkssh client's** `-R` rows against the **alkssh server
+    //! direction's** `-R` rows — both directions of the crate in one
+    //! live wire, the route-1 bridge's two legs meeting through the
+    //! alkcall fabric double (the composition the alkforge assembly
+    //! makes). A connection to the minted bound port round-trips
+    //! through: the fabric's real TCP listener → the fabric's
+    //! bind-first accept loop → the door-side `tunnel/forwarded`
+    //! arrival → the door's server-initiated `forwarded-tcpip` open
+    //! (echo args + originator) → the alkssh client's acceptor registry
+    //! → the acceptor duplex → the echo back. Real-`sshd` `-R` is
+    //! wave-5 e2e territory (never silently dropped).
+
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use async_trait::async_trait;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    use alkssh::{
+        AcceptorRegistry, DialConfig, ForwardedTcpip, TcpipForwardedAcceptor, TunnelListenSeam,
+        dial,
+    };
+
+    use super::common::harness::serve_config as harness_serve_config;
+    use super::common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
+    use super::serving::wire_fabric_with_listen;
+
+    struct EchoAcceptor {
+        opens: tokio::sync::mpsc::UnboundedSender<(String, u32, String)>,
+    }
+
+    #[async_trait]
+    impl TcpipForwardedAcceptor for EchoAcceptor {
+        async fn accept(&self, open: ForwardedTcpip) {
+            let _ = self.opens.send((
+                open.connected_address.clone(),
+                open.connected_port,
+                open.originator_address.clone(),
+            ));
+            tokio::spawn(async move {
+                let mut stream = open.stream;
+                let mut buf = vec![0u8; 4096];
+                loop {
+                    match stream.read(&mut buf).await {
+                        Ok(0) | Err(_) => return,
+                        Ok(count) => {
+                            if stream.write_all(&buf[..count]).await.is_err() {
+                                return;
+                            }
+                        }
+                    }
+                }
+            });
+        }
+    }
+
+    #[tokio::test]
+    async fn smoke_client_reverse_listen_full_circle_through_the_server_direction() {
+        let double = wire_fabric_with_listen(true, true).await;
+
+        let provider = Arc::new(MemIdentityProvider::new());
+        provider.register(
+            fixture("alpha").fingerprint,
+            identity("reverse-runner", &["ssh"]),
+        );
+        let mut serve = harness_serve_config(provider, &[fixture("bravo")], Default::default());
+        serve.channel_client = Some(Arc::clone(&double.client));
+        serve.tunnel_listen_seam = Some(TunnelListenSeam::new(Arc::clone(&double.registry)));
+        let adapter = alkssh::SshAdapter::new(serve)
+            .await
+            .expect("the -R wired adapter constructs");
+
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("the smoke's loopback listener binds");
+        let addr = listener.local_addr().expect("the smoke has an address");
+        tokio::spawn(async move {
+            loop {
+                let Ok((socket, _)) = listener.accept().await else {
+                    break;
+                };
+                if adapter.handle(socket).await.is_err() {
+                    break;
+                }
+            }
+        });
+
+        let policy = Arc::new(MemHostKeyPolicy::new());
+        policy.pin("127.0.0.1", fixture("bravo").fingerprint);
+        let config = DialConfig::new(
+            addr.to_string(),
+            "runner",
+            crate_test_key_source(),
+            Arc::clone(&policy) as Arc<dyn alkssh::HostKeyPolicy>,
+        );
+        let connection = dial(config).await.expect("the dial succeeds");
+
+        let (opens_tx, mut opens_rx) = tokio::sync::mpsc::unbounded_channel();
+        let registry: &AcceptorRegistry = connection.acceptors();
+        registry
+            .install_tcpip(Arc::new(EchoAcceptor { opens: opens_tx }))
+            .expect("the acceptor installs");
+
+        let bound = connection
+            .reverse_listen("127.0.0.1", 0)
+            .await
+            .expect("the full-circle mint serves");
+        assert!(
+            bound > 0,
+            "the server direction's fabric minted a real bound port"
+        );
+
+        let mut conn =
+            tokio::net::TcpStream::connect(("127.0.0.1", u16::try_from(bound).expect("in-range")))
+                .await
+                .expect("the fabric's listener accepts the arrival");
+        conn.write_all(b"full-circle-data")
+            .await
+            .expect("the arrival carries the bytes");
+        let mut echoed = [0u8; 16];
+        tokio::time::timeout(Duration::from_secs(30), conn.read_exact(&mut echoed))
+            .await
+            .expect("the full-circle round trip completes in time")
+            .expect("the echo reads");
+        assert_eq!(
+            &echoed, b"full-circle-data",
+            "the bytes returned through every leg: fabric → door arrival \
+             → SSH channel → client acceptor duplex → echo → back"
+        );
+
+        let (connected_address, connected_port, originator_address) =
+            tokio::time::timeout(Duration::from_secs(15), opens_rx.recv())
+                .await
+                .expect("the acceptor's handoff record arrives")
+                .expect("the acceptor sender lives");
+        assert_eq!(connected_address, "127.0.0.1");
+        assert_eq!(
+            connected_port, bound,
+            "the echo port is the minted bound one"
+        );
+        assert_eq!(
+            originator_address, "127.0.0.1",
+            "the originator carries the fabric arrival's peer"
+        );
+
+        let _ = connection.close().await;
+        let _ = double.minted_token().await; // the establisher's mint observed (the fabric's bookkeeping)
+    }
+
+    /// The smoke's dial key pose (the TestVault posture, client-side).
+    fn crate_test_key_source() -> Arc<dyn alkssh::KeySource> {
+        let vault = super::common::TestVault::unlocked();
+        let blob = vault.seal(fixture("alpha"));
+        Arc::new(alkssh::VaultKeySource::new(vault.handle(), Some(blob)))
+    }
+}
+
 #[cfg(all(feature = "tunnel", unix))]
 mod real_tool {
     //! The real-tool smoke (`w3-tunnel-direct`): one `ssh -L` (+ curl)