feat(client): land the dial surface — the authenticated connection handle + the fail-closed client handler skeleton (w4-dial)

The client direction's substrate (client.md §The dial surface), the
mirror of w2-server-mount: dial(DialConfig) runs the fail-closed
four-step sequence — decrypt-at-dial key load (every KeySource
posture a distinct no-dial outcome), the bounded russh client Config
from Limits (keepalive 15 s / 3, the wall clock always Some, window/
packet/buffer verbatim), the injected host-key policy at
check_server_key (RejectAll connects to nothing; UnknownKey maps to
DialError::HostKeyRejected — its only producer), and exactly one
publickey-only auth attempt — and returns the authenticated
SshConnection handle everything else in the wave rides.

- DialConfig amendment (the sanctioned course-correction, recorded):
  the required `username` field — the USERAUTH_REQUEST's user string;
  no crate default, never identity, never logged (ADR-004 decision 5).
- The client Handler skeleton (src/client_handler.rs): every
  server_channel_open_* hook overridden explicit-over-inherited with
  a clean AdministrativelyProhibited reject (never russh's
  accept-and-drop default, ADR-003 §4); should_accept_unknown_
  server_channel the explicit false; decline events on the crate
  vocabulary (no-acceptor / never-feature reasons).
- The in-process fake sshd (tests/common/sshd_double.rs): the
  client-direction contract vehicle — a plain russh server on a real
  loopback TCP listener, accept-set identity resolution, the auth_
  succeeded handle carrier for the server-initiated-open rows, the
  per-session outcome channel for the clean-close rows.
- tests/client_dial.rs: 10 contract rows — dial success, the five
  key-source no-dial postures, RejectAll / pin-mismatch / case-2
  policy rows, refused auth, the malformed-address dial-time refusal,
  the bounded-config observable, the six-hook decline wire rows with
  decline events, and the smoke: alkssh-client ↔ alkssh-server over a
  real TCP loopback (real-sshd dial stays wave-5 e2e).

Verification: cargo test + per-feature + --all-features (14 binaries,
all green), cargo clippy ×2 (-D warnings), cargo fmt --check,
cargo doc --no-deps — all clean. taskgraph validate: 35 tasks valid.

src/client.rs

@@ -0,0 +1,340 @@
+//! The dial surface (`w4-dial`) — the door *outward* (`client.md`
+//! §The dial surface), the client-direction mirror of
+//! [`crate::adapter`]'s mount: the russh client stack every wave-4 op
+//! row composes into.
+//!
+//! The entry ([`dial`]) takes the wave-1 [`DialConfig`] and runs the
+//! whole authenticated-dial sequence, **fail-closed at every step**:
+//!
+//! 1. **The decrypt-at-dial key load** (ADR-005 decision 2): the
+//!    injected [`KeySource::load`] runs *here and nowhere else* —
+//!    locked / absent / decryption / unknown-version / decode are all
+//!    **no-dial outcomes** ([`DialError::KeySource`], the variant
+//!    preserved). The decoded key lives for this dial; the vault is
+//!    needed only while the load runs (locking mid-session keeps
+//!    established connections streaming, POC-2 §2).
+//! 2. **The bounded russh client `Config`** ([`russh_client_config`])
+//!    from the [`Limits`]: the keepalive 15 s / 3 posture
+//!    (`reverse-runner.md` lesson 3 — OpenSSH
+//!    `ServerAliveInterval=15`/`CountMax=3` parity), the wall clock
+//!    **always `Some`** (`inactivity_timeout` — the crate never
+//!    constructs a session with no clock), and the window/packet/
+//!    channel-buffer knobs verbatim. Everything else stays russh's
+//!    own default posture (the ADR-002 surface notes, mirrored from
+//!    [`crate::adapter`]'s server mapping: rekey `Limits`
+//!    russh-enforced, preferred algorithms russh's — the ed25519-only
+//!    posture holds by the loaded key set itself). No client-side
+//!    auth-attempt knob exists at this pin and none is needed: the
+//!    family contract is exactly **one publickey attempt**.
+//! 3. **The policy call at `check_server_key`** (OQ-05) — the
+//!    wave-1 [`HostKeyPolicy`](crate::HostKeyPolicy) seam consumed by
+//!    the handler ([`crate::client_handler`]): the crate default
+//!    [`RejectAll`](crate::RejectAll) connects to nothing; a
+//!    pin-mismatch hard-declines; a case-2 assembly policy decides.
+//!    Declines surface as [`DialError::HostKeyRejected`].
+//! 4. **Publickey-only authentication** (the family contract):
+//!    exactly one `authenticate_publickey` attempt from the loaded
+//!    key — no password, no keyboard-interactive, no cert, no retry,
+//!    no fallback method (russh's handle methods for those are never
+//!    exposed through [`SshConnection`]); a refusal is
+//!    [`DialError::AuthRejected`] and the attempt is closed cleanly.
+//!
+//! **The connection handle** ([`SshConnection`]) is the type
+//! everything else in the wave rides (the op rows open channels on
+//! it; the `-R` leg installs the acceptor registry over it) — the
+//! russh `client::Handle` wrapped with the crate's surface, the
+//! policy, limits and endpoint labels carried. The full
+//! cancellation/clean-close lifecycle surface is `w4-lifecycle`'s;
+//! this task lands the raw material (the explicit [`SshConnection::
+//! close`] path; the handle primitives below it).
+//!
+//! No reconnect or restore policy is owned by the crate (OQ-12):
+//! failures are failures ([`DialError`]); reconnection strategy is
+//! assembly composition over these primitives.
+//!
+//! ## Test surface
+//!
+//! Two mechanics are re-exported `#[doc(hidden)]` (the `serve.rs`
+//! §Test-surface pattern) for the wire contract rows:
+//! [`russh_client_config`] (the bounded-knob mapping, asserted
+//! verbatim) and [`ClientHandler`](crate::client_handler::ClientHandler)
+//! (the handler skeleton; `client_handler.rs`). The wire vehicle is
+//! the in-process fake `sshd` in `tests/common/sshd_double.rs` — the
+//! client direction dials vanilla servers, so the loopback harness
+//! (the *server*-direction vehicle) only hosts the crate-to-crate
+//! smoke (see the task's smoke-posture record).
+
+use std::sync::Arc;
+
+use russh::Disconnect;
+use russh::keys::PrivateKeyWithHashAlg;
+
+use crate::client_handler::ClientHandler;
+use crate::config::{DialConfig, Limits};
+use crate::error::DialError;
+
+/// The authenticated dial — the client direction's entry point
+/// (`client.md` §The dial surface): see the module docs for the
+/// four-step sequence and their fail-closed outcomes.
+pub async fn dial(config: DialConfig) -> Result<SshConnection, DialError> {
+    let key = config.key_source.load().await?;
+
+    let (host, port) = parse_dial_address(&config.address)?;
+    let handler = ClientHandler::new(Arc::clone(&config.host_key_policy), host, port);
+
+    let russh_config = Arc::new(build_client_config(&config.limits));
+    let mut handle = russh::client::connect(russh_config, config.address.as_str(), handler).await?;
+
+    let result = handle
+        .authenticate_publickey(
+            config.username.as_str(),
+            PrivateKeyWithHashAlg::new(Arc::new(key), None),
+        )
+        .await?;
+    if !result.success() {
+        let _ = handle
+            .disconnect(Disconnect::ByApplication, "authentication refused", "en")
+            .await;
+        return Err(DialError::AuthRejected);
+    }
+
+    Ok(SshConnection {
+        handle,
+        limits: config.limits,
+        address: config.address,
+        username: config.username,
+    })
+}
+
+/// Parses the dial address into its endpoint labels — the
+/// [`PresentedHostKey`](crate::host_key::PresentedHostKey)'s
+/// `host`/`port` decision context. Malformed targets refuse before
+/// any connection ([`DialError::Address`]): plain `host:port` and the
+/// bracket form `[host]:port` (literal IPv6) are the grammar.
+fn parse_dial_address(address: &str) -> Result<(String, u16), DialError> {
+    let (host, port) = if let Some(rest) = address.strip_prefix('[') {
+        let (host, rest) = rest
+            .split_once(']')
+            .ok_or_else(|| DialError::Address(address.to_string()))?;
+        let port = rest
+            .strip_prefix(':')
+            .ok_or_else(|| DialError::Address(address.to_string()))?;
+        (host, port)
+    } else {
+        address
+            .rsplit_once(':')
+            .ok_or_else(|| DialError::Address(address.to_string()))?
+    };
+    if host.is_empty() {
+        return Err(DialError::Address(address.to_string()));
+    }
+    let port = port
+        .parse::<u16>()
+        .map_err(|_| DialError::Address(address.to_string()))?;
+    Ok((host.to_string(), port))
+}
+
+/// The [`Limits`] → russh client `Config` mapping — the bounded knobs
+/// verbatim, the wall clock always on, everything else at russh's own
+/// default posture (see the module docs). The mapping is the dial
+/// task's `bounded-config wiring observable`;
+/// `#[doc(hidden)]` re-export from the crate root.
+pub(crate) fn build_client_config(limits: &Limits) -> russh::client::Config {
+    russh::client::Config {
+        window_size: limits.window_size,
+        maximum_packet_size: limits.maximum_packet_size,
+        channel_buffer_size: limits.channel_buffer_size,
+        inactivity_timeout: Some(limits.inactivity_timeout),
+        keepalive_interval: limits.keepalive_interval,
+        keepalive_max: limits.keepalive_max,
+        ..russh::client::Config::default()
+    }
+}
+
+#[doc(hidden)]
+pub fn russh_client_config(limits: &Limits) -> russh::client::Config {
+    build_client_config(limits)
+}
+
+/// The authenticated connection handle — the type every later
+/// wave-4 op rides (`client.md`: "the authenticated connection
+/// handle"): the russh `client::Handle` wrapped with the crate's
+/// surface. The remote end is authenticated (publickey-only, single
+/// attempt); the dial key is gone (its lifetime is the dial's —
+/// everything after rides the wire, never the key).
+pub struct SshConnection {
+    handle: russh::client::Handle<ClientHandler>,
+    limits: Limits,
+    address: String,
+    username: String,
+}
+
+impl SshConnection {
+    /// The bounded-resource budget the dial ran under (carried; the
+    /// later op rows' admission surface).
+    pub fn limits(&self) -> &Limits {
+        &self.limits
+    }
+
+    /// The dialed address as the config carried it.
+    pub fn address(&self) -> &str {
+        &self.address
+    }
+
+    /// The user the connection authenticated as.
+    pub fn username(&self) -> &str {
+        &self.username
+    }
+
+    /// Whether the underlying russh session is closed.
+    pub fn is_closed(&self) -> bool {
+        self.handle.is_closed()
+    }
+
+    /// The explicit clean-close path (OQ-12's raw material; the full
+    /// cancellable-listener lifecycle surface is `w4-lifecycle`'s):
+    /// `Disconnect::ByApplication`, the honest close — the peer
+    /// observes a clean disconnect, no half-open channels.
+    pub async fn close(self) {
+        let _ = self
+            .handle
+            .disconnect(Disconnect::ByApplication, "closed by application", "en")
+            .await;
+    }
+}
+
+impl std::fmt::Debug for SshConnection {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SshConnection")
+            .field("address", &self.address)
+            .field("username", &self.username)
+            .field("limits", &self.limits)
+            .field("closed", &self.handle.is_closed())
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use russh::keys::PrivateKey;
+
+    use super::{build_client_config, parse_dial_address};
+    use crate::config::{DialConfig, Limits};
+    use crate::error::{DialError, KeySourceError};
+    use crate::host_key::RejectAll;
+    use crate::key_source::KeySource;
+
+    struct NoKeySource;
+
+    #[async_trait::async_trait]
+    impl KeySource for NoKeySource {
+        async fn load(&self) -> Result<PrivateKey, KeySourceError> {
+            Err(KeySourceError::Absent)
+        }
+    }
+
+    /// - *Contract stamp:* `client.md` §Correctness facts (the
+    ///   keepalive posture) + `overview.md` §Bounded-resources table —
+    ///   the defaults land in the russh mapping verbatim.
+    #[test]
+    fn bounded_config_mapping_carries_the_default_postures() {
+        let config = build_client_config(&Limits::default());
+
+        assert_eq!(config.window_size, 2_097_152);
+        assert_eq!(config.maximum_packet_size, 32_768);
+        assert_eq!(config.channel_buffer_size, 100);
+        assert_eq!(
+            config.inactivity_timeout,
+            Some(crate::config::DEFAULT_INACTIVITY_TIMEOUT),
+            "the wall clock is always set — no unbounded session"
+        );
+        assert_eq!(
+            config.keepalive_interval,
+            Some(Duration::from_secs(15)),
+            "the 15 s keepalive posture is the dial's default"
+        );
+        assert_eq!(config.keepalive_max, 3);
+    }
+
+    /// - *Contract stamp:* the [`Limits`] tunability contract (every
+    ///   knob is the assembly's to tune; nothing is crate-fixed).
+    #[test]
+    fn bounded_config_mapping_is_tunable_one_for_one() {
+        let limits = Limits {
+            window_size: 1_024,
+            maximum_packet_size: 4_096,
+            channel_buffer_size: 7,
+            inactivity_timeout: Duration::from_secs(1_337),
+            keepalive_interval: None,
+            keepalive_max: 9,
+            ..Limits::default()
+        };
+        let config = build_client_config(&limits);
+
+        assert_eq!(config.window_size, 1_024);
+        assert_eq!(config.maximum_packet_size, 4_096);
+        assert_eq!(config.channel_buffer_size, 7);
+        assert_eq!(config.inactivity_timeout, Some(Duration::from_secs(1_337)));
+        assert_eq!(config.keepalive_interval, None);
+        assert_eq!(config.keepalive_max, 9);
+    }
+
+    #[test]
+    fn dial_address_plain_form_parses() {
+        let (host, port) = parse_dial_address("runner.example:22").expect("parses");
+        assert_eq!(host, "runner.example");
+        assert_eq!(port, 22);
+
+        let (host, port) = parse_dial_address("127.0.0.1:4022").expect("parses");
+        assert_eq!(host, "127.0.0.1");
+        assert_eq!(port, 4022);
+    }
+
+    #[test]
+    fn dial_address_bracket_form_parses() {
+        let (host, port) = parse_dial_address("[2001:db8::1]:2222").expect("parses");
+        assert_eq!(host, "2001:db8::1");
+        assert_eq!(port, 2222);
+    }
+
+    #[test]
+    fn dial_address_malformed_forms_refuse() {
+        for bad in [
+            "runner.example",
+            ":22",
+            "[]:22",
+            "host:port",
+            "[::1:2]x",
+            "",
+        ] {
+            let err = parse_dial_address(bad).expect_err("refuses");
+            assert!(
+                matches!(err, DialError::Address(ref given) if *given == bad),
+                "expected Address, got {err:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn dial_address_out_of_range_port_refuses() {
+        let err = parse_dial_address("host:99999").expect_err("refuses");
+        assert!(matches!(err, DialError::Address(_)));
+    }
+
+    #[tokio::test]
+    async fn key_source_refusal_surfaces_the_variant_through_the_config() {
+        let config = DialConfig::new(
+            "runner.example:22",
+            "runner",
+            Arc::new(NoKeySource),
+            Arc::new(RejectAll),
+        );
+        assert!(
+            matches!(config.key_source.load().await, Err(KeySourceError::Absent)),
+            "the no-dial outcome's variant is preserved through the config"
+        );
+    }
+}

src/client_handler.rs

@@ -0,0 +1,315 @@
+//! The per-connection **client** handler — the dial surface's russh
+//! `client::Handler` impl (`w4-dial`), the mirror of
+//! [`crate::handler`]'s server-direction handler.
+//!
+//! Two responsibilities, both fail-closed:
+//!
+//! 1. **The host-key policy at `check_server_key`** (OQ-05): the hook
+//!    receives the server's presented key, computes its fingerprint on
+//!    the crate's single fingerprint path ([`crate::fingerprint`]),
+//!    labels it with the dial target (host + port, parsed from the
+//!    dial address), and hands the whole [`PresentedHostKey`] to the
+//!    injected [`HostKeyPolicy`] — the verdict rides straight through
+//!    (`Ok(policy.decide(..).await)`). The crate is policy-blind; the
+//!    built-in [`RejectAll`](crate::RejectAll) verdict makes every dial
+//!    refuse; a `false` verdict is the decline disposition russh
+//!    surfaces as its `UnknownKey` wire class
+//!    ([`DialError::HostKeyRejected`](crate::DialError)
+//!    — never an accept, never a retry).
+//! 2. **Server-initiated channel opens declined cleanly** (ADR-003 §4,
+//!    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
+//!    [`ChannelOpenFailure::AdministrativelyProhibited`] — a clean
+//!    wire-visible refusal the peer observes as an
+//!    unconfirmed open, never an accepted-then-dropped channel. 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.
+//!
+//! 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`).
+
+use russh::ChannelOpenFailure;
+use russh::client::ChannelOpenHandle;
+use russh::client::Msg;
+use russh::keys::PublicKeyOrCertificate;
+use russh::{Channel, ChannelId};
+
+use crate::host_key::{HostKeyPolicy, PresentedHostKey};
+use crate::observe;
+use crate::{DialError, fingerprint};
+
+/// 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
+/// tests drive the bare hooks, exactly the
+/// `SessionHandler` §Test-surface pattern's client mirror.
+pub struct ClientHandler {
+    policy: std::sync::Arc<dyn HostKeyPolicy>,
+    host: String,
+    port: u16,
+}
+
+impl ClientHandler {
+    /// Builds the handler over the injected policy and the dial
+    /// target's endpoint labels (host + port; the presented key's
+    /// decision context, see `check_server_key`).
+    pub fn new(
+        policy: std::sync::Arc<dyn HostKeyPolicy>,
+        host: impl Into<String>,
+        port: u16,
+    ) -> Self {
+        Self {
+            policy,
+            host: host.into(),
+            port,
+        }
+    }
+}
+
+impl russh::client::Handler for ClientHandler {
+    type Error = DialError;
+
+    /// The host-key policy's hook (OQ-05): the presented key or
+    /// certificate's fingerprint is computed on the crate's single
+    /// fingerprint path (certificates fingerprint their inner key —
+    /// the trust decision is the policy's either way) and the whole
+    /// [`PresentedHostKey`] rides the policy; the verdict is passed
+    /// through untouched. `Ok(false)` — russh's decline — is the
+    /// disposition, never an error path; the policy's own record
+    /// carries why.
+    async fn check_server_key(
+        &mut self,
+        server_public_key: &PublicKeyOrCertificate,
+    ) -> Result<bool, Self::Error> {
+        let presented = PresentedHostKey {
+            host: self.host.clone(),
+            port: self.port,
+            fingerprint: fingerprint(&server_public_key.public_key()),
+        };
+        Ok(self.policy.decide(&presented).await)
+    }
+
+    /// Unknown channel types never come through this way: the
+    /// explicit `false` (the fail-closed default, written down to
+    /// survive a pin bump).
+    async fn should_accept_unknown_server_channel(
+        &mut self,
+        _id: ChannelId,
+        _channel_type: &str,
+    ) -> bool {
+        false
+    }
+
+    async fn server_channel_open_session(
+        &mut self,
+        channel: Channel<Msg>,
+        reply: ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        observe::decline_open("session", "no-acceptor", channel.id().number());
+        reply
+            .reject(ChannelOpenFailure::AdministrativelyProhibited)
+            .await;
+        Ok(())
+    }
+
+    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: 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(())
+    }
+
+    async fn server_channel_open_forwarded_streamlocal(
+        &mut self,
+        channel: Channel<Msg>,
+        _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(())
+    }
+
+    async fn server_channel_open_direct_tcpip(
+        &mut self,
+        channel: Channel<Msg>,
+        _host_to_connect: &str,
+        _port_to_connect: u32,
+        _originator_address: &str,
+        _originator_port: u32,
+        reply: ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        observe::decline_open("direct-tcpip", "no-acceptor", channel.id().number());
+        reply
+            .reject(ChannelOpenFailure::AdministrativelyProhibited)
+            .await;
+        Ok(())
+    }
+
+    async fn server_channel_open_direct_streamlocal(
+        &mut self,
+        channel: Channel<Msg>,
+        _socket_path: &str,
+        reply: ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        observe::decline_open("direct-streamlocal", "no-acceptor", channel.id().number());
+        reply
+            .reject(ChannelOpenFailure::AdministrativelyProhibited)
+            .await;
+        Ok(())
+    }
+
+    /// A never-feature row (ADR-003 §5): x11 has no family seam — the
+    /// decline is not even registry territory.
+    async fn server_channel_open_x11(
+        &mut self,
+        channel: Channel<Msg>,
+        _originator_address: &str,
+        _originator_port: u32,
+        reply: ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        observe::decline_open("x11", "never-feature", channel.id().number());
+        reply
+            .reject(ChannelOpenFailure::AdministrativelyProhibited)
+            .await;
+        Ok(())
+    }
+
+    /// A never-feature row (ADR-003 §5): agent forwarding is the
+    /// anti-pattern of the vault custody model — always declined.
+    async fn server_channel_open_agent_forward(
+        &mut self,
+        channel: Channel<Msg>,
+        reply: ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        observe::decline_open("agent", "never-feature", channel.id().number());
+        reply
+            .reject(ChannelOpenFailure::AdministrativelyProhibited)
+            .await;
+        Ok(())
+    }
+
+    /// Wire-reachable only if
+    /// [`russh::client::Handler::should_accept_unknown_server_channel`]
+    /// ever answers `true` (it does not — the skeleton's default is
+    /// fail-closed); the defensive reject keeps the posture honest
+    /// even then.
+    async fn server_channel_open_unknown(
+        &mut self,
+        channel: Channel<Msg>,
+        reply: ChannelOpenHandle,
+        _session: &mut russh::client::Session,
+    ) -> Result<(), Self::Error> {
+        observe::decline_open("unknown", "no-acceptor", channel.id().number());
+        reply
+            .reject(ChannelOpenFailure::AdministrativelyProhibited)
+            .await;
+        Ok(())
+    }
+}
+
+/// The bare-hook observation surface for the contract tests (the
+/// `SessionHandler` §Test-surface shape, client-side): the rows call
+/// 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 super::ClientHandler;
+    use crate::host_key::HostKeyPolicy;
+
+    /// - *Contract stamp:* ADR-004 decision 6 posture mirror (the
+    ///   client-side: the policy is injected, the handler never
+    ///   decides trust), pinned at the dial wiring — here, that the
+    ///   labels ride the construction so the policy sees
+    ///   `PresentedHostKey { host, port }` verbatim, and the
+    ///   fingerprint rides the crate's single fingerprint path.
+    #[tokio::test]
+    async fn labels_ride_the_construction_into_the_policy_hook() {
+        use crate::host_key::PresentedHostKey;
+        use async_trait::async_trait;
+
+        struct Recording {
+            seen: parking_lot::Mutex<Vec<(String, String, u16)>>,
+        }
+
+        #[async_trait]
+        impl HostKeyPolicy for Recording {
+            async fn decide(&self, presented: &PresentedHostKey) -> bool {
+                self.seen.lock().push((
+                    presented.host.clone(),
+                    presented.fingerprint.clone(),
+                    presented.port,
+                ));
+                false
+            }
+        }
+
+        const ALPHA_PEM: &str = include_str!("../tests/vectors/alpha.key");
+        let expected_fp = crate::fingerprint(
+            russh::keys::decode_secret_key(ALPHA_PEM, None)
+                .expect("fixture PEM decodes")
+                .public_key(),
+        );
+
+        let policy = Arc::new(Recording {
+            seen: parking_lot::Mutex::new(Vec::new()),
+        });
+        let mut handler = ClientHandler::new(
+            Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
+            "runner.example",
+            2222,
+        );
+        let key = russh::keys::decode_secret_key(ALPHA_PEM, None)
+            .expect("fixture PEM decodes")
+            .public_key()
+            .clone();
+        let verdict = handler
+            .check_server_key(&russh::keys::PublicKeyOrCertificate::from(key))
+            .await
+            .expect("the hook completes");
+        let seen = policy.seen.lock().clone();
+
+        assert!(!verdict, "the injected policy's verdict rides through");
+        assert_eq!(
+            seen,
+            vec![("runner.example".to_string(), expected_fp, 2222)]
+        );
+    }
+}

src/config.rs

@@ -15,8 +15,9 @@
 //!   pipeline + the push-options flag), the OQ-09 `cwd`/`env` values,
 //!   and the [`Limits`].
 //! - [`DialConfig`] — the client direction's dial configuration
-//!   (`client.md` §The dial surface): address, key source, host-key
-//!   policy, and the bounded-resource knobs (carried by [`Limits`]).
+//!   (`client.md` §The dial surface): address, `username`, key source,
+//!   host-key policy, and the bounded-resource knobs (carried by
+//!   [`Limits`]).
 //!
 //! Crate rules this module encodes:
 //!
@@ -579,19 +580,41 @@ impl std::fmt::Debug for ServeConfig {
 }
 
 /// The client direction's dial configuration (`client.md` §The dial
-/// surface): everything wave-4's dial loop takes at construction.
-/// Pure data + injected handles — no russh `Config` is built here
-/// (wave 2), no dial happens (wave 4).
+/// surface): everything wave-4's dial surface takes. Pure data +
+/// injected handles — the russh client `Config` construction and the
+/// dial itself live in the crate's dial module (the `w4-dial`
+/// surface).
 ///
 /// Construction carries no validation: the key source's own
 /// load path is fail-closed ([`KeySource`] — locked/absent/decryption
 /// failures are all no-dial outcomes), and the host-key policy's
 /// default is fail-closed ([crate::host_key::RejectAll]).
+///
+/// **The `username` field** (the task's sanctioned
+/// `w3-seam-config`-pattern course-correction — the only amendment the
+/// wave-1 shape needed): the USERAUTH_REQUEST's user string. The
+/// wave-1 shape carried everything `client.md` §The dial surface names
+/// but structurally missed this one: russh's `authenticate_publickey`
+/// requires a user, and the remote end evaluates it — the family
+/// identity model stays fingerprint-only forever (`auth.md` §Host-key;
+/// the username is not identity and carries no routing semantics,
+/// OQ-03), so this field is pure dial data the assembly supplies. No
+/// crate default exists (a default account would be deployment policy
+/// smuggled into the crate); an empty string is accepted as data and
+/// is the remote end's to refuse.
 pub struct DialConfig {
     /// The dial target — `host:port` as russh's connect surface
     /// takes it (the reverse-runner shape: vanilla `sshd` on the
-    /// fleet box).
+    /// fleet box). The same address labels the host-key policy's
+    /// presented-key decisions (host = the parsed host, port = the
+    /// parsed port).
     pub address: String,
+    /// The user the dial authenticates as, verbatim on the
+    /// USERAUTH_REQUEST — not identity (the fingerprint is; ADR-004
+    /// decision 5's posture, mirrored client-side), never logged (the
+    /// crate's event vocabulary carries no username). See the type
+    /// docs.
+    pub username: String,
     /// The key source the dial authenticates with (the wave-1
     /// [`KeySource`] seam) — never a static key, never env vars;
     /// decrypt-at-dial, fail-closed per ADR-005.
@@ -609,18 +632,20 @@ pub struct DialConfig {
 }
 
 impl DialConfig {
-    /// Builds a dial-config from the address and the two injected
-    /// handles the dial direction can never default — the key source
-    /// and the host-key policy. `limits` starts at
-    /// [`Limits::default()`] (keepalive 15 s / 3 posture riding
-    /// inside).
+    /// Builds a dial-config from the address, the user it
+    /// authenticates as, and the two injected handles the dial
+    /// direction can never default — the key source and the host-key
+    /// policy. `limits` starts at [`Limits::default()`] (keepalive
+    /// 15 s / 3 posture riding inside).
     pub fn new(
         address: impl Into<String>,
+        username: impl Into<String>,
         key_source: Arc<dyn KeySource>,
         host_key_policy: Arc<dyn HostKeyPolicy>,
     ) -> Self {
         Self {
             address: address.into(),
+            username: username.into(),
             key_source,
             host_key_policy,
             limits: Limits::default(),
@@ -632,6 +657,7 @@ impl std::fmt::Debug for DialConfig {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("DialConfig")
             .field("address", &self.address)
+            .field("username", &"<present>")
             .field("key_source", &"<dyn KeySource>")
             .field("host_key_policy", &"<dyn HostKeyPolicy>")
             .field("limits", &self.limits)
@@ -963,7 +989,12 @@ mod tests {
 
     #[test]
     fn dial_config_defaults_carry_the_keepalive_posture() {
-        let dial = DialConfig::new("runner.example:22", key_source(), Arc::new(RejectAll));
+        let dial = DialConfig::new(
+            "runner.example:22",
+            "runner",
+            key_source(),
+            Arc::new(RejectAll),
+        );
 
         assert_eq!(dial.address, "runner.example:22");
         assert_eq!(dial.limits, Limits::default());
@@ -980,10 +1011,25 @@ mod tests {
         let keys = key_source();
         let policy: Arc<dyn crate::host_key::HostKeyPolicy> = Arc::new(RejectAll);
 
-        let dial = DialConfig::new("fleet.example:22", Arc::clone(&keys), Arc::clone(&policy));
+        let dial = DialConfig::new(
+            "fleet.example:22",
+            "deploy",
+            Arc::clone(&keys),
+            Arc::clone(&policy),
+        );
 
         assert!(Arc::ptr_eq(&keys, &dial.key_source));
         assert!(Arc::ptr_eq(&policy, &dial.host_key_policy));
+        assert_eq!(dial.address, "fleet.example:22");
+        assert_eq!(dial.username, "deploy");
+    }
+
+    /// an empty string is the assembly's datum (never a crate
+    /// default).
+    #[test]
+    fn dial_config_username_is_data_not_a_crate_default() {
+        let dial = DialConfig::new("fleet.example:22", "", key_source(), Arc::new(RejectAll));
+        assert_eq!(dial.username, "");
     }
 
     // The wave-3 per-feature seam handles: the construction shapes, the
@@ -1115,7 +1161,12 @@ mod tests {
 
     #[tokio::test]
     async fn dial_config_key_source_stays_fail_closed_through_the_config() {
-        let dial = DialConfig::new("fleet.example:22", key_source(), Arc::new(RejectAll));
+        let dial = DialConfig::new(
+            "fleet.example:22",
+            "deploy",
+            key_source(),
+            Arc::new(RejectAll),
+        );
 
         let err = dial.key_source.load().await.expect_err("no key source");
         assert!(matches!(err, KeySourceError::Absent));

src/error.rs

@@ -156,9 +156,84 @@ pub enum AdapterError {
     Wire(#[from] russh::Error),
 }
 
+/// The dial surface's failure classes (the client direction's error
+/// counterpart of [`AdapterError`]) — deliberately closed, every
+/// variant a no-connection outcome (ADR-005 decision 4: failures are
+/// fail-closed and distinct, never collapsed, never a fallback dial):
+///
+/// - **KeySource** — the decrypt-at-dial load refused. The
+///   [`KeySourceError`] variant is preserved verbatim (locked / absent
+///   / decryption / unknown-version / decode each stays itself — the
+///   ADR-005 decision 4 matrix read through the dial seam). No key
+///   material, no dial.
+/// - **Address** — the dial target is not a parseable `host:port`.
+///   Checked at dial time (the config carries no construction-time
+///   validation, the family contract); a malformed target is a
+///   refusal before any connection.
+/// - **HostKeyRejected** — the injected host-key policy
+///   ([`crate::HostKeyPolicy`]) declined the presented server key at
+///   `check_server_key` (a `false` verdict is the decline
+///   disposition; russh surfaces it as its `UnknownKey` wire error,
+///   which this crate maps here because that hook is the only
+///   producer). The trust decision is the policy's — its own record
+///   (its decision log) carries the reasons; the crate never
+///   second-guesses or retries.
+/// - **AuthRejected** — the remote end refused the publickey
+///   authentication (the family contract is exactly one
+///   publickey-only attempt; there is no retry, no fallback method).
+/// - **Wire** — the russh session stack's own error, mapped
+///   transparently, including the dial's connect/resolve/IO classes.
+#[derive(Debug, Error)]
+pub enum DialError {
+    /// The dial key's decrypt-at-dial load refused — every
+    /// [`KeySourceError`] variant is a no-dial outcome, preserved
+    /// un-collapsed (ADR-005 decision 4).
+    #[error("dial key source refused: {0}")]
+    KeySource(#[from] KeySourceError),
+
+    /// The dial target is not a parseable `host:port` address — the
+    /// wire never dialed. The string carries the address as given.
+    #[error("dial target is not a usable host:port address: {0}")]
+    Address(String),
+
+    /// The host-key policy declined the presented server key — the
+    /// dial aborts (a disposition the dial cannot proceed on; the
+    /// crate default [`RejectAll`](crate::RejectAll) policy makes
+    /// every dial refuse, by design).
+    #[error("host-key policy declined the presented server key")]
+    HostKeyRejected,
+
+    /// The remote end refused the publickey authentication — the
+    /// family's one-attempt publickey-only posture (no retries, no
+    /// fallback methods, never another key).
+    #[error("authentication was refused by the remote end")]
+    AuthRejected,
+
+    /// A russh session transport error (connect / KEX / IO class).
+    /// The `From<russh::Error>` conversion below is the variant's
+    /// only mapping (it carries the one semantic upgrade, see its
+    /// docs), so no `#[from]` lives on the field.
+    #[error("ssh dial wire error: {0}")]
+    Wire(russh::Error),
+}
+
+impl From<russh::Error> for DialError {
+    /// The transparent russh mapping with one semantic upgrade:
+    /// `UnknownKey` is the wire shape of a host-key policy decline
+    /// (the client's `check_server_key` hook is the only producer),
+    /// so it becomes [`DialError::HostKeyRejected`] — every other
+    /// error stays the transparent [`DialError::Wire`] class.
+    fn from(error: russh::Error) -> Self {
+        match error {
+            russh::Error::UnknownKey => Self::HostKeyRejected,
+            other => Self::Wire(other),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
-    use super::{AdapterError, KeySourceError};
+    use super::{AdapterError, DialError, KeySourceError};
 
     fn all_variants() -> [KeySourceError; 5] {
         [
@@ -224,6 +299,67 @@ mod tests {
         }
     }
 
+    fn all_dial_variants() -> Vec<DialError> {
+        let mut all: Vec<DialError> = all_variants()
+            .into_iter()
+            .map(DialError::KeySource)
+            .collect();
+        all.push(DialError::Address("not an address".into()));
+        all.push(DialError::HostKeyRejected);
+        all.push(DialError::AuthRejected);
+        all.push(DialError::Wire(russh::Error::Disconnect));
+        all
+    }
+
+    #[test]
+    fn dial_error_closed_taxonomy_no_catch_all() {
+        let exhaustive: fn(&DialError) -> &'static str = |e| match e {
+            DialError::KeySource(_) => "key-source",
+            DialError::Address(_) => "address",
+            DialError::HostKeyRejected => "host-key-rejected",
+            DialError::AuthRejected => "auth-rejected",
+            DialError::Wire(_) => "wire",
+        };
+        for v in all_dial_variants() {
+            assert!(!exhaustive(&v).is_empty());
+        }
+    }
+
+    #[test]
+    fn dial_error_key_source_variants_preserve_the_class() {
+        for source in all_variants() {
+            let wrapped = DialError::KeySource(source.clone());
+            match wrapped {
+                DialError::KeySource(seen) => {
+                    assert_eq!(seen.to_string(), source.to_string())
+                }
+                other => panic!("unexpected variant {other:?}"),
+            }
+        }
+    }
+
+    /// The host-key policy's `false` verdict is russh's `UnknownKey`
+    /// wire error — the client's `check_server_key` hook is the only
+    /// producer, so the transparent mapping upgrades exactly that one
+    /// class and leaves everything else `Wire`.
+    ///
+    /// - *Contract stamp:* `host_key.rs` (the verdict type pinned to
+    ///   the hook shape); `client.rs` (the dial surface's decline
+    ///   class).
+    #[test]
+    fn dial_error_unknown_key_maps_to_host_key_rejected() {
+        let mapped: DialError = russh::Error::UnknownKey.into();
+        assert!(
+            matches!(mapped, DialError::HostKeyRejected),
+            "expected HostKeyRejected, got {mapped:?}"
+        );
+        let mapped: DialError = russh::Error::Disconnect.into();
+        assert!(
+            matches!(mapped, DialError::Wire(russh::Error::Disconnect)),
+            "expected wire-transparent mapping, got {mapped:?}"
+        );
+    }
+
     fn all_adapter_variants() -> Vec<AdapterError> {
         let mut all: Vec<AdapterError> = all_variants()
             .into_iter()

src/lib.rs

@@ -19,13 +19,16 @@
 //! rest, decrypted at use (ADR-005), never env vars.
 //!
 //! Architecture: `docs/architecture/` (phase 1 — ADRs 001–006 accepted;
-//! the foundations, the server stack, and the server-direction mappings
-//! are landed through wave 3; the client direction, the validation
-//! suite, and the release are later implementation waves). See also
-//! `docs/research/`, `docs/plans/implementation.md`, and family ADR-003.
+//! the foundations, the server stack, the server-direction mappings, and
+//! the client direction's dial surface are landed through wave 4's dial
+//! task; the client op rows, 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;
+mod client_handler;
 mod config;
 mod dispatch;
 mod error;
@@ -43,6 +46,15 @@ mod tty_map;
 mod tunnel_map;
 
 pub use adapter::SshAdapter;
+/// 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};
+#[doc(hidden)]
+pub use client_handler::ClientHandler;
 #[cfg(feature = "git")]
 pub use config::GitSeam;
 #[cfg(feature = "session")]
@@ -50,7 +62,7 @@ pub use config::SessionSeam;
 #[cfg(feature = "tunnel")]
 pub use config::TunnelListenSeam;
 pub use config::{DialConfig, Limits, ServeConfig};
-pub use error::{AdapterError, KeySourceError};
+pub use error::{AdapterError, DialError, KeySourceError};
 pub use fingerprint::fingerprint;
 /// Contract-test surface (`#[doc(hidden)]`): the per-connection auth
 /// bridge handler — constructible so the tests drive its hooks directly

src/observe.rs

@@ -106,6 +106,11 @@
 //!   not-found pair) — the global-request and arrival rows, and on the
 //!   auth class `"method-unimplemented"` / `"cert-unsupported"` /
 //!   `"non-ed25519"` / `"fingerprint-unresolved"`.
+//!   Client-side (the dial handler's `server_channel_open_*`
+//!   skeleton, `w4-dial`): `"no-acceptor"` (the decline-all default —
+//!   a server-initiated open with no registered acceptor; the
+//!   registry is `w4-reverse-listen`'s) and `"never-feature"` (the
+//!   x11/agent rows — ADR-003 §5).
 //! - `exec_route` — only on `exec_request`-row events: the dispatch
 //!   route the parse routed to (`"git"` | `"pipe"`), so a git row's
 //!   decline is distinguishable from a pipe row's in combined builds.

tasks/implementation/w4-dial.md

@@ -1,7 +1,7 @@
 ---
 id: w4-dial
 name: Wave 4 — the client dial surface (russh client stack + the connection handle)
-status: pending
+status: completed
 depends_on: [review-wave-3]
 scope: moderate
 risk: medium
@@ -67,21 +67,21 @@ connection handle this task lands.
 
 ## Acceptance Criteria
 
-- [ ] The dial surface lands over `DialConfig` (+ any amendment
+- [x] The dial surface lands over `DialConfig` (+ any amendment
       decided and recorded); the russh client `Config` carries the
       bounded knobs (keepalive 15 s / 3 posture, inactivity, window)
-- [ ] Key source decrypt-at-dial; every `KeySourceError` failure
+- [x] Key source decrypt-at-dial; every `KeySourceError` failure
       posture is a no-dial outcome with its distinct variant surfaced
-- [ ] `check_server_key` calls the injected policy; `RejectAll`
+- [x] `check_server_key` calls the injected policy; `RejectAll`
       connects to nothing; pin-mismatch hard-declines; no accept-all
       built-in
-- [ ] The client Handler skeleton declines server-initiated opens
+- [x] The client Handler skeleton declines server-initiated opens
       cleanly (fail-closed default pinned by wire rows)
-- [ ] The authenticated connection handle lands; authentication is
+- [x] The authenticated connection handle lands; authentication is
       publickey-only from the key source
-- [ ] The in-process server double lands in `tests/common`; the
+- [x] The in-process server double lands in `tests/common`; the
       contract rows above green; 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,8 +104,139 @@ connection handle this task lands.
 
 ## Notes
 
-> To be filled by implementation agent
+### What landed (2026-10-29)
+
+- `src/client.rs`: the dial surface — `dial(DialConfig)` over
+  [`DialConfig`], the four-step sequence (key load → bounded russh
+  `Config` → policy at `check_server_key` → single publickey
+  auth), [`SshConnection`] (the connection handle: `limits()`
+  / `address()` / `username()` / `is_closed()` / `close()` — the
+  OQ-12 raw material; the full lifecycle surface is `w4-lifecycle`'s),
+  the address parse (`host:port` + the bracket form;
+  `DialError::Address` on refusal — dial-time, the family's
+  no-construction-validation shape), and the doc-hidden
+  `russh_client_config` mapping (the bounded-config observable).
+- `src/client_handler.rs`: the client Handler skeleton (the
+  mirror of `handler.rs`'s server-direction handler; a separate
+  module because `w4-op-rows` / `w4-reverse-listen` grow it) —
+  `check_server_key` wired to the injected policy
+  (certificates fingerprint their inner key; the verdict passes
+  through), every `server_channel_open_*` hook overridden
+  explicit-over-inherited with a
+  `ChannelOpenFailure::AdministrativelyProhibited` reject (never
+  russh's accept-and-drop default), `should_accept_unknown_server_channel`
+  the explicit `false`, decline events on the crate vocabulary.
+- `src/error.rs`: [`DialError`] — the dial direction's closed
+  taxonomy (`KeySource` per-variant preserved / `Address` /
+  `HostKeyRejected` / `AuthRejected` / `Wire`), with a manual
+  `From<russh::Error>` that upgrades russh's `UnknownKey` — the only
+  producer of which is the `check_server_key` decline path — into
+  `HostKeyRejected`; hence the `Wire` variant carries no `#[from]`
+  (one conversion, one place).
+- `src/config.rs`: the **DialConfig amendment** (below) + doc
+  updates; `src/observe.rs`: the client-side decline reasons
+  (`no-acceptor` — the skeleton's decline-all default; the registry
+  is `w4-reverse-listen`'s) recorded in the vocabulary; `src/lib.rs`:
+  the re-exports (`dial` / `SshConnection` / `DialError` public; the
+  handler + config mapping `#[doc(hidden)]` test surface, the
+  `serve.rs` pattern).
+- `tests/common/sshd_double.rs`: the fake `sshd` (below);
+  `tests/client_dial.rs`: the contract rows incl. the smoke.
+
+### Decided + recorded
+
+1. **The DialConfig amendment (`username`) — the task's sanctioned
+   course-correction, the only one.** The wave-1 shape carried
+   everything `client.md` §The dial surface names, but the
+   USERAUTH_REQUEST structurally requires a user string
+   (`authenticate_publickey(user, key)`) and the remote end evaluates
+   it — no dial can exist without one. Decided with the
+   `w3-seam-config` pattern: `username: String` is a required
+   `DialConfig::new` parameter (no crate default — a default account
+   would be deployment policy smuggled into the crate; the empty
+   string is accepted as data and is the remote end's to refuse). It
+   is *not identity* — the identity model stays fingerprint-only
+   (ADR-004 decision 5, mirrored client-side), it is never logged
+   (the event vocabulary carries no username), and it rides
+   [`SshConnection`] as carried dial data. Wave-1 records amended
+   (the config module + field docs).
+2. **The server double's placement/shape**
+   (`tests/common/sshd_double.rs`): a plain russh server (no alkssh
+   in its serving path — vanilla-parity) on a **real loopback TCP
+   listener** (`127.0.0.1:0` — a dial target is an address the surface
+   resolves and connects to; the duplex-pair trick is the
+   server-direction harness's, and a dial has no `&mut Handle` slot
+   to attach one to). One dial per instance (doubles are cheap); the
+   handler resolves fingerprints against an injected accept set (the
+   vanilla-`authorized_keys` stand-in) with every other auth hook an
+   explicit reject; the `auth_succeeded` hook hands the row the
+   per-connection `russh::server::Handle` through a one-shot carrier —
+   the initiator for the server-initiated-open decline rows (the
+   double calls `channel_open_*` at the client and observes each as
+   `Err(ChannelOpenFailure(AdministrativelyProhibited))`); a
+   per-session outcome channel is the clean-close rows' observable;
+   an optional client-session acceptor stub exists for later rows
+   (`next_opened_session`).
+3. **The smoke posture**: an alkssh-client ↔ alkssh-server
+   in-process loopback over a real TCP socket, landed as the row
+   `smoke_alkssh_client_dials_the_alkssh_server_direction`
+   (`tests/client_dial.rs` — both russh stacks, the full auth bridge
+   server-side, the full custody + policy chain client-side). A
+   real-`sshd` dial is wave-5 e2e territory (`docs/plans/
+   implementation.md` §Wave 5); nothing is silently dropped.
+4. **The decline-class mapping**: russh surfaces a policy decline as
+   its `UnknownKey` wire error; since the client's `check_server_key`
+   hook is the only producer of that class, the manual `From` makes
+   it the semantic [`DialError::HostKeyRejected`] — every other russh
+   error stays [`DialError::Wire`].
+5. **The skeleton's unknown-type gate** (`should_accept_unknown_
+   server_channel` → explicit `false`) is code-pinned, not
+   row-pinned: russh's server `Handle` exposes no initiator for an
+   unknown channel type, and russh's `ChannelId` is deliberately
+   unconstructible outside the library, so no wire row can reach the
+   hook. The six reachable hooks (session / direct-tcpip /
+   forwarded-tcpip / forwarded-streamlocal / x11 / agent) are
+   wire-row-pinned; the `_unknown` open hook is a defensive reject
+   behind the gate.
+6. **Auth posture**: exactly one `authenticate_publickey` attempt
+   (ed25519-only, ADR-002 decision 4 — `None` is the whole
+   hash-algorithm posture); a refusal closes the attempt cleanly
+   (`Disconnect::ByApplication`) and surfaces
+   [`DialError::AuthRejected`]. russh's other auth methods
+   (password / kbi / cert / gss / signer-injection — the signer seam
+   is dismissed by ADR-005 decision 8) are never reached from
+   [`SshConnection`].
+7. **Bounded posture**: the knob mapping keeps the crate's documented
+   postures (`keepalive` 15 s / 3, the wall clock always `Some`,
+   window/packet/buffer verbatim) and leaves
+   russh's own default postures untouched (rekey `Limits`,
+   preferred algorithms, `client_id`, `nodelay` — no client-side
+   auth-attempt knob exists at this pin and the single-attempt
+   contract makes none needed).
+
+### Verification
+
+`cargo test` (default) and `cargo test --all-features` (and
+per-feature `--features {session,git,tunnel}`) all green — 14 test
+binaries each, the new `tests/client_dial.rs` carrying 10 rows;
+`cargo clippy --all-features --all-targets` and `cargo clippy
+--all-targets` both `-D warnings` clean; `cargo fmt --check` clean;
+`cargo doc --no-deps` warning-free.
 
 ## Summary
 
-> To be filled on completion
+The client direction's substrate landed: `dial(config) ->
+Result<SshConnection, DialError>` runs the fail-closed four-step
+sequence (decrypt-at-dial key load with every no-dial posture
+distinct; the bounded russh client `Config` from `Limits`; the
+injected host-key policy at `check_server_key` — `RejectAll`
+connects to nothing; a single publickey-only auth attempt) and
+returns the authenticated connection handle everything later in the
+wave rides. One sanctioned `DialConfig` amendment: the required
+`username` field (the USERAUTH_REQUEST's user string — no crate
+default, never identity, never logged). The fail-closed client
+Handler skeleton declines every server-initiated open on the wire
+(the fake-`sshd` double in `tests/common` pins six hook rows +
+decline events), and the alkssh-client ↔ alkssh-server loopback
+smoke is landed as a row. The `w4-op-rows` chain grows this
+skeleton in place.

tests/client_dial.rs

@@ -0,0 +1,474 @@
+//! The dial surface's contract rows (`w4-dial`) — driven through the
+//! in-process fake `sshd` ([`sshd_double`](common::sshd_double), the
+//! client-direction vehicle) plus the crate-to-crate smoke against
+//! the alkssh server direction. The row/stamp discipline
+//! (`tests/common/mod.rs`'s doc) applies verbatim.
+
+mod common;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use alkssh::{
+    DialConfig, DialError, HostKeyPolicy, KeySource, KeySourceError, Limits, RejectAll,
+    ServeConfig, SshAdapter, VaultKeySource, dial, fingerprint, russh_client_config,
+};
+use common::{
+    MemHostKeyPolicy, MemHostKeySource, MemIdentityProvider, TestVault, event_capture, fixture,
+    identity, sshd_double,
+};
+use russh::{ChannelOpenFailure, Error};
+
+use sshd_double::SshdDouble;
+
+/// A vault-backed key source over the alpha fixture — the standard
+/// custody posture (the fixture PEM sealed, decrypt-at-dial behind
+/// the production source).
+fn alpha_key_source() -> Arc<dyn KeySource> {
+    let vault = TestVault::unlocked();
+    let blob = vault.seal(fixture("alpha"));
+    Arc::new(VaultKeySource::new(vault.handle(), Some(blob)))
+}
+
+/// The dial-target address of a new double; one fake `sshd` per row
+/// (the vehicle's shape decision).
+async fn spawn_double() -> (SshdDouble, sshd_double::SshdAccepts) {
+    sshd_double::spawn(vec![fixture("bravo").private_key()], true).await
+}
+
+fn port_of(double: &SshdDouble) -> u16 {
+    double
+        .address
+        .rsplit_once(':')
+        .and_then(|(_, port)| port.parse().ok())
+        .expect("the double's address carries its port")
+}
+
+/// The ADR-005 failure postures, one vault-backed source each.
+/// `127.0.0.1:9` is deliberately unbound in every no-dial row's dial
+/// (see the row's doc).
+fn postured_key_source(posture: &'static str) -> (Arc<dyn KeySource>, KeySourceError) {
+    let vault = TestVault::unlocked();
+    let alpha = fixture("alpha");
+    match posture {
+        "locked" => {
+            let blob = vault.seal(alpha);
+            vault.lock();
+            (
+                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
+                KeySourceError::Locked,
+            )
+        }
+        "absent" => (
+            Arc::new(VaultKeySource::new(vault.handle(), None)),
+            KeySourceError::Absent,
+        ),
+        "decryption" => {
+            let mut blob = vault.seal(alpha);
+            blob.data = "not the ciphertext".into();
+            (
+                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
+                KeySourceError::Decryption(String::new()),
+            )
+        }
+        "unknown-version" => {
+            let mut blob = vault.seal(alpha);
+            blob.key_version = 0;
+            (
+                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
+                KeySourceError::UnknownKeyVersion(String::new()),
+            )
+        }
+        "decode" => {
+            let blob = vault.encrypt("this is not a key");
+            (
+                Arc::new(VaultKeySource::new(vault.handle(), Some(blob))),
+                KeySourceError::Decode(String::new()),
+            )
+        }
+        other => panic!("no posture named {other}"),
+    }
+}
+
+/// - *Contract stamp:* the dial surface's success row: decrypt-at-dial
+///   → the policy accept → the single publickey attempt — an
+///   authenticated connection, the dial data carried on its surface,
+///   and the clean close the peer observes.
+#[tokio::test]
+async fn dial_success_lands_an_authenticated_connection() {
+    let (mut double, accepts) = spawn_double().await;
+    accepts.register(fixture("alpha").fingerprint);
+
+    let policy = Arc::new(MemHostKeyPolicy::new());
+    policy.pin("127.0.0.1", fixture("bravo").fingerprint);
+
+    let config = DialConfig::new(
+        double.address.clone(),
+        "runner",
+        alpha_key_source(),
+        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
+    );
+    let connection = dial(config)
+        .await
+        .expect("the dial succeeds through the fake sshd");
+
+    assert_eq!(connection.address(), double.address);
+    assert_eq!(connection.username(), "runner");
+    assert_eq!(connection.limits(), &Limits::default());
+    assert!(!connection.is_closed(), "the connection is live");
+
+    let decisions = policy.decisions();
+    assert_eq!(decisions.len(), 1, "one policy decision rode the dial");
+    assert!(decisions[0].accepted, "the pinned pair accepted");
+    assert_eq!(decisions[0].presented.host, "127.0.0.1");
+    assert_eq!(decisions[0].presented.port, port_of(&double));
+    assert_eq!(
+        decisions[0].presented.fingerprint,
+        fixture("bravo").fingerprint,
+        "the presented fingerprint is the crate's single fingerprint path value"
+    );
+
+    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();
+}
+
+/// `127.0.0.1:9` is deliberately unbound in the no-dial rows: a dial
+/// that reached the socket would fail with a wire class — only a
+/// refused key load can produce the key-source classes here.
+///
+/// - *Contract stamp:* ADR-005 decision 2 (decrypt-at-*dial*) +
+///   decision 4 (every failure mode fails closed, distinct shapes):
+///   every [`KeySourceError`] class is a no-dial outcome, each variant
+///   preserved through [`DialError::KeySource`].
+#[tokio::test]
+async fn every_key_source_posture_is_a_no_dial_outcome() {
+    for posture in [
+        "locked",
+        "absent",
+        "decryption",
+        "unknown-version",
+        "decode",
+    ] {
+        let (source, expected) = postured_key_source(posture);
+        let config = DialConfig::new("127.0.0.1:9", "runner", source, Arc::new(RejectAll));
+        let err = dial(config)
+            .await
+            .expect_err("a refused key source is a no-dial outcome");
+        assert!(
+            matches!(err, DialError::KeySource(ref e) if std::mem::discriminant(e) == std::mem::discriminant(&expected)),
+            "posture {posture}: expected {expected:?} class, got {err:?}"
+        );
+    }
+}
+
+/// - *Contract stamp:* OQ-05 resolution (`host_key.rs`): the crate's
+///   built-in default policy connects to nothing — the dial refuses
+///   as the decline class (`DialError::HostKeyRejected`), never a
+///   wire crash class, never a fallback.
+#[tokio::test]
+async fn the_crate_default_policy_connects_to_nothing() {
+    let (double, _accepts) = spawn_double().await;
+
+    let config = DialConfig::new(
+        double.address.clone(),
+        "runner",
+        alpha_key_source(),
+        Arc::new(RejectAll),
+    );
+    let err = dial(config)
+        .await
+        .expect_err("the default policy declines every presented key");
+    assert!(
+        matches!(err, DialError::HostKeyRejected),
+        "expected HostKeyRejected, got {err:?}"
+    );
+    double.shutdown();
+}
+
+/// - *Contract stamp:* OQ-05 resolution, case 1 (known beforehand):
+///   an assembly pin **hard-declines on mismatch** — the dial aborts;
+///   the policy's own log carries the context (the presented key's
+///   real fingerprint rode the policy; the crate never second-guesses).
+#[tokio::test]
+async fn pin_mismatch_hard_declines() {
+    let (double, _accepts) = spawn_double().await;
+
+    let policy = Arc::new(MemHostKeyPolicy::new());
+    policy.pin("127.0.0.1", fixture("charlie").fingerprint);
+
+    let config = DialConfig::new(
+        double.address.clone(),
+        "runner",
+        alpha_key_source(),
+        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
+    );
+    let err = dial(config)
+        .await
+        .expect_err("a mismatched pin is a hard decline");
+    assert!(matches!(err, DialError::HostKeyRejected), "got {err:?}");
+
+    let decisions = policy.decisions();
+    assert_eq!(decisions.len(), 1);
+    assert!(!decisions[0].accepted);
+    assert_eq!(
+        decisions[0].presented.fingerprint,
+        fixture("bravo").fingerprint,
+        "the presented key (not the pin) is what the policy saw"
+    );
+    double.shutdown();
+}
+
+/// - *Contract stamp:* OQ-05 resolution, case 2 (unknown beforehand):
+///   a permissive assembly policy decides — the dial proceeds; the
+///   crate ships no accept-all of its own (the permissive posture
+///   was this row's explicit choice, `MemHostKeyPolicy::
+///   with_default(true)`).
+#[tokio::test]
+async fn case2_policy_decides_and_the_dial_proceeds() {
+    let (mut double, accepts) = spawn_double().await;
+    accepts.register(fixture("alpha").fingerprint);
+
+    let config = DialConfig::new(
+        double.address.clone(),
+        "runner",
+        alpha_key_source(),
+        Arc::new(MemHostKeyPolicy::with_default(true)),
+    );
+    let connection = dial(config)
+        .await
+        .expect("the explicit permissive policy accepts");
+    assert!(!connection.is_closed());
+
+    connection.close().await;
+    let outcome = double.next_session_outcome().await;
+    assert!(matches!(outcome, Some(Ok(()))), "clean session end");
+    double.shutdown();
+}
+
+/// - *Contract stamp:* the publickey-only family contract: the
+///   remote end refusing the single attempt is the dial's
+///   [`DialError::AuthRejected`] — no retry, no fallback method, the
+///   attempt closed behind the refusal.
+#[tokio::test]
+async fn refused_authentication_is_the_dials_no_connection_outcome() {
+    let (double, _accepts) = spawn_double().await;
+
+    let config = DialConfig::new(
+        double.address.clone(),
+        "runner",
+        alpha_key_source(),
+        Arc::new(MemHostKeyPolicy::with_default(true)),
+    );
+    let err = dial(config)
+        .await
+        .expect_err("the unregistered key is refused by the remote end");
+    assert!(matches!(err, DialError::AuthRejected), "got {err:?}");
+    double.shutdown();
+}
+
+/// - *Contract stamp:* the no-construction-validation contract, the
+///   dial-time side: a malformed address refuses *at the dial* (the
+///   config carries no validation; the wire is never dialed).
+#[tokio::test]
+async fn malformed_address_refuses_at_the_dial() {
+    let config = DialConfig::new(
+        "runner.example",
+        "runner",
+        alpha_key_source(),
+        Arc::new(MemHostKeyPolicy::with_default(true)),
+    );
+    let err = dial(config).await.expect_err("refuses");
+    assert!(
+        matches!(err, DialError::Address(_)),
+        "expected Address, got {err:?}"
+    );
+}
+
+/// - *Contract stamp:* the bounded-config wiring (the task's
+///   acceptance): the dial's russh `Config` carries the
+///   [`Limits`]-mapped knobs — the keepalive 15 s / 3 posture, the
+///   always-set wall clock, the window/packet/buffer knobs verbatim —
+///   observed from outside the crate through the doc-hidden mapping.
+#[test]
+fn the_bounded_config_wiring_is_observable() {
+    let config = russh_client_config(&Limits::default());
+
+    assert_eq!(config.keepalive_interval, Some(Duration::from_secs(15)));
+    assert_eq!(config.keepalive_max, 3);
+    assert_eq!(config.inactivity_timeout, Some(Duration::from_secs(600)));
+    assert_eq!(config.window_size, 2_097_152);
+    assert_eq!(config.maximum_packet_size, 32_768);
+    assert_eq!(config.channel_buffer_size, 100);
+}
+
+/// - *Contract stamp:* ADR-003 §4 (the client-side fail-closed
+///   acceptor): the dialed client's skeleton declines **every**
+///   server-initiated open on the wire — the double's initiator
+///   observes each open as an unconfirmed refusal, never russh's
+///   accept-and-drop default; the never-feature rows (x11, agent)
+///   decline too; the declines consulted no host-key policy; and the
+///   connection lives through all of it (the clean close is the
+///   double's `Ok` session end).
+///
+///   (`should_accept_unknown_server_channel` has no wire row: russh's
+///   server surface exposes no initiator for an unknown type and the
+///   `ChannelId` is deliberately unconstructible outside the library
+///   — the skeleton's explicit `false` is code-pinned, not
+///   row-pinned; recorded in the task.)
+#[tokio::test]
+async fn server_initiated_opens_decline_cleanly_on_the_wire() {
+    event_capture::install();
+    let (mut double, accepts) = spawn_double().await;
+    accepts.register(fixture("alpha").fingerprint);
+
+    let policy = Arc::new(MemHostKeyPolicy::new());
+    policy.pin("127.0.0.1", fixture("bravo").fingerprint);
+    let config = DialConfig::new(
+        double.address.clone(),
+        "runner",
+        alpha_key_source(),
+        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
+    );
+    let connection = dial(config).await.expect("the dial succeeds");
+
+    let server = double.session_handle().await;
+    for outcome in [
+        server.channel_open_session().await,
+        server
+            .channel_open_direct_tcpip("hub.example", 4096, "originator", 0)
+            .await,
+        server
+            .channel_open_forwarded_tcpip("hub.example", 4096, "originator", 0)
+            .await,
+        server
+            .channel_open_forwarded_streamlocal("/run/hub.sock")
+            .await,
+        server.channel_open_x11("originator", 0).await,
+        server.channel_open_agent().await,
+    ] {
+        match outcome {
+            Err(Error::ChannelOpenFailure(reason)) => {
+                assert_eq!(
+                    reason,
+                    ChannelOpenFailure::AdministrativelyProhibited,
+                    "the skeleton's clean decline reason"
+                );
+            }
+            other => panic!("every server-initiated open declines cleanly, got {other:?}"),
+        }
+    }
+    assert_eq!(
+        policy.decisions().len(),
+        1,
+        "the declines consulted no host-key policy — one decision, the dial's"
+    );
+
+    let forwarded = event_capture::await_event(&[
+        ("row", "forwarded-tcpip"),
+        ("decline_class", "open"),
+        ("reason", "no-acceptor"),
+    ])
+    .await;
+    assert_eq!(
+        forwarded.field("message"),
+        Some("decline: channel open rejected")
+    );
+    assert!(forwarded.field("channel_id").is_some());
+    assert_eq!(forwarded.level, "DEBUG");
+    assert_eq!(
+        forwarded.field("feature_build"),
+        Some(event_capture::expected_build())
+    );
+
+    let never = event_capture::await_event(&[
+        ("row", "x11"),
+        ("decline_class", "open"),
+        ("reason", "never-feature"),
+    ])
+    .await;
+    assert_eq!(
+        never.field("message"),
+        Some("decline: channel open rejected")
+    );
+
+    connection.close().await;
+    let outcome = double.next_session_outcome().await;
+    assert!(
+        matches!(outcome, Some(Ok(()))),
+        "the connection stayed usable through the declines: the clean close is a clean session end"
+    );
+    double.shutdown();
+}
+
+/// The smoke posture (`w4-dial`'s decided + recorded item): the
+/// alkssh client ↔ alkssh server in-process loopback over a real TCP
+/// socket — the cheap real-stack smoke (both russh stacks, the full
+/// auth bridge server-side, the full custody + policy chain
+/// client-side). A real-`sshd` dial is wave-5 e2e territory.
+///
+/// - *Contract stamp:* the task's smoke item; `client.md` §The dial
+///   surface (the family-runner end: the dial's fingerprint resolves
+///   through the injected `IdentityProvider` server-side).
+#[tokio::test]
+async fn smoke_alkssh_client_dials_the_alkssh_server_direction() {
+    let provider = Arc::new(MemIdentityProvider::new());
+    provider.register(
+        &fingerprint(&fixture("alpha").public_key()),
+        identity("smoke-runner", &["relay:connect"]),
+    );
+    let serve = ServeConfig::new(
+        provider,
+        Arc::new(MemHostKeySource::from_fixtures(&[fixture("bravo")])),
+    );
+    let adapter = SshAdapter::new(serve).await.expect("the server 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");
+    let (outcome_tx, mut outcome_rx) = tokio::sync::mpsc::unbounded_channel();
+    tokio::spawn(async move {
+        loop {
+            let (socket, _) = match listener.accept().await {
+                Ok(accepted) => accepted,
+                Err(_) => break,
+            };
+            // The smoke has exactly one connection; the handle call
+            // runs it to completion inline (the accept loop makes no
+            // next accept until the session ends).
+            let _ = outcome_tx.send(adapter.handle(socket).await);
+        }
+    });
+
+    let policy = Arc::new(MemHostKeyPolicy::new());
+    policy.pin("127.0.0.1", fixture("bravo").fingerprint);
+    let config = DialConfig::new(
+        addr.to_string(),
+        "runner",
+        alpha_key_source(),
+        Arc::clone(&policy) as Arc<dyn HostKeyPolicy>,
+    );
+    let connection = dial(config)
+        .await
+        .expect("the alkssh client connects to the alkssh server");
+    assert_eq!(policy.decisions().len(), 1);
+    assert_eq!(
+        policy.decisions()[0].presented.fingerprint,
+        fixture("bravo").fingerprint
+    );
+    assert!(!connection.is_closed());
+
+    connection.close().await;
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+    let outcome = tokio::time::timeout_at(deadline, outcome_rx.recv()).await;
+    match outcome {
+        Ok(Some(Ok(()))) => {}
+        Ok(Some(Err(err))) => panic!("the smoke's server session must end clean, got {err:?}"),
+        Ok(None) | Err(_) => panic!("the smoke's server session never ended"),
+    }
+}

tests/common/event_capture.rs

@@ -28,7 +28,8 @@ pub struct Captured {
 }
 
 impl Captured {
-    fn field(&self, name: &str) -> Option<&str> {
+    /// The field's value by name — the rows' matching input.
+    pub fn field(&self, name: &str) -> Option<&str> {
         self.fields
             .iter()
             .find(|(n, _)| n == name)

tests/common/mod.rs

@@ -53,6 +53,12 @@
 /// `tests/common/harness.rs`).
 pub mod harness;
 
+/// The in-process fake `sshd` (`w4-dial`) — the *client-direction*
+/// contract vehicle (the dial surface, the wave-4 client chain; its
+/// own placement/shape decisions live in the module docs:
+/// `tests/common/sshd_double.rs`).
+pub mod sshd_double;
+
 /// The git-seam doubles (`w3-git-mapping`, `git`-gated — alkssh's
 /// hand-rolled minimal alkgit quartet per `alkgit/src/backend.rs`'s
 /// noop shapes; the module's own doc records the decisions:

tests/common/sshd_double.rs

@@ -0,0 +1,294 @@
+//! The in-process fake `sshd` (`w4-dial`) — the **client-direction**
+//! contract-test vehicle: a hand-rolled russh server handler standing
+//! in for the vanilla `sshd` every wave-4 dial row is pointed at (the
+//! reverse-runner shape). The alkssh loopback harness
+//! ([`super::harness`]) drives the *server* direction and is not the
+//! vehicle here — the client dials *vanilla* servers, so this double
+//! is a plain russh server (no alkssh in its serving path), on a real
+//! loopback TCP listener: a dial target is an address the dial
+//! surface resolves and connects to, which a duplex pair has no
+//! endpoint to rehearse.
+//!
+//! **Shape decisions (recorded, the task's placement item):**
+//!
+//! - **Where**: sibling of the wave-1 doubles under `tests/common/` —
+//!   the module-placing convention (test surface, test-versioned
+//!   freely, no public API export, no feature gate; the client
+//!   handler skeleton it exercises compiles in every build).
+//! - **One dial per instance**: doubles are cheap; build a fresh one
+//!   per row. The listener accepts, serves the connection to
+//!   completion on its own task, and loops (a row never re-dials the
+//!   same double).
+//! - **Identity**: the double resolves fingerprints against an
+//!   injected accept set ([`SshdAccepts`], exact-string equality —
+//!   the family's only resolution shape, ADR-004 decision 1, the
+//!   vanilla-parity stand-in for a real `sshd`'s
+//!   `authorized_keys`). Every other auth hook is an explicit reject,
+//!   never an inherited default (the ADR-004 decision 3 hygiene).
+//! - **Host keys**: the fixture host-key pair rides the
+//!   `russh::server::Config`; the presented fingerprint is the
+//!   crate's single fingerprint path value — the rows' pin rules
+//!   match on `fixture("bravo")`'s recorded value.
+//! - **The session handle**: the double's `auth_succeeded` hook hands
+//!   the per-connection [`russh::server::Handle`] to the row through
+//!   a one-shot carrier ([`SshdDouble::session_handle`]) — the row's
+//!   initiator for the server-initiated-open decline rows (server →
+//!   client `channel_open_*`; the dialed client's skeleton declines;
+//!   the server's open call observes
+//!   `Err(ChannelOpenFailure(..))` — the wire rows' observable).
+//! - **The acceptor stub**: rows that need a served session channel
+//!   (the stays-usable-after-declines row here; the op-row vehicles
+//!   later) arm the double to accept client session opens and hand
+//!   the channel out ([`SshdDouble::next_opened_session`]).
+//!
+//! The row/stamp discipline (`tests/common/mod.rs`'s doc) applies
+//! verbatim.
+
+#![allow(dead_code)]
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use parking_lot::Mutex;
+use russh::keys::PrivateKey;
+use russh::server::{Auth, Handle, Msg, PublicKeyOffer, Response, Session, VerifiedPublicKey};
+use russh::{Channel, MethodKind, MethodSet};
+
+/// 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.
+struct SshdDoubleHandler {
+    accepts: Arc<parking_lot::RwLock<Vec<String>>>,
+    accept_session_channels: bool,
+    sessions_out: Arc<Mutex<Option<tokio::sync::oneshot::Sender<Handle>>>>,
+    opened_sessions_out: tokio::sync::mpsc::UnboundedSender<Channel<Msg>>,
+}
+
+impl russh::server::Handler for SshdDoubleHandler {
+    type Error = russh::Error;
+
+    /// The accept-set resolution: a verified key whose fingerprint is
+    /// registered accepts (ADR-004 decision 1's shape, the vanilla
+    /// `authorized_keys` stand-in); anything else rejects fail-closed.
+    async fn auth_publickey(
+        &mut self,
+        _user: &str,
+        key: VerifiedPublicKey<'_>,
+    ) -> Result<Auth, Self::Error> {
+        let fingerprint = alkssh::fingerprint(key.public_key);
+        let accepted = self.accepts.read().contains(&fingerprint);
+        Ok(if accepted {
+            Auth::Accept
+        } else {
+            Auth::reject()
+        })
+    }
+
+    /// Explicit accept-and-delegate (the probe hook consumes no auth
+    /// attempt — POC-1 §4); the real disposition is
+    /// [`Self::auth_publickey`]'s.
+    async fn auth_publickey_offered(
+        &mut self,
+        _user: &str,
+        _offer: PublicKeyOffer<'_>,
+    ) -> Result<Auth, Self::Error> {
+        Ok(Auth::Accept)
+    }
+
+    /// Explicit over inherited (ADR-004 decision 3 hygiene).
+    async fn auth_none(&mut self, _user: &str) -> Result<Auth, Self::Error> {
+        Ok(Auth::reject())
+    }
+
+    /// Explicit over inherited.
+    async fn auth_password(&mut self, _user: &str, _password: &str) -> Result<Auth, Self::Error> {
+        Ok(Auth::reject())
+    }
+
+    /// Explicit over inherited.
+    async fn auth_keyboard_interactive(
+        &mut self,
+        _user: &str,
+        _submethods: &str,
+        _response: Option<Response<'_>>,
+    ) -> Result<Auth, Self::Error> {
+        Ok(Auth::reject())
+    }
+
+    /// The handle carrier: the row initiates server-side opens
+    /// through the returned [`Handle`].
+    async fn auth_succeeded(&mut self, session: &mut Session) -> Result<(), Self::Error> {
+        if let Some(carrier) = self.sessions_out.lock().take() {
+            let _ = carrier.send(session.handle());
+        }
+        Ok(())
+    }
+
+    /// The acceptor stub: armed rows accept client session opens and
+    /// hand the channel out; unarmed rows decline cleanly (the
+    /// vehicle never accept-and-drops either).
+    async fn channel_open_session(
+        &mut self,
+        channel: Channel<Msg>,
+        reply: russh::server::ChannelOpenHandle,
+        _session: &mut Session,
+    ) -> Result<(), Self::Error> {
+        if self.accept_session_channels {
+            self.opened_sessions_out
+                .send(channel)
+                .map_err(|_| russh::Error::SendError)?;
+            reply.accept().await;
+        } else {
+            reply
+                .reject(russh::ChannelOpenFailure::AdministrativelyProhibited)
+                .await;
+        }
+        Ok(())
+    }
+}
+
+/// The fake `sshd`'s accepted-identity register — exact-string
+/// fingerprints (the row registers the fixture dials with).
+pub struct SshdAccepts {
+    inner: Arc<parking_lot::RwLock<Vec<String>>>,
+}
+
+impl SshdAccepts {
+    /// Registers one accepted fingerprint (the vanilla
+    /// `authorized_keys` row's shape).
+    pub fn register(&self, fingerprint: &str) {
+        self.inner.write().push(fingerprint.to_string());
+    }
+}
+
+/// One spawned fake `sshd` — the client-direction contract vehicle.
+pub struct SshdDouble {
+    /// The dial target, `<ip>:<port>` — the row's
+    /// `DialConfig.address` value.
+    pub address: String,
+    session_carrier: Option<tokio::sync::oneshot::Receiver<Handle>>,
+    opened_sessions: tokio::sync::mpsc::UnboundedReceiver<Channel<Msg>>,
+    /// The per-connection serving outcomes (the clean-close rows'
+    /// observable); `None` after the double's end.
+    out_rx: Option<tokio::sync::mpsc::UnboundedReceiver<Result<(), russh::Error>>>,
+    accept_task: tokio::task::JoinHandle<()>,
+}
+
+/// Spawns the double on `127.0.0.1:0`: `host_keys` rides the
+/// `russh::server::Config`; `accept_session_channels` arms the
+/// session-open acceptor stub. Returns the double and its
+/// accept-set register (register before dialing — the wire's auth
+/// reads it at the dial).
+pub async fn spawn(
+    host_keys: Vec<PrivateKey>,
+    accept_session_channels: bool,
+) -> (SshdDouble, SshdAccepts) {
+    let accepts = Arc::new(parking_lot::RwLock::new(Vec::new()));
+    let (session_tx, session_rx) = tokio::sync::oneshot::channel();
+    let (opened_tx, opened_rx) = tokio::sync::mpsc::unbounded_channel();
+    let (outcomes_tx, outcomes_rx) = tokio::sync::mpsc::unbounded_channel();
+    let sessions_out = Arc::new(Mutex::new(Some(session_tx)));
+
+    let config = Arc::new(russh::server::Config {
+        methods: MethodSet::from(&[MethodKind::PublicKey][..]),
+        keys: host_keys,
+        inactivity_timeout: Some(Duration::from_secs(30)),
+        ..Default::default()
+    });
+
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("the double's loopback listener binds");
+    let addr = listener.local_addr().expect("the double has an address");
+
+    let accepts_for_loop = Arc::clone(&accepts);
+    let accept_task = tokio::spawn(async move {
+        loop {
+            let (socket, _) = match listener.accept().await {
+                Ok(accepted) => accepted,
+                Err(_) => break,
+            };
+            let config = Arc::clone(&config);
+            let handler = SshdDoubleHandler {
+                accepts: Arc::clone(&accepts_for_loop),
+                accept_session_channels,
+                sessions_out: Arc::clone(&sessions_out),
+                opened_sessions_out: opened_tx.clone(),
+            };
+            let outcomes = outcomes_tx.clone();
+            tokio::spawn(async move {
+                let session = match russh::server::run_stream(config, socket, handler).await {
+                    Ok(session) => session,
+                    Err(_) => {
+                        let _ = outcomes.send(Err(russh::Error::Disconnect));
+                        return;
+                    }
+                };
+                let _ = outcomes.send(session.await);
+            });
+        }
+    });
+
+    (
+        SshdDouble {
+            address: format!("{}:{}", addr.ip(), addr.port()),
+            session_carrier: Some(session_rx),
+            opened_sessions: opened_rx,
+            out_rx: Some(outcomes_rx),
+            accept_task,
+        },
+        SshdAccepts { inner: accepts },
+    )
+}
+
+impl SshdDouble {
+    /// The dialed session's server-side handle — the initiator for
+    /// the server-initiated-open rows. Awaits the `auth_succeeded`
+    /// carrier; the dial has authenticated by the time a row calls
+    /// this (the carrier's sender only fires inside the double's
+    /// successful auth flow), so the await is prompt.
+    pub async fn session_handle(&mut self) -> Handle {
+        let carrier = self
+            .session_carrier
+            .take()
+            .expect("the double's session handle arrives once (one dial per double)");
+        carrier
+            .await
+            .expect("the double's session handle arrives (the row dialed and authenticated)")
+    }
+
+    /// The client-initiated session channels the acceptor stub
+    /// accepted — `None` after the session's end (the observation
+    /// shape's honest empty).
+    pub async fn next_opened_session(&mut self) -> Option<Channel<Msg>> {
+        self.opened_sessions.recv().await
+    }
+
+    /// The next per-connection serving outcome — the clean-close
+    /// rows' observable (a clean `Disconnect::ByApplication` is the
+    /// sessions' `Ok`; an abrupt peer close is an `Err` wire class).
+    /// `None` after the double's loop ends.
+    pub async fn next_session_outcome(&mut self) -> Option<Result<(), russh::Error>> {
+        let receiver = self.out_rx.as_mut()?;
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
+        tokio::time::timeout_at(deadline, receiver.recv())
+            .await
+            .ok()
+            .flatten()
+    }
+
+    /// Shuts the double down (the accept loop aborts; the connection,
+    /// if still open, drops with it).
+    pub fn shutdown(mut self) {
+        self.out_rx.take();
+        self.accept_task.abort();
+    }
+}
+
+impl std::fmt::Debug for SshdDouble {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("SshdDouble")
+            .field("address", &self.address)
+            .finish()
+    }
+}