feat(tests): land the loopback harness — the in-process contract-test vehicle (w2-loopback-harness)

- tests/common/harness.rs: the recorded transport decision (tokio
  duplex pair + russh connect_stream; recorded over TCP loopback for
  determinism, no-port portability) and placement decision (sibling
  of the wave-1 doubles, the crate's contract-test vehicle convention)
- the client-side posture from the doubles: MemHostKeyPolicy bridged
  into russh's check_server_key (HarnessClientHandler, decision input
  byte-equal to the wave-4 dial wiring's), the Limits keepalive/
  inactivity posture (client_config), fixture-key auth helpers
- thin helpers: authenticate_publickey + remaining_methods,
  open_session_channel (open-failure reason observed in the Err),
  request_exec/shell/subsystem, next_message / await_exit_status /
  await_close, drive → LoopbackSession with clean unmount
- acceptance proof in tests/loopback_harness.rs: 6 contract-stamped
  rows, incl. the full in-process connect → fixture-key auth →
  authenticated-session observation (host-key decision log +
  observed decline reason) and the fail-closed host-key decline

Verification: cargo test (81 pass), cargo clippy --all-targets
-- -D warnings, cargo fmt --check, taskgraph validate — all green.

tests/common/harness.rs

@@ -0,0 +1,314 @@
+//! alkssh's loopback harness (`w2-loopback-harness`) — the in-process
+//! client/server pair that is *the* contract-test vehicle for the
+//! crate: the wave-2 decline matrix, every wave-3 mapping's contract
+//! tests, and the client-direction smokes all drive the crate through
+//! this module.
+//!
+//! **Transport decision (recorded, the `w2-loopback-harness` flag):**
+//! the server-side [`SshAdapter::handle`](alkssh::SshAdapter::handle)
+//! mount runs over a **tokio duplex pair** ([`tokio::io::duplex`],
+//! [`DUPLEX_CAPACITY`] bytes each direction), and the client side
+//! connects with [`russh::client::connect_stream`] — russh's real
+//! codec, kex and transport run over the pair verbatim, so the wire
+//! path is fully exercised. The one thing it deliberately does not
+//! carry is OS-socket semantics; TCP loopback (`127.0.0.1:0`) was the
+//! alternative and is rejected: real-socket nondeterminism (listener
+//! lifecycle, loopback timing, accept-loop flakiness under slow CI)
+//! buys only what the real-tool smokes of the mapping waves and the
+//! e2e wave already cover cheaply. The duplex pair is deterministic,
+//! needs no port, and cannot collide — the no-TCP-port-dependency
+//! shape the acceptance criteria name.
+//!
+//! **Placement decision (recorded, the `w2-loopback-harness`
+//! counterpart of the wave-1 doubles' location decision):** the
+//! harness lives as a sibling module of the wave-1 doubles, under
+//! `tests/common/`, reached with the same `mod common;` convention —
+//! test surface, test-versioned freely, never a public API export
+//! (no re-export from `lib.rs`, no feature gate). Unlike the doubles
+//! it is a *vehicle* convention, not just a seam: later waves'
+//! contract tests do not re-derive a mount/connect pair; they call
+//! [`drive`] (or compose its parts), so a transport revisit stays a
+//! one-module change.
+//!
+//! **Client-side posture (recorded):** the wave-4 dial surface does
+//! not exist yet; the harness drives a russh client directly with the
+//! doubles: a fixture key ([`authenticate_publickey`]), the
+//! [`MemHostKeyPolicy`] decision double bridged into russh's
+//! `check_server_key` hook ([`HarnessClientHandler`]), and the
+//! keepalive / wall-clock posture from [`alkssh::Limits`]
+//! ([`client_config`]). The permissive posture for routine rows is
+//! the double's `with_default(true)` — an explicit test choice; the
+//! accept-all posture exists only here, never as a crate built-in,
+//! and the fail-closed rows construct `with_default(false)`.
+//!
+//! **Host identity on the loopback (recorded):** a duplex pair has no
+//! endpoint, so the presented host key is always labeled with
+//! [`LOOPBACK_HOST`] and [`LOOPBACK_PORT`]; host-key policy rows pin
+//! rules against those values verbatim. The presented key's
+//! fingerprint is the crate's single fingerprint path
+//! ([`alkssh::fingerprint`]) — exactly the decision input the wave-4
+//! dial wiring will compute at russh's own hook, so this bridge is
+//! that wiring's rehearsal.
+//!
+//! Helpers stay thin — the russh client API is the vocabulary; these
+//! remove boilerplate (config assembly, mount/spawn, fixture-key
+//! plumbing, drain loops), never semantics: every observation is a
+//! russh-verbatim shape ([`AuthResult`], [`Error`],
+//! [`ChannelMsg`]) the rows match on directly.
+//!
+//! Like the doubles: test functions are *rows* carrying the greppable
+//! **`Contract stamp:`** doc markers (see the common module's doc).
+
+#![allow(dead_code)]
+
+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 alkssh::{
+    AdapterError, HostKeyPolicy, Limits, PresentedHostKey, ServeConfig, SshAdapter, fingerprint,
+};
+
+use super::{FixtureKey, MemHostKeyPolicy, MemHostKeySource};
+
+/// The host label the loopback presents to the host-key policy — a
+/// duplex pair has no endpoint, so every decision is labeled with
+/// this constant (see the module docs §Host identity).
+pub const LOOPBACK_HOST: &str = "loopback";
+
+/// The port label for the loopback's presented host key — `0`, the
+/// no-port marker: the transport has no TCP port by design.
+pub const LOOPBACK_PORT: u16 = 0;
+
+/// Each duplex direction's buffer — the `w2-server-mount` wire-smoke
+/// precedent's capacity, kept as the harness convention.
+pub const DUPLEX_CAPACITY: usize = 65_536;
+
+/// The harness's russh client config — the keepalive / wall-clock
+/// posture from [`Limits`]: `keepalive_interval` / `keepalive_max`
+/// mapped verbatim, `inactivity_timeout` carried as always-set (the
+/// crate never constructs a session with no wall clock,
+/// [`alkssh::Limits`]' in-crate posture extended to the harness's
+/// client direction). Everything else stays russh's own default.
+pub fn client_config(limits: Limits) -> Arc<russh::client::Config> {
+    Arc::new(russh::client::Config {
+        inactivity_timeout: Some(limits.inactivity_timeout),
+        keepalive_interval: limits.keepalive_interval,
+        keepalive_max: limits.keepalive_max,
+        ..Default::default()
+    })
+}
+
+/// 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).
+pub struct HarnessClientHandler {
+    policy: Arc<MemHostKeyPolicy>,
+}
+
+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.
+    pub fn new(policy: Arc<MemHostKeyPolicy>) -> Self {
+        Self { policy }
+    }
+}
+
+impl russh::client::Handler for HarnessClientHandler {
+    type Error = Error;
+
+    /// The policy decision, verbatim: the presented key's fingerprint
+    /// is computed on the crate's single fingerprint path, the
+    /// endpoint labels are the loopback constants, and the policy's
+    /// `bool` verdict rides russh's hook straight through.
+    async fn check_server_key(
+        &mut self,
+        server_public_key: &russh::keys::PublicKeyOrCertificate,
+    ) -> Result<bool, Self::Error> {
+        let presented = PresentedHostKey {
+            host: LOOPBACK_HOST.into(),
+            port: LOOPBACK_PORT,
+            fingerprint: fingerprint(&server_public_key.public_key()),
+        };
+        Ok(self.policy.decide(&presented).await)
+    }
+}
+
+/// Assembles the server side from the wave-1 doubles — a
+/// [`ServeConfig`] over the given identity provider and fixture host
+/// keys, with the session limits applied.
+pub fn serve_config(
+    provider: Arc<dyn IdentityProvider>,
+    host_keys: &[&'static FixtureKey],
+    limits: Limits,
+) -> ServeConfig {
+    let mut serve = ServeConfig::new(
+        provider,
+        Arc::new(MemHostKeySource::from_fixtures(host_keys)),
+    );
+    serve.limits = limits;
+    serve
+}
+
+/// Constructs the adapter from the doubles — [`serve_config`] with the
+/// default [`Limits`]; the decrypt-at-construction failure paths are
+/// the construction tests' business (`w2-server-mount`), so a harness
+/// row's construction failing is a bug, not an observation.
+pub async fn adapter(
+    provider: Arc<dyn IdentityProvider>,
+    host_keys: &[&'static FixtureKey],
+) -> SshAdapter {
+    SshAdapter::new(serve_config(provider, host_keys, Limits::default()))
+        .await
+        .expect("adapter constructs (fixture host keys load)")
+}
+
+/// One driven loopback session: the mounted server task plus the
+/// connected client handle and its policy double.
+pub struct LoopbackSession {
+    /// The connected russh client — the helpers' driver.
+    pub client: Handle<HarnessClientHandler>,
+    /// The policy double the connection's host-key decision rode —
+    /// decision-log inspection for the rows.
+    pub host_key_policy: Arc<MemHostKeyPolicy>,
+    server_task: tokio::task::JoinHandle<Result<(), AdapterError>>,
+}
+
+impl LoopbackSession {
+    /// Ends one loopback row cleanly: disconnects the client, waits
+    /// for the server task, and requires a clean session end on both
+    /// sides — the harness's every-row termination posture.
+    pub async fn unmount(self) {
+        self.client
+            .disconnect(Disconnect::ByApplication, "row done", "en")
+            .await
+            .expect("client disconnects cleanly");
+        let outcome = self.server_task.await.expect("server task ends");
+        if let Err(err) = outcome {
+            panic!("session end must be clean, got {err:?}");
+        }
+    }
+}
+
+/// Drives one loopback session: mounts the adapter over a duplex pair
+/// (spawned to completion, the assembly's accept-loop shape), connects
+/// a harness client against it and returns the pair. The adapter is
+/// consumed (the spawned task owns it) — build a fresh one per row;
+/// the doubles make that cheap.
+///
+/// Fallible at the connect step on purpose: a fail-closed host-key
+/// policy verdict is a disposition the rows observe
+/// (`Error::UnknownKey`); a routine row's `Err` is its bug.
+pub async fn drive(
+    adapter: SshAdapter,
+    host_key_policy: Arc<MemHostKeyPolicy>,
+    limits: Limits,
+) -> 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?;
+
+    Ok(LoopbackSession {
+        client,
+        host_key_policy,
+        server_task,
+    })
+}
+
+/// Authenticates one fixture key — the auth helper: the fixture
+/// plumbing (`Arc` + `PrivateKeyWithHashAlg`) removed; the
+/// [`AuthResult`] comes back verbatim for the success / failure-shape
+/// rows to match on.
+pub async fn authenticate_publickey(
+    client: &mut Handle<HarnessClientHandler>,
+    username: &str,
+    key: &FixtureKey,
+) -> AuthResult {
+    let private = Arc::new(key.private_key());
+    client
+        .authenticate_publickey(username, PrivateKeyWithHashAlg::new(private, None))
+        .await
+        .expect("the auth exchange completes over the wire")
+}
+
+/// The failure shape's advertisement half: the remaining-methods set
+/// carried by an [`AuthResult::Failure`] — `None` when the result is
+/// a success (the rows' success/failure observation shape).
+pub fn remaining_methods(result: &AuthResult) -> Option<MethodSet> {
+    match result {
+        AuthResult::Failure {
+            remaining_methods, ..
+        } => Some(remaining_methods.clone()),
+        _ => None,
+    }
+}
+
+/// Opens a session channel — the open helper. Success is a confirmed
+/// [`Channel`]; the open-failure shape is the `Err` verbatim
+/// ([`Error::ChannelOpenFailure`] carries the SSH decline reason the
+/// decline-matrix rows match on).
+pub async fn open_session_channel(
+    client: &Handle<HarnessClientHandler>,
+) -> Result<Channel<Msg>, Error> {
+    client.channel_open_session().await
+}
+
+/// Sends an exec request over a channel (`want_reply` set — the
+/// success/failure shape rides [`ChannelMsg::Success`]/[`ChannelMsg::Failure`]
+/// through [`next_message`]).
+pub async fn request_exec(channel: &Channel<Msg>, command: &str) -> Result<(), Error> {
+    channel.exec(true, command).await
+}
+
+/// Sends a shell request over a channel (`want_reply` set).
+pub async fn request_shell(channel: &Channel<Msg>) -> Result<(), Error> {
+    channel.request_shell(true).await
+}
+
+/// Sends a subsystem request over a channel (`want_reply` set).
+pub async fn request_subsystem(channel: &Channel<Msg>, name: &str) -> Result<(), Error> {
+    channel.request_subsystem(true, name).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> {
+    channel.wait().await
+}
+
+/// Awaits the channel's exit status — drains interim messages until
+/// [`ChannelMsg::ExitStatus`] (the wave-3 derivation contract's
+/// observation shape) or the channel's end, which is `None`.
+pub async fn await_exit_status(channel: &mut Channel<Msg>) -> Option<u32> {
+    loop {
+        match next_message(channel).await {
+            Some(ChannelMsg::ExitStatus { exit_status }) => return Some(exit_status),
+            Some(_) => continue,
+            None => return None,
+        }
+    }
+}
+
+/// Awaits the channel's close — drains interim messages until
+/// [`ChannelMsg::Close`] (observed: `true`) or the channel's end
+/// (`false`).
+pub async fn await_close(channel: &mut Channel<Msg>) -> bool {
+    loop {
+        match next_message(channel).await {
+            Some(ChannelMsg::Close) => return true,
+            Some(_) => continue,
+            None => return false,
+        }
+    }
+}

tests/common/mod.rs

@@ -48,6 +48,11 @@
 
 #![allow(dead_code)]
 
+/// The loopback harness — *the* contract-test vehicle (its own
+/// location/transport decisions live in its module docs:
+/// `tests/common/harness.rs`).
+pub mod harness;
+
 use std::collections::HashMap;
 
 use alkcall::core::auth::{AuthToken, Identity, IdentityProvider};

tests/loopback_harness.rs

@@ -0,0 +1,242 @@
+//! The `w2-loopback-harness` acceptance rows — the harness proved
+//! live: a full in-process connect → authenticate (fixture key through
+//! `MemIdentityProvider`) → authenticated-session observation through
+//! the wave-2 stack, plus the observation shapes (auth failure,
+//! open-failure reason, host-key verdicts) the later waves' contract
+//! tests speak through the same helpers.
+//!
+//! Every row drives the crate through the harness (`common::harness`)
+//! — the recorded transport (tokio duplex pair, no TCP port) and the
+//! recorded client posture (fixture keys, `MemHostKeyPolicy`, the
+//! `Limits` keepalive/wall-clock posture) are the harness's own
+//! decisions, exercised here.
+//!
+//! The rows are contract stamps: each names the pinning ADR § (the
+//! greppable `Contract stamp:` inventory applies).
+
+mod common;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use alkcall::core::auth::IdentityProvider;
+use common::harness::{
+    LOOPBACK_HOST, LOOPBACK_PORT, authenticate_publickey, client_config, drive,
+    open_session_channel, remaining_methods, serve_config,
+};
+use common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
+use russh::{Error, MethodSet};
+
+use alkssh::{Limits, SshAdapter};
+
+fn registered_provider(fingerprints: &[&str]) -> Arc<MemIdentityProvider> {
+    let provider = Arc::new(MemIdentityProvider::new());
+    for (index, registered) in fingerprints.iter().enumerate() {
+        provider.register(registered, identity(&format!("user-{index}"), &["ssh"]));
+    }
+    provider
+}
+
+fn permissive_policy() -> Arc<MemHostKeyPolicy> {
+    Arc::new(MemHostKeyPolicy::with_default(true))
+}
+
+/// - *Contract stamp:* ADR-001 (the mount's per-stream edge
+///   termination, run to completion over the harness's recorded
+///   in-process transport) + ADR-004 decision 1 (the fixture key's
+///   fingerprint accepts through `MemIdentityProvider`) + ADR-005
+///   decision 6 (the host key set loads decrypt-at-construction from
+///   `MemHostKeySource` — the wave-1 doubles live through the wave-2
+///   stack).
+#[tokio::test]
+async fn in_process_pair_connects_and_authenticates_end_to_end() {
+    let provider = registered_provider(&[fixture("alpha").fingerprint]);
+    let adapter = SshAdapter::new(serve_config(
+        provider as Arc<dyn IdentityProvider>,
+        &[fixture("bravo")],
+        Limits::default(),
+    ))
+    .await
+    .expect("adapter constructs");
+
+    let host_key_policy = permissive_policy();
+    let mut session = drive(adapter, host_key_policy, Limits::default())
+        .await
+        .expect("the in-process pair connects");
+
+    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
+    assert!(
+        result.success(),
+        "the fixture key authenticates: {result:?}"
+    );
+
+    let decisions = session.host_key_policy.decisions();
+    assert_eq!(
+        decisions.len(),
+        1,
+        "the host-key decision rode the MemHostKeyPolicy double: {decisions:?}"
+    );
+    assert!(decisions[0].accepted, "the permissive posture accepted");
+    assert_eq!(decisions[0].presented.host, LOOPBACK_HOST);
+    assert_eq!(decisions[0].presented.port, LOOPBACK_PORT);
+    assert_eq!(
+        decisions[0].presented.fingerprint,
+        fixture("bravo").fingerprint,
+        "the presented host key is the adapter's fixture set, fingerprinted on the crate's path"
+    );
+
+    session.unmount().await;
+}
+
+/// - *Contract stamp:* ADR-001 (the authenticated session is live and
+///   serving over the mount) + `mapping.md` §Decline matrix (the
+///   skeleton's fail-closed decline is observable through the harness
+///   as a ChannelOpenFailure reason — the *specific* reason is
+///   `w2-dispatch-decline-core`'s row, not this vehicle's).
+#[tokio::test]
+async fn authenticated_session_serves_and_declines_with_an_observed_reason() {
+    let provider = registered_provider(&[fixture("alpha").fingerprint]);
+    let adapter = SshAdapter::new(serve_config(
+        provider as Arc<dyn IdentityProvider>,
+        &[fixture("bravo")],
+        Limits::default(),
+    ))
+    .await
+    .expect("adapter constructs");
+
+    let mut session = drive(adapter, permissive_policy(), Limits::default())
+        .await
+        .expect("the in-process pair connects");
+
+    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
+    assert!(result.success(), "authenticated: {result:?}");
+
+    let opened = open_session_channel(&session.client).await;
+    assert!(
+        matches!(opened, Err(Error::ChannelOpenFailure(_))),
+        "the authenticated session answers a channel open with a decline reason: {opened:?}"
+    );
+
+    session.unmount().await;
+}
+
+/// - *Contract stamp:* ADR-004 decision 1 (`None` → reject — the
+///   failure shape rides the harness with the advertisement intact)
+///   + ADR-004 decision 4 (the remaining-methods set is exactly
+///   `{publickey}`).
+#[tokio::test]
+async fn auth_failure_shape_and_remaining_advertisement_are_observed() {
+    let provider = registered_provider(&[fixture("alpha").fingerprint]);
+    let adapter = SshAdapter::new(serve_config(
+        provider as Arc<dyn IdentityProvider>,
+        &[fixture("bravo")],
+        Limits::default(),
+    ))
+    .await
+    .expect("adapter constructs");
+
+    let mut session = drive(adapter, permissive_policy(), Limits::default())
+        .await
+        .expect("the in-process pair connects");
+
+    let result = authenticate_publickey(&mut session.client, "anyone", fixture("delta")).await;
+    assert!(!result.success(), "the unregistered key must fail");
+    let remaining =
+        remaining_methods(&result).expect("the failure shape carries the remaining-methods set");
+    assert_eq!(
+        remaining,
+        MethodSet::from(&[russh::MethodKind::PublicKey][..]),
+        "the publickey-only advertisement survives the decline"
+    );
+
+    session.unmount().await;
+}
+
+/// - *Contract stamp:* `auth.md` §Host-key policy (the crate is
+///   policy-blind; the fail-closed verdict is the policy's, and the
+///   reject disposition rides russh's hook as a hard, clean decline).
+#[tokio::test]
+async fn fail_closed_host_key_policy_verdict_disconnects_the_dial() {
+    let provider = registered_provider(&[]);
+    let adapter = SshAdapter::new(serve_config(
+        provider as Arc<dyn IdentityProvider>,
+        &[fixture("bravo")],
+        Limits::default(),
+    ))
+    .await
+    .expect("adapter constructs");
+
+    let rejecting_policy = Arc::new(MemHostKeyPolicy::with_default(false));
+    let outcome = drive(adapter, rejecting_policy, Limits::default()).await;
+
+    let err = outcome
+        .err()
+        .expect("a rejecting policy never connects (the drive errs with the decline)");
+    assert!(
+        matches!(err, Error::UnknownKey),
+        "the reject disposition is the clean dial decline: {err:?}"
+    );
+}
+
+/// - *Contract stamp:* `auth.md` §Host-key policy case 1 (the
+///   trust-anchor shape: an exact `(host, fingerprint)` pin accepts
+///   with a fail-closed default for everything else) — the policy
+///   double's rule vocabulary driven over the wire.
+#[tokio::test]
+async fn pinned_policy_rule_serves_the_trust_anchor_case() {
+    let provider = registered_provider(&[fixture("alpha").fingerprint]);
+    let adapter = SshAdapter::new(serve_config(
+        provider as Arc<dyn IdentityProvider>,
+        &[fixture("bravo")],
+        Limits::default(),
+    ))
+    .await
+    .expect("adapter constructs");
+
+    let pinned_policy = Arc::new(MemHostKeyPolicy::with_default(false));
+    pinned_policy.pin(LOOPBACK_HOST, fixture("bravo").fingerprint);
+
+    let mut session = drive(adapter, pinned_policy, Limits::default())
+        .await
+        .expect("the pin accepts the presented host key");
+    assert_eq!(
+        session.host_key_policy.decisions().len(),
+        1,
+        "one decision, from the pin"
+    );
+
+    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
+    assert!(result.success(), "the trust-anchored session authenticates");
+
+    session.unmount().await;
+}
+
+/// - *Contract stamp:* `overview.md` §Bounded-resources policy (the
+///   client direction's `Limits` postures map into the harness
+///   client config verbatim: keepalive interval/max, wall clock
+///   always set).
+#[test]
+fn client_config_maps_the_limits_postures() {
+    let limits = Limits {
+        keepalive_interval: Some(Duration::from_secs(7)),
+        keepalive_max: 11,
+        inactivity_timeout: Duration::from_secs(1_337),
+        ..Limits::default()
+    };
+    let config = client_config(limits);
+
+    assert_eq!(config.keepalive_interval, Some(Duration::from_secs(7)));
+    assert_eq!(config.keepalive_max, 11);
+    assert_eq!(config.inactivity_timeout, Some(Duration::from_secs(1_337)));
+
+    let default_config = client_config(Limits::default());
+    assert_eq!(
+        default_config.keepalive_interval,
+        alkssh::Limits::default().keepalive_interval,
+        "the default posture rides by default"
+    );
+    assert_eq!(
+        default_config.inactivity_timeout,
+        Some(alkssh::Limits::default().inactivity_timeout)
+    );
+}