feat(handler): enforce the in-crate concurrent-channel cap at the open surface (w2-bounded-resources)

- new private channel_cap module: per-session active-channel count;
  admit (check+increment at accept sites) / has_room / saturating
  release, single-writer (inline packet-loop hooks), unit-tested
- handler wires the cap at every channel_open_* hook: the session
  accept path admits-and-counts at the accept point; the tunnel rows
  cap-gate before the seam stub (a counted-but-unopened stub would
  leak the cap); x11 never consumes a count; the count releases
  through the new channel_close hook
- dispatch::cap_exceeded_open_decline -> AdministrativelyProhibited
  (the assembly's policy ceiling; OpenSSH MaxSessions parity)
- bounded-resources contract rows: cap reached -> clean decline,
  close -> recovery open succeeds (wire-order deterministic),
  zero cap refuses without poisoning, exhausted cap answers the
  tunnel row with the cap reason; max_auth_attempts verified
  behaviorally (within-ceiling failures keep the advertisement, the
  past-ceiling attempt is the NoMoreAuthMethodsAvailable teardown);
  window/packet/buffer/inactivity knobs recorded as covered by the
  config-mapping tests
- max_exec_command_len stays carried, not enforced (wave-3 grammar)

Verification: cargo test (default), --features session / git /
tunnel / --all-features, clippy --all-targets (-D warnings, also
--all-features), fmt --check, cargo doc --no-deps - all green

tests/bounded_resources.rs

@@ -0,0 +1,288 @@
+//! The `w2-bounded-resources` contract rows — the bounded-resources
+//! policy enforced through the real wire via the loopback harness: the
+//! in-crate concurrent-channel cap at every `channel_open_*` accept
+//! path, its recovery on close, its `Limits` tunability, and the
+//! behavioral verification of the wired `max_auth_attempts` russh knob
+//! (the ceiling disconnects an exceeding client). The size/window/
+//! buffer knobs (`window_size`, `maximum_packet_size`,
+//! `channel_buffer_size`), the inactivity wall clock and the remaining
+//! auth-counting posture ride the config-mapping tests
+//! (`adapter.rs` `#[cfg(test)]` + the harness's own
+//! `client_config_maps_the_limits_postures`) — recorded, not
+//! re-proven on the wire here.
+//!
+//! The cap rows are feature-gated to the builds whose opens can be
+//! accepted (session channels need `session` or `git`; the
+//! tunnel-gate ordering row additionally needs `tunnel`), per the
+//! `dispatch_core.rs` feature-build convention.
+
+#![allow(dead_code)]
+
+mod common;
+
+use std::sync::Arc;
+
+use alkcall::core::auth::IdentityProvider;
+use common::harness::{
+    DUPLEX_CAPACITY, HarnessClientHandler, LoopbackSession, authenticate_publickey, client_config,
+    drive, serve_config,
+};
+use common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
+use russh::{ChannelOpenFailure, 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
+}
+
+/// Mounts an authenticated loopback session over the doubles with the
+/// row's [`Limits`] — the bounded-resources rows' driver.
+async fn authenticated_with(limits: Limits) -> LoopbackSession {
+    let provider = registered_provider(&[fixture("alpha").fingerprint]);
+    let adapter = SshAdapter::new(serve_config(
+        provider as Arc<dyn IdentityProvider>,
+        &[fixture("bravo")],
+        limits,
+    ))
+    .await
+    .expect("adapter constructs");
+
+    let mut session = drive(
+        adapter,
+        Arc::new(MemHostKeyPolicy::with_default(true)),
+        limits,
+    )
+    .await
+    .expect("the loopback mounts");
+    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
+    assert!(
+        result.success(),
+        "the fixture identity authenticates before every bounded-resources row"
+    );
+    session
+}
+
+/// Matches an open failure's SSH reason — the wire decline reason the
+/// cap-gate rows assert.
+fn open_failure(err: Error, expected: ChannelOpenFailure, context: &str) {
+    assert!(
+        matches!(
+            &err,
+            Error::ChannelOpenFailure(reason) if *reason == expected
+        ),
+        "{context}: expected open decline with {expected:?}, got {err:?}"
+    );
+}
+
+/// - *Contract stamp:* `overview.md` §Bounded-resources policy (the
+///   auth-attempts row — the `max_auth_attempts` knob is behavioral,
+///   verified where cheap: the ceiling disconnects an exceeding
+///   client) + ADR-004 decision 6 (the mapping in `adapter.rs` makes
+///   the knob the counting ceiling; within the ceiling, failures stay
+///   ordinary `AuthResult::Failure` riding the publickey-only
+///   advertisement).
+///
+/// Drive: `max_auth_attempts: 2` (Limits-tunable, not the 10 default),
+/// three unregistered keys. The first two rejections stay ordinary
+/// failures with the advertisement intact; the third request past the
+/// ceiling is answered with the server's
+/// `Disconnect::NoMoreAuthMethodsAvailable` — client-side, the
+/// in-flight attempt sees russh's synthesized `Failure` with the
+/// **empty** remaining-methods set (no reply ever came; the
+/// advertisement is gone). The session ends cleanly server-side.
+#[tokio::test]
+async fn auth_attempt_ceiling_disconnects_an_exceeding_client() {
+    use russh::MethodKind;
+    use russh::keys::PrivateKeyWithHashAlg;
+
+    let provider = registered_provider(&[fixture("alpha").fingerprint]);
+    let limits = Limits {
+        max_auth_attempts: 2,
+        ..Limits::default()
+    };
+    let adapter = SshAdapter::new(serve_config(
+        provider as Arc<dyn IdentityProvider>,
+        &[fixture("bravo")],
+        limits,
+    ))
+    .await
+    .expect("adapter constructs");
+
+    let (client_io, server_io) = tokio::io::duplex(DUPLEX_CAPACITY);
+    let server_task = tokio::spawn(async move { adapter.handle(server_io).await });
+    let mut client = russh::client::connect_stream(
+        client_config(limits),
+        client_io,
+        HarnessClientHandler::new(Arc::new(MemHostKeyPolicy::with_default(true))),
+    )
+    .await
+    .expect("the in-process pair connects");
+
+    let normal_advertisement = MethodSet::from(&[MethodKind::PublicKey][..]);
+    let within = authenticate_publickey(&mut client, "anyone", fixture("delta")).await;
+    assert!(
+        !within.success(),
+        "within the ceiling, a rejection stays an ordinary failure: {within:?}"
+    );
+    assert_eq!(
+        remaining_ads(&within),
+        Some(normal_advertisement.clone()),
+        "the within-ceiling failure still carries the advertisement"
+    );
+    let last_within = authenticate_publickey(&mut client, "anyone", fixture("charlie")).await;
+    assert!(
+        !last_within.success(),
+        "the ceiling's last rejection still rides as an ordinary failure: {last_within:?}"
+    );
+    assert_eq!(
+        remaining_ads(&last_within),
+        Some(normal_advertisement),
+        "the advertisement survives every within-ceiling rejection"
+    );
+
+    let exceeding = async {
+        let key = Arc::new(fixture("bravo").private_key());
+        client
+            .authenticate_publickey("anyone", PrivateKeyWithHashAlg::new(key, None))
+            .await
+    }
+    .await;
+    match &exceeding {
+        Ok(result) => {
+            assert!(
+                !result.success(),
+                "the attempt past the ceiling never re-answers auth: {result:?}"
+            );
+            assert_eq!(
+                remaining_ads(result),
+                Some(MethodSet::empty()),
+                "the past-ceiling answer is the torn-down shape: no advertisement rides"
+            );
+        }
+        Err(_) => {
+            // The client loop already exited on the disconnect — the
+            // equally-valid torn-down shape (the send side is gone).
+        }
+    }
+
+    drop(client);
+    server_task
+        .await
+        .expect("server task ends")
+        .expect("the ceiling disconnect ends the session cleanly");
+}
+
+/// The remaining-methods set of an [`AuthResult`] — `None` on success
+/// (the within-ceiling vs past-ceiling observation shape).
+fn remaining_ads(result: &russh::client::AuthResult) -> Option<russh::MethodSet> {
+    match result {
+        russh::client::AuthResult::Failure {
+            remaining_methods, ..
+        } => Some(remaining_methods.clone()),
+        _ => None,
+    }
+}
+
+#[cfg(any(feature = "session", feature = "git"))]
+#[tokio::test]
+async fn concurrent_channel_cap_declines_and_recovers_over_the_wire() {
+    use common::harness::open_session_channel;
+
+    let session = authenticated_with(Limits {
+        max_concurrent_channels: 2,
+        ..Limits::default()
+    })
+    .await;
+
+    let first = open_session_channel(&session.client)
+        .await
+        .expect("the first open admits under the cap");
+    let _second = open_session_channel(&session.client)
+        .await
+        .expect("the second open admits under the cap");
+
+    let err = open_session_channel(&session.client)
+        .await
+        .expect_err("the third open is refused at the cap — never queued");
+    open_failure(
+        err,
+        ChannelOpenFailure::AdministrativelyProhibited,
+        "the cap-exceeded open",
+    );
+
+    first.close().await.expect("the close rides the wire");
+
+    // The close reply is not observable client-side at this pin: russh
+    // removes a locally-closed channel from its map before the
+    // server's CHANNEL_CLOSE arrives back, so no message — and no
+    // `None` — is ever delivered (the harness's `await_close` only
+    // observes *server-initiated* closes). The recovery observation is
+    // the open itself, and it is deterministic: close and open ride
+    // the client loop's single FIFO event queue, they are written in
+    // that order on the wire, and the server's packet loop processes
+    // them in order — the counted close releases before the recovery
+    // open is admitted. Had the release not happened, this open
+    // declines and this expect fails.
+    let _third = open_session_channel(&session.client)
+        .await
+        .expect("a close recovers the cap; the next open succeeds again");
+
+    session.unmount().await;
+}
+
+#[cfg(any(feature = "session", feature = "git"))]
+#[tokio::test]
+async fn zero_cap_refuses_every_open_and_poisons_nothing() {
+    use common::harness::open_session_channel;
+
+    let session = authenticated_with(Limits {
+        max_concurrent_channels: 0,
+        ..Limits::default()
+    })
+    .await;
+
+    for attempt in 1..=2 {
+        let err = open_session_channel(&session.client)
+            .await
+            .expect_err("a zero cap refuses the very first open — no half-open, no queue");
+        open_failure(
+            err,
+            ChannelOpenFailure::AdministrativelyProhibited,
+            &format!("the zero-cap open #{attempt}"),
+        );
+    }
+
+    session.unmount().await;
+}
+
+#[cfg(all(feature = "tunnel", any(feature = "session", feature = "git")))]
+#[tokio::test]
+async fn an_exhausted_cap_answers_the_tunnel_row_with_the_cap_reason() {
+    use common::harness::{open_direct_tcpip, open_session_channel};
+
+    let session = authenticated_with(Limits {
+        max_concurrent_channels: 1,
+        ..Limits::default()
+    })
+    .await;
+
+    let _first = open_session_channel(&session.client)
+        .await
+        .expect("the first open admits — the cap is now exhausted");
+
+    let err = open_direct_tcpip(&session.client, "internal.example", 5432)
+        .await
+        .expect_err("the cap gates the tunnel row before the seam stub");
+    open_failure(
+        err,
+        ChannelOpenFailure::AdministrativelyProhibited,
+        "the cap-exhausted tunnel open",
+    );
+
+    session.unmount().await;
+}