gunnar_ssh/

server.rs

//! The russh server: authentication, the exec dispatch, and the channel
//! hand-off (the ACL model, the two-listener split).
//!
//! # The shape of a connection
//!
//! 0. **`auth_none`**, the anonymous read path, and the only place
//!    gunnar reads the SSH username. See below.
//! 1. **`auth_publickey`**: russh has already verified that the client owns the
//!    offered key. gunnar fingerprints it (SHA-256 over the wire blob, OpenSSH's
//!    printed form, [`crate::access::fingerprint_of_key`]) and asks
//!    `gunnar-core`'s identity table through [`Control`]. The SSH *username* is
//!    logged and otherwise ignored: identity comes from the key, so
//!    `git@host:ns/repo.git` works and `git` means nothing (the ACL model).
//! 2. **`channel_open_session`**. The channel is kept, because the exec request
//!    arrives as a separate message that only carries a [`ChannelId`], and the
//!    stream can only be taken from the `Channel` itself. A watchdog is armed
//!    here so a session that authenticates and then never execs cannot sit on
//!    the connection.
//! 3. **`env_request`**, answered against [`ENV_ALLOWLIST`], which holds
//!    exactly one name: `GIT_PROTOCOL`. Everything else is refused. See below.
//! 4. **`exec_request`**: [`Dispatch::authorise`] parses, resolves and checks
//!    the grant. Only then is the channel's stream handed to [`GitTransport`].
//! 5. **everything else**. `shell_request`, `subsystem_request`,
//!    `pty_request` and `x11_request` are answered with a channel failure.
//!    **No shell, ever.**
//!
//! # Why the transport is spawned and not awaited
//!
//! russh calls handler methods from the session's own read loop (see
//! `russh::server::Session::run`). Awaiting a whole clone inside `exec_request`
//! would stop the loop that has to deliver the client's bytes to the channel;
//! the transfer would deadlock on its first read. russh's own `sftp_server`
//! example awaits inline; gunnar does not.
//!
//! # The exit status, and why the channel is split rather than consumed
//!
//! `Channel::into_stream()` wraps the channel in russh's `ChannelCloseOnDrop`,
//! which spawns a task sending `ChannelMsg::Close` when the transport lets the
//! stream go. That `Close` raced the `exit-status` the server sent next, and
//! won often enough that a *successful* push reported failure. Measured
//! 2026-08-03: a complete 244-byte advertisement delivered with `ssh` exiting
//! **255**. See [`crate::channel`].
//!
//! The channel is therefore `split()` instead. The read half becomes an owned
//! [`ChannelReader`], the write half stays with the server, and `exit-status`
//! and `close` go out through **the same sender the data went through**, in
//! order, after the transport has finished with it. Nothing sleeps and nothing
//! races.
//!
//! # One span per connection, and what hangs off it
//!
//! ```text
//!   ssh.connection { conn, peer, principal, repo }     gunnar::ssh
//!     └── git.transfer { service, access }             gunnar::git
//!           ├── git.upload_pack | git.receive_pack     gunnar::git
//!           └── policy.gate { ref_name }               gunnar::policy
//! ```
//!
//! `principal` and `repo` are declared empty when the connection opens and
//! **recorded** when `auth_publickey` and `exec_request` learn them, so a line
//! emitted at any depth carries all four fields without any call site restating
//! them. That is the difference from what was here before: connection identity
//! was an ad-hoc `?self.peer` copy-pasted to fourteen call sites, there was no
//! id at all, and two lines from the same connection could not be tied
//! together.
//!
//! Two hand-offs would drop the span if they were not made explicit, and each
//! is a place where an operator would silently lose half a connection:
//!
//! * `tokio::spawn`: the exec watchdog and the transfer task both take
//!   `.instrument(span)`, because a spawned task does not inherit a
//!   thread-local span;
//! * `spawn_blocking`: `gunnar-cli`'s `WireTransport::serve` captures
//!   `Span::current()` and enters it on the blocking thread, which is where the
//!   whole git conversation actually happens.
//!
//! # How an anonymous session arrives, and why it is `auth_none` on one
//! # username
//!
//! SSH has no unauthenticated mode: a client must complete *some* userauth
//! method before it may open a channel. Three shapes were available.
//!
//! | | why not |
//! |---|---|
//! | accept `none` for **every** username | **This is the one that would break the server.** OpenSSH's client always sends a `none` request first, to learn the method list. A server that answers it with SUCCESS authenticates *every* stock client as anonymous before it ever offers its key, so every existing push would start failing, and every existing user would silently lose their identity. Measured against the russh dispatch: `server/encrypted.rs` calls `auth_none` on that first probe regardless of the advertised `methods`. |
//! | a **well-known published key** | The anonymous-FTP shape. It works, but a stranger has to obtain a key file before they can clone, which is worse than the problem it solves, and gunnar would then hold a key whose private half is public in the same table that authenticates real users. |
//! | **`none`, on one reserved username** | Taken. |
//!
//! `ssh://anonymous@host/ns/repo.git` gets the `none` method accepted and a
//! session whose principal is
//! [`Principal::anonymous`](gunnar_core::Principal::anonymous). Every other
//! username gets `none` refused, exactly as russh's default handler refused it
//! before this existed, and the client falls through to publickey unchanged.
//!
//! **Why it cannot become a bypass**, in the order the objections come:
//!
//! * *Does it weaken private repositories?* No. The username selects the
//!   authentication method; the **authorisation** answer still comes from
//!   `AuthTable::may` (the ACL model), where the anonymous principal is one rule, `Read` on
//!   `Public`, and is refused everything else including `may_namespace` and
//!   `may_server`. There is no second ACL (one implementation).
//! * *Can a caller claim to be someone by picking a username?* No. The
//!   anonymous principal is the **only** identity a username can select, it is
//!   never in the identity table, [`Principal::new`] and
//!   `AuthTable::add_principal` refuse its id, so no key and no grant can ever
//!   attach to it.
//! * *Can an authenticated user be silently downgraded?* No. `none` is refused
//!   for their username, so nothing changes for them; and a user who *does*
//!   type `anonymous@` asked for strictly less access, which is not an
//!   escalation.
//! * *Does the username now reach an authorisation decision, against the ACL model?* It
//!   reaches exactly one, in the direction of *removing* everything. The ACL model's
//!   rule exists so that a username cannot be used to become somebody; this
//!   uses it to become nobody.
//!
//! **`none` is not advertised** in [`RusshConfig::methods`], which stays
//! publickey-only. Advertising it would tell every client that `none` is
//! available when it is available only for one name (the failure list is
//! supposed to be true), and russh calls `auth_none` on the client's opening
//! probe either way, which is the only place it is ever needed.
//!
//! [`SshConfig::allow_anonymous`](crate::SshConfig::allow_anonymous) closes the
//! whole path at the listener for an appliance that wants it shut.
//!
//! # `GIT_PROTOCOL`, and why it is an allowlist of one
//!
//! A git client announces protocol v2 over SSH by sending
//! `GIT_PROTOCOL=version=2` as a channel environment request. Refusing every
//! `env_request`, which this server used to do, meant every session fell back
//! to v0. Stock `git` copes; `gix`'s blocking v0 client does not, because it
//! requires `multi_ack_detailed` and `gunnar_wire::caps` correctly declines to
//! advertise a capability it does not implement. The measured consequence was
//! that `gunnar clone` could not clone from `gunnar serve` at all.
//!
//! Accepting the variable is **not** the same as accepting an environment. An
//! SSH server that passes arbitrary client-supplied names through has handed
//! the client `LD_PRELOAD`, `PATH` and every other lever there is. [`ENV_ALLOWLIST`]
//! names the one variable gunnar reads; every other name is refused with a
//! channel failure, exactly as before. The accepted value is not an input to
//! any authorisation decision: it reaches [`ServiceRequest::git_protocol`] and
//! nothing else, and it only ever selects the *framing* the wire answers in.

use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

/// How long [`BoundServer::shutdown`] waits for sessions to answer the
/// disconnect before it aborts the listener. Longer than the gRPC grace (2 s):
/// a git transfer mid-pack deserves a moment to finish its write.
pub const SSH_SHUTDOWN_GRACE: Duration = Duration::from_secs(10);

use russh::keys::ssh_key::PublicKey;
use russh::server::{Auth, Config as RusshConfig, Handler, Msg, Server, Session};
use russh::{Channel, ChannelId, Disconnect, MethodKind, MethodSet};
use tokio::net::TcpListener;
use tokio::task::JoinHandle;
use tracing::Instrument;

use gunnar_core::observe::{RequestId, ServerMetrics, TARGET_GIT, TARGET_SSH};
use gunnar_core::RepoRef;

use crate::access::{fingerprint_of_key, Access, Fingerprint, Principal};
use crate::admission::{Admission, AdmissionStats};
use crate::channel::{
    serve_until_session_ends, session_ended_error, session_over, ChannelReader, SessionBound,
};
use crate::command::GitService;
use crate::command::{self, CommandError};
use crate::config::{SshConfig, ANONYMOUS_SSH_USER};
use crate::control::Control;
use crate::governor::OpClass;
use crate::seam::{Delivery, GitTransport, ServiceRequest};

// ───────────────────────────────────────────────────────────────────────────
// The authorisation decision, without any SSH in it
// ───────────────────────────────────────────────────────────────────────────

/// Why an exec request was not served.
#[derive(Debug, thiserror::Error)]
pub enum DispatchError {
    /// The command did not parse. **The `bash` case, and every other one.**
    #[error("{0}")]
    Command(#[from] CommandError),
    /// The repository does not exist.
    #[error("no such repository: {0}")]
    NoSuchRepo(RepoRef),
    /// The principal does not hold enough access.
    #[error("{principal} holds {held} on {repo}, needs {needed}")]
    Denied {
        /// Who asked.
        principal: String,
        /// What they asked for.
        repo: RepoRef,
        /// What the command requires.
        needed: Access,
        /// What they actually hold, rendered as `none` when they hold nothing.
        held: String,
    },
    /// **This box is a read-only STANDBY** and the write is refused.
    ///
    /// A distinct arm, not a [`DispatchError::Denied`]: the writer holds the
    /// grant — on a primary this push would land — so "you do not have access"
    /// would be a lie that sends them to an administrator about permissions.
    /// The honest sentence is that this box does not take client writes; push
    /// to the primary. It is only ever reached by a principal who already
    /// passed the grant check, so it leaks no repository the writer could not
    /// already see.
    #[error("{principal}: this box is a standby; writes go to the primary ({repo})")]
    Standby {
        /// Who asked.
        principal: String,
        /// The repository they tried to write.
        repo: RepoRef,
        /// The box is a browse replica, so the
        /// refusal says so: a stranger who found a public repository here and
        /// tried to push to it should learn what this box is, not merely that
        /// it is frozen.
        browse: bool,
    },
    /// **This box is a TWIN and serves no git at all** — not a read, not a
    /// write, not to anybody (2026-09-21, lane T4).
    ///
    /// Its own arm and not a [`DispatchError::Standby`], because a standby is a
    /// box that serves reads and refuses writes, and a twin is a box that
    /// serves NEITHER: `serve.rs` says why in one line — *a read would grow the
    /// caches the twin must not hold* — and the twin's whole value is that its
    /// disk holds the primary's log and nothing derived.
    ///
    /// **Why this exists at the door rather than only at the bind.** The bind
    /// is chosen ONCE, in `serve::start`, from `stores.is_twin()` at that
    /// instant. A box becomes a twin over the wire, through
    /// `Bootstrap.BecomeTwin`, long after that — and the owner's one-medium
    /// ruling makes that the ONLY way it happens, because a twin is now
    /// installed from the same image as a main with no per-box parameter to
    /// say which it is. MEASURED under KVM by lane T3, 2026-09-21: a box
    /// conferred at runtime kept `0.0.0.0:2222` bound for the rest of its life,
    /// the `this box is a TWIN` warning never printed, and its volume ended up
    /// holding five `objects.exploded` cache files. A socket that cannot be
    /// moved without a restart is not a reason to serve git from it in the
    /// meantime.
    #[error("{principal}: this box is a TWIN; it serves no git to anyone — clone and push to the primary it follows ({repo})")]
    Twin {
        /// Who asked.
        principal: String,
        /// What they asked for.
        repo: RepoRef,
    },
    /// **Anonymous reads are not served here** (`--anonymous refuse`); the
    /// browse replica serves them, and `browse_host` names it when the
    /// operator did. Raised before the repository is even looked up, so a
    /// stranger learns nothing about what exists on the primary.
    #[error("anonymous reads are refused on this box ({repo}); browse host: {browse_host:?}")]
    AnonymousElsewhere {
        /// The repository the stranger asked for.
        repo: RepoRef,
        /// Where anonymous reads are served, when the operator named it.
        browse_host: Option<String>,
    },
    /// **The gate chain refuses this**: the disk is under its floor (every
    /// build), or (`metered`) a Suspended or Retention tenant's push or LFS
    /// upload, an anonymous read of a lapsed tenant's public repository, or a
    /// push at the pack quota.
    ///
    /// Its own arm for the same reason [`DispatchError::Standby`] is one: the
    /// caller holds the grant, so "you do not have access" would be a lie
    /// that sends a paying customer to an administrator about permissions.
    /// The honest sentence names the fact — free bytes and the floor, or the
    /// state, the date and where to renew — and it is only ever reached AFTER
    /// the grant check admitted the request, so it leaks nothing about a
    /// repository the caller could not already see.
    #[error("{namespace}: {line}")]
    Plan {
        /// The tenant.
        namespace: String,
        /// What the client is told, verbatim.
        line: String,
    },
    /// The identity table or the catalog could not be read.
    ///
    /// **Deliberately not folded into [`DispatchError::Denied`].** The tables
    /// live in storage (the Arrow-IPC ref log) and a read can fail; answering a storage
    /// fault with "you do not have access" would turn an outage into a silent,
    /// convincing permission denial that nobody would think to investigate.
    /// The ACL model is explicit that `may(...)` returns `Result<bool>` for this
    /// reason, and this variant is where that `Result` lands.
    #[error("the control plane could not answer: {0}")]
    Store(#[from] gunnar_core::Error),
}

impl DispatchError {
    /// What the client is told.
    ///
    /// `None` means "answer with a channel failure and say nothing": the
    /// request was not even a git command, so there is no git client on the
    /// other end that would understand a pkt-line.
    ///
    /// **"No such repository" and "no permission" produce the same string on
    /// purpose.** A client that could tell them apart could enumerate private
    /// repositories one guess at a time. The server log keeps the distinction;
    /// the wire does not.
    ///
    /// Those two arms are **one arm**, and were two identical `format!`s until
    /// 2026-08-05. That is the twinning one implementation forbids, and this is the case it
    /// forbids it for: the property is that the two strings are equal, so two
    /// copies of the string put the property one careless edit away from being
    /// false, in the direction where the wire starts leaking which private
    /// repositories exist.
    /// `a_missing_repository_and_a_forbidden_one_are_one_string_to_the_client`
    /// guards it from the far side of a real session; this makes the guard hard
    /// to need.
    pub fn client_message(&self) -> Option<String> {
        match self {
            // A verb gunnar knows and refuses BY NAME gets its sentence; the
            // rest of the parse failures are not git clients at all.
            DispatchError::Command(CommandError::NotServed { verb, why }) => {
                Some(format!("{verb}: {why}"))
            }
            DispatchError::Command(_) => None,
            DispatchError::NoSuchRepo(repo) | DispatchError::Denied { repo, .. } => Some(format!(
                "repository not found, or you do not have access: {repo}"
            )),
            // A standby refusal is told plainly — the writer is authorised, so
            // there is no repository to hide from them, and "push to the
            // primary" is the one thing they need to hear to succeed.
            // A twin refusal is told plainly for the same reason, and it is told
            // to EVERYONE including a stranger: it names no repository and
            // reveals nothing this box holds — "this is a twin" is a fact about
            // the box, not about its contents.
            DispatchError::Twin { .. } => Some(
                "this box is a TWIN: it holds a copy of the primary's transaction log and serves \
                 no git to anyone, read or write. Clone and push to the primary it follows"
                    .to_string(),
            ),
            DispatchError::Standby { browse: false, .. } => {
                Some("this box is a standby; push to the primary".to_string())
            }
            DispatchError::Standby { browse: true, .. } => Some(
                "this box is a browse replica: it serves public repositories read-only and \
                 takes no writes; push to the primary"
                    .to_string(),
            ),
            DispatchError::AnonymousElsewhere {
                browse_host: Some(host),
                repo,
            } => Some(format!(
                "anonymous reads are served by the browse replica, not here: clone \
                 {repo} from {host} (or authenticate with a registered key)"
            )),
            DispatchError::AnonymousElsewhere {
                browse_host: None, ..
            } => Some(
                "this server does not serve anonymous reads; authenticate with a registered \
                 key, or use the browse replica"
                    .to_string(),
            ),
            // The gate's refusal is told in full: the state, the date and the
            // URL are the three things a lapsed tenant needs to get going again,
            // and the free bytes and the floor are what a full box's operator
            // needs; hiding any of them would turn a fact into a support ticket.
            DispatchError::Plan { line, .. } => Some(line.clone()),
            // A fault, not a decision. The client is told the server broke,
            // because telling it "you do not have access" would be a lie that
            // sends the user to their administrator about permissions.
            DispatchError::Store(_) => {
                Some("the gunnar control plane is unavailable; try again".to_string())
            }
        }
    }
}

/// Parse → resolve → check, with no SSH anywhere in it.
///
/// Kept separate from the [`Handler`] so the security decision can be tested
/// directly, exhaustively and without a handshake, and so there is exactly one
/// place where it is made.
#[derive(Clone, Debug)]
pub struct Dispatch {
    control: Control,
}

impl Dispatch {
    /// Build a dispatch over the control plane.
    pub fn new(control: Control) -> Self {
        Dispatch { control }
    }

    /// The control plane this dispatch authenticates and authorises against.
    pub fn control(&self) -> &Control {
        &self.control
    }

    /// Decide whether `principal` may run `raw` on this server.
    ///
    /// Parse (which includes the sandbox), resolve, then check the grant. A
    /// storage failure at either lookup comes back as
    /// [`DispatchError::Store`] rather than as a denial. See
    /// [`crate::control`].
    ///
    /// **The client's environment is not a parameter here, and must not
    /// become one.** The returned [`ServiceRequest`] carries
    /// `git_protocol: None`; the handler attaches the allowlisted value with
    /// [`ServiceRequest::with_git_protocol`] *after* this decision has already
    /// been made, so there is no path by which anything the client typed into
    /// its environment can reach the grant check.
    pub fn authorise(
        &self,
        principal: &Principal,
        key_fingerprint: Option<&Fingerprint>,
        raw: &[u8],
    ) -> Result<ServiceRequest, DispatchError> {
        let parsed = command::parse(raw)?;
        self.authorise_service(principal, key_fingerprint, parsed.service, parsed.repo)
    }

    /// The same decision, entered from a **URL path** rather than an exec line.
    ///
    /// Smart HTTP does not send `git-upload-pack '<path>'`; it sends
    /// `POST /<path>/git-upload-pack`. The verb and the path arrive already
    /// separated, and the path has no shell quoting on it, so there is nothing
    /// for [`command::parse`] to do — but everything **after** the parse is
    /// identical and must stay identical: resolve, ask the ACL, and refuse in
    /// the one way that does not tell a caller whether a repository exists.
    ///
    /// So this is where that decision lives and [`Dispatch::authorise`] is now
    /// a parser in front of it. Two entry points, one answer (one implementation). A second
    /// copy of this in `gunnar-http` is precisely the thing that ends with SSH
    /// and HTTP disagreeing about who may push, and the disagreement being
    /// found by whoever pushes.
    ///
    /// `repo` is a [`RepoRef`], which is to say it has already been through
    /// [`gunnar_core::resolve_repo_path`]. The sandbox is not optional on this
    /// path either: the caller cannot construct a `RepoRef` without it.
    pub fn authorise_service(
        &self,
        principal: &Principal,
        key_fingerprint: Option<&Fingerprint>,
        service: GitService,
        path: RepoRef,
    ) -> Result<ServiceRequest, DispatchError> {
        let required = service.required_access();
        // **A TWIN serves no git**, before anything else is asked — before the
        // anonymous policy, before the catalog, before the grant. A read on a
        // twin grows exactly the caches the twin exists not to hold, and the
        // box's own `browse` flag (which `become_twin` sets) would otherwise
        // OVERRIDE `--anonymous refuse` and serve a stranger. This is a live
        // read of the same flag `ServerStores` flips, so the refusal begins the
        // instant `Bootstrap.BecomeTwin` returns and does not wait for a
        // restart to move the listener.
        if self.control.is_twin() {
            return Err(DispatchError::Twin {
                principal: principal.id.clone(),
                repo: path,
            });
        }
        // **`--anonymous refuse`** (the browse replica), before the catalog is consulted:
        // a stranger on a primary that has a browse replica is sent there, and
        // learns nothing about what this box holds on the way.
        if principal.id == gunnar_core::ANONYMOUS_PRINCIPAL_ID && self.control.refuses_anonymous() {
            return Err(DispatchError::AnonymousElsewhere {
                repo: path,
                browse_host: self.control.browse_host().map(str::to_owned),
            });
        }
        let (repo, prospective) = match self.control.resolve(&path)? {
            Some(repo) => (repo, false),
            // The repository does not exist. For exactly one shape of request
            // — a PUSH, by a principal the control plane already permits to
            // create repositories here — absence is the first-repository door
            // rather than a refusal: the repository is created and the push
            // proceeds through the unchanged grant check below.
            //
            // # Why a door here, and not a fourth verb or the control plane
            //
            // The shipped appliance runs `--no-grpc` (gRPC v1 is
            // unauthenticated, loopback-only), has no shell by construction,
            // and its exec surface is three git verbs that all want a
            // repository — so after `--admin-key` enrolled the bootstrap
            // admin (a8ba0665), the admin could authenticate and then do
            // NOTHING, forever (measured under KVM, 2026-08-25). The
            // alternatives lost on the code, not on taste:
            //
            // * **a create verb** would be a wire surface no git client
            //   speaks, would need its own client incantation, and does not
            //   fit the dispatch's own shape: `SERVICES` carries per-service
            //   *repo grant levels* and this decision compares levels — a
            //   verb whose target does not exist has no repo to hold a level
            //   on, so it would be a special case beside the table, and a
            //   probe-discoverable one for every stranger;
            // * **turning gRPC on** is the door the unauthenticated v1 and
            //   `gunnar-grpc/tests/loopback_default.rs` exist to keep shut on
            //   a mailed medium: `Bootstrap.Enroll` with no credential is a
            //   race between the customer and whoever else finds the box.
            //
            // Push-to-create adds no verb, no parse production and no
            // vocabulary: `Control::create_for_push` asks the exact two
            // questions `Repos.CreateNamespace`/`Repos.Create` gate on, so
            // this door and the RPC door can never disagree (one implementation) — and it
            // sits HERE, where the two entry points (exec line, smart-HTTP
            // URL) have already fanned in, so SSH and HTTP answer a push to a
            // missing repository identically.
            //
            // Only a WRITE: ReceivePack, and (2026-09-28) an LFS upload.
            // Reading never creates: a `git clone` of a typo, or an LFS
            // download, stays an error.
            //
            // **Why the LFS upload too.** git-lfs's pre-push hook runs AFTER
            // git has taken receive-pack's advertisement and BEFORE it sends
            // a command, and it asks `git-lfs-transfer <repo> upload` first.
            // MEASURED 2026-09-28 (git-lfs 3.8.0, and on a real box): into a
            // repository that did not exist yet that exec was the fused
            // not-found, the hook died — `batch request: exec request failed
            // on channel 0: exit status 255` — and the push that would have
            // created the repository never sent its commands. The upload is
            // the same act as the push (it puts content in the repository),
            // by the same principal, and it is asked the same two questions
            // (`push_create_spec`), so it opens the same door and nobody else
            // learns anything new. Nothing is written here either: the
            // transport creates the repository when the first OBJECT arrives
            // (`gunnar-server` `ProspectiveLfs`), so a client that only
            // batches, or hangs up, leaves nothing.
            //
            // For everyone `create_for_push` answers `None` for, the refusal
            // below is bit-identical to what it always was — the one fused
            // "repository not found, or you do not have access" — so the
            // door's existence is not observable to a key it does not admit.
            //
            // **Nothing is written here.** The door answers with the
            // repository it WOULD create (`prospective`), the transport
            // advertises that, and the catalog row and the store are made
            // only once the client has sent commands that pass the hash
            // check — see `Control::push_create_spec` for the ghost
            // repository this order closes.
            None if service == GitService::ReceivePack
                || service == GitService::LfsTransfer(crate::command::Operation::Upload) =>
            {
                match self.control.push_create_spec(principal, &path)? {
                    Some(spec) => {
                        // The name filter (2026-10-05) — asked only of a
                        // pusher the door admits, so its sentence tells
                        // nobody else anything.
                        if let Some(line) = self.control.push_create_name_refusal(principal, &path)
                        {
                            return Err(DispatchError::Plan {
                                namespace: path.namespace().to_owned(),
                                line,
                            });
                        }
                        (spec.prospect()?, true)
                    }
                    None => return Err(DispatchError::NoSuchRepo(path)),
                }
            }
            None => return Err(DispatchError::NoSuchRepo(path)),
        };
        let held = self.control.access_for(principal, &repo)?;
        // **A pull mirror takes no person's push** (GUN-FIELD-TEST, 2026-10-31). Its references
        // are the upstream's: a pushed commit was accepted and the next sync moved the branch
        // back over it, without a word to anybody. Asked for a write by a non-`Machine`
        // principal only (the standby freeze's rule: a box's own machine identities are not
        // people), and read only then, so a read costs nothing new.
        let mirror_of = if required.permits(Access::Write)
            && principal.kind != gunnar_core::PrincipalKind::Machine
        {
            self.control.pull_mirror_of(&path)?
        } else {
            None
        };
        match held {
            // **The standby freeze.** Checked here, INSIDE the authorised arm,
            // so it is only ever reached by a writer the grant check already
            // admitted — an unauthorised push still gets the fused not-found and
            // learns nothing new. A write-requiring service by a non-`Machine`
            // principal on a frozen box is refused: this box has abdicated or is
            // a configured standby, and a client write here would be a split
            // the primary never sees. The primary's own replication push is a
            // `Machine` identity and sails through — that is what a standby is
            // for.
            Some(level)
                if level.permits(required)
                    && required.permits(Access::Write)
                    && self.control.write_frozen_for(principal) =>
            {
                Err(DispatchError::Standby {
                    principal: principal.id.clone(),
                    repo: path,
                    browse: self.control.is_browse(),
                })
            }
            // The pull mirror (above): inside the authorised arm, so a stranger still gets the
            // fused not-found and learns nothing about what this box mirrors.
            Some(level) if level.permits(required) && mirror_of.is_some() => {
                Err(DispatchError::Plan {
                    namespace: path.namespace().to_owned(),
                    line: format!(
                        "{path} is a pull mirror of {}: its references are the upstream's, and \
                         the next sync moves them back over anything pushed here. Push to the \
                         upstream, or stop mirroring (`gunnar control mirror remove {path}`) to \
                         make this repository its own",
                        mirror_of.as_deref().unwrap_or_default()
                    ),
                })
            }
            // **An encrypted repository takes no LFS upload** (wiring tracer B
            // 2026-10-08). git-lfs uploads in the pre-push hook, BEFORE the
            // pack the encrypted gate reads, and an LFS object is the
            // plaintext of a `filter=lfs` file that never went through
            // `filter=gunnar-crypt`: refused here, for SSH and smart HTTP
            // alike, before a byte is stored. Asked inside the authorised arm,
            // so a stranger still gets the fused not-found. A download is never
            // refused.
            Some(level)
                if level.permits(required)
                    && repo.encrypted
                    && service == GitService::LfsTransfer(crate::command::Operation::Upload) =>
            {
                Err(DispatchError::Plan {
                    namespace: path.namespace().to_owned(),
                    line: format!(
                        "{path} is encrypted, so it takes no LFS objects: LFS stores files as \
                         they are, outside your members' encryption. Track the file with \
                         `filter=gunnar-crypt` instead of `filter=lfs`"
                    ),
                })
            }
            Some(level) if level.permits(required) => {
                // **The gate chain**, asked LAST: after the grant (so a
                // refusal leaks nothing to a stranger) and after the standby
                // freeze (so a standby stays one whatever the gates say). A
                // write asks `Write`; an anonymous read asks `AnonymousRead`;
                // an authenticated read asks nothing at all, because the
                // owner's clone is allowed in every state — data is never held
                // hostage. See `crate::plan`.
                let PlanOutcome {
                    notice: plan_notice,
                    ceiling: pack_ceiling,
                    pack_mark,
                    allowance: lfs_allowance,
                } = self.ask_plan(principal, service, required, &repo)?;
                Ok(ServiceRequest {
                    service,
                    repo,
                    prospective,
                    principal: principal.clone(),
                    access: level,
                    // The key that authenticated this session, carried
                    // rather than guessed back out of the identity table. A
                    // parameter and not a builder, so it cannot be forgotten into
                    // `None`.
                    key_fingerprint: key_fingerprint.cloned(),
                    git_protocol: None,
                    delivery: Delivery::Session,
                    // Derived from `key_fingerprint` by the transport rather
                    // than spelled here: see `ServiceRequest::credential`.
                    credential: None,
                    plan_notice,
                    pack_ceiling,
                    pack_mark,
                    lfs_allowance,
                    cancel: crate::channel::Cancel::new(),
                })
            }
            other => Err(DispatchError::Denied {
                principal: principal.id.clone(),
                repo: path,
                needed: required,
                held: match other {
                    Some(level) => level.to_string(),
                    None => "none".to_string(),
                },
            }),
        }
    }

    /// **Ask the gate chain**, for a request the grant check already admitted.
    /// `Ok(Some(line))` is an allowed request that carries a warning (Grace);
    /// `Err(Plan)` is the refusal, with the line the client is told.
    ///
    /// The three shapes, and why each asks what it asks:
    ///
    /// * a **write** (`git-receive-pack`, LFS upload) asks
    ///   [`PlanAction::Write`](crate::plan::PlanAction::Write), and if that is
    ///   allowed asks [`PlanGate::pack_quota`](crate::plan::PlanGate::pack_quota)
    ///   — both before a single pack byte is read;
    /// * an **anonymous read** asks
    ///   [`PlanAction::AnonymousRead`](crate::plan::PlanAction::AnonymousRead):
    ///   a lapsed tenant's public surface is what costs the provider and earns
    ///   nothing;
    /// * an **authenticated read** asks nothing. The owner clones their own
    ///   repositories in every state, and so does anyone they granted. That is
    ///   rule 1 of `gunnar_core::entitlement`, and it is enforced here by the
    ///   absence of a question rather than by an answer.
    ///
    /// **The chain.** Gates are asked in the order `Control::with_plan_gate`
    /// added them — disk first — and the FIRST `Refuse` is the answer; a later
    /// gate is not consulted, so a full box is told it is full and not also
    /// that its plan lapsed. Each gate is asked BOTH its questions before the
    /// next gate is asked anything: its ladder (`verdict`), then, for a write
    /// that grows data, its quota (`pack_quota`). Within one gate that order is
    /// what tells a Suspended tenant they are suspended rather than also full;
    /// across gates it is what makes "disk first" mean disk first — the disk
    /// gate only ever refuses from `pack_quota`, and a chain that asked every
    /// ladder before any quota would let a plan's ladder outrank it whatever
    /// the order (`tests/disk_headroom.rs` caught exactly that, red first).
    /// Warnings from every allowing gate are joined.
    ///
    /// With no gate wired (`Control::with_plan_gate` never called) every
    /// namespace is Free and this is a no-op. A `Machine` principal —
    /// replication, the primary's copy arriving — is asked only of the gates
    /// that say they apply to it ([`PlanGate::applies_to_machine`]): the disk
    /// does, a plan does not, because refusing a lapsed tenant's replication
    /// would make their standby diverge from the primary.
    fn ask_plan(
        &self,
        principal: &Principal,
        service: GitService,
        required: Access,
        repo: &gunnar_core::Repo,
    ) -> Result<PlanOutcome, DispatchError> {
        use crate::plan::{PlanAction, PlanVerdict};
        let machine = principal.kind == gunnar_core::PrincipalKind::Machine;
        let gates: Vec<_> = self
            .control
            .plan_gates()
            .iter()
            .filter(|gate| !machine || gate.applies_to_machine())
            .collect();
        if gates.is_empty() {
            return Ok(PlanOutcome::default());
        }
        let anonymous = principal.id == gunnar_core::ANONYMOUS_PRINCIPAL_ID;
        let action = if required.permits(Access::Write) {
            PlanAction::Write
        } else if anonymous {
            PlanAction::AnonymousRead
        } else {
            return Ok(PlanOutcome::default());
        };
        let refused = |line: String| DispatchError::Plan {
            namespace: repo.namespace.clone(),
            line,
        };
        // `service` is named here so a future fourth write-shaped verb has to
        // decide whether it grows the tenant's data.
        let grows_data = matches!(
            service,
            GitService::ReceivePack | GitService::LfsTransfer(crate::command::Operation::Upload)
        );
        let mut notices: Vec<String> = Vec::new();
        let mut ceiling: Option<crate::plan::PackCeiling> = None;
        let mut allowance: Option<gunnar_lfs::Allowance> = None;
        // **Marked BEFORE any gate walks the disk, and that ordering is the
        // contract**, not an accident of where the line sits. A gate's
        // `lfs_allowance` measures the namespace's bytes off the disk; an upload
        // admitted DURING that walk may or may not have landed bytes for the
        // walk to see. Taking the mark first puts such an admission after the
        // mark, so it is charged — and charged twice if the walk saw it too,
        // which refuses a shade early. Taking the mark after the walk would put
        // it before the mark and possibly outside the walk, which admits too
        // much. Only one of those two roundings is a quota that leaks. See
        // `gunnar_lfs::ledger`.
        let mark = gunnar_lfs::AllowanceLedger::shared().mark(&repo.namespace);
        // …and the same standing in the PACK meter, taken in the same breath and
        // for the same reason: `pack_ceiling` below measures the namespace's
        // packfile bytes off the disk, and a push admitted during that walk has
        // to land after this mark to be counted.
        //
        // **Two marks and not one, because they are two meters.**
        // `caps["pack_bytes"]` and `caps["lfs_bytes"]` are capped separately on
        // the signed fact and reported separately by `Usage.Read`, and the two
        // numbers these marks are judged against are computed from different
        // caps. One shared counter would charge an LFS upload against a push's
        // ceiling — cutting a push mid-transfer over room its own cap never
        // lost — which is a quota that refuses wrongly, the same defect as one
        // that admits wrongly. They share a disk; they do not share a cap. See
        // `gunnar_lfs::AllowanceLedger::shared_pack`.
        let pack_mark = gunnar_lfs::AllowanceLedger::shared_pack().mark(&repo.namespace);
        for gate in &gates {
            match gate.verdict(&repo.namespace, action) {
                PlanVerdict::Allow => {}
                PlanVerdict::Warn(line) => notices.push(line),
                PlanVerdict::Refuse(line) => return Err(refused(line)),
            }
            // The quota, only for a write this gate's ladder allowed.
            if action == PlanAction::Write && grows_data {
                if let PlanVerdict::Refuse(line) = gate.pack_quota(&repo.namespace) {
                    return Err(refused(line));
                }
                // **The smallest ceiling in the chain wins**, which is the
                // byte-wise spelling of first-`Refuse`-wins one line above: a
                // push has to satisfy every gate, so it is held to the
                // tightest one, and the line the client reads is that gate's
                // own. Asked only where `pack_quota` is asked, so a read and a
                // deletion are never given a ceiling to trip over.
                if let Some(asked) = gate.pack_ceiling(&repo.namespace) {
                    let tighter = match &ceiling {
                        Some(held) => asked.bytes < held.bytes,
                        None => true,
                    };
                    if tighter {
                        ceiling = Some(asked);
                    }
                }
                // …and the same tightest-wins rule for the LFS wire, for the
                // same reason: a transfer has to satisfy every gate, so it is
                // held to the smallest room any of them will admit, and the
                // sentence the tenant reads is that gate's own.
                if let Some(asked) = gate.lfs_allowance(&repo.namespace) {
                    let tighter = match &allowance {
                        Some(held) => asked.bytes < held.bytes,
                        None => true,
                    };
                    if tighter {
                        allowance = Some(asked);
                    }
                }
            }
        }
        Ok(PlanOutcome {
            notice: (!notices.is_empty()).then(|| notices.join("; ")),
            // The tightest gate's number and the standing taken before any gate
            // looked, carried side by side rather than folded into one type:
            // see `ServiceRequest::pack_mark` for why the ceiling stays exactly
            // what `PlanGate::pack_ceiling` returned.
            pack_mark: ceiling.is_some().then_some(pack_mark),
            ceiling,
            // The tightest gate's number, bound to the standing taken above it.
            // `mark` is dropped unused when no gate asked for an allowance,
            // which costs one map read and holds nothing: there is no
            // reservation in a mark to leak.
            allowance: allowance
                .map(|allowance| gunnar_lfs::NamespaceAllowance::new(mark, allowance)),
        })
    }
}

/// What [`Dispatch::ask_plan`] learned: everything the gate chain wants carried
/// into the transfer, as opposed to the refusals, which are returned as errors
/// and never reach here.
///
/// A struct and not a tuple because both fields are `Option` and both are
/// about the plan: a `(Option<String>, Option<PackCeiling>)` at the call site
/// is two anonymous maybes that the next field added would make three.
#[derive(Default)]
struct PlanOutcome {
    /// See [`crate::ServiceRequest::plan_notice`].
    notice: Option<String>,
    /// See [`crate::ServiceRequest::pack_ceiling`].
    ceiling: Option<crate::plan::PackCeiling>,
    /// See [`crate::ServiceRequest::pack_mark`]. `Some` exactly when `ceiling`
    /// is: a mark with no number to be judged against is bookkeeping nobody
    /// reads, and a number with no mark is the bug this closed.
    pack_mark: Option<gunnar_lfs::Mark>,
    /// See [`crate::ServiceRequest::lfs_allowance`]. A third `Option` on this
    /// struct, and the reason the struct exists rather than a tuple.
    allowance: Option<gunnar_lfs::NamespaceAllowance>,
}

// ───────────────────────────────────────────────────────────────────────────
// The server
// ───────────────────────────────────────────────────────────────────────────

struct Shared {
    dispatch: Dispatch,
    transport: Arc<dyn GitTransport>,
    exec_grace: Option<Duration>,
    /// Whether `auth_none` on [`ANONYMOUS_SSH_USER`] is accepted.
    allow_anonymous: bool,
    admission: Arc<Admission>,
    metrics: Arc<ServerMetrics>,
    /// **Are the stores open?** See [`GunnarSsh::with_readiness`].
    ready: Arc<std::sync::atomic::AtomicBool>,
    /// [`SshConfig::max_auth_attempts`], carried here because the CONNECTION
    /// enforces it and russh 0.61.2 does not. See
    /// [`Connection::auth_attempts`].
    config_max_auth_attempts: usize,
    /// [`SshConfig::session_end_grace`] and
    /// [`SshConfig::session_end_grace_crowded`].
    session_end_grace: (Duration, Duration),
    /// [`SshConfig::transfer_idle_timeout`] (GUN-SSH-SLOT-HANG, [`crate::idle`]).
    transfer_idle: Option<Duration>,
    /// The sealed box's tunnel-address door ([`TUNNEL_ADDRESS_VERB`]), or
    /// `None` on a build with no WireGuard tunnel.
    tunnel_door: Option<TunnelAddressDoor>,
    /// The repository-settings door ([`REPO_VERB`]), or `None` on a server
    /// with no control plane (`serve --no-grpc`).
    repo_door: Option<RepoDoor>,
}

/// **The one control verb on this listener** (2026-09-21, T17):
/// `gunnar-tunnel-address [a.b.c.d/prefix]`.
///
/// The sealed installer runs `serve --no-grpc`, so its SSH door is its only
/// door, and a box's WireGuard tunnel address is per box and not on the
/// medium. So it arrives here, held to **server-wide Admin** and answered by
/// the closure `gunnar-server` injects — the SAME function the gRPC doors call.
/// No argument is a read. Output is `key value` lines on stdout, exit 0; a
/// refusal is `error: <why>` on stdout, exit 1 (stdout and not stderr, so a
/// client reading only the data band still gets the sentence). Not a git
/// verb, so it never enters [`Dispatch::authorise`] or the transfer path.
pub const TUNNEL_ADDRESS_VERB: &str = "gunnar-tunnel-address";

/// **The operator's fire drill** (stinkies ssh-double-exec-error, 2026-09-30): ONE exec
/// verb, held to server-wide Admin like [`TUNNEL_ADDRESS_VERB`], that says exactly one
/// ERROR naming itself a drill and who asked — what the estate's watch raises on purpose
/// to prove its alert path end to end. Nothing is wrong when it is said; no other caller
/// can raise the server's ERROR at will.
pub const FIRE_DRILL_VERB: &str = "gunnar-fire-drill";

/// **A member's SSH public keys, for an encrypted repository's key ring**
/// (2026-10-06): `gunnar-recipient-keys '<namespace>/<repository>' '<member>'`.
///
/// What `gunnar crypt add-member NAME --from-server` asks when it is given no
/// control address — on an installed estate the forge's SSH door (tcp/2222
/// through the front) is the one door a member's key opens; the control plane
/// is bound to the WireGuard overlay. The SAME rule as `Repos.RecipientKeys`:
/// the caller reads the repository (otherwise it is told what a stranger is
/// told, the one "not found, or no access" sentence), and so must the person
/// named — gunnar hands out keys only for someone who could read it anyway,
/// never a directory of everybody's keys. Output is one OpenSSH public key
/// line per key on stdout, exit 0; a refusal is `error: <why>` on stdout,
/// exit [`EXIT_DENIED`] (stdout, like [`TUNNEL_ADDRESS_VERB`]'s). An anonymous
/// session is refused the channel. Not a git verb: it never enters
/// [`Dispatch::authorise`] or the transfer path, and it changes nothing.
pub const RECIPIENT_KEYS_VERB: &str = "gunnar-recipient-keys";

/// **Who this key is** (wiring tracer B 2026-10-08): `gunnar-whoami` answers the principal id
/// the session authenticated as, one line, exit 0 — what `gunnar crypt init` names its first
/// member by (the account name `add-member --from-server` looks members up by). Reveals nothing
/// the key's holder does not already know. An anonymous session is refused the channel. Not a
/// git verb.
pub const WHOAMI_VERB: &str = "gunnar-whoami";

/// **git-lfs's fallback verb**, answered with a sentence instead of a channel
/// failure. Gunnar serves Git LFS over pure SSH only (`git-lfs-transfer`,
/// git-lfs >= 3.0); git-lfs asks this verb only when a pure-SSH session failed
/// or was refused, and then prints what is on stderr after `batch request:`.
pub const LFS_AUTHENTICATE_VERB: &str = "git-lfs-authenticate";

/// What [`LFS_AUTHENTICATE_VERB`] answers, on stderr with exit 1. It must never
/// say `not found` / `no such file`: git-lfs reads those as "no such verb" and
/// goes on to a guessed HTTPS endpoint, which Gunnar does not serve.
pub const LFS_AUTHENTICATE_MESSAGE: &str =
    "Gunnar serves Git LFS over pure SSH only (git-lfs-transfer, \
git-lfs 3.0 or newer); there is no HTTPS LFS endpoint. The pure SSH transfer just failed or was \
refused (the server was busy or restarting): run the same command again.\n";

/// **A repository's settings, over the forge's door** (owner, 2026-10-14):
/// `gunnar-repo show '<namespace>/<repository>'`,
/// `gunnar-repo visibility '<namespace>/<repository>' public|private`,
/// `gunnar-repo default-branch '<namespace>/<repository>' '<refs/heads/…>'`.
///
/// What the client `gunnar repo show|visibility|default-branch` asks. On an
/// installed estate the forge's SSH door (tcp/2222 through the front) is the
/// one door a person's key opens; the control plane is bound to the WireGuard
/// overlay. The verb decides NOTHING itself: it hands the authenticated
/// principal and the ask to [`RepoDoor`], which `gunnar-server` answers with
/// the SAME `Repos.Get` / `Repos.SetVisibility` / `Repos.SetDefaultBranch`
/// handlers the control plane mounts — so the account-admin rule for
/// visibility, the repository-Admin rule for the default branch, the audit
/// event and the browse replica's notice are the gRPC door's, not a copy.
/// Output is `key value` lines on stdout, exit 0; a refusal is
/// `error: <why>` on stdout, exit [`EXIT_DENIED`]. An anonymous session is
/// refused the channel. Not a git verb: it never enters
/// [`Dispatch::authorise`] or the transfer path.
pub const REPO_VERB: &str = "gunnar-repo";

/// What a [`REPO_VERB`] line asks.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RepoAction {
    /// `Repos.Get`.
    Show,
    /// `Repos.SetVisibility`: `true` is public.
    Visibility(bool),
    /// `Repos.SetDefaultBranch`, the branch exactly as sent (the server
    /// validates it; the client expands a short name).
    DefaultBranch(String),
    /// `Repos.Create` of an empty repository in this object format (`true`
    /// is sha256): what a client that knows its own format asks before its
    /// first push, since a push-made repository can only be sha1.
    Create {
        /// `true`: a sha256 repository; `false`: sha1.
        sha256: bool,
    },
}

/// One [`REPO_VERB`] ask, with who asked: the principal the session's key
/// authenticated as (re-read at the exec, never cached) and that key's
/// fingerprint.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RepoAsk {
    /// The principal the session's key authenticated as.
    pub principal: String,
    /// That key's fingerprint.
    pub fingerprint: String,
    /// The repository the verb names.
    pub repo: RepoRef,
    /// The verb itself.
    pub action: RepoAction,
}

/// What answers [`REPO_VERB`]: the ask in, the `key value` lines or the
/// refusal's sentence out.
pub type RepoDoor = Arc<
    dyn Fn(
            RepoAsk,
        )
            -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<String, String>> + Send>>
        + Send
        + Sync,
>;

/// Parse a [`REPO_VERB`] exec line. `None`: not this verb. `Some(Err)`: this
/// verb, malformed. Each argument may be single-quoted the way git quotes a
/// path; the path goes through the one sandbox
/// ([`gunnar_core::resolve_repo_path`]).
pub fn parse_repo_verb(raw: &[u8]) -> Option<Result<(RepoRef, RepoAction), String>> {
    let text = std::str::from_utf8(raw).ok()?;
    let mut tokens = text.split(' ').filter(|t| !t.is_empty());
    if tokens.next()? != REPO_VERB {
        return None;
    }
    let usage = || {
        format!(
            "the line is not `{REPO_VERB} show <namespace>/<repository>`, \
             `{REPO_VERB} visibility <namespace>/<repository> public|private` or \
             `{REPO_VERB} default-branch <namespace>/<repository> <branch>` or \
             `{REPO_VERB} create <namespace>/<repository> sha1|sha256`"
        )
    };
    if raw.len() > crate::command::MAX_EXEC_LEN || raw.iter().any(|b| *b < 0x20 || *b == 0x7f) {
        return Some(Err(usage()));
    }
    let mut args = Vec::new();
    for token in tokens {
        match crate::command::unquote(token) {
            Ok(arg) => args.push(arg),
            Err(_) => return Some(Err(usage())),
        }
    }
    let (sub, path, value) = match args.as_slice() {
        [sub, path] => (sub.as_str(), path.as_str(), None),
        [sub, path, value] => (sub.as_str(), path.as_str(), Some(value.as_str())),
        _ => return Some(Err(usage())),
    };
    let action = match (sub, value) {
        ("show", None) => RepoAction::Show,
        ("visibility", Some("public")) => RepoAction::Visibility(true),
        ("visibility", Some("private")) => RepoAction::Visibility(false),
        ("visibility", Some(other)) => {
            return Some(Err(format!(
                "visibility is `public` or `private`, not `{}`",
                other.escape_default()
            )))
        }
        ("default-branch", Some(branch)) if !branch.is_empty() => {
            RepoAction::DefaultBranch(branch.to_string())
        }
        ("create", Some("sha256")) => RepoAction::Create { sha256: true },
        ("create", Some("sha1")) => RepoAction::Create { sha256: false },
        ("create", Some(other)) => {
            return Some(Err(format!(
                "an object format is `sha1` or `sha256`, not `{}`",
                other.escape_default()
            )))
        }
        _ => return Some(Err(usage())),
    };
    let repo = match gunnar_core::resolve_repo_path(path) {
        Ok(repo) => repo,
        Err(err) => return Some(Err(err.to_string())),
    };
    Some(Ok((repo, action)))
}

/// Parse a [`RECIPIENT_KEYS_VERB`] exec line. `None`: not this verb.
/// `Some(Err)`: this verb, malformed. The path and the member may each be
/// single-quoted the way git quotes a path; the path goes through the one
/// sandbox ([`gunnar_core::resolve_repo_path`]) and the member through the
/// principal id rule.
pub fn parse_recipient_keys_verb(raw: &[u8]) -> Option<Result<(RepoRef, String), String>> {
    let text = std::str::from_utf8(raw).ok()?;
    let (verb, rest) = text.split_once(' ').unwrap_or((text, ""));
    if verb != RECIPIENT_KEYS_VERB {
        return None;
    }
    let usage =
        || format!("the line is not `{RECIPIENT_KEYS_VERB} <namespace>/<repository> <member>`");
    if raw.len() > crate::command::MAX_EXEC_LEN || raw.iter().any(|b| *b < 0x20 || *b == 0x7f) {
        return Some(Err(usage()));
    }
    let Some((path, member)) = rest.trim().rsplit_once(' ') else {
        return Some(Err(usage()));
    };
    let (Ok(path), Ok(member)) = (
        crate::command::unquote(path.trim()),
        crate::command::unquote(member.trim()),
    ) else {
        return Some(Err(usage()));
    };
    let repo = match gunnar_core::resolve_repo_path(&path) {
        Ok(repo) => repo,
        Err(err) => return Some(Err(err.to_string())),
    };
    if gunnar_core::Principal::new(
        member.clone(),
        member.clone(),
        gunnar_core::PrincipalKind::User,
    )
    .is_err()
    {
        return Some(Err(format!(
            "`{}` is not an account name",
            member.escape_default()
        )));
    }
    Some(Ok((repo, member)))
}

/// **The answer to [`RECIPIENT_KEYS_VERB`]** for `caller`: `(exit status,
/// stdout)`. Reads the identity table and the catalog; changes nothing.
pub fn recipient_keys_answer(
    control: &Control,
    caller: &str,
    repo: &RepoRef,
    member: &str,
) -> (u32, String) {
    if control.is_twin() {
        let said = DispatchError::Twin {
            principal: caller.to_string(),
            repo: repo.clone(),
        }
        .client_message()
        .unwrap_or_default();
        return (EXIT_DENIED, format!("error: {said}\n"));
    }
    let hidden = || {
        let said = DispatchError::NoSuchRepo(repo.clone())
            .client_message()
            .unwrap_or_default();
        (EXIT_DENIED, format!("error: {said}\n"))
    };
    // The member is told Gunnar could not answer; what broke is the operator's (the journal).
    let fault = |err: gunnar_core::Error| {
        tracing::error!(target: "gunnar::ssh", event = "ssh.recipient_keys.fault", %err, "gunnar-recipient-keys: the identity table or the catalog could not be read");
        (
            EXIT_DENIED,
            "error: Gunnar could not answer this just now; try again in a while\n".to_string(),
        )
    };
    let entry = match control.catalog().get(repo) {
        Ok(Some(entry)) => entry,
        Ok(None) => return hidden(),
        Err(err) => return fault(err),
    };
    match control.auth().may(caller, &entry, gunnar_core::Level::Read) {
        Ok(true) => {}
        Ok(false) => return hidden(),
        Err(err) => return fault(err),
    }
    match control.auth().may(member, &entry, gunnar_core::Level::Read) {
        Ok(true) => {}
        Ok(false) => {
            return (
                EXIT_DENIED,
                format!(
                    "error: {member} cannot read {}/{}: grant them access before adding them as a member of its key ring\n",
                    repo.namespace(),
                    repo.name()
                ),
            )
        }
        Err(err) => return fault(err),
    }
    // **The member's ONE encryption key** (CRYPT-YUBI 2026-10-26): a ring is wrapped to the
    // person's encryption public key, never to an SSH key.
    let Some(keys) = control.encryption_keys() else {
        return (
            EXIT_DENIED,
            "error: this Gunnar keeps no encryption keys\n".to_string(),
        );
    };
    match keys.public_of(member) {
        Ok(Some(public)) => (0, format!("{}\n", public.trim())),
        Ok(None) => (
            EXIT_DENIED,
            format!(
                "error: {member} has no encryption key yet: {member} makes one with `gunnar crypt key new` \
                 (or by adding a security key on their Account page), then the ring can be wrapped to it\n"
            ),
        ),
        Err(gunnar_core::encryption_key::KeyRefusal::Store(err)) => fault(err),
        Err(other) => (EXIT_DENIED, format!("error: {other}\n")),
    }
}

/// **The answer to [`gunnar_crypt::door::VERB`]** (`gunnar-user-key`, CRYPT-YUBI) for
/// `caller`, the session's own principal: `(exit status, stdout)`. Reads or changes the
/// caller's own encryption key only; there is no name to give.
pub fn user_key_answer(
    control: &Control,
    principal: &gunnar_core::Principal,
    request: gunnar_crypt::door::Request,
) -> (u32, String) {
    let caller = principal.id.as_str();
    use gunnar_core::encryption_key::{KeyRefusal, UnlockMethod};
    use gunnar_crypt::door::Request;
    if control.is_twin() {
        return (
            EXIT_DENIED,
            "error: this is a twin; ask the primary\n".to_string(),
        );
    }
    let Some(keys) = control.encryption_keys() else {
        return (
            EXIT_DENIED,
            "error: this Gunnar keeps no encryption keys\n".to_string(),
        );
    };
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0);
    let said = |r: Result<gunnar_core::encryption_key::EncryptionKey, KeyRefusal>| match r {
        Ok(key) => (0, key.to_answer().text()),
        Err(KeyRefusal::Store(err)) => {
            tracing::error!(target: "gunnar::ssh", event = "ssh.user_key.fault", %err, "gunnar-user-key: the store could not be read or written");
            (
                EXIT_DENIED,
                "error: Gunnar could not answer this just now; try again in a while\n".to_string(),
            )
        }
        Err(other) => (EXIT_DENIED, format!("error: {other}\n")),
    };
    match request {
        Request::Get => match keys.get(caller) {
            Ok(Some(key)) => (0, key.to_answer().text()),
            Ok(None) => (0, String::new()),
            Err(e) => said(Err(e)),
        },
        // The account's audit, the same kinds the gRPC door writes (RA4-3).
        Request::Create { public_key, method } => {
            let kind = method.kind.clone();
            let r = UnlockMethod::from_line(method)
                .and_then(|m| keys.create(caller, &public_key, m, now));
            if r.is_ok() {
                control.encryption_key_event(
                    principal,
                    "encryption-key.created",
                    &format!("first method {kind}"),
                );
                tracing::info!(target: "gunnar::audit", event = "access.encryption_key_created", principal = %caller, method = %kind, "an encryption key was created");
            }
            said(r)
        }
        Request::Add(method) => {
            let kind = method.kind.clone();
            let r = UnlockMethod::from_line(method).and_then(|m| keys.add_method(caller, m, now));
            if r.is_ok() {
                control.encryption_key_event(principal, "encryption-key.method-added", &kind);
                tracing::info!(target: "gunnar::audit", event = "access.unlock_method_added", principal = %caller, method = %kind, "an unlock method was added");
            }
            said(r)
        }
        Request::Remove(id) => {
            let r = keys.remove_method(caller, &id);
            if r.is_ok() {
                control.encryption_key_event(principal, "encryption-key.method-removed", &id);
                tracing::info!(target: "gunnar::audit", event = "access.unlock_method_removed", principal = %caller, method_id = %id, "an unlock method was removed");
            }
            said(r)
        }
    }
}

/// What answers [`TUNNEL_ADDRESS_VERB`]: the address (empty = a read) in, the
/// answer's lines or the refusal out.
pub type TunnelAddressDoor = Arc<dyn Fn(&str) -> Result<String, String> + Send + Sync>;

/// Parse a [`TUNNEL_ADDRESS_VERB`] exec line. `None`: not this verb. `Some(Err)`:
/// this verb, malformed. Printable ASCII only, at most one argument, short.
pub fn parse_tunnel_address_verb(raw: &[u8]) -> Option<Result<String, String>> {
    let text = std::str::from_utf8(raw).ok()?;
    let mut tokens = text.split(' ').filter(|t| !t.is_empty());
    if tokens.next()? != TUNNEL_ADDRESS_VERB {
        return None;
    }
    if raw.len() > 64 || !raw.iter().all(|b| (0x20..0x7f).contains(b)) {
        return Some(Err(
            "the line is not `gunnar-tunnel-address [a.b.c.d/prefix]`".to_string(),
        ));
    }
    let address = tokens.next().unwrap_or("").to_string();
    if tokens.next().is_some() {
        return Some(Err(
            "one argument at most: `gunnar-tunnel-address [a.b.c.d/prefix]`".to_string(),
        ));
    }
    Some(Ok(address))
}

/// **The key exchanges gunnar's SSH door offers, post-quantum FIRST**
/// (2026-10-06, the PQ transport lane).
///
/// `mlkem768x25519-sha256` (draft-ietf-sshm-mlkem-hybrid-kex: ML-KEM-768 and
/// X25519, the session key from both) is what OpenSSH ≥ 9.9 offers first and
/// 10.0 makes its default — russh 0.63 runs it on RustCrypto's `ml-kem`. The
/// classical exchanges stay after it, so an older client still connects
/// (preferred, never required). `sntrup761x25519-sha512` is not here: russh
/// does not implement it.
///
/// **gunnar's own list, not russh's default**, though the two agree today:
/// a russh bump that reorders or drops its `Preferred::DEFAULT` must not
/// change what this door offers without a red test here. SSH lets the
/// CLIENT's order decide; the server's job is to offer the hybrid at all.
pub static KEX_ORDER: &[russh::kex::Name] = &[
    russh::kex::MLKEM768X25519_SHA256,
    russh::kex::CURVE25519,
    russh::kex::CURVE25519_PRE_RFC_8731,
    russh::kex::DH_GEX_SHA256,
    russh::kex::DH_G18_SHA512,
    russh::kex::DH_G17_SHA512,
    russh::kex::DH_G16_SHA512,
    russh::kex::DH_G15_SHA512,
    russh::kex::DH_G14_SHA256,
    russh::kex::EXTENSION_SUPPORT_AS_CLIENT,
    russh::kex::EXTENSION_SUPPORT_AS_SERVER,
    russh::kex::EXTENSION_OPENSSH_STRICT_KEX_AS_CLIENT,
    russh::kex::EXTENSION_OPENSSH_STRICT_KEX_AS_SERVER,
];

/// russh's algorithm preferences with gunnar's [`KEX_ORDER`]; ciphers, MACs
/// and host-key algorithms stay russh's.
pub fn preferred() -> russh::Preferred {
    russh::Preferred {
        kex: std::borrow::Cow::Borrowed(KEX_ORDER),
        ..russh::Preferred::DEFAULT
    }
}

/// gunnar's SSH listener.
pub struct GunnarSsh {
    config: SshConfig,
    shared: Arc<Shared>,
    /// **A connection that arrives at a refused LOCAL address is dropped**
    /// before a byte is spoken (`gunnar serve --wg-only`: the utility
    /// network's address — every box-to-box connection runs over the tunnel).
    refused_local: Option<RefusedLocal>,
}

/// Is a connection that arrived at this local address refused?
pub type RefusedLocal = Arc<dyn Fn(std::net::IpAddr) -> bool + Send + Sync>;

impl GunnarSsh {
    /// Build a server over the control plane and the wire seam.
    pub fn new(config: SshConfig, control: Control, transport: Arc<dyn GitTransport>) -> Self {
        let exec_grace = config.exec_grace;
        let allow_anonymous = config.allow_anonymous;
        let config_max_auth_attempts = config.max_auth_attempts;
        let session_end_grace = (config.session_end_grace, config.session_end_grace_crowded);
        let transfer_idle = config.transfer_idle_timeout;
        // The memory budget travels with the two concurrency numbers: one
        // constructor, so a configuration cannot arrive with a governed
        // concurrency limit and an unmetered budget. `None` falls back to the
        // shipped constant rather than to "unlimited" — an operator who did not
        // set a budget wants the default, not the removal of a gate.
        let budget = config
            .memory_budget_bytes
            .unwrap_or(crate::governor::DEFAULT_MEMORY_BUDGET_BYTES);
        let wait = config
            .admission_wait
            .unwrap_or(crate::governor::DEFAULT_ADMISSION_WAIT);
        let admission = Arc::new(
            if config.per_principal_limit_is_hard {
                Admission::hard_capped(
                    config.max_concurrent_transfers,
                    config.max_transfers_per_principal,
                )
                .with_budget(budget)
            } else {
                Admission::with_memory_budget(
                    config.max_concurrent_transfers,
                    config.max_transfers_per_principal,
                    budget,
                )
            }
            // The anonymous door and its width are configured together, so
            // an appliance that opens the door cannot forget the ceiling.
            .with_anonymous_ceiling(config.max_anonymous_transfers)
            .with_wait(wait)
            .with_keyed_wait(config.keyed_admission_wait),
        );
        // The usage share (`admit-share`): the configured policy, unseeded
        // until `with_flow` hands it the durable ledger to seed from.
        #[cfg(feature = "admit-share")]
        let admission = Arc::new(
            Arc::try_unwrap(admission)
                .expect("the gate is not shared yet")
                .with_policy(config.admit.build()),
        );
        GunnarSsh {
            config,
            refused_local: None,
            shared: Arc::new(Shared {
                dispatch: Dispatch::new(control),
                transport,
                exec_grace,
                allow_anonymous,
                admission,
                metrics: Arc::new(ServerMetrics::new()),
                // Ready unless somebody hands us a flag to watch. A listener
                // built by a test has nothing to wait for, and a default of
                // "not ready" would make every such test hang on a sentence
                // about a boot that never happened.
                ready: Arc::new(std::sync::atomic::AtomicBool::new(true)),
                config_max_auth_attempts,
                session_end_grace,
                transfer_idle,
                tunnel_door: None,
                repo_door: None,
            }),
        }
    }

    /// **Watch a readiness flag**, and answer git traffic with a *retryable*
    /// sentence until it is set.
    ///
    /// # The boot window — MEASURED 2026-08-30
    ///
    /// After a power cut the appliance answered ssh :2222 and grpc :50051
    /// **0.6–0.7 s before its stores had finished their post-crash open**, and
    /// a read in that window came back `the repository is unavailable; try
    /// again`. That sentence is `gunnar_wire::Error::Store`'s, it arrives with
    /// a *denial* exit status, and it is what a client is told when a
    /// repository is genuinely broken — so a harness could not tell "not yet"
    /// from "gone". The first power-cut door reported **64 acked pushes
    /// missing**; a boot one second later found every one of them, and the same
    /// window wedged a whole bench run's clone and fetch phases.
    ///
    /// The fix is not to bind later — PID 1 must answer quickly or an operator
    /// watching a console sees a dead box — but to answer **honestly**. While
    /// the flag is clear every git exec is refused in git's own protocol with
    /// [`EXIT_BUSY`] (75, EX_TEMPFAIL — the same status the admission gate
    /// uses, which is the status that means *retry*) and a sentence containing
    /// the word **starting**, which is the word a script may key on. Nothing is
    /// served, nothing is opened, and nothing looks like a missing repository.
    ///
    /// A flag and not a trait: `gunnar-ssh` may not read the operating system,
    /// and "have the stores finished opening" is `gunnar-server`'s fact to
    /// publish, not this crate's to discover.
    pub fn with_readiness(mut self, ready: Arc<std::sync::atomic::AtomicBool>) -> Self {
        // `Arc::get_mut` cannot be used: `bind` has not run, but `Shared` is
        // already behind an `Arc`. Rebuilding it is one allocation at start-up
        // and keeps every field in one place.
        let shared = Arc::try_unwrap(self.shared).unwrap_or_else(|shared| {
            panic!(
                "readiness must be set before the listener is shared: {} strong refs",
                Arc::strong_count(&shared)
            )
        });
        self.shared = Arc::new(Shared { ready, ..shared });
        self
    }

    /// **Refuse every connection that arrives at a local address `refused`
    /// names** — closed at accept, nothing said, logged once per peer at
    /// debug. `gunnar serve --wg-only` names the utility address: the public's
    /// git clients still reach SSH, a box over plain IP does not.
    pub fn with_refused_local(mut self, refused: RefusedLocal) -> Self {
        self.refused_local = Some(refused);
        self
    }

    /// **Answer [`TUNNEL_ADDRESS_VERB`]** with `door` — `gunnar-server` hands in
    /// its `appliance::tunnel_address_answer`, the function both gRPC doors
    /// call. Without one the verb is answered "no WireGuard tunnel".
    pub fn with_tunnel_address_door(mut self, door: TunnelAddressDoor) -> Self {
        let shared = Arc::try_unwrap(self.shared).unwrap_or_else(|shared| {
            panic!(
                "the tunnel door must be set before the listener is shared: {} strong refs",
                Arc::strong_count(&shared)
            )
        });
        self.shared = Arc::new(Shared {
            tunnel_door: Some(door),
            ..shared
        });
        self
    }

    /// **Answer [`REPO_VERB`]** with `door` — `gunnar-server` hands in the
    /// control plane's own `Repos` handlers. Without one the verb is answered
    /// "this server answers repository settings only on its control plane".
    pub fn with_repo_door(mut self, door: RepoDoor) -> Self {
        let shared = Arc::try_unwrap(self.shared).unwrap_or_else(|shared| {
            panic!(
                "the repository door must be set before the listener is shared: {} strong refs",
                Arc::strong_count(&shared)
            )
        });
        self.shared = Arc::new(Shared {
            repo_door: Some(door),
            ..shared
        });
        self
    }

    /// **Give the admission gate the per-namespace flow ledger.**
    ///
    /// The gate is what all three front doors — this exec line, smart HTTP and
    /// the gRPC transfer gate — ask before a transfer, so it is the one place a
    /// refusal can be charged to a tenant once rather than three times. The
    /// exec line's own `no-grant` denial reaches the same ledger back through
    /// `Admission::flow`.
    ///
    /// A listener built without one refuses exactly as it always did and counts
    /// the refusal only in its own `AdmissionStats`, which is what a test that
    /// drives `gunnar-ssh` alone wants — and what every build before the ledger
    /// existed did.
    ///
    /// The gate is rebuilt rather than mutated for the reason
    /// [`Self::with_readiness`] rebuilds `Shared`: `bind` has not run, but the
    /// `Arc` is already there.
    pub fn with_flow(mut self, flow: Arc<gunnar_core::FlowLedger>) -> Self {
        let shared =
            Arc::get_mut(&mut self.shared).expect("the shared state is not published until `bind`");
        let gate = Arc::try_unwrap(std::mem::replace(
            &mut shared.admission,
            Arc::new(Admission::unlimited()),
        ))
        .unwrap_or_else(|gate| {
            panic!(
                "the flow ledger must be set before the gate is shared: {} strong refs",
                Arc::strong_count(&gate)
            )
        });
        // admit-share: the policy is rebuilt SEEDED from this ledger — its
        // table is then a cache of the ledger, filled once per namespace.
        #[cfg(feature = "admit-share")]
        let gate = gate.with_policy(
            self.config
                .admit
                .build_seeded(Some(Admission::flow_seed(Arc::clone(&flow)))),
        );
        shared.admission = Arc::new(gate.with_flow(flow));
        self
    }

    /// Share the server-wide counters with the rest of the process.
    ///
    /// **Not a second set of counters.** `gunnar serve` mints one
    /// [`ServerMetrics`] and hands the same `Arc` to this listener and to the
    /// git transport, so "connections refused" and "pushes rejected by gate"
    /// are numbers about one server rather than two crates' private tallies.
    /// A listener built without this one still counts, into a set nobody else
    /// can read, which is what a test that only drives `gunnar-ssh` wants.
    pub fn with_metrics(mut self, metrics: Arc<ServerMetrics>) -> Self {
        let shared =
            Arc::get_mut(&mut self.shared).expect("the shared state is not published until `bind`");
        shared.metrics = metrics;
        self
    }

    /// The counters this listener writes to.
    pub fn metrics(&self) -> &Arc<ServerMetrics> {
        &self.shared.metrics
    }

    /// The russh configuration this server runs with.
    ///
    /// `methods` is publickey **only**, and stays that way even with the
    /// anonymous path open: `none` is available for exactly one username,
    /// so advertising it to everybody would make the failure list a lie and
    /// invite every client to try something that will not work. russh calls
    /// [`Handler::auth_none`] on the client's opening probe regardless of what
    /// is advertised, which is the only place it is needed.
    ///
    /// `password` is not offered and there is no handler for it, so the
    /// rejecting default is never the last line of defence.
    fn russh_config(&self, host_key: russh::keys::PrivateKey) -> RusshConfig {
        RusshConfig {
            server_id: russh::SshId::Standard(self.config.server_id.clone().into()),
            preferred: preferred(),
            methods: MethodSet::from(&[MethodKind::PublicKey][..]),
            auth_rejection_time: self.config.auth_rejection_time,
            auth_rejection_time_initial: Some(Duration::ZERO),
            keys: vec![host_key],
            max_auth_attempts: self.config.max_auth_attempts,
            inactivity_timeout: self.config.inactivity_timeout,
            keepalive_interval: self.config.keepalive_interval,
            nodelay: true,
            ..RusshConfig::default()
        }
    }

    /// Bind the listener and start accepting.
    ///
    /// Returns once the socket is bound, so a caller (or a test) knows the
    /// real address before the first client connects. That matters for
    /// [`SshConfig::ephemeral_loopback`], where the port is chosen by the
    /// kernel.
    pub async fn bind(self, host_key: russh::keys::PrivateKey) -> std::io::Result<BoundServer> {
        let listener = TcpListener::bind(self.config.bind).await?;
        let local_addr = listener.local_addr()?;
        let russh_config = Arc::new(self.russh_config(host_key));
        let shared = self.shared.clone();
        let admission = Arc::clone(&self.shared.admission);
        let metrics = Arc::clone(&self.shared.metrics);
        let (handle, _) = tokio::sync::broadcast::channel::<String>(1);

        let join = tokio::spawn(accept_loop(
            listener,
            russh_config,
            Factory { shared },
            handle.clone(),
            self.refused_local.clone(),
        ));

        tracing::info!(
            target: TARGET_SSH,
            %local_addr,
            max_concurrent_transfers = ?admission.max_concurrent(),
            max_transfers_per_principal = ?admission.max_per_principal(),
            per_principal_is_hard = admission.per_principal_is_hard(),
            "gunnar ssh listener bound"
        );
        Ok(BoundServer {
            local_addr,
            handle,
            join,
            admission,
            metrics,
        })
    }
}

/// How long the accept loop waits after the kernel refused it a descriptor,
/// before it asks again. A session ending in that window frees what the next
/// `accept` needs; nothing shorter gains anything, nothing longer is felt.
pub const ACCEPT_BACKOFF: Duration = Duration::from_millis(100);

/// Is this `accept` failure the kind the loop must outlive?
///
/// The resource ones — no descriptor left in the process (`EMFILE`) or in the
/// system (`ENFILE`), no buffer or memory for the socket — and the one the
/// client causes, `ECONNABORTED` (it hung up between SYN and `accept`).
/// Everything else (`EBADF`, `EINVAL`, a closed listener) is the listener
/// itself being gone, and the loop must return that.
pub fn accept_error_is_transient(err: &std::io::Error) -> bool {
    // Linux errno values, spelled here rather than through a `libc` crate this
    // crate does not otherwise carry: ENOMEM 12, ENFILE 23, EMFILE 24,
    // ECONNABORTED 103, ENOBUFS 105.
    matches!(err.raw_os_error(), Some(12 | 23 | 24 | 103 | 105))
        || matches!(
            err.kind(),
            std::io::ErrorKind::ConnectionAborted | std::io::ErrorKind::OutOfMemory
        )
}

/// **The accept loop, owned here so that it cannot die of `EMFILE`.**
///
/// russh's `run_on_socket` is the same loop with one difference that cost a
/// whole appliance: its `Err(e) => return Err(e)` on `accept`. MEASURED
/// 2026-08-29 on the rig: the 202nd repository touched since boot ran the
/// process out of descriptors, the next `accept` failed with `EMFILE`, the
/// loop returned, and every later client got `kex_exchange_identification:
/// read: Connection reset by peer` until reboot — the kernel kept completing
/// handshakes into a backlog nobody drained — while gRPC on its own listener
/// answered as if nothing were wrong. A server that has no descriptor for the
/// next client must say so and try again in a moment, not resign.
///
/// So on a transient failure ([`accept_error_is_transient`]) it logs a WARN
/// on `gunnar::ssh` — a row on the appliance dashboard — sleeps
/// [`ACCEPT_BACKOFF`] and continues. Everything else is what russh did: one
/// task per connection, `set_nodelay`, `run_stream`, the session raced against
/// the shutdown broadcast so `BoundServer::shutdown` still disconnects every
/// live session by application, and session errors go through
/// [`note_session_end`] exactly as `Server::handle_session_error` did.
async fn accept_loop(
    listener: TcpListener,
    config: Arc<RusshConfig>,
    mut factory: Factory,
    shutdown: tokio::sync::broadcast::Sender<String>,
    refused_local: Option<RefusedLocal>,
) -> std::io::Result<()> {
    let mut shutdown_rx = shutdown.subscribe();
    let mut refused_in_a_row: u32 = 0;
    loop {
        tokio::select! {
            _ = shutdown_rx.recv() => {
                tracing::debug!(target: TARGET_SSH, "ssh listener shutdown requested");
                return Ok(());
            }
            accepted = listener.accept() => {
                let (socket, peer) = match accepted {
                    Ok(pair) => pair,
                    Err(err) if accept_error_is_transient(&err) => {
                        refused_in_a_row += 1;
                        tracing::warn!(
                            target: TARGET_SSH,
                            event = "ssh.accept.refused",
                            error = %err,
                            refused_in_a_row,
                            "ssh accept failed: {err}; the listener stays up and retries in {} ms",
                            ACCEPT_BACKOFF.as_millis()
                        );
                        tokio::time::sleep(ACCEPT_BACKOFF).await;
                        continue;
                    }
                    Err(err) => {
                        tracing::error!(
                            target: TARGET_SSH,
                            event = "ssh.accept.failed",
                            error = %err,
                            "ssh accept failed with an error the listener cannot outlive"
                        );
                        return Err(err);
                    }
                };
                if let (Some(refused), Ok(local)) = (&refused_local, socket.local_addr()) {
                    if refused(local.ip()) {
                        tracing::debug!(target: TARGET_SSH, event = "ssh.accept.plain_ip_refused", %peer, %local, "a connection at the utility address is refused (--wg-only): boxes speak over the tunnel");
                        drop(socket);
                        continue;
                    }
                }
                if refused_in_a_row > 0 {
                    tracing::info!(
                        target: TARGET_SSH,
                        event = "ssh.accept.recovered",
                        refused_in_a_row,
                        "ssh accept works again"
                    );
                    refused_in_a_row = 0;
                }
                let handler = factory.new_client(Some(peer));
                let config = Arc::clone(&config);
                let mut session_shutdown = shutdown.subscribe();
                tokio::spawn(async move {
                    if config.nodelay {
                        if let Err(err) = socket.set_nodelay(true) {
                            tracing::debug!(target: TARGET_SSH, %err, "set_nodelay failed");
                        }
                    }
                    let socket = HandshakeQuickAck::new(socket);
                    let session = match russh::server::run_stream(config, socket, handler).await {
                        Ok(session) => session,
                        Err(err) => {
                            note_session_end(&err);
                            return;
                        }
                    };
                    let handle = session.handle();
                    tokio::select! {
                        reason = session_shutdown.recv() => {
                            let reason = reason.unwrap_or_default();
                            let _ = handle
                                .disconnect(russh::Disconnect::ByApplication, reason, String::new())
                                .await;
                        }
                        ended = session => {
                            if let Err(err) = ended {
                                note_session_end(&err);
                            }
                        }
                    }
                });
            }
        }
    }
}

/// **The accepted socket, with `TCP_QUICKACK` re-armed before each of its first reads**
/// (GUN-PERF, 2026-10-31).
///
/// OpenSSH turns `TCP_NODELAY` on only after authentication, so in the key exchange its
/// KEXINIT and its key-exchange init are two writes under Nagle: the second waits for
/// the first to be acknowledged. This server has already sent its own KEXINIT by then
/// and has nothing to carry that ACK, so the kernel delays it, about 40 ms, on most
/// connections. Every `git` over SSH paid it (`ls-remote` 50 to 70 ms over SSH against
/// 15 to 20 ms over HTTP). Quick-ACK mode is not sticky (the kernel leaves it again
/// after it sends), so it is re-armed before each read of the handshake, and only
/// there: the bulk transfer after it keeps the kernel's own ACK pacing.
struct HandshakeQuickAck {
    inner: tokio::net::TcpStream,
    reads_left: u8,
}

impl HandshakeQuickAck {
    /// The key exchange and the authentication are well inside this many reads.
    const HANDSHAKE_READS: u8 = 32;

    fn new(inner: tokio::net::TcpStream) -> Self {
        HandshakeQuickAck {
            inner,
            reads_left: Self::HANDSHAKE_READS,
        }
    }

    /// A send is what puts the kernel back into delayed-ACK mode (it reads the exchange
    /// as interactive), and the client answers a send: so in the handshake each send is
    /// followed by quick-ACK again, before that answer arrives.
    fn rearm_after_send(&self, wrote: &std::task::Poll<std::io::Result<usize>>) {
        if self.reads_left > 0 && matches!(wrote, std::task::Poll::Ready(Ok(n)) if *n > 0) {
            let _ = self.inner.set_quickack(true);
        }
    }
}

impl tokio::io::AsyncRead for HandshakeQuickAck {
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        if self.reads_left > 0 {
            self.reads_left -= 1;
            // Only slower when it fails; never a reason to drop the connection.
            let _ = self.inner.set_quickack(true);
        }
        std::pin::Pin::new(&mut self.inner).poll_read(cx, buf)
    }
}

impl tokio::io::AsyncWrite for HandshakeQuickAck {
    fn poll_write(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        let wrote = std::pin::Pin::new(&mut self.inner).poll_write(cx, buf);
        self.rearm_after_send(&wrote);
        wrote
    }
    fn poll_write_vectored(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        bufs: &[std::io::IoSlice<'_>],
    ) -> std::task::Poll<std::io::Result<usize>> {
        let wrote = std::pin::Pin::new(&mut self.inner).poll_write_vectored(cx, bufs);
        self.rearm_after_send(&wrote);
        wrote
    }
    fn is_write_vectored(&self) -> bool {
        self.inner.is_write_vectored()
    }
    fn poll_flush(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.inner).poll_flush(cx)
    }
    fn poll_shutdown(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        std::pin::Pin::new(&mut self.inner).poll_shutdown(cx)
    }
}

/// A bound, running listener.
pub struct BoundServer {
    local_addr: SocketAddr,
    /// The shutdown broadcast every live session and the accept loop listen on.
    handle: tokio::sync::broadcast::Sender<String>,
    join: JoinHandle<std::io::Result<()>>,
    admission: Arc<Admission>,
    metrics: Arc<ServerMetrics>,
}

impl BoundServer {
    /// The address actually bound, with the kernel's choice of port filled in.
    pub fn local_addr(&self) -> SocketAddr {
        self.local_addr
    }

    /// What the admission gate has done since this listener started.
    ///
    /// This is the visibility half of the fix. A limit whose saturation cannot
    /// be read is the same undiagnosable "git is slow" the accidental 512-thread
    /// ceiling produced; `admitted`, `refused_*` and `peak_in_flight` are what
    /// an operator, or a test, looks at instead.
    pub fn admission(&self) -> AdmissionStats {
        self.admission.stats()
    }

    /// The admission gate itself, so a **second transport can route through the
    /// SAME one** (one implementation).
    ///
    /// `gunnar-http` is a second front door onto the same server, and a git
    /// transfer over HTTP has to count against the same server-wide and
    /// per-principal ceilings — and the same memory budget — as one over SSH,
    /// or an appliance's real concurrency is silently twice the number written
    /// on the box. Handing out the `Arc` (rather than rebuilding a gate from
    /// the same numbers) is what makes it one gate with one set of counters,
    /// not two that happen to agree: on 2026-08-31 the HTTP front was found with no gate
    /// at all, and two gates would have been the next-worst answer.
    pub fn admission_gate(&self) -> Arc<Admission> {
        Arc::clone(&self.admission)
    }

    /// The server-wide counters.
    pub fn metrics(&self) -> gunnar_core::observe::MetricsSnapshot {
        self.metrics.snapshot()
    }

    /// The counter block itself, for a caller that wants to keep reading it
    /// after this listener is consumed by [`BoundServer::shutdown`].
    pub fn metrics_handle(&self) -> &Arc<ServerMetrics> {
        &self.metrics
    }

    /// Ask every session to disconnect, then wait for the accept loop to stop
    /// — for at most [`SSH_SHUTDOWN_GRACE`].
    ///
    /// **Bounded, the way the gRPC half already is** (`GRPC_SHUTDOWN_GRACE` in
    /// gunnar-server): the accept loop only ends when its sessions do, and a
    /// session whose peer never answers the disconnect — MEASURED 2026-09-28 on
    /// loki, a primary's replication session into a standby being stopped, 1 run
    /// in 17 under a parallel test binary — held `shutdown` for an hour. A
    /// shutdown that can hang forever is a restart, a drain or a growth that can
    /// hang forever. Past the grace the listener task is aborted and the reason
    /// is logged; the sessions go with the process.
    pub async fn shutdown(self, reason: &str) -> std::io::Result<()> {
        self.shutdown_within(reason, SSH_SHUTDOWN_GRACE).await
    }

    /// [`Self::shutdown`] with an explicit grace (tests use a short one).
    pub async fn shutdown_within(self, reason: &str, grace: Duration) -> std::io::Result<()> {
        let _ = self.handle.send(reason.to_string());
        let abort = self.join.abort_handle();
        match tokio::time::timeout(grace, self.join).await {
            Ok(Ok(result)) => result,
            Ok(Err(err)) if err.is_cancelled() => Ok(()),
            Ok(Err(err)) => Err(std::io::Error::other(err)),
            Err(_elapsed) => {
                tracing::warn!(
                    target: TARGET_SSH,
                    ?grace,
                    "the SSH listener did not stop within its grace: a session did not \
                     answer the disconnect. Aborting the listener so the shutdown completes."
                );
                abort.abort();
                Ok(())
            }
        }
    }
}

struct Factory {
    shared: Arc<Shared>,
}

impl Server for Factory {
    type Handler = Connection;

    fn new_client(&mut self, peer_addr: Option<SocketAddr>) -> Connection {
        // ── ONE span per connection, and one id ────────────────────────────
        //
        // `principal` and `repo` are declared `Empty` and filled in by
        // `auth_publickey` and `exec_request` as they become known, so every
        // line emitted under this connection carries all three WITHOUT the
        // call site repeating them. Before this, connection identity in the
        // SSH logs was an ad-hoc `?self.peer` copy-pasted to fourteen call
        // sites, and there was no way at all to tie two lines together.
        let conn = RequestId::next();
        let span = tracing::info_span!(
            target: TARGET_SSH,
            "ssh.connection",
            %conn,
            peer = tracing::field::Empty,
            principal = tracing::field::Empty,
            repo = tracing::field::Empty,
        );
        if let Some(peer) = peer_addr {
            span.record("peer", tracing::field::display(peer));
        }
        self.shared.metrics.add_connections_opened(1);
        span.in_scope(|| tracing::debug!(target: TARGET_SSH, "ssh connection accepted"));
        Connection {
            shared: self.shared.clone(),
            span,
            principal: None,
            key_fingerprint: None,
            channels: HashMap::new(),
            lfs_slots: HashMap::new(),
            exec_seen: Arc::new(AtomicBool::new(false)),
            misuse_said: false,
            auth_attempts: 0,
            session_over: tokio::sync::watch::channel(false).0,
        }
    }

    /// The session loop stopped. Usually because the client hung up, which is
    /// how every git client ends a transfer and is not news.
    ///
    /// A free function does the deciding, for the same reason
    /// [`note_teardown`] is one: the rule can then be asserted without standing
    /// up a russh session, and what is asserted is the line an operator reads.
    fn handle_session_error(&mut self, error: anyhow::Error) {
        note_session_end(&error);
    }
}

/// How many `git-lfs-transfer` channels of one connection share one admission slot
/// ([`Connection::lfs_slots`]): git-lfs's pool is `lfs.concurrenttransfers` + 1 (9 by
/// default); 32 leaves room for a client that raised it, and bounds what one slot carries.
pub const LFS_CHANNELS_PER_SLOT: usize = 32;

/// **How long a transfer waits for a slot before it is told it waits**
/// (GUN-SSH-SLOT-HANG): a wait the client cannot see is a hang to whoever runs it.
const WAIT_NOTICE_AFTER: Duration = Duration::from_millis(500);

/// One open session channel, before its exec request arrives.
struct OpenChannel {
    /// The channel itself. The exec request carries only a [`ChannelId`], and
    /// the byte stream can only be taken from the `Channel`.
    channel: Channel<Msg>,
    /// The value of `GIT_PROTOCOL`, if the client sent it and
    /// [`ENV_ALLOWLIST`] accepted the name. Kept beside the channel it was sent
    /// on so that it cannot outlive it or be read for a different one.
    git_protocol: Option<String>,
}

/// One client connection.
struct Connection {
    shared: Arc<Shared>,
    /// The `ssh.connection` span every line from this connection is emitted
    /// under. Cloned (cheap: it is an id, not the data) rather than borrowed
    /// at each entry point, because entering it borrows `self` immutably while
    /// every handler needs `&mut self`.
    span: tracing::Span,
    /// Set by `auth_publickey`, and the only thing that authorises anything.
    principal: Option<Principal>,
    /// The key `auth_publickey` authenticated this session with.
    ///
    /// Beside `principal` rather than inside it, because a principal is a
    /// server-side identity that may hold several keys and this is the one that
    /// made *this* connection. Stays `None` for the anonymous path, which
    /// authenticated with no key at all.
    key_fingerprint: Option<Fingerprint>,
    channels: HashMap<ChannelId, OpenChannel>,
    /// **The admission slot this connection's `git-lfs-transfer` channels share**, per
    /// repository and operation (GHUNT-7, 2026-10-28). Stock git-lfs multiplexes its
    /// pure-SSH pool over ONE connection (`lfs.ssh.automultiplex`): `concurrenttransfers`
    /// (8) + 1 channels, opened at once and kept open, mostly idle. A slot per channel let
    /// one git-lfs on a box whose limit is under nine wait for slots its own idle channels
    /// held: the push failed after the keyed wait, or hung. One client, one slot; a slot
    /// carries at most [`LFS_CHANNELS_PER_SLOT`] channels, past that a channel asks the
    /// gate as any transfer does.
    lfs_slots: HashMap<(String, GitService), std::sync::Weak<crate::admission::AdmissionGuard>>,
    exec_seen: Arc<AtomicBool>,
    /// A request for a channel this connection does not hold open (a client's second
    /// `exec` on one channel) was said already: WARN once per connection, then quiet —
    /// a client's protocol misuse is never the server's ERROR (stinkies ssh-double-exec-error).
    misuse_said: bool,
    /// **How many publickey offers this connection has had refused.**
    ///
    /// Counted here because russh does not count it. `SshConfig::    /// max_auth_attempts` is handed to russh 0.61.2's `Config`, which declares
    /// the field, defaults it to 10 and `Debug`-prints it — and never compares
    /// it to anything. VERIFIED by reading the whole of that crate's source:
    /// the only three occurrences are the declaration, the default and the
    /// `Debug` impl. (russh 0.63 does compare it, and carries tests named for
    /// exactly this; gunnar is pinned to 0.61.2 through an exact `ssh-key`
    /// dependency, and a limit that works is not worth an SSH-stack bump to
    /// obtain.)
    ///
    /// So until 2026-09-11 a connection could offer keys without bound: a
    /// dictionary of public keys against the identity table, at
    /// `auth_rejection_time` apiece, on one TCP connection that never had to be
    /// re-established.
    auth_attempts: usize,
    /// **Set when this session ends** — by `Drop`, because russh drops the
    /// handler on every way a session ends (hang-up, unanswered keepalives,
    /// inactivity, protocol error, shutdown). Every transfer's writer watches
    /// it ([`SessionBound`]): russh's own writer parks on the SSH window and
    /// nothing wakes it when the session dies, which left a fetch to a dead
    /// client holding its fan worker and admission slot forever (cov-13
    /// finding 4, 2026-10-06).
    session_over: tokio::sync::watch::Sender<bool>,
}

impl Drop for Connection {
    fn drop(&mut self) {
        self.session_over.send_replace(true);
    }
}

/// Why a session that authenticated may no longer act.
///
/// Two variants and not a `bool`, for [`Control::principal_for_key`]'s reason:
/// "nobody holds this key any more" and "the identity table could not be read"
/// call for entirely different operator action, and both refuse.
#[derive(Debug)]
enum Stale {
    /// The key that opened this session no longer resolves to anyone, or
    /// resolves to somebody else. Revoked, expired, or re-registered.
    Revoked,
    /// The identity table could not be read. Refused, because failing open is
    /// not on the table — the same answer `auth_publickey` gives.
    Unreadable(gunnar_core::Error),
}

impl Connection {
    /// **Who this session may act as right now — asked again, per exec.**
    ///
    /// `auth_publickey` caches the principal, and until 2026-09-11 every exec
    /// on the connection was authorised out of that cache. Revocation
    /// therefore took effect immediately for a NEW connection and never for an
    /// open one: `DocRef::read` caches nothing, so the identity table was
    /// current the whole time and nothing asked it. A long-lived session — and
    /// git over SSH holds one open for as long as a client wants to keep
    /// pushing — outlived the credential that opened it, for as long as it
    /// liked. An operator who revokes a leaked key and sees the revocation take
    /// effect on their own next connection has every reason to believe it took
    /// effect everywhere.
    ///
    /// Cheap enough to do per exec: it is the same single
    /// `principal_by_fingerprint` lookup the connection already pays once at
    /// auth, and an exec is about to fork a git process and move a pack.
    ///
    /// A key that now belongs to a DIFFERENT identity is refused rather than
    /// adopted. This session authenticated as somebody; if the key has since
    /// been registered to somebody else, this is not that session, and
    /// continuing as the new holder would make re-registration a way to inherit
    /// an open connection.
    ///
    /// The anonymous path has no fingerprint — it authenticated with no key at
    /// all — so there is nothing to re-read, and `Principal::anonymous()` is not
    /// in the identity table to be revoked from. It keeps the cached value, and
    /// what anonymity may DO is still `Dispatch::authorise`'s answer.
    fn live_principal(&self) -> Result<Principal, Stale> {
        let Some(cached) = self.principal.clone() else {
            return Err(Stale::Revoked);
        };
        let Some(fingerprint) = self.key_fingerprint.as_ref() else {
            return Ok(cached);
        };
        match self
            .shared
            .dispatch
            .control()
            .principal_for_key(fingerprint)
        {
            Ok(Some(now)) if now.id == cached.id => Ok(now),
            Ok(_) => Err(Stale::Revoked),
            Err(err) => Err(Stale::Unreadable(err)),
        }
    }

    /// **Count a refused publickey offer, and say whether this connection has
    /// had its last one.**
    ///
    /// One place, called from every refusing arm of `auth_publickey`, so a new
    /// refusal reason cannot arrive uncounted — which is how a limit stops
    /// applying to exactly the case somebody added last.
    ///
    /// `0` is no limit, which is russh's own reading of the same field and the
    /// reading an operator who writes `0` expects.
    ///
    /// The `none` probe is deliberately NOT counted. Every OpenSSH client opens
    /// with one to learn the method list, so counting it would spend one of the
    /// six on the ordinary first packet of an ordinary session — and it carries
    /// no key, so it cannot be the dictionary attempt the limit is for.
    fn refused_auth(&mut self) -> bool {
        self.auth_attempts += 1;
        let max = self.shared.config_max_auth_attempts;
        max > 0 && self.auth_attempts >= max
    }

    /// The error that ends a connection which has offered too many keys.
    ///
    /// An `Err` out of a handler, because russh 0.61.2's `auth_publickey` is
    /// handed no session and no handle, so there is nothing to call
    /// `disconnect` on — and the run loop treating a handler error as the end
    /// of the session is exactly the outcome wanted. It is a `warn!`, not a
    /// `debug!`: one of these is a client with a full agent, and a hundred is
    /// somebody reading the identity table one key at a time.
    fn too_many_auth_attempts(&self) -> anyhow::Error {
        tracing::warn!(
            target: TARGET_SSH,
            event = "ssh.auth.too_many",
            attempts = self.auth_attempts,
            max = self.shared.config_max_auth_attempts,
            "disconnecting: too many refused publickey offers on one connection"
        );
        self.shared.metrics.add_connections_refused(1);
        anyhow::anyhow!(
            "too many authentication failures ({} refused publickey offers)",
            self.auth_attempts
        )
    }
}

impl Handler for Connection {
    type Error = anyhow::Error;

    /// **The anonymous read path.** Accept the `none` method for exactly
    /// one username, and refuse it for every other.
    ///
    /// The refusing branch is byte-for-byte what russh's default handler did
    /// before this method existed: `Auth::reject()`, which makes russh drop
    /// `MethodKind::None` from the advertised set and let the client proceed to
    /// publickey. So a client that does not ask for anonymity is on exactly the
    /// path it was on yesterday.
    ///
    /// The accepting branch does **not** consult the catalog, the repository or
    /// the ACL: it establishes *who* is asking (nobody), and *what nobody may
    /// do* is `Dispatch::authorise`'s answer, through
    /// [`AuthTable::may`](gunnar_core::AuthTable::may), like everyone else.
    /// Deciding here would be a second ACL.
    async fn auth_none(&mut self, user: &str) -> Result<Auth, Self::Error> {
        let span = self.span.clone();
        let _entered = span.enter();
        if !self.shared.allow_anonymous || user != ANONYMOUS_SSH_USER {
            // Not a refusal worth a `warn!`: EVERY OpenSSH client opens with a
            // `none` probe to learn the method list, so this is the ordinary
            // first packet of an ordinary authenticated session.
            tracing::debug!(
                target: TARGET_SSH,
                event = "ssh.auth.none.refused",
                reason = if self.shared.allow_anonymous {
                    "not-the-anonymous-user"
                } else {
                    "anonymous-access-disabled"
                },
                decorative_ssh_user = %user,
                "the none method is refused; publickey is the only way in"
            );
            return Ok(Auth::reject());
        }

        let anonymous = Principal::anonymous();
        span.record("principal", tracing::field::display(&anonymous.id));
        self.shared.metrics.add_connections_anonymous(1);
        tracing::info!(
            target: TARGET_SSH,
            event = "ssh.auth.anonymous",
            // Named so nobody later reads this line as "a key was accepted".
            // It is on `gunnar::ssh` at `info` because "who is reading my
            // public repositories" must be answerable from a production log.
            authenticated = false,
            "anonymous session accepted: no key, public repositories only"
        );
        self.principal = Some(anonymous);
        Ok(Auth::Accept)
    }

    async fn auth_publickey(
        &mut self,
        user: &str,
        key: russh::server::VerifiedPublicKey<'_>,
    ) -> Result<Auth, Self::Error> {
        // russh 0.65: the key arrives with the algorithm its signature was verified
        // under; the identity is the key alone, as before.
        let public_key: &PublicKey = key.public_key;
        let span = self.span.clone();
        let _entered = span.enter();
        // Over the key's WIRE BLOB, not over its type: russh's `ssh-key` is
        // 0.7.0-rc.10 and `gunnar-core`'s is 0.6.7, so this is the only value
        // that can reach the identity table. It is byte-for-byte the
        // string `ssh-keygen -lf` prints for the same key.
        //
        // `None` means the key would not re-encode. That used to be an
        // `expect()`, a panic in the auth handler, over client-supplied bytes.
        // It is a refusal now; `fingerprint_of_key` has already
        // logged why.
        let Some(fingerprint) = fingerprint_of_key(public_key) else {
            let refusal = refuse_unencodable_key(&self.shared.metrics, user);
            if self.refused_auth() {
                return Err(self.too_many_auth_attempts());
            }
            return Ok(refusal);
        };
        let started = std::time::Instant::now();
        // `peer` is a field of the enclosing span from here down, so it is not
        // repeated on any line below (one implementation, and the whole point of the span).
        match self
            .shared
            .dispatch
            .control()
            .principal_for_key(&fingerprint)
        {
            Ok(Some(principal)) => {
                span.record("principal", tracing::field::display(&principal.id));
                self.shared.metrics.add_connections_accepted(1);
                tracing::info!(
                    target: TARGET_SSH,
                    event = "ssh.auth.accepted",
                    %fingerprint,
                    // Logged so an operator can see what was sent, and named so
                    // nobody later mistakes it for an input to the decision.
                    decorative_ssh_user = %user,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "publickey accepted"
                );
                self.principal = Some(principal);
                // Kept, not recomputed and not dropped. This is the
                // value the identity table was probed with one line above, so
                // it is the key that authenticated, by construction.
                self.key_fingerprint = Some(fingerprint);
                Ok(Auth::Accept)
            }
            Ok(None) => {
                self.shared.metrics.add_connections_refused(1);
                tracing::warn!(
                    target: TARGET_SSH,
                    event = "ssh.auth.refused",
                    reason = "unknown-key",
                    %fingerprint,
                    decorative_ssh_user = %user,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    "publickey refused: no principal holds this key"
                );
                if self.refused_auth() {
                    return Err(self.too_many_auth_attempts());
                }
                Ok(Auth::reject())
            }
            // The key is still refused, since failing open is not on the table, but
            // this is a FAULT, not "unknown key", and the log has to say so.
            // An identity table that has become unreadable looks exactly like
            // an empty one from the client's side (the ACL model).
            Err(err) => {
                self.shared.metrics.add_connections_refused(1);
                self.shared.metrics.add_auth_faults(1);
                tracing::error!(
                    target: TARGET_SSH,
                    event = "ssh.auth.refused",
                    reason = "identity-table-unreadable",
                    %fingerprint,
                    decorative_ssh_user = %user,
                    elapsed_ms = started.elapsed().as_millis() as u64,
                    %err,
                    "publickey refused: the identity table could not be read"
                );
                if self.refused_auth() {
                    return Err(self.too_many_auth_attempts());
                }
                Ok(Auth::reject())
            }
        }
    }

    async fn channel_open_session(
        &mut self,
        channel: Channel<Msg>,
        // russh 0.63 (2026-10-03): the open is answered through this handle, not
        // a returned bool; dropping it unanswered would refuse the channel.
        reply: russh::server::ChannelOpenHandle,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        // The exec request arrives later, carrying only a `ChannelId`, and the
        // byte stream can only be taken from the `Channel` itself, so it is
        // kept here. russh's own sftp example does the same.
        self.channels.insert(
            channel.id(),
            OpenChannel {
                channel,
                git_protocol: None,
            },
        );

        // A session that authenticates, opens a channel and then does nothing is
        // not idle, because it answers keepalives, so the inactivity timeout never
        // fires and the connection would be held indefinitely. Arm a watchdog.
        if let Some(grace) = self.shared.exec_grace {
            let handle = session.handle();
            let exec_seen = self.exec_seen.clone();
            // `.instrument`, not a captured `peer`: a spawned task does NOT
            // inherit the thread-local span, so the watchdog's line would
            // otherwise be the one line in the whole connection that could not
            // be tied back to it.
            let span = self.span.clone();
            tokio::spawn(
                async move {
                    tokio::time::sleep(grace).await;
                    if !exec_seen.load(Ordering::SeqCst) {
                        tracing::info!(
                            target: TARGET_SSH,
                            event = "ssh.watchdog.disconnect",
                            ?grace,
                            "disconnecting: authenticated but never asked for a git command"
                        );
                        note_teardown(
                            "watchdog-disconnect",
                            handle
                                .disconnect(
                                    Disconnect::ByApplication,
                                    "no git command requested".to_string(),
                                    String::new(),
                                )
                                .await,
                        );
                    }
                }
                .instrument(span),
            );
        }
        reply.accept().await;
        Ok(())
    }

    async fn exec_request(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.exec_seen.store(true, Ordering::SeqCst);
        let span = self.span.clone();
        let _entered = span.enter();

        // **The credential is re-read here, not remembered.** See
        // [`Connection::live_principal`]: russh will not dispatch an exec
        // before authentication succeeds, so an unauthenticated session is
        // unreachable rather than checked — but a session whose key was
        // REVOKED while it was open is entirely reachable, and until
        // 2026-09-11 it was authorised out of a cache filled at auth time.
        let principal = match self.live_principal() {
            Ok(principal) => principal,
            Err(stale) => {
                self.shared.metrics.add_exec_refused(1);
                let (reason, detail) = match &stale {
                    Stale::Revoked => ("key-revoked", String::new()),
                    Stale::Unreadable(err) => ("identity-table-unreadable", format!("{err}")),
                };
                tracing::warn!(
                    target: TARGET_SSH,
                    event = "ssh.exec.denied",
                    reason,
                    fingerprint = self.key_fingerprint.as_ref().map(tracing::field::display),
                    detail = %detail,
                    "exec denied: the key that opened this session no longer authorises it"
                );
                // Answered in git's own protocol, like every other denial on
                // this path: an `ERR ` pkt-line is what `git` prints verbatim
                // after `remote: `. A bare `channel_failure` here would put the
                // client back on "Could not read from remote repository", which
                // is the failure `report_error` was found unused for.
                session.channel_success(channel)?;
                session.data(channel, err_packet(REVOKED_MESSAGE)?)?;
                session.exit_status_request(channel, EXIT_DENIED)?;
                session.close(channel)?;
                // And the session goes with it. Refusing the one exec would
                // leave the connection open for the next one, which is the
                // finding restated a request later: a revoked key must not keep
                // a channel it can keep asking on.
                note_teardown(
                    "revoked-key-disconnect",
                    session
                        .handle()
                        .disconnect(
                            Disconnect::ByApplication,
                            "the key that opened this session no longer authorises it".to_string(),
                            String::new(),
                        )
                        .await,
                );
                return Ok(());
            }
        };

        // ── the operator's fire drill: ONE ERROR on purpose ─────────────
        //
        // What the estate's watch raises to prove its alert path end to end (estate-box
        // 8A). Held to server-wide Admin before it says anything; an anonymous session
        // is refused the channel like the tunnel verb below.
        if std::str::from_utf8(data).map(str::trim) == Ok(FIRE_DRILL_VERB) {
            if self.key_fingerprint.is_none() {
                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {FIRE_DRILL_VERB} needs a key");
                session.channel_failure(channel)?;
                return Ok(());
            }
            let (exit, out) = match self
                .shared
                .dispatch
                .control()
                .auth()
                .may_server(&principal.id, gunnar_core::Level::Admin)
            {
                Ok(true) => {
                    tracing::error!(
                        target: TARGET_SSH,
                        event = "ssh.fire_drill",
                        principal = %principal.id,
                        "fire drill: {} asked for this ERROR to prove the alert path end to end — nothing is wrong",
                        principal.id
                    );
                    (
                        0,
                        "fire drill: one ERROR said on this server's log (nothing is wrong)\n"
                            .to_string(),
                    )
                }
                Ok(false) => {
                    tracing::warn!(target: TARGET_SSH, event = "ssh.exec.denied", reason = "not-server-admin", principal = %principal.id, "{FIRE_DRILL_VERB} refused");
                    (EXIT_DENIED, format!("error: {FIRE_DRILL_VERB} needs server-wide Admin, which {} does not hold\n", principal.id))
                }
                Err(err) => (
                    EXIT_DENIED,
                    format!("error: the identity table could not be read: {err}\n"),
                ),
            };
            session.channel_success(channel)?;
            session.data(channel, out.into_bytes())?;
            session.exit_status_request(channel, exit)?;
            session.close(channel)?;
            return Ok(());
        }

        // ── git-lfs's fallback: answered, never a channel failure (2026-10-25) ──
        if std::str::from_utf8(data)
            .ok()
            .and_then(|l| l.split_whitespace().next())
            == Some(LFS_AUTHENTICATE_VERB)
        {
            tracing::info!(
                target: TARGET_SSH,
                event = "ssh.lfs_authenticate.answered",
                principal = %principal.id,
                command = %String::from_utf8_lossy(data),
                "git-lfs asked for its HTTPS fallback: its pure-SSH transfer failed or was refused just before"
            );
            session.channel_success(channel)?;
            session.extended_data(channel, 1, LFS_AUTHENTICATE_MESSAGE.as_bytes().to_vec())?;
            session.exit_status_request(channel, 1)?;
            session.close(channel)?;
            return Ok(());
        }

        // ── who this key is (wiring tracer B 2026-10-08) ──
        if std::str::from_utf8(data).map(str::trim) == Ok(WHOAMI_VERB) {
            if self.key_fingerprint.is_none() {
                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {WHOAMI_VERB} needs a key");
                session.channel_failure(channel)?;
                return Ok(());
            }
            session.channel_success(channel)?;
            session.data(channel, format!("{}\n", principal.id).into_bytes())?;
            session.exit_status_request(channel, 0)?;
            session.close(channel)?;
            return Ok(());
        }

        // ── a repository's settings (owner, 2026-10-14) ──
        //
        // Decided by the control plane's own `Repos` handlers through
        // [`RepoDoor`]: this listener only proves who asks (the key, re-read
        // above) and carries the answer. An anonymous session is refused the
        // channel, before the pause and the plan gates (not a transfer).
        if let Some(parsed) = parse_repo_verb(data) {
            let Some(fingerprint) = self.key_fingerprint.as_ref().map(|f| f.to_string()) else {
                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {REPO_VERB} needs a key");
                session.channel_failure(channel)?;
                return Ok(());
            };
            let (exit, out) = match parsed {
                Err(why) => (1, format!("error: {why}\n")),
                Ok((repo, action)) => match self.shared.repo_door.clone() {
                    None => (
                        1,
                        "error: this server answers repository settings only on its control plane\n".to_string(),
                    ),
                    Some(door) => {
                        let ask = RepoAsk { principal: principal.id.clone(), fingerprint, repo: repo.clone(), action: action.clone() };
                        let answer = door(ask).await;
                        tracing::info!(
                            target: TARGET_SSH,
                            event = "ssh.control",
                            verb = REPO_VERB,
                            principal = %principal.id,
                            repo = %repo,
                            action = ?action,
                            decision = if answer.is_ok() { "allow" } else { "deny" },
                            "repository settings asked over the forge's door"
                        );
                        match answer {
                            Ok(said) => (0, if said.ends_with('\n') { said } else { format!("{said}\n") }),
                            Err(why) => (EXIT_DENIED, format!("error: {why}\n")),
                        }
                    }
                },
            };
            session.channel_success(channel)?;
            session.data(channel, out.into_bytes())?;
            session.exit_status_request(channel, exit)?;
            session.close(channel)?;
            return Ok(());
        }

        // ── a member's keys for an encrypted repository's ring (2026-10-06) ──
        //
        // A read of the identity table, held to the same rule as
        // `Repos.RecipientKeys` (see [`recipient_keys_answer`]). Before the
        // pause and the plan gates for the tunnel verb's reason: not a
        // transfer. An anonymous session is refused the channel.
        if let Some(parsed) = parse_recipient_keys_verb(data) {
            if self.key_fingerprint.is_none() {
                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {RECIPIENT_KEYS_VERB} needs a key");
                session.channel_failure(channel)?;
                return Ok(());
            }
            let (exit, out) = match parsed {
                Err(why) => (1, format!("error: {why}\n")),
                Ok((repo, member)) => {
                    let (exit, out) = recipient_keys_answer(
                        self.shared.dispatch.control(),
                        &principal.id,
                        &repo,
                        &member,
                    );
                    tracing::info!(
                        target: TARGET_SSH,
                        event = "ssh.control",
                        verb = RECIPIENT_KEYS_VERB,
                        principal = %principal.id,
                        repo = %repo,
                        member = %member,
                        decision = if exit == 0 { "allow" } else { "deny" },
                        "recipient keys asked over the forge's door"
                    );
                    (exit, out)
                }
            };
            session.channel_success(channel)?;
            session.data(channel, out.into_bytes())?;
            session.exit_status_request(channel, exit)?;
            session.close(channel)?;
            return Ok(());
        }

        // ── the caller's own encryption key (CRYPT-YUBI 2026-10-26) ──────
        //
        // The session's principal reads or changes ITS OWN key and methods: the
        // public half, and wraps that are ciphertext to this server. Not a
        // transfer; an anonymous session is refused the channel.
        if let Some(parsed) = gunnar_crypt::door::parse_request(data) {
            if self.key_fingerprint.is_none() {
                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {} needs a key", gunnar_crypt::door::VERB);
                session.channel_failure(channel)?;
                return Ok(());
            }
            let (exit, out) = match parsed {
                Err(why) => (1, format!("error: {why}\n")),
                Ok(request) => {
                    let verb = match &request {
                        gunnar_crypt::door::Request::Get => "get",
                        gunnar_crypt::door::Request::Create { .. } => "create",
                        gunnar_crypt::door::Request::Add(_) => "add",
                        gunnar_crypt::door::Request::Remove(_) => "remove",
                    };
                    let (exit, out) =
                        user_key_answer(self.shared.dispatch.control(), &principal, request);
                    tracing::info!(
                        target: TARGET_SSH,
                        event = "ssh.control",
                        verb = gunnar_crypt::door::VERB,
                        action = verb,
                        principal = %principal.id,
                        decision = if exit == 0 { "allow" } else { "deny" },
                        "own encryption key over the forge's door"
                    );
                    (exit, out)
                }
            };
            session.channel_success(channel)?;
            session.data(channel, out.into_bytes())?;
            session.exit_status_request(channel, exit)?;
            session.close(channel)?;
            return Ok(());
        }

        // ── the ONE control verb: the tunnel address (T17) ─────────────
        //
        // Before the pause and the plan gates: those are about transfers, and
        // this is not one. Never through `Dispatch::authorise`, which knows
        // only git verbs. An anonymous session gets the channel refused like a
        // shell would; everyone else is held to server-wide Admin BEFORE the
        // door runs.
        if let Some(parsed) = parse_tunnel_address_verb(data) {
            if self.key_fingerprint.is_none() {
                tracing::warn!(target: TARGET_SSH, event = "ssh.exec.refused", reason = "anonymous-control-verb", "exec refused: {TUNNEL_ADDRESS_VERB} needs a key");
                session.channel_failure(channel)?;
                return Ok(());
            }
            let (exit, out) = match self
                .shared
                .dispatch
                .control()
                .auth()
                .may_server(&principal.id, gunnar_core::Level::Admin)
            {
                Ok(false) => {
                    tracing::warn!(target: TARGET_SSH, event = "ssh.exec.denied", reason = "not-server-admin", principal = %principal.id, "{TUNNEL_ADDRESS_VERB} refused");
                    (EXIT_DENIED, format!("error: {TUNNEL_ADDRESS_VERB} needs server-wide Admin, which {} does not hold\n", principal.id))
                }
                Err(err) => (
                    EXIT_DENIED,
                    format!("error: the identity table could not be read: {err}\n"),
                ),
                Ok(true) => match parsed {
                    Err(why) => (1, format!("error: {why}\n")),
                    Ok(address) => match &self.shared.tunnel_door {
                        None => (
                            1,
                            "error: this box has no WireGuard tunnel (built without wg-client)\n"
                                .to_string(),
                        ),
                        Some(door) => match door(&address) {
                            Ok(answer) => {
                                tracing::warn!(target: TARGET_SSH, event = "ssh.control", verb = TUNNEL_ADDRESS_VERB, principal = %principal.id, address = %address, decision = "allow", "tunnel address door answered");
                                (
                                    0,
                                    if answer.ends_with('\n') {
                                        answer
                                    } else {
                                        format!("{answer}\n")
                                    },
                                )
                            }
                            Err(why) => {
                                tracing::warn!(target: TARGET_SSH, event = "ssh.control", verb = TUNNEL_ADDRESS_VERB, principal = %principal.id, address = %address, decision = "deny", reason = %why, "tunnel address door refused");
                                (1, format!("error: {why}\n"))
                            }
                        },
                    },
                },
            };
            session.channel_success(channel)?;
            session.data(channel, out.into_bytes())?;
            session.exit_status_request(channel, exit)?;
            session.close(channel)?;
            return Ok(());
        }

        // ── the PAUSED door, before the plan gates ──────────────────────
        //
        // MEASURED under KVM (bench R1, 2026-09-15): with the gate paused for
        // a growth and `/data` already unmounted, a push knocked, the plan
        // gates ran FIRST, and the disk-floor check statvfs'd the read-only
        // guard tmpfs under the mountpoint — the client was told "/data has
        // 64.0 KiB free … no push until space is reclaimed (`gunnar control
        // repo delete --purge`)". The wrong sentence, with advice to delete
        // repositories, on a box that is growing. The pause is the box's own
        // state and says nothing about any repository, so it is answered
        // before the command is even resolved: the same words and the same
        // EX_TEMPFAIL for every principal and every path.
        if let Some(pause) = self.shared.admission.paused() {
            self.shared.admission.note_refused_paused();
            tracing::info!(
                target: TARGET_GIT,
                event = "git.transfer.refused",
                reason = pause.reason(),
                command = %String::from_utf8_lossy(data),
                "transfer REFUSED: the admission gate is paused"
            );
            session.channel_success(channel)?;
            session.data(channel, err_packet(pause.words())?)?;
            session.exit_status_request(channel, EXIT_BUSY)?;
            session.close(channel)?;
            return Ok(());
        }

        let request =
            match self
                .shared
                .dispatch
                .authorise(&principal, self.key_fingerprint.as_ref(), data)
            {
                Ok(request) => request,
                Err(err) => {
                    self.shared.metrics.add_exec_refused(1);
                    // **THE counting point for a tenant's `auth_failures`**: a
                    // principal that authenticated and then held no grant on
                    // the repository it named.
                    //
                    // Only `Denied`, and deliberately only `Denied`. The other
                    // arms are a different sentence: `NoSuchRepo` hides a
                    // typo behind the same wire message on purpose and would
                    // charge a namespace that may not exist; `Standby` and
                    // `Plan` are the box and the plan refusing, not the
                    // credential; `Command` never named a repository at all.
                    // And the OTHER half of "authentication failed" — a key
                    // that resolves to nobody — happens before any exec line
                    // and belongs to no namespace, so it stays where it is, on
                    // the box's own `connections_refused`.
                    if let (DispatchError::Denied { repo, .. }, Some(flow)) =
                        (&err, self.shared.admission.flow())
                    {
                        flow.auth_failure(repo.namespace(), gunnar_core::now_ms());
                    }
                    let raw = String::from_utf8_lossy(data);
                    match err.client_message() {
                        // Not a git command at all: `bash`, `scp`, an empty
                        // string. Refuse the channel request itself; the client is
                        // not speaking a protocol that could read a reply.
                        // git-lfs asks SSH for its credentials (`git-lfs-authenticate`,
                        // `git-lfs-transfer`) before every transfer and goes to its
                        // https endpoint when refused: a client's known first step,
                        // said at INFO (RUN A 2026-10-08: seven a minute during one LFS
                        // push failed the log gate as WARN spam). Anything else stays WARN.
                        None if is_lfs_over_ssh(&raw) => {
                            tracing::info!(
                                target: TARGET_SSH,
                                event = "ssh.exec.refused",
                                reason = "lfs-over-ssh",
                                command = %raw,
                                "exec refused: git-lfs asks SSH first and uses its https endpoint"
                            );
                            session.channel_failure(channel)?;
                        }
                        None => {
                            tracing::warn!(
                                target: TARGET_SSH,
                                event = "ssh.exec.refused",
                                reason = "not-a-git-command",
                                command = %raw,
                                %err,
                                "exec refused"
                            );
                            session.channel_failure(channel)?;
                        }
                        // A git client asked for something it may not have. Accept
                        // the channel and answer in its own protocol, so it prints
                        // the reason instead of "Could not read from remote
                        // repository".
                        Some(message) => {
                            tracing::warn!(
                                target: TARGET_SSH,
                                event = "ssh.exec.denied",
                                // The distinction the *wire* deliberately hides:
                                // "no such repository" and "no permission" are one
                                // string to the client so a caller cannot enumerate
                                // private repositories. The log keeps it.
                                reason = match &err {
                                    DispatchError::NoSuchRepo(_) => "no-such-repository",
                                    DispatchError::Denied { .. } => "no-grant",
                                    DispatchError::Standby { .. } => "standby-read-only",
                                    DispatchError::Twin { .. } => "twin-serves-no-git",
                                    DispatchError::AnonymousElsewhere { .. } => {
                                        "anonymous-elsewhere"
                                    }
                                    DispatchError::Plan { .. } => "gate-refused",
                                    DispatchError::Store(_) => "control-plane-unavailable",
                                    DispatchError::Command(_) => "not-a-git-command",
                                },
                                command = %raw,
                                %err,
                                "exec denied"
                            );
                            session.channel_success(channel)?;
                            session.data(channel, err_packet(&message)?)?;
                            session.exit_status_request(channel, EXIT_DENIED)?;
                            session.close(channel)?;
                        }
                    }
                    return Ok(());
                }
            };

        let Some(open) = self.channels.remove(&channel) else {
            // The CLIENT's misuse (a second exec on one channel, or a channel never
            // opened): refused, said at WARN once per connection — any caller that may
            // run one git command could otherwise raise the server's ERROR at will.
            if std::mem::replace(&mut self.misuse_said, true) {
                tracing::debug!(target: TARGET_SSH, ?channel, "exec request for a channel that is not open (said already on this connection)");
            } else {
                tracing::warn!(
                    target: TARGET_SSH,
                    event = "ssh.exec.misuse",
                    ?channel,
                    "exec request for a channel that is not open (a second exec on one channel, or one never opened) — refused; said once per connection"
                );
            }
            session.channel_failure(channel)?;
            return Ok(());
        };

        // The repository is known from here on, so it goes on the CONNECTION
        // span: every line the transfer emits underneath inherits it, and the
        // "which repository was that connection touching" question is answered
        // even for a line that has nothing to do with git.
        span.record("repo", tracing::field::display(&request.repo));

        // Attached *after* the decision above, never before it: an environment
        // variable selects the framing of the answer and nothing else.
        let request = request.with_git_protocol(open.git_protocol);

        // ── the explicit ceiling ───────────────────────────────────────────
        //
        // After authorisation, deliberately. "The server is busy" is a fact
        // about the server, and telling it to a client that may not touch this
        // repository at all would leak that the repository exists. The order
        // also means the gate meters *transfers gunnar was going to serve*,
        // which is what the counter is supposed to mean.
        //
        // It refuses rather than queues; `crate::admission` sets out why at
        // length. In one line: a git client retries, and a fast refusal with a
        // sentence in it beats a socket that goes quiet while still holding the
        // session, the channel and a slot.
        // The class the governor meters this operation as. `git-upload-pack` is
        // clone, fetch AND ls-remote on the wire and cannot be told apart before
        // the stream is open, which is *after* this decision by design; see
        // `OpClass::for_ssh_service` for what that costs and why the middle
        // class is the right guess.
        // ── the BOOT WINDOW door, before admission ─────────────────────────
        //
        // After the repository is known and authorised, so a stranger cannot
        // learn from this whether a repository exists, and before a slot is
        // taken, so a boot does not spend its gate on requests it cannot serve.
        // See `GunnarSsh::with_readiness` for what was measured.
        // A KEYED caller is queued for the stores to open, up to its keyed
        // wait: a restart (a release, an upgrade) lasts seconds, and git-lfs's
        // pure-SSH transfer cannot retry a refusal — it falls back to
        // `git-lfs-authenticate` and fails the push (gunnar.rs 2026-10-25). A
        // stranger is still answered at once.
        if !self.shared.ready.load(std::sync::atomic::Ordering::Relaxed)
            && request.principal.id != gunnar_core::ANONYMOUS_PRINCIPAL_ID
        {
            let until =
                std::time::Instant::now() + self.shared.admission.wait_for(&request.principal.id);
            while !self.shared.ready.load(std::sync::atomic::Ordering::Relaxed)
                && std::time::Instant::now() < until
            {
                tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            }
        }
        if !self.shared.ready.load(std::sync::atomic::Ordering::Relaxed) {
            tracing::info!(
                target: TARGET_GIT,
                event = "git.transfer.refused",
                reason = "starting",
                service = %request.service,
                "transfer REFUSED: the stores are still opening after a restart"
            );
            session.channel_success(channel)?;
            session.data(channel, err_packet(STARTING_MESSAGE)?)?;
            session.exit_status_request(channel, EXIT_BUSY)?;
            session.close(channel)?;
            return Ok(());
        }

        let class = OpClass::for_ssh_service(request.service);
        // One git-lfs, one slot (GHUNT-7, `Connection::lfs_slots`): a `git-lfs-transfer`
        // channel joins the slot an earlier channel of this connection holds for the same
        // repository and operation, while it carries fewer than LFS_CHANNELS_PER_SLOT.
        // Each channel holds the slot twice (the task and its metered stream).
        let lfs_key = matches!(request.service, GitService::LfsTransfer(_))
            .then(|| (request.repo.to_string(), request.service));
        let joined = lfs_key
            .as_ref()
            .and_then(|k| self.lfs_slots.get(k))
            .and_then(std::sync::Weak::upgrade)
            .filter(|slot| Arc::strong_count(slot) < 2 * LFS_CHANNELS_PER_SLOT);
        // `admit_within`, not `try_admit`: over the limit the transfer parks for
        // a bounded moment instead of being refused on the spot. A refused
        // client reconnects, and a reconnect is a whole SSH handshake — more
        // than the wait costs. `crate::governor::DEFAULT_ADMISSION_WAIT` has the
        // measurement; `--admission-wait-ms 0` is the old behaviour exactly.
        let admitted = if let Some(slot) = joined {
            slot
        } else {
            // GUN-SSH-SLOT-HANG: a wait is never silent. Past WAIT_NOTICE_AFTER the client
            // is told on stderr that it waits for a slot, how full the gate is, and how
            // long it waits before it is refused by name.
            let admitted_or_refused = {
                let admit = self.shared.admission.admit_within(
                    &request.principal.id,
                    &request.repo.namespace,
                    class,
                );
                tokio::pin!(admit);
                tokio::select! {
                    biased;
                    done = &mut admit => done,
                    () = tokio::time::sleep(WAIT_NOTICE_AFTER) => {
                        let stats = self.shared.admission.stats();
                        let bound = self.shared.admission.wait_for(&request.principal.id).as_secs();
                        let notice = format!(
                            "gunnar: waiting for a transfer slot: {} of {} transfers are in use; \
                             refused by name after {bound} s if none frees\n",
                            stats.in_flight, stats.limit
                        );
                        let _ = session.extended_data(channel, 1, notice.into_bytes());
                        admit.await
                    }
                }
            };
            match admitted_or_refused {
                Ok((guard, waited)) => {
                    // AGGREGATE, at most once a second. A row per waiting client is
                    // the appliance's whole twenty-row console under load (the
                    // measured 5411 in one 256-client run); a row per second
                    // carrying the totals is the same fact, readable.
                    if !waited.is_zero() && self.shared.admission.wait_row_due() {
                        let stats = self.shared.admission.stats();
                        tracing::info!(
                            target: TARGET_GIT,
                            event = "git.transfer.waited",
                            class = class.name(),
                            waited_ms = waited.as_millis() as u64,
                            waited_total = stats.admitted_after_wait,
                            mean_wait_ms = stats.mean_wait_ms(),
                            peak_wait_ms = stats.peak_wait_ms,
                            in_flight = stats.in_flight,
                            limit = stats.limit,
                            "transfers are WAITING for a slot and are then served: {} so far, mean \
                             {} ms, worst {} ms against a bound of {} ms",
                            stats.admitted_after_wait,
                            stats.mean_wait_ms(),
                            stats.peak_wait_ms,
                            self.shared.admission.wait().as_millis() as u64
                        );
                    }
                    let slot = Arc::new(guard);
                    if let Some(k) = lfs_key {
                        self.lfs_slots.retain(|_, w| w.strong_count() > 0);
                        self.lfs_slots.insert(k, Arc::downgrade(&slot));
                    }
                    slot
                }
                Err(refusal) => {
                    let stats = self.shared.admission.stats();
                    tracing::warn!(
                        target: TARGET_GIT,
                        event = "git.transfer.refused",
                        // Which gate, not just "a gate": the anonymous ceiling
                        // fires with the box nearly empty, and a line that
                        // still said `admission-gate-full` there would send an
                        // operator to raise a limit that was never reached.
                        reason = refusal.reason(),
                        service = %request.service,
                        class = class.name(),
                        %refusal,
                        in_flight = stats.in_flight,
                        peak_in_flight = stats.peak_in_flight,
                        // The governed limit, so a log line says what was actually
                        // in force rather than what was configured. Without it the
                        // shed is invisible again.
                        limit = stats.limit,
                        reserved_bytes = stats.reserved_bytes,
                        refused_total = stats.refused(),
                        // The bound the transfer waited out before this. Without it
                        // a reader cannot tell an instant refusal from one the
                        // server spent two seconds trying to avoid.
                        waited_ms = self.shared.admission.wait().as_millis() as u64,
                        peak_wait_ms = stats.peak_wait_ms,
                        "transfer REFUSED by the admission gate"
                    );
                    // Answered in the client's own protocol, like a denial, so git
                    // prints the reason instead of "Could not read from remote
                    // repository". Saturation the operator can see and the user
                    // cannot is only half a fix.
                    session.channel_success(channel)?;
                    session.data(channel, err_packet(&refusal.client_message())?)?;
                    session.exit_status_request(channel, EXIT_BUSY)?;
                    session.close(channel)?;
                    return Ok(());
                }
            }
        };

        // ── the span the whole transfer runs under ─────────────────────────
        //
        // A CHILD of `ssh.connection`, so it inherits `conn`, `peer`,
        // `principal` and `repo` and states none of them again. `service` is
        // the one thing that is new. Everything `gunnar-cli`'s transport and
        // `gunnar-policy`'s gate emit lands underneath it, in the same
        // process, without either of them being handed an id to repeat.
        let transfer = tracing::info_span!(
            target: TARGET_GIT,
            "git.transfer",
            service = %request.service,
            access = %request.access,
        );
        transfer.in_scope(|| {
            tracing::info!(
                target: TARGET_GIT,
                event = "git.transfer.started",
                git_protocol = ?request.git_protocol,
                "serving"
            )
        });
        let in_flight = self.shared.metrics.transfer_started();
        let metrics = Arc::clone(&self.shared.metrics);
        let started = std::time::Instant::now();

        session.channel_success(channel)?;

        let transport = self.shared.transport.clone();
        // Split, NOT `into_stream()`. `into_stream` wraps the channel in
        // russh's `ChannelCloseOnDrop`, which spawns a `Close` the instant the
        // transport lets the stream go, and that `Close` raced (and beat) the
        // `exit-status` sent next, which is how a successful push came to
        // report failure. Neither half closes the channel on drop, so the two
        // messages below are ordered by the sender's own queue.
        let (read_half, write_half) = open.channel.split();
        // METERED, so the memory ledger learns this transfer's real size as it
        // happens rather than being charged a flat per-class constant at
        // admission. `crate::admission::Metered` is the only place a byte count
        // reaches the governor, and it wraps the duplex the transport was
        // already going to be handed — no second copy, no second truth.
        // IDLE-BOUND (GUN-SSH-SLOT-HANG): a transfer that moves no bytes either way
        // while it waits on the client is ended after `--transfer-idle-secs`, and its
        // slot is given back; `idle` says afterwards whether that is how it ended.
        let (joined_stream, idle) = crate::idle::Idle::maybe(
            tokio::io::join(
                ChannelReader::new(read_half),
                // Bound to the session: a write parked on the SSH window ends
                // with an error when the session does (see `SessionBound`).
                SessionBound::new(write_half.make_writer(), self.session_over.subscribe()),
            ),
            self.shared.transfer_idle,
        );
        let stream = crate::admission::Metered::new(joined_stream, Arc::clone(&admitted));
        let idle_secs = self
            .shared
            .transfer_idle
            .map(|t| t.as_secs())
            .unwrap_or_default();
        // Spawned, not awaited: see the module docs. Awaiting here would stop
        // the session loop that has to feed this very stream.
        // ── THE STANDING FAN, not a fresh thread ────────────────────────
        //
        // This used to be `tokio::spawn` alone, and the task itself was never
        // the cost — tasks are cheap. The cost is behind it: `SyncIoBridge`
        // hands the transfer to tokio's BLOCKING pool, which spawns an OS
        // thread on demand, per operation, up to 512 of them. That is
        // per-request initialisation, and it is what the fans replace.
        //
        // The future is driven by `Handle::block_on` on a worker that was
        // already alive and parked before this client connected. A full queue
        // refuses here, which is the only place a refusal is free.
        let handle = tokio::runtime::Handle::current();
        let pipelines = crate::pipeline::fans();
        let over = self.session_over.subscribe();
        let (calm_grace, crowded_grace) = self.shared.session_end_grace;
        let body = async move {
            // ── the CoDel dequeue ──────────────────────────────────────
            //
            // The FIRST line of the task, deliberately. Everything between
            // `try_admit` above and this point is the transfer WAITING:
            // tokio's ready queue, and behind it the blocking-pool handoff
            // that `SyncIoBridge` needs before a single byte moves. That
            // wait is the queue the record called invisible, and this call is
            // what makes the governor able to see it. Measured here rather
            // than at the end because sojourn is wait time; service time
            // filed as sojourn would make every large clone look like a
            // standing queue.
            admitted.entered_service();
            // ★ The WHOLE transfer is bound to the session, not only its writes
            // (`SessionBound`): a client that goes silent while main is still
            // building the pack has no write parked on the window to wake, and
            // the build can take minutes (the UpCloud mockup's LD1/LD2,
            // 2026-10-09: 2 min 10 s on TCG). When the session ends — the
            // keepalives unanswered, a hang-up — the transfer ends with it and
            // gives back its fan worker, admission slot and store handle.
            let cancel = request.cancel.clone();
            let served = serve_until_session_ends(transport.serve(request, Box::new(stream)), over.clone(), &cancel, || {
                // Stinky row 19 (2026-10-10): a hung-up client's worker is one
                // a queued client waits for, so the grace is short when any do.
                crate::channel::grace_for(calm_grace, crowded_grace, pipelines.queued(class))
            }, || {
                crate::seam::TransportError::Io(session_ended_error())
            })
            .await;
            // ── closed for idleness: counted, said, and told to the client ──
            if idle.timed_out() {
                metrics.add_transfers_idle_closed(1);
                tracing::warn!(
                    target: TARGET_GIT,
                    event = "git.transfer.idle_closed",
                    idle_secs,
                    "transfer CLOSED: no bytes moved for {idle_secs} s while it waited on the \
                     client; its admission slot is free again"
                );
                let told = format!(
                    "gunnar: closed this git transfer: {} for {idle_secs} s while it waited on \
                     this client; its transfer slot is free again\n",
                    crate::idle::IDLE_CLOSED
                );
                let sent = tokio::select! {
                    sent = write_half.extended_data_bytes(1, told.into_bytes()) => sent,
                    () = session_over(over.clone()) => Err(russh::Error::IO(session_ended_error())),
                };
                note_teardown("idle-notice", sent);
            }
            let status = match served {
                Ok(()) => 0,
                Err(err) => {
                    metrics.add_transfers_failed(1);
                    // ── the refusal the client used to never hear ───────
                    //
                    // Until 2026-08-05 this arm logged the whole reason and
                    // sent the client an exit status and nothing else, so
                    // every post-authorisation refusal — the repository's
                    // store lock held by a `gunnar gc`, a storage fault on
                    // open, an unwired service — reached the user as
                    // `fatal: the remote end hung up unexpectedly`. The two
                    // gates ABOVE this one, in this same function, both
                    // answer in the client's own protocol; this one did not,
                    // and the asymmetry is what made it invisible.
                    //
                    // `client_message()` is `None` whenever bytes may
                    // already have gone out, so this cannot append a
                    // pkt-line to a half-written pack — see
                    // `TransportError::client_message`.
                    let told = err.client_message();
                    tracing::warn!(
                        target: TARGET_GIT,
                        event = "git.transfer.failed",
                        %err,
                        // Whether the user was given a reason, said in the
                        // log, because "the client was told" and "the client
                        // saw a dead socket" must not be the same line.
                        client_told = told.is_some(),
                        "transport ended with an error"
                    );
                    if let Some(message) = told {
                        match err_packet(&message) {
                            Ok(packet) => {
                                // Raced against the session's end: `data_bytes`
                                // waits for SSH window like the stream does.
                                let sent = tokio::select! {
                                    sent = write_half.data_bytes(packet) => sent,
                                    () = session_over(over.clone()) => Err(russh::Error::IO(session_ended_error())),
                                };
                                note_teardown("refusal-packet", sent)
                            }
                            Err(err) => tracing::error!(
                                target: TARGET_GIT,
                                %err,
                                "the refusal could not be encoded, so the client is \
                                 getting only an exit status"
                            ),
                        }
                    }
                    EXIT_TRANSPORT_FAILED
                }
            };
            // The channel is still open here, nothing has been dropped that
            // could have closed it, so the status arrives before the close,
            // through the same sender the data went through.
            if write_half.exit_status(status).await.is_err() {
                // WARN only when the client missed a verdict it needed. A
                // client that took everything it asked for and hung up
                // before the `exit-status` has lost nothing — it is how
                // `ls-remote` and `fetch` end — and warning about it made
                // `grep -i error` useless on a completely healthy server:
                // MEASURED 2026-08-06, 9 successful transfers, 9 of these.
                // A client that missed a NON-ZERO status did lose
                // something: it believes the transfer worked.
                if status == 0 {
                    tracing::debug!(
                        target: TARGET_GIT,
                        event = "git.exit_status.undelivered",
                        ?channel,
                        status,
                        "the client hung up before the exit status; it had already \
                         been served"
                    );
                } else {
                    tracing::warn!(
                        target: TARGET_GIT,
                        event = "git.exit_status.undelivered",
                        ?channel,
                        status,
                        "the client was gone before a NON-ZERO exit status, so it \
                         does not know this transfer failed"
                    );
                }
            }
            note_teardown("channel-close", write_half.close().await);
            // The duration an operator asked "why was that clone slow"
            // about: the WHOLE transfer, from admission to the last byte
            // acknowledged, not the state machine's own view of itself.
            tracing::info!(
                target: TARGET_GIT,
                event = "git.transfer.finished",
                status,
                elapsed_ms = started.elapsed().as_millis() as u64,
                "transfer finished"
            );
            // Moved into this task and dropped here, so the slot is held for the
            // whole transfer and released by `Drop` rather than by a call that
            // an early return could skip. The in-flight gauge is released the
            // same way, and for the same reason.
            drop(admitted);
            drop(in_flight);
        }
        .instrument(transfer);

        // ── WHAT THE FAN IS ACTUALLY CARRYING ───────────────────────────────
        //
        // `Pipelines` has counted `served`, `queued` and `workers` since it was
        // written and NOTHING reported them, so the only evidence the fans were
        // doing anything was the absence of refusals — which is not evidence,
        // it is silence. Asked "is the queue filling", the honest answer was an
        // inference from the workload's shape.
        //
        // `submit` already returns the queue depth after the push, so this
        // costs one integer. Emitted only when an operation ACTUALLY QUEUED
        // (depth > 1, i.e. it did not go straight to a free worker), because a
        // line per transfer on a 102-client arm is noise that hides the signal.
        // `served` and `workers` ride along so one line answers "how deep, out
        // of how many, having done how much".
        match pipelines.submit(
            class,
            Box::new(move || {
                handle.block_on(body);
            }),
        ) {
            Ok(depth) if depth > 1 => tracing::info!(
                target: TARGET_GIT,
                event = "git.pipeline.queued",
                class = ?class,
                depth,
                workers = pipelines.workers(class),
                served = pipelines.served(class),
                "an operation waited for a worker"
            ),
            Ok(_) => {}
            Err(refused) => {
                // The fan is saturated. Say so in the client's own protocol
                // rather than dropping the channel — a dead socket is how a
                // refusal becomes "the remote end hung up unexpectedly".
                tracing::warn!(
                    target: TARGET_GIT,
                    event = "git.transfer.refused",
                    class = ?class,
                    depth = pipelines.queued(class),
                    workers = pipelines.workers(class),
                    served = pipelines.served(class),
                    %refused,
                    "the standing fan for this class refused the operation"
                );
                return Err(anyhow::anyhow!("{refused}"));
            }
        }
        Ok(())
    }

    // ── everything below is the "no shell, ever" half ──────────────────────
    //
    // russh's defaults for these return `Ok(())` **without** answering the
    // channel request, which leaves a client waiting rather than failing. Each
    // one is answered explicitly.

    async fn shell_request(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.refuse(channel, session, "shell").await
    }

    async fn subsystem_request(
        &mut self,
        channel: ChannelId,
        name: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        // Includes `sftp`. gunnar serves git and nothing else.
        self.span.in_scope(|| {
            tracing::warn!(
                target: TARGET_SSH,
                event = "ssh.request.refused",
                subsystem = %name,
                "subsystem refused"
            )
        });
        self.refuse(channel, session, "subsystem").await
    }

    async fn pty_request(
        &mut self,
        channel: ChannelId,
        _term: &str,
        _col_width: u32,
        _row_height: u32,
        _pix_width: u32,
        _pix_height: u32,
        _modes: &[(russh::Pty, u32)],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.refuse(channel, session, "pty").await
    }

    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> {
        self.refuse(channel, session, "x11").await
    }

    async fn env_request(
        &mut self,
        channel: ChannelId,
        variable_name: &str,
        variable_value: &str,
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        // An ALLOWLIST, not a filter: everything that is not named here is
        // refused, so adding a variable is a deliberate edit to one array and
        // the default for anything new is "no". See the module docs for why an
        // SSH server that passes an environment through is an attack surface.
        let span = self.span.clone();
        let _entered = span.enter();
        if !ENV_ALLOWLIST.contains(&variable_name) {
            // Same refusal, same event, same fields — two levels. See
            // [`ENV_ROUTINE`] for the 406-of-406 measurement that split them.
            if env_refusal_is_routine(variable_name) {
                tracing::debug!(
                    target: TARGET_SSH,
                    event = "ssh.env.refused",
                    reason = "not-allowlisted",
                    routine = true,
                    variable = %variable_name,
                    "env refused: gunnar reads exactly one environment variable"
                );
            } else {
                tracing::warn!(
                    target: TARGET_SSH,
                    event = "ssh.env.refused",
                    reason = "not-allowlisted",
                    routine = false,
                    variable = %variable_name,
                    "env refused: gunnar reads exactly one environment variable"
                );
            }
            session.channel_failure(channel)?;
            return Ok(());
        }

        // A value gunnar will not read is not worth holding. `version=2` is
        // nine bytes; the cap is generous and exists so a client cannot make
        // the server keep an unbounded string per channel.
        if variable_value.len() > MAX_ENV_VALUE {
            tracing::warn!(
                target: TARGET_SSH,
                event = "ssh.env.refused",
                reason = "value-too-long",
                variable = %variable_name,
                length = variable_value.len(),
                "env refused: the value is implausibly long"
            );
            session.channel_failure(channel)?;
            return Ok(());
        }

        let Some(open) = self.channels.get_mut(&channel) else {
            tracing::warn!(
                target: TARGET_SSH,
                ?channel,
                "env request for a channel that was never opened"
            );
            session.channel_failure(channel)?;
            return Ok(());
        };
        if variable_name != ENV_GIT_PROTOCOL {
            // `ENV_ALLOWLIST` grew a name and nothing here was taught to carry
            // it. Fail closed and say so, rather than quietly filing an unknown
            // variable in the `GIT_PROTOCOL` slot.
            tracing::error!(
                target: TARGET_SSH,
                variable = %variable_name,
                "env allowlisted but not carried anywhere; refusing"
            );
            session.channel_failure(channel)?;
            return Ok(());
        }
        tracing::debug!(
            target: TARGET_SSH,
            variable = %variable_name,
            value = %variable_value,
            "env accepted"
        );
        open.git_protocol = Some(variable_value.to_string());
        session.channel_success(channel)?;
        Ok(())
    }

    async fn channel_close(
        &mut self,
        channel: ChannelId,
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        self.channels.remove(&channel);
        Ok(())
    }
}

impl Connection {
    async fn refuse(
        &mut self,
        channel: ChannelId,
        session: &mut Session,
        what: &'static str,
    ) -> Result<(), anyhow::Error> {
        self.span.in_scope(|| {
            tracing::warn!(
                target: TARGET_SSH,
                event = "ssh.request.refused",
                request = what,
                "refused: gunnar has no shell"
            )
        });
        session.channel_failure(channel)?;
        Ok(())
    }
}

/// The session loop ended: say so in words an operator's `grep -i error` will
/// not find, unless it really was an error.
///
/// See [`peer_hung_up`]. `error=early eof` fired on **every healthy fetch**
/// before this existed.
fn note_session_end(error: &anyhow::Error) {
    if peer_hung_up(error.as_ref()) {
        tracing::debug!(
            target: TARGET_SSH,
            event = "ssh.session.closed",
            reason = "peer-hung-up",
            "the client closed the connection"
        );
        return;
    }
    tracing::debug!(target: TARGET_SSH, %error, "ssh session ended with an error");
}

/// Note that a best-effort teardown step did not work.
///
/// # Best-effort stays best-effort. What changes is that it stops being silent
///
/// Closing a channel and disconnecting a watchdogged session are both allowed
/// to fail: the peer may already be gone, which is the ordinary case and not
/// worth an operator's attention. Neither ever became an error, and neither
/// should. But a **systemic** failure to tear sessions down, from descriptor
/// exhaustion, the failure mode the unbounded handle cache used to
/// produce, was previously indistinguishable from a quiet server, because
/// `let _ =` leaves nothing behind at any level.
///
/// `debug!` rather than `warn!`: one of these is normal and a thousand is not,
/// and the thing that tells them apart is a count in a journal, not a severity
/// on a line that fires every time a client hangs up first.
/// The one encoder for a client-visible refusal: `ERR <message>` as a pkt-line.
///
/// Three call sites now answer a refusal in the client's own protocol — the
/// dispatch denial, the admission gate and a transfer refused before its first
/// byte — and a fourth is a matter of time. One implementation: the framing is written once,
/// so a refusal cannot acquire a subtly different shape depending on which gate
/// produced it. `ERR` is git's own out-of-band error line, which every git
/// version since 1.6 prints verbatim instead of "Could not read from remote
/// repository".
fn err_packet(message: &str) -> anyhow::Result<Vec<u8>> {
    let mut packet = Vec::new();
    gunnar_pktline::write_text(&mut packet, format!("ERR {message}").as_bytes())
        .map_err(|e| anyhow::anyhow!("encoding ERR packet: {e}"))?;
    Ok(packet)
}

///
/// # …and "the peer is already gone" is not a failure at all
///
/// Closing a channel whose peer has already hung up returns
/// `russh::Error::SendError`, and until 2026-08-06 that was logged as
/// `ssh.teardown.failed err=Channel send error` on **every healthy transfer**
/// (MEASURED that day: 9 successful transfers, 6 of these). It is the same
/// fact `handle_session_error` and the undelivered exit status already report,
/// counted a third time in the third set of words with "err" in it. It is
/// separated here by [`peer_hung_up`] so the systemic case this function exists
/// for — descriptor exhaustion, a wedged sender — is still the only thing that
/// says `failed`.
fn note_teardown<E: std::error::Error + 'static>(step: &'static str, result: Result<(), E>) {
    if let Err(err) = result {
        if peer_hung_up(&err) {
            tracing::debug!(
                target: TARGET_SSH,
                event = "ssh.teardown.peer_gone",
                step,
                "the peer had already gone, so this teardown step had nothing to say \
                 it to"
            );
            return;
        }
        tracing::debug!(
            target: TARGET_SSH,
            event = "ssh.teardown.failed",
            step,
            %err,
            "a session teardown step did not complete"
        );
    }
}

/// **Is this failure just "the peer hung up first"?**
///
/// The one classifier for that question (one implementation). Three places used to answer it
/// independently, and all three answered *"no"*, which is why a completely
/// healthy `gunnar ls-remote` left this in the journal:
///
/// ```text
///   DEBUG ssh session ended with an error   error=early eof
///   WARN  the client was gone before the exit status  channel=ChannelId(2) status=0
///   DEBUG ssh.teardown.failed  step="channel-close"  err=Channel send error
///   INFO  transfer finished  status=0  elapsed_ms=106
/// ```
///
/// MEASURED on the reference desktop 2026-08-06 from the live server's own journal, at
/// `10:55:10`, for a transfer that succeeded. Nothing was damaged — `status=0`,
/// `transfer finished` — but three lines carrying `error`, `err` and `failed`
/// per healthy call means `grep -i error` finds only noise, and that is the
/// first command an operator runs.
///
/// # Classified on the cause, never on the message text
///
/// A `contains("early eof")` would be a guard that silently stops matching the
/// day tokio rewords its `read_exact`, and the failure would be invisible
/// because it fails in the *quiet* direction — back to noise, not to a crash.
/// So this walks the source chain and matches concrete things:
/// `io::ErrorKind`, and russh's own named variants. `early eof` is
/// `ErrorKind::UnexpectedEof` (tokio's `read_exact`) and `Channel send error`
/// is `russh::Error::SendError`; neither string appears below.
///
/// # What it deliberately does NOT swallow
///
/// Only the socket going away. A key exchange failure, a MAC failure, a
/// protocol desync, a decryption failure — every one of those is still an error
/// and still logged as one. `tests/teardown_noise.rs` asserts both directions,
/// because a classifier that answered `true` for everything would produce a
/// beautifully quiet journal for a server that was on fire.
fn peer_hung_up(err: &(dyn std::error::Error + 'static)) -> bool {
    use std::io::ErrorKind;

    let mut cause: Option<&(dyn std::error::Error + 'static)> = Some(err);
    while let Some(e) = cause {
        if let Some(io) = e.downcast_ref::<std::io::Error>() {
            if matches!(
                io.kind(),
                ErrorKind::UnexpectedEof
                    | ErrorKind::BrokenPipe
                    | ErrorKind::ConnectionReset
                    | ErrorKind::ConnectionAborted
                    | ErrorKind::NotConnected
            ) {
                return true;
            }
        }
        if let Some(r) = e.downcast_ref::<russh::Error>() {
            if matches!(
                r,
                // The channel's sender is gone: nothing left to send to.
                russh::Error::SendError
                    // russh's own names for a hung-up peer.
                    | russh::Error::HUP
                    | russh::Error::Disconnect
            ) {
                return true;
            }
        }
        cause = e.source();
    }
    false
}

/// Refuse a key that would not re-encode.
///
/// A free function taking only the counters, so the decision this makes (a
/// **rejection**, and a counted one) can be asserted without standing up a
/// russh session. The alternative, an `expect()`, aborted the auth handler.
///
/// `auth_faults` is deliberately **not** moved: that counter means "the
/// identity table could not be read", and this key never reached it. The `warn`
/// [`crate::access::fingerprint_of_key`] emits carries
/// `reason = "key-does-not-re-encode"`, which is what tells the two apart.
fn refuse_unencodable_key(metrics: &gunnar_core::observe::ServerMetrics, user: &str) -> Auth {
    metrics.add_connections_refused(1);
    tracing::warn!(
        target: TARGET_SSH,
        event = "ssh.auth.refused",
        reason = "key-does-not-re-encode",
        decorative_ssh_user = %user,
        "publickey refused: the offered key did not re-encode"
    );
    Auth::reject()
}

/// The one environment variable a git client may send: how it asks for
/// protocol v2 over SSH.
pub const ENV_GIT_PROTOCOL: &str = "GIT_PROTOCOL";

/// **Every** environment variable this server accepts from a client.
///
/// An allowlist, checked by `env_request`. `LD_PRELOAD`, `PATH`, `GIT_DIR` and
/// everything else a client might try is refused with a channel failure, the
/// same answer this server gave to *all* environment requests before
/// [`ENV_GIT_PROTOCOL`] was needed. Nothing here is ever executed, exported to
/// a process or consulted by the grant check; gunnar has no shell and spawns
/// nothing, so the only reason an accepted name exists is that the *wire* reads
/// it.
pub const ENV_ALLOWLIST: [&str; 1] = [ENV_GIT_PROTOCOL];

/// The refused names that mean **nothing at all**, because OpenSSH sends them
/// on its own without being asked.
///
/// Still refused — [`ENV_ALLOWLIST`] is unchanged and this list grants nothing.
/// What it changes is the *level* the refusal is reported at, and the reason it
/// has to exist is a measurement.
///
/// MEASURED on the reference desktop 2026-08-05, an 18-repository ingest over real SSH with stock
/// git 2.53.0 as the client, 203 connections, `RUST_LOG=info`:
///
/// | WARN lines | of which `ssh.env.refused` for `LANG`/`COLORTERM` |
/// |---|---|
/// | 406 | **406 (100%)** |
///
/// Zero warnings of any other kind were emitted in the whole run. OpenSSH's
/// shipped `ssh_config` carries `SendEnv LANG LC_*`, and desktop terminals add
/// `COLORTERM`, so every `git` over `ssh` produces two of these before it has
/// asked for anything. An operator who greps for `WARN` on a live gunnar
/// therefore reads noise at a 100% false-positive rate and will not see the
/// first real one — which is the same disease as a check that cannot go red,
/// wearing the other colour.
///
/// A client naming something NOT on this list is a different event and keeps
/// its `warn`: `LD_PRELOAD`, `PATH`, `GIT_DIR`, `GIT_ALTERNATE_OBJECT_DIRECTORIES`
/// are what an attacker tries, and nothing sends them by accident.
pub const ENV_ROUTINE: [&str; 4] = ["LANG", "LANGUAGE", "COLORTERM", "TERM"];

/// Whether `name` is a routine refusal ([`ENV_ROUTINE`], or any `LC_*`, which
/// OpenSSH sends as a glob).
pub fn env_refusal_is_routine(name: &str) -> bool {
    ENV_ROUTINE.contains(&name) || name.starts_with("LC_")
}

/// The longest environment value the server will hold.
///
/// `version=2` is nine bytes. This is a bound on what one channel can make the
/// server remember, not a syntax check; the grammar is `gunnar-wire`'s.
pub const MAX_ENV_VALUE: usize = 256;

/// Exit status reported when access was refused. Matches what `git` itself
/// reports for a failed remote helper.
pub const EXIT_DENIED: u32 = 128;

/// **What a client is told when the key that opened its session stopped
/// authorising it.**
///
/// One string, named here rather than spelled at the refusal site, so the test
/// that asserts what reaches the client reads the same constant the server
/// sends. It says what happened and what to do, because "denied" on a
/// connection that worked a second ago is otherwise indistinguishable from the
/// server having broken.
///
/// It does NOT distinguish a revoked key from an unreadable identity table, for
/// the reason the whole of this surface merges "no such repository" with "no
/// permission": the difference is operator information, and the log keeps it.
pub const REVOKED_MESSAGE: &str =
    "the key that opened this session no longer authorises it; reconnect";

/// Exit status reported when the wire failed mid-transfer.
pub const EXIT_TRANSPORT_FAILED: u32 = 1;

/// Exit status reported when the admission gate refused a transfer.
///
/// `EX_TEMPFAIL` from `sysexits.h`, and deliberately **not** [`EXIT_DENIED`]:
/// "try again" and "you may not" are different answers, and a retry script that
/// cannot tell them apart either hammers a permission error forever or gives up
/// on a transient one.
pub const EXIT_BUSY: u32 = 75;

/// What a client is told while the server's stores are still opening.
///
/// **The word `starting` is load-bearing** and is asserted by
/// `tests/readiness.rs`: it is what separates this from
/// `gunnar_wire::Error::Store`'s `the repository is unavailable; try again`,
/// which is what a client hears when a repository is genuinely broken. The
/// power-cut door read the second sentence during the first condition and
/// reported 64 acked pushes missing that were all present a second later. A
/// harness, a CI script and a person all have to be able to tell "not yet" from
/// "gone", and the only place that distinction can be made is here, in the
/// words.
///
/// It carries [`EXIT_BUSY`], not [`EXIT_DENIED`], for the same reason: 75 means
/// retry.
pub const STARTING_MESSAGE: &str =
    "the server is starting: its stores are still opening after a restart, and it will serve \
     this in a moment; retry";

#[cfg(test)]
mod tests {
    use super::*;

    /// `gunnar-recipient-keys`: the path through the one sandbox (git's
    /// quoting taken off), the member through the principal id rule, and
    /// nothing else on the line; another verb is not this one.
    #[test]
    fn the_recipient_keys_line_takes_a_sandboxed_path_and_an_account_name() {
        let ok = |line: &str| {
            parse_recipient_keys_verb(line.as_bytes())
                .expect("this verb")
                .expect("well formed")
        };
        let (repo, member) = ok("gunnar-recipient-keys 'team/vault.git' 'bob'");
        assert_eq!(
            (repo.namespace(), repo.name(), member.as_str()),
            ("team", "vault", "bob")
        );
        let (repo, member) = ok("gunnar-recipient-keys team/vault bob");
        assert_eq!(
            (repo.namespace(), repo.name(), member.as_str()),
            ("team", "vault", "bob")
        );
        let (repo, _) = ok("gunnar-recipient-keys '/team/vault.git' bob");
        assert_eq!(repo.name(), "vault");
        assert!(parse_recipient_keys_verb(b"git-upload-pack 'team/vault.git'").is_none());
        assert!(parse_recipient_keys_verb(b"gunnar-recipient-keysx a/b c").is_none());
        for bad in [
            "gunnar-recipient-keys",
            "gunnar-recipient-keys 'team/vault.git'",
            "gunnar-recipient-keys '../etc/passwd' bob",
            "gunnar-recipient-keys team/vault 'bo b'",
            "gunnar-recipient-keys team/vault anonymous",
            "gunnar-recipient-keys team/vault 'bob",
            "gunnar-recipient-keys team/vault bob\u{1}",
        ] {
            assert!(
                matches!(parse_recipient_keys_verb(bad.as_bytes()), Some(Err(_))),
                "{bad:?} is refused"
            );
        }
        let long = format!("gunnar-recipient-keys team/vault {}", "b".repeat(600));
        assert!(matches!(
            parse_recipient_keys_verb(long.as_bytes()),
            Some(Err(_))
        ));
    }

    /// **The door offers the post-quantum hybrid first**, from gunnar's own
    /// list: `mlkem768x25519-sha256`, then the classical fallbacks, and the
    /// strict-kex markers (Terrapin) kept.
    #[test]
    fn the_door_offers_mlkem768x25519_first_from_gunnars_own_list() {
        let p = preferred();
        assert!(
            std::ptr::eq(p.kex.as_ref().as_ptr(), KEX_ORDER.as_ptr()),
            "the server's list is gunnar's KEX_ORDER, not whatever russh defaults to"
        );
        let names: Vec<&str> = p.kex.iter().map(|n| n.as_ref()).collect();
        assert_eq!(
            names[..3],
            [
                "mlkem768x25519-sha256",
                "curve25519-sha256",
                "curve25519-sha256@libssh.org"
            ],
            "{names:?}"
        );
        assert!(
            names.contains(&"kex-strict-s-v00@openssh.com"),
            "strict kex stays: {names:?}"
        );
        assert_eq!(
            p.cipher,
            russh::Preferred::DEFAULT.cipher,
            "only the key exchange is gunnar's own"
        );
    }

    /// The accept loop outlives a descriptor shortage and a client that hung
    /// up mid-handshake; it returns on a listener that is gone.
    #[test]
    fn transient_accept_errors_are_named() {
        for errno in [24, 23, 105, 12, 103] {
            assert!(
                accept_error_is_transient(&std::io::Error::from_raw_os_error(errno)),
                "errno {errno}"
            );
        }
        for errno in [9, 22, 88] {
            assert!(
                !accept_error_is_transient(&std::io::Error::from_raw_os_error(errno)),
                "errno {errno}"
            );
        }
        assert!(!accept_error_is_transient(&std::io::Error::other("closed")));
    }

    /// A teardown that fails is **reported**, and one that works is
    /// silent.
    ///
    /// Driven through the helper rather than through a live channel: making
    /// `Channel::close()` fail on demand means a peer that has vanished at an
    /// exact moment, which is a race to arrange and a race to assert on. What
    /// the helper decides is the whole of what changed, `let _ =` against a
    /// line in the journal, and both directions are asserted, because a
    /// `note_teardown` that logged unconditionally would put a line in an
    /// operator's journal on every clean disconnect and make the signal
    /// useless.
    /// Capture every `tracing` line `emit` produces, rendered as
    /// `<target> <field>=<value> …`.
    ///
    /// One capture for every test in this module that reads the journal (one
    /// implementation): there were two copies of this layer forty lines apart, and a second
    /// copy of a capture is a second chance to filter out the line under test.
    fn captured_lines(emit: impl FnOnce()) -> Vec<String> {
        use std::sync::{Arc, Mutex};
        use tracing_subscriber::layer::SubscriberExt;
        use tracing_subscriber::Layer as _;

        #[derive(Clone, Default)]
        struct Capture(Arc<Mutex<Vec<String>>>);
        struct CaptureLayer(Capture);
        impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CaptureLayer {
            fn on_event(
                &self,
                event: &tracing::Event<'_>,
                _: tracing_subscriber::layer::Context<'_, S>,
            ) {
                struct Render(String);
                impl tracing::field::Visit for Render {
                    fn record_debug(
                        &mut self,
                        field: &tracing::field::Field,
                        value: &dyn std::fmt::Debug,
                    ) {
                        self.0.push_str(&format!("{}={value:?} ", field.name()));
                    }
                }
                let mut render = Render(String::new());
                event.record(&mut render);
                self.0 .0.lock().expect("capture").push(format!(
                    "{} {}",
                    event.metadata().target(),
                    render.0
                ));
            }
        }

        let capture = Capture::default();
        let subscriber = tracing_subscriber::registry().with(
            CaptureLayer(capture.clone())
                .with_filter(tracing_subscriber::filter::LevelFilter::TRACE),
        );
        tracing::subscriber::with_default(subscriber, emit);
        let lines = capture.0.lock().expect("capture").clone();
        lines
    }

    #[test]
    fn a_failed_teardown_step_is_reported_and_a_clean_one_is_not() {
        let lines = captured_lines(|| {
            note_teardown::<std::io::Error>("channel-close", Ok(()));
            note_teardown(
                "watchdog-disconnect",
                Err(std::io::Error::other("the peer was already gone")),
            );
        });

        assert_eq!(
            lines.len(),
            1,
            "expected exactly one line — the failure — got {lines:?}"
        );
        assert!(
            lines[0].starts_with(gunnar_core::observe::TARGET_SSH),
            "the line is not on the ssh target: {}",
            lines[0]
        );
        assert!(
            lines[0].contains("watchdog-disconnect") && lines[0].contains("already gone"),
            "the line names neither the step nor the error: {}",
            lines[0]
        );
    }

    /// **RED-WHEN-BROKEN: a real failure is still reported as one.**
    ///
    /// The dangerous direction of [`peer_hung_up`] is not missing a hung-up
    /// peer — that only restores the noise. It is answering `true` for
    /// everything, which gives a beautifully quiet journal for a server that is
    /// on fire, and does it in a way no other test would notice.
    ///
    /// So every case is asserted in both directions, against **rendered log
    /// output** rather than against the predicate, because the predicate is
    /// private and the log line is what an operator has.
    #[test]
    fn a_real_failure_is_still_reported_as_one() {
        // Benign: the peer went away. Each of these is a shape MEASURED in the
        // live journal on 2026-08-06 for a transfer that succeeded.
        let benign: Vec<(&str, Box<dyn std::error::Error + 'static>)> = vec![
            // `err=Channel send error` — the channel's sender is gone.
            ("russh SendError", Box::new(russh::Error::SendError)),
            ("russh HUP", Box::new(russh::Error::HUP)),
            ("russh Disconnect", Box::new(russh::Error::Disconnect)),
            // `error=early eof` — tokio's `read_exact` on a closed socket.
            (
                "early eof",
                Box::new(std::io::Error::new(
                    std::io::ErrorKind::UnexpectedEof,
                    "early eof",
                )),
            ),
            (
                "broken pipe",
                Box::new(std::io::Error::from(std::io::ErrorKind::BrokenPipe)),
            ),
            (
                "connection reset",
                Box::new(std::io::Error::from(std::io::ErrorKind::ConnectionReset)),
            ),
        ];
        for (what, err) in benign {
            let lines = captured_lines(|| note_teardown("channel-close", Err(BoxedError(err))));
            assert_eq!(lines.len(), 1, "{what}: expected one line, got {lines:?}");
            let line = lines[0].to_ascii_lowercase();
            assert!(
                line.contains("peer_gone"),
                "{what} was not classified as a hung-up peer: {}",
                lines[0]
            );
            assert!(
                !line.contains("error") && !line.contains("err=") && !line.contains("fail"),
                "{what} is benign but its line is still found by `grep -i error`: {}",
                lines[0]
            );
        }

        // NOT benign: the protocol itself went wrong. These must keep saying so
        // — this is the half that stops the classifier becoming a blanket yes.
        let real: Vec<(&str, Box<dyn std::error::Error + 'static>)> = vec![
            ("key exchange", Box::new(russh::Error::Kex)),
            ("packet auth", Box::new(russh::Error::PacketAuth)),
            ("decryption", Box::new(russh::Error::DecryptionError)),
            ("inconsistent state", Box::new(russh::Error::Inconsistent)),
            ("no common algo", Box::new(russh::Error::WrongChannel)),
            (
                "permission denied on disk",
                Box::new(std::io::Error::from(std::io::ErrorKind::PermissionDenied)),
            ),
            (
                "out of descriptors",
                Box::new(std::io::Error::from(std::io::ErrorKind::Other)),
            ),
        ];
        for (what, err) in real {
            let lines = captured_lines(|| note_teardown("channel-close", Err(BoxedError(err))));
            assert_eq!(lines.len(), 1, "{what}: expected one line, got {lines:?}");
            assert!(
                lines[0].contains("ssh.teardown.failed"),
                "{what} was swallowed as benign, so the classifier is a blanket yes: {}",
                lines[0]
            );
            assert!(
                lines[0].contains("err="),
                "{what} lost the error itself, so the line names no cause: {}",
                lines[0]
            );
        }
    }

    /// A boxed error that is still `std::error::Error`, so the table above can
    /// hold two concrete types in one `Vec` and still reach the classifier's
    /// `downcast_ref` through `source()`.
    #[derive(Debug)]
    struct BoxedError(Box<dyn std::error::Error + 'static>);

    impl std::fmt::Display for BoxedError {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            self.0.fmt(f)
        }
    }

    impl std::error::Error for BoxedError {
        fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
            Some(self.0.as_ref())
        }
    }

    /// The session-level half of the same rule: an ended session says nothing
    /// an operator greps for when the client simply hung up, and still says
    /// `error` when it did not.
    ///
    /// Asserted through the real `Handler` method rather than the predicate,
    /// because the method is what russh calls.
    #[test]
    fn a_hung_up_session_is_not_logged_as_an_error() {
        let quiet = captured_lines(|| {
            note_session_end(&anyhow::Error::new(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                "early eof",
            )));
        });
        assert_eq!(quiet.len(), 1, "expected one line, got {quiet:?}");
        let line = quiet[0].to_ascii_lowercase();
        assert!(
            line.contains("peer-hung-up") && !line.contains("error"),
            "a client hanging up is still logged as an error: {}",
            quiet[0]
        );

        let loud = captured_lines(|| {
            note_session_end(&anyhow::Error::new(russh::Error::Kex));
        });
        assert_eq!(loud.len(), 1, "expected one line, got {loud:?}");
        assert!(
            loud[0].to_ascii_lowercase().contains("error"),
            "a key-exchange failure was swallowed as a hung-up peer: {}",
            loud[0]
        );
    }

    /// An unencodable key is **refused and counted**, never accepted.
    ///
    /// The assertion is on the applied output, the `Auth` value that goes back
    /// to russh and the counter an operator reads, not on "it did not panic".
    /// A body that returned `Auth::Accept` would let a key nobody could
    /// fingerprint onto the server, which is the one outcome worse than the
    /// panic this replaced.
    #[test]
    fn a_key_that_does_not_re_encode_is_rejected_and_counted() {
        let metrics = gunnar_core::observe::ServerMetrics::default();
        let auth = refuse_unencodable_key(&metrics, "decorative");

        assert!(
            matches!(auth, Auth::Reject { .. }),
            "an unencodable key must be rejected, got {auth:?}"
        );
        let snapshot = metrics.snapshot();
        assert_eq!(
            snapshot.connections_refused, 1,
            "the refusal must be visible to an operator"
        );
        // Not an identity-table fault, and the counters must not blur the two.
        assert_eq!(snapshot.auth_faults, 0);
        assert_eq!(snapshot.connections_accepted, 0);
    }
}

/// Pure: whether an exec line is git-lfs asking SSH (`git-lfs-authenticate …`,
/// `git-lfs-transfer …`) — refused, and said at INFO.
pub fn is_lfs_over_ssh(command: &str) -> bool {
    let first = command.split_whitespace().next().unwrap_or("");
    // The transfer verb is named through the dispatch table's own LFS row, never spelled here
    // (`command_parsing::the_dispatch_table_is_the_only_place_a_verb_is_named`).
    // `git-lfs-authenticate` is no verb gunnar dispatches: it is git-lfs's own probe.
    first == "git-lfs-authenticate"
        || crate::command::SERVICES.iter().any(|spec| {
            spec.verb == first && matches!(spec.service, crate::command::GitService::LfsTransfer(_))
        })
}

#[cfg(test)]
mod lfs_over_ssh_tests {
    /// ★ git-lfs's SSH probe is a client's known step, not a defect (RUN A 2026-10-08).
    #[test]
    fn git_lfs_asking_ssh_is_known_and_anything_else_is_not() {
        assert!(super::is_lfs_over_ssh(
            "git-lfs-authenticate /cbc/vault upload"
        ));
        assert!(super::is_lfs_over_ssh(
            "git-lfs-transfer 'acme/site' download"
        ));
        for other in [
            "bash",
            "scp -t /etc",
            "",
            "git-lfs-authenticatex /a upload",
            "echo git-lfs-authenticate",
        ] {
            assert!(!super::is_lfs_over_ssh(other), "{other:?}");
        }
    }
}