//! The per-connection server handler — the ADR-004 dispositional auth
//! bridge, plus the parse-only dispatch and the fail-closed decline
//! core (`w2-dispatch-decline-core`; `w2-auth-bridge` filled the auth
//! hooks).
//!
//! ## 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 `tunnel`-featured rows decline at the
//! family-seam stub ([`crate::dispatch::tunnel_seam_stub_decline`])
//! until wave 3's route-1 bridge replaces it. 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
//! (a git-parseable command routes to the git row, anything else to
//! the session pipe row — [`crate::dispatch::classify_exec`]);
//! `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 channel handle
//! a confirmed `session` open produces is *not* consumed this wave —
//! the store-and-take handoff (`channels.remove(&id)` → `into_stream`
//! in the spawned serving task) is wave-3 mapping material; the
//! routing decisions around it are already this wave's, because the
//! routing stubs spawn their decline at the seam.
//!
//! ## 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 session-row seam stubs clone
//! `session.handle()`, spawn their decline through the clone
//! ([`crate::serve`]), 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 derivation rules are wave-3 `mapping.md` material), pinned
//! end-to-end by the mechanics rows in `tests/spawn_return.rs`.
//!
//! The open-level declines (unfeatured/never-feature rows, the tunnel
//! seam stub) stay inline by class: they answer the open *before* any
//! channel is owned — nothing is served yet, so nothing spawns. Wave
//! 3's accept-only-after-fabric-open flow is where the tunnel row
//! gains its spawned serving path, per the same constraint.
//!
//! 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` / the cancel forms) decline — they are
//! `tunnel` rows whose serving flow is wave-3 material, so featured
//! builds decline them too this wave.
//!
//! ## 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 → administratively
//!   prohibited; the tunnel seam stub → 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 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); a
//!   *serving-path* decline (the wave-3 seam stubs) answers failure
//!   and closes the channel — the vanilla-clean shape for a row that
//!   owns the channel. The serving-path decline now rides the spawned
//!   serving task (ADR-006 decision 4, §The spawn-and-return
//!   structural constraint above); the wire shape is unchanged.
//!
//! ## 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 through the
//! `channel_close` hook so the cap recovers. The order of the gates:
//! the feature gate first (an unfeatured build declines everything
//! before the cap matters), the cap gate before any accept — and
//! before the tunnel seam stub, whose decline opens nothing and so
//! must not consume a count. The `max_exec_command_len` cap stays
//! reserved for wave 3's grammar (carried by [`crate::Limits`], not
//! enforced here); 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::sync::Arc;

use alkcall::core::auth::Identity;
use russh::keys::Algorithm;
use russh::server::{
    Auth, ChannelOpenHandle, Handle, Msg, PublicKeyOffer, Session, VerifiedPublicKey,
};
use russh::{Channel, ChannelId, Pty};

use crate::channel_cap::ChannelCap;
use crate::config::ServeConfig;
use crate::dispatch;
use crate::error::AdapterError;
use crate::fingerprint::fingerprint;
use crate::observe;
use crate::serve;

/// 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>,
    channel_cap: ChannelCap,
    resolved_identity: Option<Identity>,
}

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: ChannelCap::default(),
            resolved_identity: None,
        }
    }

    /// 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 — never queued,
    /// never half-open. Called exactly at accept sites: a decline path
    /// (unfeatured, never-feature, the tunnel seam stub) never touches
    /// the count, and the only writer is this connection's packet loop
    /// ([`ChannelCap`]'s single-writer reasoning).
    fn admit_channel_open(&mut self) -> bool {
        self.channel_cap
            .admit(self.serve.limits.max_concurrent_channels)
    }

    /// Whether the cap is already exhausted — the gate slot for the
    /// rows whose decline is ordered *before* the accept decision.
    /// This wave that is the tunnel rows, whose seam stub declines
    /// without ever opening a channel (a counted admission it never
    /// opens would leak against the cap — declined-first-alloc-never,
    /// the ADR-006 dispatch rule, applies to the budget too). Wave 3's
    /// accept-after-fabric-open flow replaces the stub and moves the
    /// admission ([`Self::admit_channel_open`]) to the accept point.
    fn channel_cap_exhausted(&self) -> bool {
        !self
            .channel_cap
            .has_room(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.
    fn serves_session_rows(&self) -> bool {
        cfg!(feature = "session")
    }

    /// The git-row gate (git-parseable exec): only the `git` feature
    /// serves this row.
    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.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;
        } 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`): unfeatured declines
    /// at the open; featured routes to the tunnel seam stub — the
    /// route-1 bridge is wave-3 material, and the stub's decline is
    /// the explicit `ConnectFailed` reject shape ADR-006 decision 1
    /// pins (accept only after the fabric open succeeds; the skeleton
    /// never opens the fabric). The in-crate cap gates before the stub
    /// ([`Self::channel_cap_exhausted`]): a cap-exhausted build
    /// answers the cap reason — the stub opens nothing, so admission
    /// moves to wave 3's real accept point.
    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(());
        }
        if self.channel_cap_exhausted() {
            observe::decline_open(ROW, "cap-exceeded", channel.id().number());
            reply.reject(dispatch::cap_exceeded_open_decline()).await;
            return Ok(());
        }
        tunnel_row_stub(ROW, channel.id().number(), reply).await;
        Ok(())
    }

    /// The `forwarded-tcpip` open gate (`tunnel` — the `-R` data
    /// leg): the same gate/cap/stub split as `direct-tcpip`. The
    /// harness's russh client cannot initiate this channel type (the
    /// server opens it toward the client), so its wire rows are
    /// wave-3 acceptor material; the gate itself is straight-line
    /// identical to the other tunnel rows.
    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(());
        }
        if self.channel_cap_exhausted() {
            observe::decline_open(ROW, "cap-exceeded", channel.id().number());
            reply.reject(dispatch::cap_exceeded_open_decline()).await;
            return Ok(());
        }
        tunnel_row_stub(ROW, channel.id().number(), reply).await;
        Ok(())
    }

    /// The `direct-streamlocal` open gate (`tunnel`): the same
    /// gate/cap/stub split as the tcp rows. The runtime-Unix gate (OQ-11)
    /// gates *serving* on Unix; this wave the stub declines
    /// everywhere, so the runtime gate has nothing to ride yet.
    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(());
        }
        if self.channel_cap_exhausted() {
            observe::decline_open(ROW, "cap-exceeded", channel.id().number());
            reply.reject(dispatch::cap_exceeded_open_decline()).await;
            return Ok(());
        }
        tunnel_row_stub(ROW, channel.id().number(), reply).await;
        Ok(())
    }

    /// The parse-only exec dispatch (ADR-006 point 3 as amended):
    /// parse first ([`crate::dispatch::classify_exec`]) — before any
    /// allocation, decline first, alloc never — then route on the
    /// *feature graph* the route carries: a git-parseable command to
    /// the git row (a session-only build declines here — the git
    /// parse is its gate), 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> {
        let command = std::str::from_utf8(data).unwrap_or("");
        match dispatch::classify_exec(command) {
            dispatch::ExecRoute::Git => {
                if self.serves_git_row() {
                    git_row_stub(channel, session);
                } else {
                    observe::decline_request(
                        "exec_request",
                        "unfeatured",
                        channel.number(),
                        Some("git"),
                    );
                    decline_unfeatured_request(channel, session).await;
                }
            }
            dispatch::ExecRoute::Pipe => {
                if self.serves_session_rows() {
                    pipe_row_stub(channel, session);
                } else {
                    observe::decline_request(
                        "exec_request",
                        "unfeatured",
                        channel.number(),
                        Some("pipe"),
                    );
                    decline_unfeatured_request(channel, session).await;
                }
            }
        }
        Ok(())
    }

    /// The `pty_request` row route (`session`): featured routes to
    /// the session-row seam stub; unfeatured declines the capability.
    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> {
        if self.serves_session_rows() {
            pty_row_stub(channel, session);
        } else {
            observe::decline_request("pty_request", "unfeatured", channel.number(), None);
            decline_unfeatured_request(channel, session).await;
        }
        Ok(())
    }

    /// The `shell_request` row route (`session`): the same shape as
    /// `pty_request`. The OQ-08 unset-shell check is the serving
    /// path's consumption (wave 3): this wave the stub declines the
    /// row regardless, so the unset-declines posture already holds.
    async fn shell_request(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        if self.serves_session_rows() {
            shell_row_stub(channel, session);
        } else {
            observe::decline_request("shell_request", "unfeatured", channel.number(), None);
            decline_unfeatured_request(channel, session).await;
        }
        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): a `tunnel`
    /// row whose serving flow is wave-3 material — the gate exists
    /// ([`Self::serves_tunnels`]) but even a featured build declines
    /// this wave (never advertise a listen that cannot serve). The
    /// bound-port `port = 0` picking rule rides the wave-3 flow.
    async fn tcpip_forward(
        &mut self,
        _address: &str,
        _port: &mut u32,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        observe::decline_global_request("tcpip_forward", "unimplemented");
        Ok(false)
    }

    /// Nothing is ever registered this wave — the cancel form
    /// declines accordingly.
    async fn cancel_tcpip_forward(
        &mut self,
        _address: &str,
        _port: u32,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        observe::decline_global_request("cancel_tcpip_forward", "unimplemented");
        Ok(false)
    }

    /// The `streamlocal-forward` listen request: the same decline as
    /// `tcpip-forward`, plus its runtime-Unix gate (OQ-11) riding the
    /// wave-3 serving flow it will gate.
    async fn streamlocal_forward(
        &mut self,
        _socket_path: &str,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        observe::decline_global_request("streamlocal_forward", "unimplemented");
        Ok(false)
    }

    /// Nothing is ever registered this wave — the cancel form
    /// declines accordingly.
    async fn cancel_streamlocal_forward(
        &mut self,
        _socket_path: &str,
        _session: &mut Session,
    ) -> Result<bool, Self::Error> {
        observe::decline_global_request("cancel_streamlocal_forward", "unimplemented");
        Ok(false)
    }
}

/// 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 family-seam decline shape for a **routed-but-not-yet-served**
/// row — now the spawned serving task's body: fail the request and
/// close the channel, through the cloned handle the task was handed.
/// The vanilla-clean shape for a serving path that owns the channel
/// (wave 3 replaces these stubs with the real serving paths; a
/// declined seam never leaks a half-served channel, since the decline
/// IS the row this wave). 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`, reason `seam-stub`), and —
/// after the wire reporting has completed — ended, or failed when a
/// report send could not complete (the one warn in the vocabulary).
async fn seam_stub_decline(
    handle: Handle,
    channel: ChannelId,
    row: &'static str,
    exec_route: Option<&'static str>,
) {
    observe::serving_path_started(row, channel.number(), exec_route);
    observe::decline_serving(row, "seam-stub", 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 the seam-stub 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.
fn spawn_seam_stub(
    channel: ChannelId,
    session: &Session,
    row: &'static str,
    exec_route: Option<&'static str>,
) {
    let handle = session.handle();
    serve::spawn_serving_path(seam_stub_decline(handle, channel, row, exec_route));
}

/// The git row's seam stub — wave 3 lands the alkgit seam here
/// (`parse command → Service + repo → serve_duplex_resolved`), spawned
/// per ADR-006 decision 4; the skeleton declines at the seam, on its
/// own task.
fn git_row_stub(channel: ChannelId, session: &Session) {
    spawn_seam_stub(channel, session, "exec_request", Some("git"));
}

/// The session pipe row's seam stub — wave 3 lands the alktty pipe
/// seam here (`wrapper ++ [cmd]`, the OQ-08 argv append), spawned per
/// ADR-006 decision 4; the skeleton declines at the seam, on its own
/// task.
fn pipe_row_stub(channel: ChannelId, session: &Session) {
    spawn_seam_stub(channel, session, "exec_request", Some("pipe"));
}

/// The session pty row's seam stub — wave 3 lands the alktty
/// `NegotiateRequest.tty` construction here
/// (`drive_session_pre_negotiated`), spawned per ADR-006 decision 4;
/// the skeleton declines at the seam, on its own task.
fn pty_row_stub(channel: ChannelId, session: &Session) {
    spawn_seam_stub(channel, session, "pty_request", None);
}

/// The session shell row's seam stub — wave 3 lands the alktty
/// `NegotiateRequest.cmd` = serve-config shell argv here (OQ-08, no
/// crate default), spawned per ADR-006 decision 4; the skeleton
/// declines at the seam, on its own task.
fn shell_row_stub(channel: ChannelId, session: &Session) {
    spawn_seam_stub(channel, session, "shell_request", None);
}

/// The tunnel row's seam stub: the explicit `ConnectFailed` reject
/// ([`crate::dispatch::tunnel_seam_stub_decline`]) — wave 3 lands the
/// route-1 bridge here (`TunnelSession::open_direct` → accept only
/// after the fabric open succeeds → `take_halves` pump). The
/// open-level decline event is emitted here: the stub's decline
/// reason is the `connect-failed` class ADR-006 decision 1 pins.
async fn tunnel_row_stub(row: &'static str, channel_id: u32, reply: ChannelOpenHandle) {
    observe::decline_open(row, "connect-failed", channel_id);
    reply.reject(dispatch::tunnel_seam_stub_decline()).await;
}