feat(tunnel): land the -R listen flow + the forwarded-tcpip data leg (w3-tunnel-listen)

The tunnel branch's other half is served for real against published
alktunnels 0.4.1 / alkcall 0.10.1, with POC-3 §C.4 as the template:

- tcpip_forward/streamlocal_forward (+ both cancel forms) map the
  vanilla listen request family-ward through TunnelListener::open on
  the held ChannelClient (bounded inline dial — the global request's
  reply + the &mut u32 bound-port out-param are the hook's); the
  wire bind endpoint rides as the resource string verbatim (OpenSSH
  client-side normalization parity). A reply without the minted
  listen token/bound report declines and reaps its fabric listener.
- the door serves the fabric's inbound tunnel/forwarded opens: the
  serve config gains TunnelListenSeam (the seam-config second
  course-correction — the shared ServingConfig registry the assembly
  wired at from_connection_with_serving); alkssh registers the
  forwarded openable (ChannelCore over a clone of the held client's
  manager) at SshAdapter::new — a failed registration is a failed
  construction (AdapterError::ListenRegistration).
- the forwarded-tcpip data leg: token correlation through the shared
  ListenRouter → the in-crate per-session channel cap at the accept
  surface (the acceptor vocabulary) → Handle::
  channel_open_forwarded_tcpip streaming the listen request's own
  args + the actually-bound port (sshd parity) with
  TunnelForwardedParams.peer as the originator → pump_bidi(ssh,
  fabric). Correlation-miss → teardown-at-birth, never a half-open
  SSH channel. Unix arrivals ride forwarded-streamlocal@openssh.com.
- bounded bookkeeping: the per-session listener map packet-loop-only
  with a dedicated max_concurrent_channels-ceiling counter; the
  route map bounded by construction (mints at accepted listens,
  removed at cancels + session teardown — the handler's Drop).
- the harness grew the client-side accept hook
  (server_channel_open_forwarded_tcpip/_streamlocal,
  acceptor-driven) + a hand-rolled tracing subscriber; 18 contract
  rows + both real-tool smokes (ssh -L, ssh -R + curl) green in
  tests/tunnel_mapping.rs; the wave-2 stub listen rows migrated per
  build (unfeatured / seam-unconfigured).

Verified: cargo test across default/git/session/sftp/tunnel and all
combos + --all-features; clippy -D warnings (default, all-features,
combos); cargo fmt --check; cargo doc --no-deps clean;
taskgraph validate (26 tasks).

tests/common/harness.rs

@@ -76,7 +76,7 @@ use std::sync::Arc;
 use alkcall::core::auth::IdentityProvider;
 use russh::client::{AuthResult, Handle, Msg};
 use russh::keys::PrivateKeyWithHashAlg;
-use russh::{Channel, ChannelMsg, Disconnect, Error, MethodSet};
+use russh::{Channel, ChannelMsg, ChannelOpenFailure, Disconnect, Error, MethodSet};
 
 use alkssh::{
     AdapterError, HostKeyPolicy, Limits, PresentedHostKey, ServeConfig, SshAdapter, fingerprint,
@@ -112,20 +112,86 @@ pub fn client_config(limits: Limits) -> Arc<russh::client::Config> {
     })
 }
 
+/// One server-initiated forwarded open the rows observe at the
+/// harness client (`w3-tunnel-listen`'s accept-hook harness row — the
+/// `-R` data leg arrives here). The `Channel` only carries data after
+/// the row accepts (the harness hook answers the open itself).
+#[derive(Debug)]
+pub enum ForwardedOpen {
+    /// The `forwarded-tcpip` row — the echo args + originator the door
+    /// opened with, verbatim as the wire carried them.
+    Tcpip {
+        connected_address: String,
+        connected_port: u32,
+        originator_address: String,
+        originator_port: u32,
+        channel: Channel<Msg>,
+    },
+    /// The `forwarded-streamlocal@openssh.com` row (the unix listen
+    /// leg's arrival shape).
+    Streamlocal {
+        socket_path: String,
+        channel: Channel<Msg>,
+    },
+}
+
+/// The harness client's acceptor — what a vanilla client's matching
+/// `-R` listener stands-in as: when installed, server-initiated
+/// forwarded opens are accepted and handed to the row through the
+/// channel the acceptor was built with; **no acceptor installed → the
+/// open declines cleanly** (`AdministrativelyProhibited` — the
+/// family's client-side contract, ADR-003 symmetry: a vanillla stand-
+/// in with no matching forward never accept-and-drops).
+pub struct ForwardedAcceptor {
+    opens: tokio::sync::mpsc::UnboundedSender<ForwardedOpen>,
+}
+
+impl ForwardedAcceptor {
+    /// Builds the acceptor over its event channel; the receiving half
+    /// rides the rows (`drain_forwarded`).
+    pub fn new() -> (Self, tokio::sync::mpsc::UnboundedReceiver<ForwardedOpen>) {
+        let (opens, receiver) = tokio::sync::mpsc::unbounded_channel();
+        (Self { opens }, receiver)
+    }
+
+    fn hand_off(&self, open: ForwardedOpen) -> Result<(), Error> {
+        self.opens
+            .send(open)
+            .map_err(|_| Error::ChannelOpenFailure(ChannelOpenFailure::UnknownChannelType))
+    }
+}
+
 /// The harness's russh client handler — the [`MemHostKeyPolicy`]
 /// decision double bridged into russh's `check_server_key` hook, with
 /// the decision input exactly as the wave-4 dial wiring will shape it
-/// ([`PresentedHostKey`] over the crate's fingerprint path).
+/// ([`PresentedHostKey`] over the crate's fingerprint path), plus the
+/// `w3-tunnel-listen` accept hook: the server-initiated forwarded
+/// opens the contract rows observe (the ADR-003 §4 client-side
+/// acceptor symmetry).
 pub struct HarnessClientHandler {
     policy: Arc<MemHostKeyPolicy>,
+    acceptor: Option<Arc<ForwardedAcceptor>>,
 }
 
 impl HarnessClientHandler {
     /// Builds the handler over a policy double — `with_default(true)`
     /// for the routine permissive posture, `with_default(false)` (or a
-    /// pinned rule) for the fail-closed rows.
+    /// pinned rule) for the fail-closed rows — with the forwarded
+    /// acceptor absent (opens decline cleanly; install one with
+    /// [`HarnessClientHandler::with_forwarded_acceptor`]).
     pub fn new(policy: Arc<MemHostKeyPolicy>) -> Self {
-        Self { policy }
+        Self {
+            policy,
+            acceptor: None,
+        }
+    }
+
+    /// Installs the acceptor (the matching-`-R`-listener stand-in).
+    pub fn with_forwarded_acceptor(self, acceptor: Arc<ForwardedAcceptor>) -> Self {
+        Self {
+            policy: self.policy,
+            acceptor: Some(acceptor),
+        }
     }
 }
 
@@ -147,6 +213,67 @@ impl russh::client::Handler for HarnessClientHandler {
         };
         Ok(self.policy.decide(&presented).await)
     }
+
+    /// The `forwarded-tcpip` accept hook: accept-into-the-acceptor
+    /// (the row observes the echoed args + originator verbatim and
+    /// pumps the `Channel`), or the clean decline when no acceptor is
+    /// installed.
+    async fn server_channel_open_forwarded_tcpip(
+        &mut self,
+        channel: Channel<Msg>,
+        connected_address: &str,
+        connected_port: u32,
+        originator_address: &str,
+        originator_port: u32,
+        reply: russh::client::ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        match self.acceptor.as_ref() {
+            Some(acceptor) => {
+                let open = ForwardedOpen::Tcpip {
+                    connected_address: connected_address.to_owned(),
+                    connected_port,
+                    originator_address: originator_address.to_owned(),
+                    originator_port,
+                    channel,
+                };
+                acceptor.hand_off(open)?;
+                reply.accept().await;
+            }
+            None => {
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+            }
+        }
+        Ok(())
+    }
+
+    /// The `forwarded-streamlocal` accept hook: the unix leg's shape.
+    async fn server_channel_open_forwarded_streamlocal(
+        &mut self,
+        channel: Channel<Msg>,
+        socket_path: &str,
+        reply: russh::client::ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        match self.acceptor.as_ref() {
+            Some(acceptor) => {
+                let open = ForwardedOpen::Streamlocal {
+                    socket_path: socket_path.to_owned(),
+                    channel,
+                };
+                acceptor.hand_off(open)?;
+                reply.accept().await;
+            }
+            None => {
+                reply
+                    .reject(ChannelOpenFailure::AdministrativelyProhibited)
+                    .await;
+            }
+        }
+        Ok(())
+    }
 }
 
 /// Assembles the server side from the wave-1 doubles — a
@@ -218,16 +345,28 @@ pub async fn drive(
     adapter: SshAdapter,
     host_key_policy: Arc<MemHostKeyPolicy>,
     limits: Limits,
+) -> Result<LoopbackSession, Error> {
+    drive_with_acceptor(adapter, host_key_policy, limits, None).await
+}
+
+/// [`drive`] with the client-side forwarded acceptor installed — the
+/// `w3-tunnel-listen` rows' vehicle (the harness handler's accept
+/// hook consumes the installed acceptor; `None` = the clean-decline
+/// posture [`drive`] carries).
+pub async fn drive_with_acceptor(
+    adapter: SshAdapter,
+    host_key_policy: Arc<MemHostKeyPolicy>,
+    limits: Limits,
+    acceptor: Option<Arc<ForwardedAcceptor>>,
 ) -> Result<LoopbackSession, Error> {
     let (client_io, server_io) = tokio::io::duplex(DUPLEX_CAPACITY);
     let server_task = tokio::spawn(async move { adapter.handle(server_io).await });
 
-    let client = russh::client::connect_stream(
-        client_config(limits),
-        client_io,
-        HarnessClientHandler::new(Arc::clone(&host_key_policy)),
-    )
-    .await?;
+    let mut handler = HarnessClientHandler::new(Arc::clone(&host_key_policy));
+    if let Some(acceptor) = acceptor {
+        handler = handler.with_forwarded_acceptor(acceptor);
+    }
+    let client = russh::client::connect_stream(client_config(limits), client_io, handler).await?;
 
     Ok(LoopbackSession {
         client,
@@ -353,7 +492,9 @@ pub async fn request_agent_forward(channel: &Channel<Msg>) -> Result<(), Error>
 
 /// Requests a `tcpip-forward` listen (`-R` ingress) — the
 /// success/failure shape is the result verbatim
-/// ([`Error::RequestDenied`] on a declined listen).
+/// ([`Error::RequestDenied`] on a declined listen; a `port = 0`
+/// request's actually-bound number rides the success value — sshd
+/// parity).
 pub async fn tcpip_forward(
     client: &Handle<HarnessClientHandler>,
     address: &str,
@@ -362,6 +503,40 @@ pub async fn tcpip_forward(
     client.tcpip_forward(address, port).await
 }
 
+/// Requests a `cancel-tcpip-forward`.
+pub async fn cancel_tcpip_forward(
+    client: &Handle<HarnessClientHandler>,
+    address: &str,
+    port: u32,
+) -> Result<(), Error> {
+    client.cancel_tcpip_forward(address, port).await
+}
+
+/// Requests a `streamlocal-forward` listen (the unix `-R` ingress).
+pub async fn streamlocal_forward(
+    client: &Handle<HarnessClientHandler>,
+    socket_path: &str,
+) -> Result<(), Error> {
+    client.streamlocal_forward(socket_path).await
+}
+
+/// Requests a `cancel-streamlocal-forward`.
+pub async fn cancel_streamlocal_forward(
+    client: &Handle<HarnessClientHandler>,
+    socket_path: &str,
+) -> Result<(), Error> {
+    client.cancel_streamlocal_forward(socket_path).await
+}
+
+/// Drains the acceptor's event channel — the row's observation shape
+/// (`None` once the acceptor's sender went, i.e. after the session's
+/// end).
+pub async fn next_forwarded_open(
+    receiver: &mut tokio::sync::mpsc::UnboundedReceiver<ForwardedOpen>,
+) -> Option<ForwardedOpen> {
+    receiver.recv().await
+}
+
 /// Observes the next channel message — `None` after the channel has
 /// closed (russh's EOF-of-conversation shape).
 pub async fn next_message(channel: &mut Channel<Msg>) -> Option<ChannelMsg> {