//! The per-connection server handler — the ADR-004 dispositional auth
//! bridge, plus the exec-grammar dispatch and the fail-closed decline
//! core (`w2-dispatch-decline-core`; `w2-auth-bridge` filled the auth
//! hooks; `w3-git-mapping` replaced the skeleton probe with the full
//! grammar, landed the git row's alkgit dispatch, and with it the
//! store-and-take channel handoff).
//!
//! ## The auth bridge (ADR-004, decisions 1–5)
//!
//! russh performs all cryptographic work: [`Handler::auth_publickey`]
//! fires *after* signature and ownership verification, so this handler
//! never sees or verifies a signature — its surface is dispositions
//! only. The single resolution path: ed25519-only gate → the crate's
//! one fingerprint path ([`crate::fingerprint::fingerprint`]) → the
//! injected `IdentityProvider::resolve_from_fingerprint` → `Some`
//! accepts with the resolved identity stored in per-connection handler
//! state (it rides the session for the wave-3 seam propagation —
//! [`Handler::auth_succeeded`] is the ride-along hook); `None` rejects
//! fail-closed — the resolved-identity-`None` path never creates a
//! weakly authorized session.
//!
//! Every other method is an **explicit** reject — [`Handler::auth_none`],
//! [`Handler::auth_password`] and [`Handler::auth_keyboard_interactive`]
//! each implemented, never inherited — so the posture survives a pin
//! bump that could silently change russh's defaults (ADR-004 decision
//! 3). The non-ed25519 algorithm decline is explicit-over-inherited too
//! (ADR-002 decision 4). [`Handler::auth_publickey_offered`]
//! accepts-and-delegates explicitly (the probe hook consumes no attempt
//! — POC-1 verified; the delegation keeps `max_auth_attempts` counting
//! honest).
//!
//! The username carries no semantics (OQ-03 resolved): resolution is
//! fingerprint-only — the bridge core takes no username parameter at
//! all, structurally refusing any identity input from the user string.
//! Auth declines are *dispositions* (`Auth::Reject`), not errors — the
//! error model stays key-source-only.
//!
//! ## The parse-only dispatch + the fail-closed decline core
//!
//! (ADR-006 point 3 as amended, ADR-003; the routing decision and the
//! exec-parse skeleton probe live in [`crate::dispatch`].) Every
//! `channel_open_*` hook gates per the feature graph: `session`
//! channels open only where a session-serving feature exists (`session`
//! ∨ `git` — the parse path lives inside session channels);
//! `direct-tcpip` / `forwarded-tcpip` / `direct-streamlocal` gate on
//! the `tunnel` feature; **`x11` always declines** — never
//! feature-gated (ADR-003 §5). Unfeatured and never-feature rows
//! decline **at the open** with a clean wire reason
//! ([`crate::dispatch::unfeatured_open_decline`]) — never a
//! half-open channel. The `direct-tcpip` / `direct-streamlocal`
//! featured rows are the route-1 bridge (`w3-tunnel-direct`,
//! [`crate::tunnel_map`]): the pre-admit decline classes (the unset
//! fabric handle under the feature — the unset-declines posture; the
//! unreachable-in-practice `no-identity`; an impossible wire target),
//! the cap admission at the hook, then the spawned bridge — accept
//! only after the fabric open succeeds, `ConnectFailed` on failure
//! ([`crate::dispatch::connect_failed_open_decline`], the wave-2
//! seam stub's reason surviving as the real decline).
//! An **inbound** `forwarded-tcpip` open — arriving *from* the vanilla
//! client — declines with the same explicit `ConnectFailed` as the
//! unvanilla-arrival class (`w3-tunnel-listen`: the real `-R` data
//! leg is the door's own server-initiated opens, driven by the
//! [`crate::tunnel_map`] arrival pump; a vanilla client never
//! initiates the type, OpenSSH parity).
//! The bare
//! `auth-agent@openssh.com` channel *type*
//! (`forwarded-streamlocal@openssh.com` likewise) has no russh handler
//! hook at this pin — russh rejects those opens inline
//! (administratively prohibited), which is the decline the never-
//! feature contract needs, but it is an inherited posture: a pin bump
//! that changed it would surface here as an accepted-then-dropped
//! channel (third-party-upstream note; file upstream first).
//!
//! The session rows are **parse-only**: `exec_request` parses first
//! (the full grammar, [`crate::dispatch::parse_exec`] — a git route to
//! the git row, a declined-with-reason grammar class to its honest
//! decline, anything else to the session pipe row); `pty_request` /
//! `shell_request` route to the session rows. Nothing is served before
//! a request exists — no request, no serving, so a request-less
//! session channel declines as unfeatured by construction.
//! **Decline first, alloc never**: every decline here precedes any
//! channel/resource allocation (a leaked channel alloc is a
//! capability leak — the POC-3 stage-A rule); the per-length
//! `max_exec_command_len` cap is enforced before the parse itself.
//!
//! The store-and-take channel handoff (`w3-git-mapping`)
//!
//! A confirmed `session` open **stores** its [`Channel`] in
//! per-connection handler state ([`Self::take_stored_channel`]'s
//! supplier: `channel_open_session` accepts, then inserts — the
//! POC-3 §A.4 accepted-but-held shape); the git row's `exec_request`
//! **takes** it (`channels.remove(&id)` → `into_stream()` in the
//! spawned serving task) and the session's packet loop keeps owning
//! the store: `channel_close` removes whatever it held (a channel
//! that served or declined without an exec frees there), and every
//! mutation happens on the loop's task — single-writer, no lock. A
//! git exec on a channel whose stored handle is already gone
//! (re-served, or never held — unvanilla) declines serving-path.
//!
//! ## The session rows' real serving paths (`w3-tty-session`)
//!
//! The pty/shell/pipe rows replaced their wave-3 skeleton stubs with
//! the alktty translation bridge ([`crate::tty_map`]): the pty row
//! stores the negotiated `TerminalParamsWire` ([`Self::ptys`]'s
//! supplier — taken by the later shell/exec request on the same
//! channel, `ssh -t host "cmd"`'s pty-then-exec); the shell and pipe
//! rows take the stored channel the same way the git row does and
//! spawn `serve_tty_row` over the configured seam — locally-built
//! negotiation, unset-fields-decline (OQ-08: `shell_argv` unset →
//! the shell row declines, `exec_wrapper_argv` unset → the pipe row
//! declines, `session_seam` unset → both rows and the pty row
//! decline `seam-unconfigured`; every one of them the serving-path
//! failure+close shape from a spawned task). A routed request
//! answers `channel_success` inline (the request routed; the exit
//! chunk reports the outcome later, status-before-close from the
//! spawned task).
//!
//! ## The control rows' per-channel surface (`w3-tty-session`)
//!
//! `window_change_request` and `signal` arrive on the packet loop
//! **after** the serving task owns the wire — the spawned bridge is
//! reached through [`Self::bridges`]' per-channel surface
//! ([`crate::tty_map::BridgeSurface`]: the bounded control mpsc's
//! `try_send` — never awaited on the packet loop — plus the
//! per-channel cancel). A resize for a channel whose pty is stored
//! but not yet serving updates the stored params (the final size
//! wins); signal names are accepted verbatim (the backend owns the
//! semantics). Unfeatured builds emit the request-level decline
//! event with no reply — the control rows are fire-and-forget, no
//! SSH reply exists to send. [`Self::channel_close`] releases the
//! surface (firing the cancel: the alktty kill-on-drop contract is
//! the cancelled session's cleanup) along with the channel and pty
//! stores. The **connection teardown** signal ([`crate::tty_map::
//! Teardown`]) fires from this handler's `Drop` so a serving task
//! can never outlive the russh session that spawned it.
//!
//! ## The spawn-and-return structural constraint (ADR-006 decision 4)
//!
//! russh at this pin runs handler callbacks **inline inside the
//! session's packet loop** — a blocking handler suspends all packet
//! routing and deadlocks against the peer's follow-up bytes (POC-3
//! proved by construction with receive-pack). Every serving path is
//! therefore spawned through [`crate::serve::spawn_serving_path`] and
//! the handler returns fast: the serving rows (the git row's alkgit
//! dispatch, the session rows' config-driven declines and
//! [`crate::tty_map`]'s bridge) all clone `session.handle()`, spawn
//! through the clone, and are gone from the hook's stack — no hook
//! awaits a serving path inline. The decline keeps its wire shape
//! (failure then close), riding the spawn. Exit/status reporting from
//! spawned tasks rides the same cloned handle
//! ([`crate::serve::report_exit_status_then_close`] — the mechanics;
//! the tty rows' derivation rules live in [`crate::tty_map`]), pinned
//! end-to-end by the mechanics rows in `tests/spawn_return.rs` and the
//! mapping rows in `tests/git_mapping.rs` / `tests/tty_mapping.rs`.
//!
//! The open-level declines (unfeatured/never-feature rows, the
//! direct rows' pre-admit classes) stay inline by class: they answer
//! the open *before* any channel is owned — nothing is served yet,
//! so nothing spawns. The route-1 bridge (`w3-tunnel-direct`) moved
//! the direct rows' accept past the fabric open: the accept and the
//! pump ride the spawned task, per the same constraint. The `-R`
//! listen rows' family-ward dial is the one **inline await** on the
//! loop (`w3-tunnel-listen`'s recorded decision: the global request's
//! reply + the `&mut u32` bound-port out-param are the hook's own —
//! no spawned-task answer exists at this pin — and the fabric's
//! answer rides the fabric leg's connections, not this session's
//! routing; bounded by `LISTEN_DIAL_TIMEOUT`), while everything under
//! the reply — the arrivals, the pumps — stays spawned.
//!
//! Request-level declines: `subsystem_request` declines every name —
//! `sftp` is the OQ-13 placeholder (the feature exists, the route
//! does not until alkfs lands); git service names carry only a name,
//! never a repository id, so git can never dispatch as a subsystem
//! (ADR-006 decision 2); everything else is unimplemented. `env_request`
//! declines unconditionally (OQ-09 — no allowlist ships: wire env is
//! not family config). `x11_request` and `agent_request` decline
//! (ADR-003 §5). The global listen requests (`tcpip_forward`
//! / `streamlocal_forward` + their cancel forms) — the `-R` rows
//! (`w3-tunnel-listen`): the real flow under `tunnel` with the listen
//! seam configured (the family-ward `TunnelListener::open` dial, the
//! arrivals' arrival pump through the shared router); every decline
//! class keeps a global-request-level reason (see the hooks).
//!
//! ## Wire decline reasons (the wave-1 error-model reservation)
//!
//! Two decline classes, chosen for vanilla-client observability (POC-1
//! §5: channel-scoped decline, auth intact, no hangs):
//!
//! - **Open-level declines** — channel-open failure codes (`x11`
//!   never-feature, unfeatured feature rows, the tunnel rows'
//!   identity-less guard and beyond-cap refusals → administratively
//!   prohibited; the direct rows' cannot-bridge classes (the fabric
//!   entry unconfigured under the feature, the failed fabric open, an
//!   impossible wire target), the off-Unix `direct-streamlocal`
//!   decline, and the unvanilla inbound `forwarded-tcpip` open →
//!   connect failed). These
//!   happen before the channel exists, so the SSH open-failure code is
//!   the honest, richest reason surface.
//! - **Request-level declines** — [`Session::channel_failure`]: the
//!   SSH request-reply vocabulary has no failure text, so a request
//!   decline's "reason" is the routing decision itself (documented
//!   above, in [`crate::dispatch`]'s grammar, and per feature build in
//!   `mapping.md`'s decline matrix). A *capability* decline answers
//!   failure and leaves the channel usable (the POC-1 §5
//!   session-survives-declines posture) — including the git grammar's
//!   declinables (`unknown-service` / `service-only` /
//!   `invalid-repo-id`, the parsed reply shapes, `w3-git-mapping`);
//!   a *serving-path* decline (a routable row that owns the request's
//!   whole outcome: the seam-unconfigured git row, the session rows'
//!   unset-field and unset-seam declines — `w3-tty-session`'s OQ-08
//!   shapes — and the identity-less guard) answers failure and closes
//!   the channel — the vanilla-clean shape for a row whose serving
//!   path cannot exist. The serving-path decline rides the spawned
//!   serving task (ADR-006 decision 4, §The spawn-and-return
//!   structural constraint above); the wire shape is unchanged. The
//!   `channel-not-held` guard is the inline request-level shape (the
//!   channel that could not be held stays usable). The real serve's
//!   own outcomes are not declines: the substrate maps them (denial
//!   band before any protocol byte; exit-status then close — the
//!   `exit_code` mapping decision recorded in [`crate::git_map`];
//!   the tty bridge's negotiation-error and exit-chunk decisions in
//!   [`crate::tty_map`]).
//!
//! ## The event vocabulary on every decline (`w3-decline-tracing`)
//!
//! Every decline site here emits its event through
//! [`crate::observe`] — the wave-2 gate's booked finding: the
//! deciding context (row, feature build, channel id, decline class,
//! reason) rides the tracing event, not just the routing decision
//! documented per build. The vocabulary and its safety posture (no
//! secrets, no username, no fingerprint — structurally) live in that
//! module; the serving-path lifecycle (started / ended / failure)
//! hangs off the spawned stub body and is what the wave-3 mappings'
//! real serving paths inherit.
//!
//! ## Bounded resources at the handler surface (`w2-bounded-resources`)
//!
//! (AGENTS.md convention 9; `overview.md` §Bounded-resources policy —
//! the "concurrent channels per session — ours" row.) The in-crate
//! concurrent-channel cap ([`crate::channel_cap::ChannelCap`]) is
//! enforced at every `channel_open_*` hook: an accept path beyond
//! `Limits::max_concurrent_channels` is refused fail-closed at the
//! open ([`crate::dispatch::cap_exceeded_open_decline`] — never
//! queued, never half-open), and the count releases so the cap
//! recovers: through the `channel_close` hook for established
//! channels, and — `w3-tunnel-direct` — inside the route-1 bridge's
//! spawned task on a fabric-open failure (the admitted-but-rejected
//! shape: ADR-006 decision 1 rejects the open, so no `channel_close`
//! can ever fire for it; the release mechanism is the shared
//! [`ChannelCap`] `Arc`, recorded in that module). The `-R` leg adds
//! its two budget surfaces (`w3-tunnel-listen`, recorded in
//! `channel_cap.rs`): the arrivals admit into the owning route's
//! count inside the spawned arrival tasks (a beyond-cap arrival never
//! opens an SSH channel — teardown-at-birth), and the listen count is
//! its own dedicated counter against the same ceiling (the
//! listener/channels mix stays unconstrained; both stay bounded). The
//! order of the
//! gates: the feature gate first (an unfeatured build declines
//! everything before the cap matters), then the pure-decline classes
//! (the tunnel rows' unset-handle/unix-gate/identity/invalid-target
//! checks precede the admission — decline-first-alloc-never applies
//! to the budget too), the cap admission before any accept; the
//! admitted count is held across the fabric open and released exactly
//! once on either path. The `max_exec_command_len` cap is
//! enforced at `exec_request` *before* the parse and any allocation
//! (`w3-git-mapping`): an over-length wire command declines
//! request-level, unfeatured-or-not. The per-identity policy layer
//! arrives with the alkcall bridge.
//!
//! ## Test surface
//!
//! The handler type is deliberately constructible outside this module
//! (re-exported `#[doc(hidden)]` from the crate root) so the wave-1
//! contract tests drive these hooks directly against the in-memory
//! doubles — the wire-facing `VerifiedPublicKey`/`PublicKeyOffer` hook
//! arguments are `#[non_exhaustive]` (only russh constructs them), so
//! the bridge core is its own method ([`SessionHandler::resolve_bridge`])
//! the tests call with ordinary key material; the wire-level rows
//! exercise the hook wiring itself, end to end.

use std::collections::HashMap;
use std::sync::Arc;

use alkcall::core::auth::Identity;
use russh::keys::Algorithm;
#[cfg(any(feature = "git", feature = "session"))]
use russh::server::Handle;
use russh::server::{Auth, ChannelOpenHandle, Msg, PublicKeyOffer, Session, VerifiedPublicKey};
use russh::{Channel, ChannelId, Pty, Sig};

use crate::channel_cap::ChannelCap;
use crate::config::ServeConfig;
#[cfg(feature = "session")]
use crate::config::SessionSeam;
use crate::dispatch::{self, ExecRoute};
use crate::error::AdapterError;
use crate::fingerprint::fingerprint;
#[cfg(feature = "git")]
use crate::git_map;
use crate::observe;
#[cfg(any(feature = "git", feature = "session", feature = "tunnel"))]
use crate::serve;
#[cfg(feature = "session")]
use crate::tty_map;
#[cfg(feature = "tunnel")]
use crate::tunnel_map;
#[cfg(feature = "session")]
use alktty::negotiation::NegotiateRequest;

/// The per-connection handler (one per mounted stream): russh constructs
/// nothing here — [`crate::SshAdapter::handle`] builds one per accepted
/// stream and drives it through `russh::server::run_stream`.
///
/// Carries the serve-config the auth bridge and dispatch consume, the
/// in-crate concurrent-channel cap (`ChannelCap` — see the private
/// `channel_cap` module; enforced at every `channel_open_*` accept
/// path, bounded resources, AGENTS.md convention 9), and the resolved
/// identity once a publickey auth has been accepted (see
/// [`Self::resolved_identity`] — the wave-3 seam reads it).
pub struct SessionHandler {
    serve: Arc<ServeConfig>,
    /// The in-crate concurrent-channel cap (`w2-bounded-resources`),
    /// shared as an `Arc` since `w3-tunnel-direct`: the route-1 bridge's
    /// spawned task releases the count on a fabric-open failure (the
    /// admitted-but-rejected shape — no `channel_close` can ever fire
    /// for it; see `channel_cap.rs`'s release-sites record). Admissions
    /// have two sites since `w3-tunnel-listen` — the packet loop's
    /// open hooks and the `-R` arrival tasks — and releases three; the
    /// mutex makes them sound (the reasoning, recorded in
    /// `channel_cap.rs`).
    channel_cap: Arc<ChannelCap>,
    /// The `-R` listen leg's per-session bookkeeping
    /// (`w3-tunnel-listen`): the minted-listen map (token → pending
    /// listen; written only by the packet loop's hooks — the listen
    /// global requests mints, the cancel forms + this handler's
    /// teardown remove, single-writer/no-lock like the `channels`
    /// store) and the listen count's dedicated
    /// `max_concurrent_channels`-ceiling counter (a listen is at least
    /// one future channel's potential; its own counter keeps the
    /// listener/channels mix unconstrained while both stay bounded —
    /// no unbounded listener map).
    #[cfg(feature = "tunnel")]
    listeners: HashMap<u32, tunnel_map::PendingListen>,
    #[cfg(feature = "tunnel")]
    listener_cap: tunnel_map::ListenerCap,
    resolved_identity: Option<Identity>,
    /// The store-and-take handoff (`w3-git-mapping`): confirmed
    /// session channels held from the open until their row takes or
    /// the close releases them. Written only by the packet loop's
    /// hooks (`channel_open_session` inserts; the git row's and the
    /// session rows' serving dispatches remove for their taken
    /// channel; `channel_close` removes) — single-writer, no lock.
    channels: HashMap<ChannelId, Channel<Msg>>,
    /// The session rows' negotiated pty params (`w3-tty-session`):
    /// `pty_request` stores, the later `shell_request`/`exec_request`
    /// on the same channel consumes (`ssh -t host "cmd"`'s
    /// pty-then-exec), the close releases. Packet-loop only.
    #[cfg(feature = "session")]
    ptys: HashMap<ChannelId, alktty::negotiation::TerminalParamsWire>,
    /// The per-channel control surfaces (`w3-tty-session`): installed
    /// when a session row's serving task spawns, forwarded to by
    /// `window_change_request`/`signal`, fired-and-removed by
    /// `channel_close`. Packet-loop only.
    #[cfg(feature = "session")]
    bridges: HashMap<ChannelId, tty_map::BridgeSurface>,
    /// The connection-teardown signal (`w3-tty-session`): fired by
    /// this handler's [`Drop`](Self::drop) so a serving task never
    /// outlives the russh session that spawned it (the alktty
    /// kill-on-drop contract is the cancelled session's cleanup).
    #[cfg(feature = "session")]
    teardown: tty_map::Teardown,
}

impl Drop for SessionHandler {
    /// Fires the teardown signal (`w3-tty-session`): every spawned
    /// **session-row** serving task of this connection watches it and
    /// cancels its alktty session (the kill-on-drop cleanup) instead
    /// of outliving the connection. The git serving task does not
    /// watch this signal — it ends through the channel duplex (stream
    /// EOF / send failure) once the russh session dies, bounded by the
    /// alkgit seam's own `session_wall_clock` (the alkgit limits the
    /// assembly configured). The `tunnel` build adds the `-R`
    /// leg's teardown (`w3-tunnel-listen`): the session's pending
    /// listens' tokens leave the shared router (a fabric arrival after
    /// the session is gone is the correlation-miss class —
    /// teardown-at-birth) and the listener objects drop here (the
    /// fabric-side reap). Any in-flight arrival already past the
    /// route lookup fails its dead-handle open and releases its
    /// admission — the never-established path.
    fn drop(&mut self) {
        #[cfg(feature = "session")]
        self.teardown.fire();
        #[cfg(feature = "tunnel")]
        {
            let tokens: Vec<u32> = self.listeners.keys().copied().collect();
            if let Some(router) = &self.serve.listen_router {
                router.remove_tokens(&tokens);
            }
        }
    }
}

impl SessionHandler {
    /// Builds the handler over the serving configuration — the
    /// identity provider inside is the assembly-supplied
    /// `alkcall::core::auth::IdentityProvider`; the bridge only
    /// resolves through it.
    pub fn new(serve: Arc<ServeConfig>) -> Self {
        Self {
            serve,
            channel_cap: Arc::new(ChannelCap::default()),
            resolved_identity: None,
            channels: HashMap::new(),
            #[cfg(feature = "session")]
            ptys: HashMap::new(),
            #[cfg(feature = "session")]
            bridges: HashMap::new(),
            #[cfg(feature = "session")]
            teardown: tty_map::Teardown::new(),
            #[cfg(feature = "tunnel")]
            listeners: HashMap::new(),
            #[cfg(feature = "tunnel")]
            listener_cap: tunnel_map::ListenerCap::default(),
        }
    }

    /// Takes a stored channel handle (the store half of the
    /// store-and-take handoff): the git row's and the session rows'
    /// dispatches remove the handle the open stored, handing the
    /// stream into the spawned serving task. `None` when nothing is
    /// held — a re-served or never-held channel (unvanilla; declined
    /// serving-path).
    #[cfg(any(feature = "git", feature = "session"))]
    fn take_stored_channel(&mut self, channel: ChannelId) -> Option<Channel<Msg>> {
        self.channels.remove(&channel)
    }

    /// The identity this connection currently rides — `Some` only after
    /// a fingerprint resolution accepted a publickey auth (the bridge's
    /// fail-closed rejecting paths never leave a weakly-authorized
    /// state behind). The wave-3 seam propagation consumes this.
    pub fn resolved_identity(&self) -> Option<&Identity> {
        self.resolved_identity.as_ref()
    }

    /// The accept-path admission under the in-crate concurrent-channel
    /// cap (bounded resources, `w2-bounded-resources`): room under
    /// `Limits::max_concurrent_channels` admits and counts the channel;
    /// beyond the cap the open is refused fail-closed with a clean
    /// wire reason ([`crate::dispatch::cap_exceeded_open_decline`]) —
    /// never queued, never half-open. Called exactly at accept sites —
    /// the packet loop owns every admit call (the single admit writer,
    /// [`ChannelCap`]'s recorded reasoning); a pure-decline path
    /// (unfeatured, never-feature, the tunnel rows' pre-admit classes)
    /// never touches the count. Since `w3-tunnel-direct`, the tunnel
    /// rows admit **at the hook and hold the count across the fabric
    /// open**: a fabric failure releases inside the spawned bridge
    /// task (the admitted-but-rejected shape — no `channel_close` can
    /// ever fire); a success releases when the established channel
    /// closes ([`Self::channel_close`]). The release sites are disjoint
    /// per admission by the wire shape.
    fn admit_channel_open(&self) -> bool {
        self.channel_cap
            .admit(self.serve.limits.max_concurrent_channels)
    }

    /// The ADR-004 decision 1 bridge core, structured out of the
    /// [`Handler::auth_publickey`] hook so the contract tests can drive
    /// it directly: the hook's russh-owned argument shape
    /// (`russh::server::VerifiedPublicKey`) is `#[non_exhaustive]` —
    /// only russh constructs it — but this method takes the two facts
    /// the hook receives (the verified key and the signature's
    /// algorithm) as ordinary arguments.
    ///
    /// Ed25519-only gate first (ADR-002 decision 4 — `SkEd25519` and
    /// every other scheme share the non-ed25519 decline, which is
    /// explicit-over-inherited so it survives a pin bump), then the
    /// single fingerprint path → the injected provider's
    /// `resolve_from_fingerprint`. No username parameter exists — the
    /// user string is never an identity input (OQ-03). `None` is the
    /// fail-closed reject disposition; `Some` carries the identity the
    /// hook stores in handler state.
    pub fn resolve_bridge(
        &self,
        public_key: &russh::keys::PublicKey,
        signature_algorithm: &Algorithm,
    ) -> Option<Identity> {
        if *signature_algorithm != Algorithm::Ed25519
            || public_key.algorithm() != Algorithm::Ed25519
        {
            observe::decline_auth("auth_publickey", "non-ed25519");
            return None;
        }
        let identity = self
            .serve
            .identity_provider
            .resolve_from_fingerprint(&fingerprint(public_key));
        if identity.is_none() {
            observe::decline_auth("auth_publickey", "fingerprint-unresolved");
        }
        identity
    }

    /// The `session` channel-open gate (ADR-003 feature graph):
    /// session channels open only where a session-serving feature
    /// exists — `session` ∨ `git`, because git's parse path lives
    /// inside session channels.
    fn serves_session_channels(&self) -> bool {
        cfg!(feature = "session") || cfg!(feature = "git")
    }

    /// The session-row gate (`pty`/`shell`/pipe-exec): only the
    /// `session` feature serves these rows.
    #[cfg(feature = "session")]
    fn serves_session_rows(&self) -> bool {
        cfg!(feature = "session")
    }

    /// The git-row gate (git-parseable exec): only the `git` feature
    /// serves this row.
    #[cfg(feature = "git")]
    fn serves_git_row(&self) -> bool {
        cfg!(feature = "git")
    }

    /// The tunnel-row gate (`direct-tcpip` / `forwarded-tcpip` /
    /// `direct-streamlocal` opens and the listen flow): only the
    /// `tunnel` feature serves these rows.
    fn serves_tunnels(&self) -> bool {
        cfg!(feature = "tunnel")
    }
}

impl russh::server::Handler for SessionHandler {
    type Error = AdapterError;

    /// ADR-004 decision 1 (as amended for ADR-002 decision 4): the
    /// verified key bridges fingerprint →
    /// `IdentityProvider::resolve_from_fingerprint` →
    /// accept-with-identity / reject fail-closed; non-ed25519 keys
    /// decline explicitly before any resolution. The resolved identity
    /// lands in the per-connection handler state.
    async fn auth_publickey(
        &mut self,
        _user: &str,
        key: VerifiedPublicKey<'_>,
    ) -> Result<Auth, Self::Error> {
        match self.resolve_bridge(key.public_key, key.algorithm) {
            Some(identity) => {
                self.resolved_identity = Some(identity);
                Ok(Auth::Accept)
            }
            None => Ok(Auth::reject()),
        }
    }

    /// ADR-004 decision 2: accept-and-delegate, explicit — the probe
    /// hook consumes no auth attempt (POC-1 verified), so keeping it at
    /// `Auth::Accept` keeps the `max_auth_attempts` counting honest;
    /// 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)
    }

    /// ADR-004 decision 3: explicit family-contract decline — never the
    /// inherited default (a pin bump could silently change it).
    async fn auth_none(&mut self, _user: &str) -> Result<Auth, Self::Error> {
        observe::decline_auth("auth_none", "method-unimplemented");
        Ok(Auth::reject())
    }

    /// ADR-004 decision 3: explicit family-contract decline — password
    /// auth does not exist on this door; ADR-005's key custody is the
    /// whole story.
    async fn auth_password(&mut self, _user: &str, _password: &str) -> Result<Auth, Self::Error> {
        observe::decline_auth("auth_password", "method-unimplemented");
        Ok(Auth::reject())
    }

    /// ADR-004 decision 3 + OQ-04 (resolved): explicit decline — the
    /// CA-trust story does not exist in the family identity model;
    /// future support is new-ADR territory.
    async fn auth_openssh_certificate(
        &mut self,
        _user: &str,
        _certificate: &russh::keys::Certificate,
    ) -> Result<Auth, Self::Error> {
        observe::decline_auth("auth_openssh_certificate", "cert-unsupported");
        Ok(Auth::reject())
    }

    /// ADR-004 decision 3: explicit family-contract decline.
    async fn auth_keyboard_interactive(
        &mut self,
        _user: &str,
        _submethods: &str,
        _response: Option<russh::server::Response<'_>>,
    ) -> Result<Auth, Self::Error> {
        observe::decline_auth("auth_keyboard_interactive", "method-unimplemented");
        Ok(Auth::reject())
    }

    /// A counted channel closed — the in-crate concurrent-channel
    /// cap's decrement (`w2-bounded-resources`): every fired
    /// `channel_close` corresponds to exactly one previously-admitted
    /// open (russh only routes `CHANNEL_CLOSE` for established
    /// channels — a declined open never established one — and consumes
    /// the close at most once), so the count recovers through this
    /// single release point, saturating in [`ChannelCap::release`] on
    /// top. The spawned serving paths never touch the count directly:
    /// their closes ride the cloned handle, the client's close reply
    /// cycles back through the packet loop, and this hook fires — the
    /// single-writer reasoning in [`ChannelCap`].
    async fn channel_close(
        &mut self,
        channel: ChannelId,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.channels.remove(&channel);
        #[cfg(feature = "session")]
        {
            self.ptys.remove(&channel);
            if let Some(surface) = self.bridges.remove(&channel) {
                tty_map::fire_cancel(&surface.cancel);
            }
        }
        self.channel_cap.release();
        Ok(())
    }

    /// The session-channel open gate (ADR-003 feature graph,
    /// `session` ∨ `git`), then the in-crate concurrent-channel cap
    /// ([`Self::admit_channel_open`] — bounded resources,
    /// `w2-bounded-resources`): unfeatured builds decline at the open
    /// — never a half-open channel — and a featured open beyond the
    /// cap is refused fail-closed with a clean wire reason
    /// ([`crate::dispatch::cap_exceeded_open_decline`]), never queued.
    /// A confirmed open accepts the channel and counts it; the count
    /// releases when the channel closes ([`Self::channel_close`]). The
    /// parse-only dispatch serves the accepted channel through its
    /// requests (nothing serves request-less — ADR-006 as amended).
    async fn channel_open_session(
        &mut self,
        channel: Channel<Msg>,
        reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        if !self.serves_session_channels() {
            observe::decline_open("session", "unfeatured", channel.id().number());
            reply.reject(dispatch::unfeatured_open_decline()).await;
            return Ok(());
        }
        if self.admit_channel_open() {
            reply.accept().await;
            self.channels.insert(channel.id(), channel);
        } else {
            observe::decline_open("session", "cap-exceeded", channel.id().number());
            reply.reject(dispatch::cap_exceeded_open_decline()).await;
        }
        Ok(())
    }

    /// The x11 channel open: **always declines** — a never-feature
    /// (ADR-003 §5), declined in every build, never feature-gated. The
    /// cap never applies here: a count is consumed only at an accept
    /// path, and this hook never has one.
    async fn channel_open_x11(
        &mut self,
        channel: Channel<Msg>,
        _originator_address: &str,
        _originator_port: u32,
        reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        observe::decline_open("x11", "never-feature", channel.id().number());
        reply.reject(dispatch::unfeatured_open_decline()).await;
        Ok(())
    }

    /// The `direct-tcpip` open gate (`tunnel`) and its route-1 bridge
    /// dispatch (`w3-tunnel-direct`): unfeatured declines at the open;
    /// featured rows build the bridge target from the SSH open
    /// arguments (`SubstrateAddr::HostPort` + `Substrate::Tcp` — an
    /// out-of-range wire port is an impossible target, declined
    /// pre-admission) and route to [`Self::dispatch_direct_open`]:
    /// pre-admit decline classes first (the unset fabric handle
    /// declining under the feature — the unset-declines posture — and
    /// the unreachable-in-practice `no-identity` guard), then the cap
    /// admission **at the hook, held across the fabric open**, then
    /// the spawned bridge per spawn-and-return —
    /// `TunnelSession::open_direct` → accept only after the fabric
    /// open succeeds (`ConnectFailed` on failure, the count released
    /// in the task), `take_halves()` pumped against the taken channel.
    async fn 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 Session,
    ) -> Result<(), Self::Error> {
        const ROW: &str = "direct-tcpip";
        if !self.serves_tunnels() {
            observe::decline_open(ROW, "unfeatured", channel.id().number());
            reply.reject(dispatch::unfeatured_open_decline()).await;
            return Ok(());
        }
        #[cfg(feature = "tunnel")]
        {
            let target = u16::try_from(port_to_connect).ok().map(|port| {
                alktunnels::params::SubstrateAddr::HostPort {
                    host: host_to_connect.to_owned(),
                    port,
                }
            });
            direct_open_dispatch(
                self,
                ROW,
                channel,
                reply,
                target,
                alktunnels::params::Substrate::Tcp,
            )
            .await;
        }
        #[cfg(not(feature = "tunnel"))]
        {
            let _ = (host_to_connect, port_to_connect);
        }
        Ok(())
    }

    /// The `forwarded-tcpip` open gate (`tunnel`): an **inbound** open
    /// of this channel type — arriving *from* the vanilla client — is
    /// not the `-R` leg at all (the `-R` data leg is the door's own
    /// server-initiated opens via [`crate::tunnel_map`]'s arrival
    /// pump; a vanilla client never initiates `forwarded-tcpip`,
    /// OpenSSH parity). The gate + the unvanilla decline stay
    /// straight-line and inline: nothing serves, nothing was ever in
    /// flight, so no admission is touched, and the explicit
    /// `ConnectFailed` reject keeps the wave-2 wire shape (the event's
    /// class records it as the unvanilla-arrival decline).
    async fn channel_open_forwarded_tcpip(
        &mut self,
        channel: Channel<Msg>,
        _host_to_connect: &str,
        _port_to_connect: u32,
        _originator_address: &str,
        _originator_port: u32,
        reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        const ROW: &str = "forwarded-tcpip";
        if !self.serves_tunnels() {
            observe::decline_open(ROW, "unfeatured", channel.id().number());
            reply.reject(dispatch::unfeatured_open_decline()).await;
            return Ok(());
        }
        observe::decline_open(ROW, "unvanilla-inbound", channel.id().number());
        reply.reject(dispatch::connect_failed_open_decline()).await;
        Ok(())
    }

    /// The `direct-streamlocal` open gate (`tunnel`) and its route-1
    /// bridge dispatch: the `direct-tcpip` shape with the OQ-11
    /// **runtime-Unix gate** — the row compiles wherever `tunnel`
    /// compiles (universal compile, no per-platform feature fork), but
    /// off-Unix it declines at the open **before any fabric call**
    /// (never a partial bridge — the gate precedes the unset-handle
    /// check, the identity guard, and the cap admission), and on-Unix
    /// it rides the same bridge with `SubstrateAddr::Path` +
    /// `Substrate::Unix`.
    async fn channel_open_direct_streamlocal(
        &mut self,
        channel: Channel<Msg>,
        _socket_path: &str,
        reply: ChannelOpenHandle,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        const ROW: &str = "direct-streamlocal";
        if !self.serves_tunnels() {
            observe::decline_open(ROW, "unfeatured", channel.id().number());
            reply.reject(dispatch::unfeatured_open_decline()).await;
            return Ok(());
        }
        #[cfg(feature = "tunnel")]
        {
            #[cfg(unix)]
            {
                let target = Some(alktunnels::params::SubstrateAddr::Path {
                    path: _socket_path.to_owned(),
                });
                direct_open_dispatch(
                    self,
                    ROW,
                    channel,
                    reply,
                    target,
                    alktunnels::params::Substrate::Unix,
                )
                .await;
            }
            #[cfg(not(unix))]
            {
                observe::decline_open(ROW, "unix-unavailable", channel.id().number());
                reply.reject(dispatch::connect_failed_open_decline()).await;
            }
        }
        #[cfg(not(feature = "tunnel"))]
        {
            let _ = _socket_path;
        }
        Ok(())
    }

    /// The exec-grammar dispatch (ADR-006 point 3 as amended; the
    /// full grammar, `w3-git-mapping`): the per-length cap checks the
    /// wire bytes first (pre-parse, pre-allocation — decline first,
    /// alloc never), then [`crate::dispatch::parse_exec`] routes:
    /// a git route to the git row (feature-gated — a session-only
    /// build declines it as its gate), a grammar declinable to its
    /// honest request-level decline, anything else to the session
    /// pipe row. Invalid UTF-8 parses to the pipe row (parse-then-
    /// fail).
    async fn exec_request(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        if data.len() > self.serve.limits.max_exec_command_len {
            observe::decline_request("exec_request", "exec-oversize", channel.number(), None);
            decline_unfeatured_request(channel, session).await;
            return Ok(());
        }
        let command = std::str::from_utf8(data).unwrap_or("");
        match dispatch::parse_exec(command) {
            ExecRoute::Git { service, repo } => {
                #[cfg(feature = "git")]
                if self.serves_git_row() {
                    git_row_dispatch(self, channel, session, service, repo);
                    return Ok(());
                }
                #[cfg(not(feature = "git"))]
                let _ = (&service, &repo);
                observe::decline_request(
                    "exec_request",
                    "unfeatured",
                    channel.number(),
                    Some("git"),
                );
                decline_unfeatured_request(channel, session).await;
            }
            ExecRoute::Declined(decline) => {
                observe::decline_request(
                    "exec_request",
                    decline.reason(),
                    channel.number(),
                    Some("git"),
                );
                decline_unfeatured_request(channel, session).await;
            }
            ExecRoute::Pipe => {
                #[cfg(feature = "session")]
                if self.serves_session_rows() {
                    pipe_row_dispatch(self, channel, session, command.to_owned());
                    return Ok(());
                }
                #[cfg(not(feature = "session"))]
                let _ = &command;
                observe::decline_request(
                    "exec_request",
                    "unfeatured",
                    channel.number(),
                    Some("pipe"),
                );
                decline_unfeatured_request(channel, session).await;
            }
        }
        Ok(())
    }

    /// The `pty_request` row (`session`): the args →
    /// `TerminalParamsWire` (ttymodes → the reserved `{}` JSON,
    /// OQ-44), stored per channel for the later shell/exec request on
    /// the same channel — after the seam check (a pty whose session
    /// can never serve is a declined allocation,
    /// `seam-unconfigured`). Unfeatured builds decline the
    /// capability. The successful row answers `channel_success`
    /// (POC-3 §D.1's shape — serving starts at the shell/exec
    /// request).
    async fn pty_request(
        &mut self,
        channel: ChannelId,
        term: &str,
        col_width: u32,
        row_height: u32,
        pix_width: u32,
        pix_height: u32,
        _modes: &[(Pty, u32)],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        #[cfg(feature = "session")]
        match self.serve.session_seam.clone() {
            Some(_) => {
                self.ptys.insert(
                    channel,
                    tty_map::terminal_params(term, col_width, row_height, pix_width, pix_height),
                );
                let _ = session.channel_success(channel);
            }
            None => {
                spawn_seam_stub(channel, session, "pty_request", None, "seam-unconfigured");
            }
        }
        #[cfg(not(feature = "session"))]
        {
            let _ = (term, col_width, row_height, pix_width, pix_height);
            observe::decline_request("pty_request", "unfeatured", channel.number(), None);
            decline_unfeatured_request(channel, session).await;
        }
        Ok(())
    }

    /// The `shell_request` row (`session`): the serve-config shell
    /// argv, verbatim, as `NegotiateRequest.cmd`; the stored pty
    /// params consumed (`None` → pipe-mode shell — decided: a pty-less
    /// shell request is served, `ssh -T host` is legitimate vanilla);
    /// unset-declines first (OQ-08: `shell_argv` unset →
    /// `shell-unconfigured`; the seam unset → `seam-unconfigured`
    /// — both the spawned serving-path shape). A routed request
    /// answers `channel_success` inline; the exit chunk reports the
    /// outcome, status-before-close from the spawned task.
    async fn shell_request(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        #[cfg(feature = "session")]
        match (
            self.serve.shell_argv.clone(),
            self.serve.session_seam.clone(),
        ) {
            (None, _) => {
                spawn_seam_stub(
                    channel,
                    session,
                    "shell_request",
                    None,
                    "shell-unconfigured",
                );
            }
            (_, None) => {
                spawn_seam_stub(channel, session, "shell_request", None, "seam-unconfigured");
            }
            (Some(shell_argv), Some(seam)) => {
                let tty = self.ptys.remove(&channel);
                let Some(taken) = self.take_stored_channel(channel) else {
                    observe::decline_request(
                        "shell_request",
                        "channel-not-held",
                        channel.number(),
                        None,
                    );
                    let _ = session.channel_failure(channel);
                    return Ok(());
                };
                let request = tty_map::negotiate_request(
                    seam.backend.clone(),
                    tty,
                    shell_argv,
                    self.serve.session_cwd.clone(),
                    self.serve.session_env.clone(),
                );
                spawn_tty_row(
                    self,
                    channel,
                    session,
                    "shell_request",
                    None,
                    taken,
                    request,
                    seam,
                );
            }
        }
        #[cfg(not(feature = "session"))]
        {
            observe::decline_request("shell_request", "unfeatured", channel.number(), None);
            decline_unfeatured_request(channel, session).await;
        }
        Ok(())
    }

    /// `window_change_request` (`session`): the resize control row —
    /// forwarded into the serving bridge's bounded control surface
    /// (`try_send`, never awaited on the packet loop); for a channel
    /// whose pty is stored but not yet serving, the stored params
    /// resize instead (the final size wins — vanilla sshd's own
    /// ordering); with neither, the event is dropped. Fire-and-forget
    /// on the wire — no SSH reply exists, so a decline sends none
    /// (the event is the record). Signal names: see [`Self::signal`].
    async fn window_change_request(
        &mut self,
        channel: ChannelId,
        col_width: u32,
        row_height: u32,
        pix_width: u32,
        pix_height: u32,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        #[cfg(feature = "session")]
        {
            let cols = tty_map::clamp_u16(col_width);
            let rows = tty_map::clamp_u16(row_height);
            let pix_w = tty_map::clamp_u16(pix_width);
            let pix_h = tty_map::clamp_u16(pix_height);
            if let Some(surface) = self.bridges.get(&channel) {
                let message = alktty::ControlMessage::Resize {
                    cols,
                    rows,
                    pixel_width: pix_w,
                    pixel_height: pix_h,
                };
                let _ = surface.forward_control(&message);
            } else if let Some(params) = self.ptys.get_mut(&channel) {
                params.cols = cols;
                params.rows = rows;
                params.pixel_width = pix_w;
                params.pixel_height = pix_h;
                params.modes = serde_json::json!({});
            }
        }
        #[cfg(not(feature = "session"))]
        {
            let _ = (col_width, row_height, pix_width, pix_height);
            observe::decline_request(
                "window_change_request",
                "unfeatured",
                channel.number(),
                None,
            );
        }
        Ok(())
    }

    /// The `signal` row (`session`): accepted verbatim — no
    /// alkssh-side name filter (the backend owns the semantics;
    /// alktty's control handle applies its own default kill for
    /// names it rejects, and the pipe-mode pid-only vs PTY
    /// process-group scope is backend semantics) — so there is no
    /// decline shape: the request is fire-and-forget on the SSH
    /// wire, and the bridge's bounded surface is the only path. An
    /// unfeatured build emits the decline event with no reply.
    async fn signal(
        &mut self,
        channel: ChannelId,
        signal: Sig,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        #[cfg(feature = "session")]
        {
            if let Some(surface) = self.bridges.get(&channel) {
                let message = alktty::ControlMessage::Signal {
                    name: tty_map::sig_control_name(&signal),
                };
                let _ = surface.forward_control(&message);
            }
        }
        #[cfg(not(feature = "session"))]
        {
            let _ = signal;
            observe::decline_request("signal", "unfeatured", channel.number(), None);
        }
        Ok(())
    }

    /// Every subsystem name declines (unimplemented on this door):
    /// `subsystem("sftp")` is the OQ-13 placeholder — the feature
    /// exists, the route does not until alkfs lands; git service
    /// names carry only a name, never a repository id, so git can
    /// never dispatch as a subsystem (ADR-006 decision 2, POC-3
    /// §A.3's forced decline); everything else is unimplemented. Wire
    /// shape: `channel_failure`, the channel left usable (the
    /// POC-1 §5 session-survives-declines posture).
    async fn subsystem_request(
        &mut self,
        channel: ChannelId,
        _name: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        observe::decline_request("subsystem_request", "unimplemented", channel.number(), None);
        decline_unfeatured_request(channel, session).await;
        Ok(())
    }

    /// `env_request` declines unconditionally (OQ-09: no allowlist
    /// ships — wire env is not family config; the assembly supplies
    /// the session env through [`crate::ServeConfig::session_env`]).
    async fn env_request(
        &mut self,
        channel: ChannelId,
        _variable_name: &str,
        _variable_value: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        observe::decline_request("env_request", "declined-by-design", channel.number(), None);
        decline_unfeatured_request(channel, session).await;
        Ok(())
    }

    /// The `x11-req` request on a session channel: always declines
    /// (never-feature, ADR-003 §5).
    async fn x11_request(
        &mut self,
        channel: ChannelId,
        _single_connection: bool,
        _x11_auth_protocol: &str,
        _x11_auth_cookie: &str,
        _x11_screen_number: u32,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        observe::decline_request("x11_request", "never-feature", channel.number(), None);
        decline_unfeatured_request(channel, session).await;
        Ok(())
    }

    /// Agent forwarding on a session session-channel: always declines
    /// (never-feature, ADR-003 §5 — agent forwarding is the
    /// anti-pattern of the alkvault custody model). `false` is the
    /// decline; russh replies the channel failure.
    async fn agent_request(
        &mut self,
        channel: ChannelId,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        observe::decline_request("agent_request", "never-feature", channel.number(), None);
        Ok(false)
    }

    /// The `tcpip-forward` listen request (`-R` ingress, `w3-tunnel-
    /// listen`): the gate is [`Self::serves_tunnels`]; the real flow
    /// rides [`listen_request_dispatch`] (`tunnel` builds) — the
    /// family-ward `TunnelListener::open` dial, the bound-port out-param
    /// (`port = 0` → the fabric's actually-bound number, the sshd
    /// parity), the route + listener registration. Unfeatured builds
    /// decline with the global-request class; never a minted listen a
    /// door cannot serve.
    async fn tcpip_forward(
        &mut self,
        address: &str,
        port: &mut u32,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        #[cfg(feature = "tunnel")]
        {
            listen_request_dispatch(
                self,
                "tcpip_forward",
                address,
                alktunnels::params::Substrate::Tcp,
                Some(port),
                _session,
            )
            .await
        }
        #[cfg(not(feature = "tunnel"))]
        {
            let _ = (address, port);
            observe::decline_global_request("tcpip_forward", "unfeatured");
            Ok(false)
        }
    }

    /// The `cancel-tcpip-forward` cancel form (`-KR`, mux teardown,
    /// `ExitOnForwardFailure`): 1:1 into the family listener's teardown —
    /// the pending listen matched by (the wire address as the request
    /// carried it, the bound port), the route removed **before** any
    /// teardown (a racing arrival is already the correlation-miss
    /// class), the listener count released, the fabric channel's
    /// teardown spawned (`TunnelListener::close`). `false` for a
    /// (address, port) pair with no pending listen — the sshd parity
    /// (`cancel of a not-found forward`).
    async fn cancel_tcpip_forward(
        &mut self,
        address: &str,
        port: u32,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        #[cfg(feature = "tunnel")]
        {
            cancel_listen_dispatch(self, "cancel_tcpip_forward", address, Some(port))
        }
        #[cfg(not(feature = "tunnel"))]
        {
            let _ = (address, port);
            observe::decline_global_request("cancel_tcpip_forward", "unfeatured");
            Ok(false)
        }
    }

    /// The `streamlocal-forward` listen request: the same listen flow
    /// as `tcpip_forward` over the unix substrate — with the OQ-11
    /// runtime-Unix gate riding the serving flow it serves (off-Unix
    /// it declines before any fabric call; universal compile, no
    /// per-platform feature fork). No port out-param (the unix listen
    /// carries no port semantics; the path echoes verbatim).
    async fn streamlocal_forward(
        &mut self,
        socket_path: &str,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        #[cfg(feature = "tunnel")]
        {
            if !cfg!(unix) {
                observe::decline_global_request("streamlocal_forward", "unix-unavailable");
                return Ok(false);
            }
            listen_request_dispatch(
                self,
                "streamlocal_forward",
                socket_path,
                alktunnels::params::Substrate::Unix,
                None,
                _session,
            )
            .await
        }
        #[cfg(not(feature = "tunnel"))]
        {
            let _ = socket_path;
            observe::decline_global_request("streamlocal_forward", "unfeatured");
            Ok(false)
        }
    }

    /// The `cancel-streamlocal-forward` form: the tcpip cancel's shape
    /// over the unix substrate (path-match only; no port).
    async fn cancel_streamlocal_forward(
        &mut self,
        socket_path: &str,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        #[cfg(all(feature = "tunnel", unix))]
        {
            cancel_listen_dispatch(self, "cancel_streamlocal_forward", socket_path, None)
        }
        #[cfg(all(feature = "tunnel", not(unix)))]
        {
            let _ = socket_path;
            observe::decline_global_request("cancel_streamlocal_forward", "unix-unavailable");
            Ok(false)
        }
        #[cfg(not(feature = "tunnel"))]
        {
            let _ = socket_path;
            observe::decline_global_request("cancel_streamlocal_forward", "unfeatured");
            Ok(false)
        }
    }
}

/// The `tcpip_forward` / `streamlocal_forward` global requests' real
/// flow (`w3-tunnel-listen`; `mapping.md` §Global-request rows): the
/// gate order is decline-first-alloc-never, each consumer checked
/// before the next allocation is taken:
///
/// 1. **The `no-identity` guard** — unreachable on the wire (russh
///    authorizes before requests serve) and fail-closed: never
///    anonymous listens.
/// 2. **The seam checks** — the fabric handle or the listen seam
///    (`listen_router` absent: an egress-only door — the seam unset or
///    the registration never ran) decline `seam-unconfigured` before
///    any dial.
/// 3. **The OQ-11 unix gate** — the streamlocal variant's off-Unix
///    decline rides its hook (checked before this fn runs, so the
///    unreachable-on-Unix-only row never here).
/// 4. **The listener cap** — the listen count's dedicated
///    `max_concurrent_channels`-ceiling admission; beyond it
///    `cap-exceeded`.
///
/// Then the family-ward dial ([`crate::tunnel_map::open_listen`], the
/// `resource = <the wire bind endpoint>` naming convention) — awaited
/// inline (the recorded inline decision: the global request's reply +
/// the `&mut u32` out-param are the hook's; the fabric answer rides
/// the fabric leg, not this session's packet loop; the dial is
/// bounded by `LISTEN_DIAL_TIMEOUT`, so the stall is bounded). The
/// reply's `bound` port writes the out-param (`port = 0` → the
/// actually-bound number, sshd parity); the route registers into the
/// shared router — the register is the collision pre-check
/// (`w4-router-collision`): a mint on an already-held token refuses
/// fail-closed (`token-collision`, the minted listener torn down, the
/// cap released — a collided mint can never steal a live route; the
/// decline rides `Ok(false)`, never the error taxonomy, since a
/// global-request hook's `Err` would kill the whole session) — and
/// the pending listen into the session's
/// packet-loop-only map. `Ok(true)`'s answer is the only success
/// shape; every decline path answers the vanilla client's
/// `RequestDenied` via `Ok(false)`.
#[cfg(feature = "tunnel")]
async fn listen_request_dispatch(
    handler: &mut SessionHandler,
    row: &'static str,
    wire_endpoint: &str,
    substrate: alktunnels::params::Substrate,
    port: Option<&mut u32>,
    session: &mut Session,
) -> Result<bool, AdapterError> {
    if handler.resolved_identity().is_none() {
        observe::decline_global_request(row, "no-identity");
        return Ok(false);
    }
    let (Some(client), Some(router)) = (
        handler.serve.channel_client.as_ref(),
        handler.serve.listen_router.as_ref(),
    ) else {
        observe::decline_global_request(row, "seam-unconfigured");
        return Ok(false);
    };
    if !handler
        .listener_cap
        .admit(handler.serve.limits.max_concurrent_channels)
    {
        observe::decline_global_request(row, "cap-exceeded");
        return Ok(false);
    }
    let minted = match tunnel_map::open_listen(
        client,
        tunnel_map::listen_resource(wire_endpoint, substrate, port.as_deref().copied()),
        substrate,
    )
    .await
    {
        Ok(minted) => minted,
        Err(_) => {
            handler.listener_cap.release();
            observe::decline_global_request(row, "listen-failed");
            return Ok(false);
        }
    };
    let bound = minted.bound;
    let echo = match (&substrate, &bound) {
        (
            alktunnels::params::Substrate::Tcp,
            alktunnels::params::SubstrateAddr::HostPort {
                port: bound_port, ..
            },
        ) => tunnel_map::RouteEcho::Tcp {
            address: wire_endpoint.to_owned(),
            port: u32::from(*bound_port),
        },
        (alktunnels::params::Substrate::Unix, alktunnels::params::SubstrateAddr::Path { path }) => {
            tunnel_map::RouteEcho::Unix { path: path.clone() }
        }
        _ => {
            handler.listener_cap.release();
            observe::decline_global_request(row, "listen-failed");
            return Ok(false);
        }
    };
    let token = minted.token;
    let route = tunnel_map::ListenRoute {
        echo,
        substrate,
        handle: Arc::new(session.handle()),
        cap: Arc::clone(&handler.channel_cap),
        cap_value: handler.serve.limits.max_concurrent_channels,
    };
    if let Err(_collision) = router.register(token, route) {
        handler.listener_cap.release();
        observe::decline_global_request(row, "token-collision");
        serve::spawn_serving_path(async move {
            minted.listener.close().await;
        });
        return Ok(false);
    }
    if let (Some(port), Some(bound_port)) = (port, tunnel_map::bound_port(&bound)) {
        *port = bound_port;
    }
    handler.listeners.insert(
        token,
        tunnel_map::PendingListen {
            wire_address: wire_endpoint.to_owned(),
            bound_port: tunnel_map::bound_port(&bound).unwrap_or(0),
            substrate,
            listener: minted.listener,
        },
    );
    Ok(true)
}

/// The `cancel_tcpip_forward` / `cancel_streamlocal_forward` global
/// requests' real flow (`w3-tunnel-listen`): 1:1 into the family
/// listener's teardown. Match rule (sshd parity): the wire endpoint as
/// the original request carried it + the bound port for the tcp
/// variants (the client sends the same normalized string it sent
/// originally, and the *allocated* port for `port = 0` mints — the
/// value the listen reply carried); the path alone for the unix
/// variant. Order: the entry leaves the session map and the route
/// leaves the shared router before any family teardown spawns (a
/// racing arrival that has not yet cloned its route is already the
/// correlation-miss class — no post-cancel half-serve; one that
/// cloned just before completes independently — the recorded
/// caveat), then the listener count releases, then the family
/// teardown spawns (`TunnelListener::close` inside the spawn — a
/// bounded local shutdown + entry reap, not a serving path the hook
/// may await). `Ok(true)` iff a pending listen matched.
#[cfg(feature = "tunnel")]
fn cancel_listen_dispatch(
    handler: &mut SessionHandler,
    row: &'static str,
    wire_endpoint: &str,
    port: Option<u32>,
) -> Result<bool, AdapterError> {
    if handler.resolved_identity().is_none() {
        observe::decline_global_request(row, "no-identity");
        return Ok(false);
    }
    let token = handler.listeners.iter().find_map(|(token, pending)| {
        let port_matched = match port {
            Some(port) => pending.bound_port == port,
            None => pending.substrate == alktunnels::params::Substrate::Unix,
        };
        (pending.wire_address == wire_endpoint && port_matched).then_some(*token)
    });
    let Some(token) = token else {
        observe::decline_global_request(row, "no-such-listen");
        return Ok(false);
    };
    let Some(entry) = handler.listeners.remove(&token) else {
        observe::decline_global_request(row, "no-such-listen");
        return Ok(false);
    };
    if let Some(router) = &handler.serve.listen_router {
        router.cancel(&token);
    }
    handler.listener_cap.release();
    serve::spawn_serving_path(async move {
        entry.listener.close().await;
    });
    Ok(true)
}

/// The request-level decline shape for an unfeatured/unimplemented
/// capability: fail the request and leave the channel usable (POC-1
/// §5 — session survives declines, no state poisoning). The decline
/// event is emitted by the caller (the hook carries the row/reason
/// context); this is the wire shape only.
async fn decline_unfeatured_request(channel: ChannelId, session: &mut Session) {
    let _ = session.channel_failure(channel);
}

/// The spawned serving-path decline body (`w3-git-mapping`'s shape,
/// reused by `w3-tty-session`'s config-driven declines): fail the
/// request and close the channel, through the cloned handle the task
/// was handed — the vanilla-clean shape for a routed row that owns
/// the request's whole outcome. The helper name keeps the wave-3
/// history (the decline IS the path it spawns). Decline-first-alloc-
/// never holds structurally: nothing is allocated before the failure
/// lands — the spawn's handle clone is the only resource ever taken.
///
/// The serving-path lifecycle events ride this body
/// ([`crate::observe`]): started (the task's first action), the
/// serving-path decline (class `serving`, the caller's vocabulary
/// reason), and — after the wire reporting has completed — ended, or
/// failed when a report send could not complete (the one warn in the
/// vocabulary).
#[cfg(any(feature = "git", feature = "session"))]
async fn seam_stub_decline(
    handle: Handle,
    channel: ChannelId,
    row: &'static str,
    exec_route: Option<&'static str>,
    reason: &'static str,
) {
    observe::serving_path_started(row, channel.number(), exec_route);
    observe::decline_serving(row, reason, channel.number(), exec_route);
    let error = match handle.channel_failure(channel).await {
        Ok(()) => handle.close(channel).await.err(),
        Err(error) => Some(error),
    };
    match error {
        Some(error) => observe::serving_path_failed(row, channel.number(), &error, exec_route),
        None => observe::serving_path_ended(row, channel.number(), exec_route),
    }
}

/// Spawns a serving-path decline as the row's serving path (ADR-006
/// decision 4): the handle is cloned here, the decline runs on its own
/// task, and the hook returns — no serving path awaits inline. The
/// row never serves, so the decline *is* the path: failure + close,
/// event-shaped ([`crate::observe`]) with the caller's reason.
#[cfg(any(feature = "git", feature = "session"))]
fn spawn_seam_stub(
    channel: ChannelId,
    session: &mut Session,
    row: &'static str,
    exec_route: Option<&'static str>,
    reason: &'static str,
) {
    let handle = session.handle();
    serve::spawn_serving_path(seam_stub_decline(handle, channel, row, exec_route, reason));
}

/// The git row's routing stage (feature-gated, `w3-git-mapping`):
/// validation of the normalized id ([`crate::git_map::repo_id_validates`] —
/// reply-shaped `invalid-repo-id`), the seam-unconfigured decline
/// ([`Self::serves_git_row`] alone configures nothing — the
/// unset-declines contract), then the store-and-take handoff and the
/// spawned alkgit dispatch ([`crate::git_map::serve_git_row`], the
/// resolved identity riding along). A routed exec answers
/// `channel_success` inline (the request-reply vocabulary: the request
/// routed; the serving outcome reports through exit-status later —
/// the POC-3 §A.5 shape), never awaited serving.
#[cfg(feature = "git")]
fn git_row_dispatch(
    handler: &mut SessionHandler,
    channel: ChannelId,
    session: &mut Session,
    service: dispatch::GitServiceRoute,
    repo: String,
) {
    if !git_map::repo_id_validates(&repo) {
        observe::decline_request(
            "exec_request",
            "invalid-repo-id",
            channel.number(),
            Some("git"),
        );
        let _ = session.channel_failure(channel);
        return;
    }
    let Some(seam) = handler.serve.git_seam.clone() else {
        spawn_seam_stub(
            channel,
            session,
            "exec_request",
            Some("git"),
            "seam-unconfigured",
        );
        return;
    };
    let Some(taken) = handler.take_stored_channel(channel) else {
        observe::decline_request(
            "exec_request",
            "channel-not-held",
            channel.number(),
            Some("git"),
        );
        let _ = session.channel_failure(channel);
        return;
    };
    #[cfg(feature = "session")]
    {
        let _ = handler.ptys.remove(&channel);
    }
    let identity = handler.resolved_identity.clone();
    let handle = session.handle();
    serve::spawn_serving_path(git_map::serve_git_row(
        handle, taken, service, repo, identity, seam,
    ));
    let _ = session.channel_success(channel);
}

/// The session pipe row's dispatch (`w3-tty-session`): unset-declines
/// first (OQ-08 — `exec_wrapper_argv` unset → `wrapper-unconfigured`;
/// the seam unset → `seam-unconfigured`; both the spawned serving-path
/// shape, never a fallback), then the store-and-take handoff
/// (`channel-not-held` → the inline unvanilla decline, the git row's
/// shape), the wire command appended to the wrapper argv (`wrapper ++
/// [cmd]` — never naively split), the stored pty params consumed if a
/// pty preceded, and the spawned bridge. A routed exec answers
/// `channel_success` inline (the request routed; the exit chunk
/// reports the serving outcome later).
#[cfg(feature = "session")]
fn pipe_row_dispatch(
    handler: &mut SessionHandler,
    channel: ChannelId,
    session: &mut Session,
    command: String,
) {
    let Some(mut cmd) = handler.serve.exec_wrapper_argv.clone() else {
        spawn_seam_stub(
            channel,
            session,
            "exec_request",
            Some("pipe"),
            "wrapper-unconfigured",
        );
        return;
    };
    let Some(seam) = handler.serve.session_seam.clone() else {
        spawn_seam_stub(
            channel,
            session,
            "exec_request",
            Some("pipe"),
            "seam-unconfigured",
        );
        return;
    };
    let tty = handler.ptys.remove(&channel);
    let Some(taken) = handler.take_stored_channel(channel) else {
        observe::decline_request(
            "exec_request",
            "channel-not-held",
            channel.number(),
            Some("pipe"),
        );
        let _ = session.channel_failure(channel);
        return;
    };
    cmd.push(command);
    let request = tty_map::negotiate_request(
        seam.backend.clone(),
        tty,
        cmd,
        handler.serve.session_cwd.clone(),
        handler.serve.session_env.clone(),
    );
    spawn_tty_row(
        handler,
        channel,
        session,
        "exec_request",
        Some("pipe"),
        taken,
        request,
        seam,
    );
}

/// Installs the per-channel bridge surface, clones the session
/// handle, and spawns the spawned serving path — the session rows'
/// spawn step (ADR-006 decision 4: the handler returns fast; the
/// routed request answers `channel_success` inline).
#[cfg(feature = "session")]
#[allow(clippy::too_many_arguments)]
fn spawn_tty_row(
    handler: &mut SessionHandler,
    channel: ChannelId,
    session: &mut Session,
    row: &'static str,
    exec_route: Option<&'static str>,
    taken: Channel<Msg>,
    request: NegotiateRequest,
    seam: SessionSeam,
) {
    let (surface, chunk_rx, cancel_rx) = tty_map::build_surface();
    let stdin_tx = surface.sender.clone();
    handler.bridges.insert(channel, surface);
    let identity = handler.resolved_identity.clone();
    let handle = session.handle();
    serve::spawn_serving_path(tty_map::serve_tty_row(
        row,
        exec_route,
        handle,
        taken,
        request,
        identity,
        seam,
        stdin_tx,
        chunk_rx,
        cancel_rx,
        handler.teardown.subscribe(),
    ));
    let _ = session.channel_success(channel);
}

/// The direct tunnel rows' dispatch (`w3-tunnel-direct`): the
/// pre-admit decline classes first — **decline first, alloc never**
/// applies to the cap budget too, so every consumer is checked before
/// the admission is taken. Order (the recorded admission mechanics:
/// the admission is taken at the hook — the packet-loop's single
/// admit-writer site, no race window against the cap check — and held
/// across the fabric open):
///
/// 1. **The unset fabric handle decline** (`seam-unconfigured`, the
///    unset-declines posture — the feature alone configures nothing):
///    `ConnectFailed` ([`crate::dispatch::connect_failed_open_decline`])
///    — the bridge cannot be attempted at all.
/// 2. **The `no-identity` guard** — unreachable on the wire (russh
///    authorizes before channels exist) and fail-closed: never
///    anonymous serving (the git/tty rows' shape, discharged here for
///    the tunnel row — the parked `w2-auth-bridge` item; the
///    `set_identity` mechanics the rides records live in
///    [`crate::tunnel_map`]).
/// 3. **The impossible-target decline** (`invalid-target` — an
///    out-of-range wire port): `ConnectFailed`.
///
/// Then the cap admission ([`Self::admit_channel_open`]; beyond it →
/// `cap-exceeded`), and the spawn: the admitted count rides into
/// [`crate::tunnel_map::spawn_direct_row`], which releases it exactly
/// once on the failure path (no `channel_close` can ever fire for a
/// rejected open) and never on the success path (whose close fires
/// [`Self::channel_close`] when the established channel closes).
#[cfg(feature = "tunnel")]
async fn direct_open_dispatch(
    handler: &SessionHandler,
    row: &'static str,
    channel: Channel<Msg>,
    reply: ChannelOpenHandle,
    target: Option<alktunnels::params::SubstrateAddr>,
    substrate: alktunnels::params::Substrate,
) {
    let channel_number = channel.id().number();
    let Some(client) = handler.serve.channel_client.clone() else {
        observe::decline_open(row, "seam-unconfigured", channel_number);
        reply.reject(dispatch::connect_failed_open_decline()).await;
        return;
    };
    if handler.resolved_identity().is_none() {
        observe::decline_open(row, "no-identity", channel_number);
        reply
            .reject(russh::ChannelOpenFailure::AdministrativelyProhibited)
            .await;
        return;
    }
    let Some(target) = target else {
        observe::decline_open(row, "invalid-target", channel_number);
        reply.reject(dispatch::connect_failed_open_decline()).await;
        return;
    };
    if !handler.admit_channel_open() {
        observe::decline_open(row, "cap-exceeded", channel_number);
        reply.reject(dispatch::cap_exceeded_open_decline()).await;
        return;
    }
    tunnel_map::spawn_direct_row(
        row,
        channel,
        reply,
        target,
        substrate,
        client,
        Arc::clone(&handler.channel_cap),
    );
}