//! The consumer's TCP front door (feature `local`, ADR-005/ADR-014;
//! consumer.md §The Front Doors): the ssh `-D` local exposure — a real
//! `socks5://127.0.0.1:port` endpoint vanilla clients (curl, browsers,
//! tun2proxy) point at. The door is the vanilla↔ACL boundary; the RFC
//! codec lives only here (ADR-003's "codec at front doors only").
//!
//! Two door variants (ADR-005), selected by constructor config:
//!
//! - **The NoAuth door** (POC #1's raw-pipe shape): per accepted
//!   connection, a literal raw pipe onto a channel `BiStream`
//!   (`copy_bidirectional`, **zero protocol translation**). The
//!   producer runs the whole RFC conversation through the pipe — the
//!   door adds nothing. The loopback-only / wrapper-aware shape.
//! - **The cred-bearing door** (RFC 1929): the door terminates the
//!   local RFC conversation itself — the forked
//!   `accept_password_auth` shape (`start` →
//!   `negotiate_auth(&[PasswordAuthentication])` →
//!   `read_username_password` → check → `accept`/`reject`) — then
//!   maps the accepted credential through the **static config table**
//!   (`user:pass →` channel identity, ADR-005) and **re-originates
//!   the SOCKS conversation in-band** via the token-carrying consumer
//!   sessions (`Socks5ChannelSession::connect_with_token` /
//!   `BindSession::bind_with_token`), re-encoding the in-band replies
//!   to the vanilla client and piping the post-auth data-plane bytes
//!   raw. **The door never pipes raw post-auth**: the vanilla
//!   client's command bytes were consumed by the door's own RFC 1929
//!   half; the in-band path re-speaks them from the parsed target.
//!   The SOCKS password never crosses the channel as a SOCKS
//!   password; the channel identity never reaches the vanilla client.
//!
//! **The identity mechanism** (the CF-005 seam; decided at
//! implementation, recorded in the task Notes): the credential→identity
//! mapping yields an alkcall [`Identity`] (scopes included); the
//! per-credential identity rides the open-op call's `auth_token`
//! payload (CF-005 precedence (1): payload `auth_token` →
//! `identity_provider` resolution — alkcall `dispatch::
//! resolve_identity`, ADR-017 §7), never a per-credential
//! `ChannelClient`. The token is the mapped identity's id — the
//! resolution seam's vocabulary; the password itself never rides it.
//! The producer resolves it through its `ServingConfig
//! .identity_provider` and the ACL gate reads the resolved scopes
//! exactly as for any per-call opener identity (CF-005/CF-006). The
//! door checks credentials itself — no forked `AuthMethod` impl
//! (ADR-005).
//!
//! **The accept path owns the idle defense** (ADR-014): an accepted
//! connection that sends no first byte within the configured window
//! is closed door-side (an accepted socket is not a channel — the cap
//! regime does not bound it); plus the max-concurrent-sessions cap
//! (an accept beyond the bound is refused door-side, no channel
//! opened). The two v1 knobs are constructor config.
//!
//! **The BIND door**: the same listener shape; a vanilla BIND
//! conversation rides [`BindSession`] (reply#1's advertised address
//! passes through to the vanilla client unchanged — the conversation
//! is the client's own; the door adds nothing, consumer.md §The BIND
//! door's reply#2 fidelity). The BIND door rides the same door
//! variant split (NoAuth raw-pipe vs cred-bearing re-origination).
//!
//! Wiring (consumer.md §Wiring): `FrontDoor::builder()
//! .credentials(cred_table).bind("127.0.0.1:1080").build()` →
//! `serve(channel_client)` — the NoAuth door omits `.credentials`.
//! `serve` runs the accept loop; per-connection failures are logged +
//! isolated (the loop continues — the alktunnels per-accept isolation
//! posture).

use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;

use alkcall::channels::client::ChannelClient;
use alkcall::core::auth::Identity;
use tokio::io::AsyncWriteExt as _;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::Semaphore;

use crate::consumer::{TakenInner, TokenClient};
use crate::rfc1928::consts;
use crate::rfc1928::server::{new_reply, states, Socks5ServerProtocol, SocksServerError};
use crate::rfc1928::util::target_addr::TargetAddr;
use crate::rfc1928::{ReplyError, Socks5Command, SocksError};

/// One static credential-table entry (ADR-005): `user:pass →` channel
/// identity. The table is the door's credential→identity mapping —
/// static, declarative, assembly-layer.
#[derive(Clone, Debug)]
pub struct DoorCredential {
    /// The RFC 1929 username the vanilla client presents.
    pub username: String,
    /// The RFC 1929 password the vanilla client presents.
    pub password: String,
    /// The channel identity the accepted credential maps to (scopes
    /// included — the open gate and the BIND command gate read them
    /// producer-side). The identity's `id` doubles as the open-op
    /// `auth_token` (the resolution seam's vocabulary — the password
    /// never rides it).
    pub identity: Identity,
}

/// The credential→identity static config table (ADR-005): exact
/// `username`/`password` lookup. A miss rejects the RFC 1929
/// conversation door-side (no channel opened — the ACL boundary held;
/// the vanilla client got the RFC 1929 rejection).
#[derive(Clone, Debug, Default)]
pub struct CredentialTable {
    entries: HashMap<(String, String), Identity>,
}

impl CredentialTable {
    /// Build the table from [`DoorCredential`] entries.
    pub fn new(entries: impl IntoIterator<Item = DoorCredential>) -> Self {
        Self {
            entries: entries
                .into_iter()
                .map(|c| ((c.username, c.password), c.identity))
                .collect(),
        }
    }

    /// The exact-match lookup: `(username, password) → Option<identity>`.
    pub fn lookup(&self, username: &str, password: &str) -> Option<&Identity> {
        self.entries
            .get(&(username.to_string(), password.to_string()))
    }

    /// The number of entries (observability).
    pub fn len(&self) -> usize {
        self.entries.len()
    }

    /// Whether the table is empty (an empty cred-bearing table
    /// rejects every conversation door-side).
    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }
}

/// Why one accepted connection's serving task ended (the door's
/// per-connection teardown diagnostic — logged + isolated).
#[derive(Debug)]
enum DoorExit {
    /// The vanilla client went away (clean end, both variants).
    ClientGone,
    /// The data plane (local pipe ↔ in-band session) ended.
    PlaneEnded,
    /// The local RFC half failed (a rejected credential, a malformed
    /// conversation, protocol garbage) — the vanilla client's fault.
    LocalRfc(SocksError),
    /// The in-band open failed (the producer refused the mapped
    /// identity, the target dial refused, the channel open errored).
    Open(String),
    /// A plain I/O failure on the local socket.
    Io(std::io::Error),
}

impl std::fmt::Display for DoorExit {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DoorExit::ClientGone => f.write_str("client gone"),
            DoorExit::PlaneEnded => f.write_str("data plane ended"),
            DoorExit::LocalRfc(err) => write!(f, "local rfc: {err}"),
            DoorExit::Open(msg) => write!(f, "open failed: {msg}"),
            DoorExit::Io(err) => write!(f, "io: {err}"),
        }
    }
}

impl From<SocksServerError> for DoorExit {
    fn from(err: SocksServerError) -> Self {
        DoorExit::LocalRfc(SocksError::ServerError(err))
    }
}

/// The TCP front door (feature `local`): the vanilla↔ACL boundary.
/// Variant selection is constructor config (ADR-005): credentials
/// configured = the cred-bearing door (RFC 1929 terminates locally,
/// re-origination in-band); credentials absent = the NoAuth door (the
/// literal raw pipe). The accept path's defense knobs (ADR-014) are
/// constructor config: the idle-first-byte window and the
/// max-concurrent-sessions cap.
pub struct FrontDoor {
    bind_addr: SocketAddr,
    credentials: Option<CredentialTable>,
    idle_first_byte: Duration,
    max_sessions: usize,
}

impl FrontDoor {
    /// The builder (consumer.md §Wiring).
    pub fn builder() -> FrontDoorBuilder {
        FrontDoorBuilder::default()
    }
}

/// The builder for [`FrontDoor`] (consumer.md §Wiring: `builder()
/// .credentials(cred_table).bind("127.0.0.1:1080").build()`). The
/// NoAuth door omits `.credentials`.
#[derive(Default)]
pub struct FrontDoorBuilder {
    bind_addr: Option<SocketAddr>,
    credentials: Option<CredentialTable>,
    idle_first_byte: Option<Duration>,
    max_sessions: Option<usize>,
}

impl FrontDoorBuilder {
    /// The credential→identity static config table (ADR-005).
    /// Configuring it selects the cred-bearing door (RFC 1929);
    /// omitting it selects the NoAuth door (the raw-pipe shape).
    pub fn credentials(mut self, table: CredentialTable) -> Self {
        self.credentials = Some(table);
        self
    }

    /// The local bind address — the door's explicit bind decision
    /// (the never-binds posture: the door is the assembly layer's
    /// spelled-out choice; loopback is the documented posture).
    pub fn bind(mut self, addr: impl Into<SocketAddr>) -> Self {
        self.bind_addr = Some(addr.into());
        self
    }

    /// The idle-first-byte window (ADR-014's accept-path defense
    /// knob 1): an accepted connection that sends no first byte
    /// within this window is closed door-side. The window covers the
    /// first byte of the local conversation — the NoAuth door's
    /// greeting bytes (read producer-side through the pipe) and the
    /// cred-bearing door's greeting (read by the door's own RFC 1929
    /// half).
    pub fn idle_first_byte(mut self, window: Duration) -> Self {
        self.idle_first_byte = Some(window);
        self
    }

    /// The max-concurrent-sessions cap (ADR-014's accept-path defense
    /// knob 2): an accept beyond the bound is refused door-side — the
    /// socket is closed immediately, no channel opened.
    pub fn max_sessions(mut self, cap: usize) -> Self {
        self.max_sessions = Some(cap);
        self
    }

    pub fn build(self) -> FrontDoor {
        FrontDoor {
            bind_addr: self.bind_addr.expect("the door binds an address"),
            credentials: self.credentials,
            idle_first_byte: self.idle_first_byte.unwrap_or(Duration::from_secs(30)),
            max_sessions: self.max_sessions.unwrap_or(256),
        }
    }
}

impl FrontDoor {
    /// Bind the configured address and run the accept loop
    /// (consumer.md §Wiring: `door.serve(channel_client)`). Per-
    /// connection failures are logged + isolated — the loop continues
    /// (the alktunnels per-accept isolation posture). Runs until the
    /// listener fails fatally; the caller owns the task's lifetime.
    pub async fn serve(self, client: Arc<ChannelClient>) -> std::io::Result<()> {
        let listener = TcpListener::bind(self.bind_addr).await?;
        tracing::info!(bind = %self.bind_addr, "front door: listening");
        self.serve_on(listener, client).await
    }

    /// Run the accept loop over an existing listener (the
    /// test/assembly seam — the same accept loop without the bind).
    pub async fn serve_on(
        self,
        listener: TcpListener,
        client: Arc<ChannelClient>,
    ) -> std::io::Result<()> {
        let credentials = self.credentials;
        let idle = self.idle_first_byte;
        let live = Arc::new(Semaphore::new(self.max_sessions));

        // The cred-bearing door's channel surface: the client's call
        // connection is taken once for the door's lifetime (one
        // transport, one call connection, many per-credential
        // identities riding the open-op `auth_token` payloads). The
        // NoAuth door keeps the client's own calling surface
        // (`open_channel` — the raw-pipe shape never carries a
        // token).
        let token_client = if credentials.is_some() {
            Some(Arc::new(TokenClient::wrap(&client).await))
        } else {
            None
        };

        loop {
            let (stream, peer) = match listener.accept().await {
                Ok(accepted) => accepted,
                Err(err) => {
                    tracing::warn!(error = %err, "front door: accept failed");
                    continue;
                }
            };

            // The session cap (ADR-014 knob 2): refuse beyond the
            // bound — `try_acquire` keeps the accept loop free (a
            // permit returns when a serving task ends and drops its
            // guard).
            let Ok(permit) = live.clone().try_acquire_owned() else {
                tracing::warn!(peer = %peer, "front door: session cap reached — refusing");
                drop(stream);
                continue;
            };

            let client = Arc::clone(&client);
            let per_accept_credentials = credentials.clone();
            let per_accept_tokens = token_client.clone();
            tokio::spawn(async move {
                let _permit = permit;
                let exit = serve_connection(
                    stream,
                    peer,
                    &client,
                    per_accept_credentials.as_ref(),
                    per_accept_tokens.as_ref(),
                    idle,
                )
                .await;
                tracing::debug!(peer = %peer, "front door: connection ended: {exit}");
            });
        }
    }
}

/// Serve one accepted connection through the door variant split: the
/// idle-first-byte gate first (ADR-014 knob 1 — the door's accept
/// path owns the defense; an accepted socket is not a channel), then
/// the variant.
async fn serve_connection(
    stream: TcpStream,
    peer: SocketAddr,
    client: &Arc<ChannelClient>,
    credentials: Option<&CredentialTable>,
    token_client: Option<&Arc<TokenClient>>,
    idle: Duration,
) -> DoorExit {
    // The idle-first-byte defense: an accepted connection that sends
    // no first byte within the window is closed door-side (ADR-014's
    // slowloris sliver). The peek waits for readability without
    // consuming — the variant's own protocol read picks the byte up.
    let mut first = [0u8; 1];
    match tokio::time::timeout(idle, stream.peek(&mut first)).await {
        Ok(Ok(_)) => {}
        Ok(Err(err)) => return DoorExit::Io(err),
        Err(_) => {
            tracing::debug!(peer = %peer, "front door: idle-first-byte timeout");
            return DoorExit::ClientGone;
        }
    }

    match (credentials, token_client) {
        // The NoAuth door: the literal raw pipe (POC #1's shape —
        // zero protocol translation).
        (None, _) => raw_pipe(stream, client).await,
        // The cred-bearing door: RFC 1929 terminates locally,
        // re-origination rides the token sessions.
        (Some(table), Some(tokens)) => cred_bearing(stream, client, table, tokens).await,
        // Unreachable by construction (`serve_on` builds the pair) —
        // the arm exists to keep the match exhaustive without a
        // panic path (convention 2).
        (Some(_), None) => {
            DoorExit::Open("the cred-bearing door's token surface was not built".to_string())
        }
    }
}

/// The NoAuth door's raw pipe (POC #1's `front_door_raw_pipe_vanilla_
/// client` shape): open a channel, then `copy_bidirectional` the
/// accepted socket onto the channel `BiStream` — the producer runs
/// the whole RFC conversation through the pipe; the door adds
/// nothing.
async fn raw_pipe(mut stream: TcpStream, client: &Arc<ChannelClient>) -> DoorExit {
    let (_channel_id, send, recv) = match client
        .open_channel(
            crate::params::OP_SOCKS5_OPEN,
            crate::params::Socks5Params::as_json(),
            crate::params::SOCKS5_ALPN,
        )
        .await
    {
        Ok(open) => open,
        Err(err) => return DoorExit::Open(err.to_string()),
    };
    let mut bidi = alkcall::core::types::BiStream::from_joined(recv, send);
    let _ = tokio::io::copy_bidirectional(&mut stream, &mut bidi).await;
    DoorExit::PlaneEnded
}

/// The cred-bearing door (ADR-005): terminate the vanilla RFC 1929
/// conversation's auth locally (the forked surface), map the accepted
/// credential through the static table, re-originate the SOCKS
/// conversation in-band via the token-carrying consumer sessions,
/// re-encode the in-band replies to the vanilla client, and pipe the
/// post-auth data-plane bytes raw.
///
/// The door never pipes raw post-auth: the vanilla client's command
/// bytes were consumed by the door's own RFC half; the in-band path
/// re-speaks them from the parsed target. The password never crosses
/// the channel as a SOCKS password (the in-band handshake is NoAuth —
/// the mapped identity is the authorization); the channel identity
/// never reaches the vanilla client.
async fn cred_bearing(
    mut stream: TcpStream,
    _client: &Arc<ChannelClient>,
    table: &CredentialTable,
    token_client: &Arc<TokenClient>,
) -> DoorExit {
    // The local RFC 1929 half (the forked surface) + the command read.
    let outcome = local_rfc1929_half(&mut stream, table).await;
    let (cmd, target, identity) = match outcome {
        Ok(triple) => triple,
        Err(exit) => return exit,
    };

    // The in-band re-origination: the per-credential identity rides
    // the open-op call's `auth_token` payload (CF-005 precedence
    // (1)). The token is the mapped identity's id — the resolution
    // seam's vocabulary; the password itself never rides it.
    let token = identity.id.as_bytes().to_vec();

    match cmd {
        Socks5Command::TCPConnect => {
            let session = match crate::consumer::Socks5ChannelSession::connect_with_token(
                token_client,
                target,
                token,
            )
            .await
            {
                Ok(session) => session,
                Err(err) => return DoorExit::Open(err.to_string()),
            };
            pipe_connect_plane(stream, session).await
        }
        Socks5Command::TCPBind => {
            let session =
                match crate::consumer::BindSession::bind_with_token(token_client, target, token)
                    .await
                {
                    Ok(session) => session,
                    Err(err) => return DoorExit::Open(err.to_string()),
                };
            pipe_bind_plane(stream, session).await
        }
        // UDP ASSOCIATE through the TCP door is the UDP door's shape
        // (the datagram plane's RFC codec boundary — ADR-003), not
        // this one's: refuse RFC-faithfully — the vanilla client gets
        // the refused reply through the door's own codec (no phantom
        // data plane, no in-band session opened).
        Socks5Command::UDPAssociate => {
            let reply = new_reply(
                &ReplyError::CommandNotSupported,
                SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
            );
            let _ = stream.write_all(&reply).await;
            let _ = stream.flush().await;
            DoorExit::LocalRfc(SocksError::ReplyError(ReplyError::CommandNotSupported))
        }
    }
}

/// The door's local RFC 1929 half (the forked surface — ADR-005's
/// accept_password_auth shape): `negotiate_auth(&[
/// PasswordAuthentication])` → `read_username_password` → check
/// through the static table → `accept`/`reject`, then
/// `read_command` for the re-origination's parsed target. The check
/// rides the forked `CheckResult`'s `bool` impl; the mapped identity
/// is recorded by the same table lookup the check ran (one table,
/// one truth) and re-yielded to the caller.
async fn local_rfc1929_half(
    stream: &mut TcpStream,
    table: &CredentialTable,
) -> Result<(Socks5Command, TargetAddr, Identity), DoorExit> {
    // The check closure is sync (`FnMut(String, String) -> bool` over
    // the forked `CheckResult`'s `bool` impl); the mapped identity is
    // recorded through a sync witness (a `std` mutex — the closure
    // runs once, synchronously).
    let found: Arc<std::sync::Mutex<Option<Identity>>> = Arc::new(std::sync::Mutex::new(None));
    let found_for_check = Arc::clone(&found);

    let (proto, _ok) = Socks5ServerProtocol::<_, states::Authenticated>::accept_password_auth(
        stream,
        |user, pass| {
            let mapped = table.lookup(&user, &pass).cloned();
            // A poisoned lock would mean a panic mid-check — the
            // `unwrap_or_else(into_inner)` convention keeps the door
            // alive (AGENTS.md convention 2).
            let mut slot = found_for_check.lock().unwrap_or_else(|e| e.into_inner());
            let accepted = mapped.is_some();
            *slot = mapped;
            // The check IS the table lookup: a miss rejects (the
            // forked `accept_password_auth` writes the RFC 1929
            // rejection when the checker returns `false`).
            accepted
        },
    )
    .await
    .map_err(DoorExit::from)?;

    let identity = found
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .take()
        .ok_or_else(|| DoorExit::from(SocksServerError::AuthenticationRejected))?;

    let (_proto, cmd, target) = proto.read_command().await.map_err(DoorExit::from)?;
    Ok((cmd, target, identity))
}

/// Pipe the re-originated CONNECT session: the vanilla reply (the
/// in-band reply's BND.ADDR re-encoded — the RFC codec at the door
/// boundary only, ADR-003), then the data plane ↔ the accepted socket.
/// The door already consumed the vanilla client's greeting + command
/// bytes; the post-auth bytes ride the session's data plane raw from
/// here. The pump shape is the taken inner stream
/// (`Socks5Stream<BiStream>` is itself `AsyncRead + AsyncWrite` — the
/// forked client type's intended use) against the local socket via
/// `copy_bidirectional`; the session stays the channel-teardown owner
/// (the `into_inner` pairing; dropped at the end — the EOF sentinel
/// reaches the producer's pump).
async fn pipe_connect_plane(
    mut stream: TcpStream,
    session: crate::consumer::Socks5ChannelSession,
) -> DoorExit {
    // The vanilla CONNECT reply: the in-band reply's BND.ADDR
    // re-encoded (the vanilla client is waiting for it — the door's
    // RFC half consumed its local bytes; the in-band conversation is
    // fresh and re-originated). The sentinel address is the
    // RFC-compatibility surface the producer already chose; the door
    // adds nothing.
    let reply = match session.reply_addr() {
        Some(TargetAddr::Ip(sa)) => new_reply(&ReplyError::Succeeded, sa),
        Some(TargetAddr::Domain(_, port)) => new_reply(
            &ReplyError::Succeeded,
            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port),
        ),
        None => {
            session.close().await;
            return DoorExit::PlaneEnded;
        }
    };
    if stream.write_all(&reply).await.is_err() {
        session.close().await;
        return DoorExit::ClientGone;
    }
    let _ = stream.flush().await;

    let Ok(TakenInner { session, inner }) = session.into_inner() else {
        return DoorExit::PlaneEnded;
    };
    let _ = tokio::io::copy_bidirectional(&mut stream, &mut { inner }).await;
    let _ = session;
    DoorExit::PlaneEnded
}

/// Pipe the re-originated BIND session (the BIND door, cred-bearing
/// variant): reply#1's advertised address is re-encoded to the
/// vanilla client from the session's captured `bind_addr` (the
/// conversation is the vanilla client's own — the door's RFC half
/// consumed the local greeting + command bytes, and the in-band
/// session's reply#1 address passes through re-encoded; consumer.md
/// §The BIND door's reply#2 fidelity). The quiet wait rides
/// `await_inbound` (the take-guarded reply#2 reader); reply#2's peer
/// is likewise re-encoded; the data plane passes through raw. The
/// session stays the channel-teardown owner.
async fn pipe_bind_plane(
    mut stream: TcpStream,
    mut session: crate::consumer::BindSession,
) -> DoorExit {
    // Reply#1 (the vanilla form): the in-band reply's BND.ADDR
    // re-encoded — the advertised address passes through unchanged.
    let bind_addr = session.reply1().map(|reply1| reply1.bind_addr.clone());
    let Some(bind_addr) = bind_addr else {
        session.close().await;
        return DoorExit::PlaneEnded;
    };
    let wire = match &bind_addr {
        TargetAddr::Ip(sa) => new_reply(&ReplyError::Succeeded, *sa),
        TargetAddr::Domain(name, port) => {
            let mut wire = vec![
                consts::SOCKS5_VERSION,
                ReplyError::Succeeded.as_u8(),
                0x00,
                consts::SOCKS5_ADDR_TYPE_DOMAIN_NAME,
            ];
            wire.push(name.len() as u8);
            wire.extend_from_slice(name.as_bytes());
            wire.extend_from_slice(&port.to_be_bytes());
            wire
        }
    };
    if write_vanilla_reply(&mut stream, &wire).await.is_err() {
        session.close().await;
        return DoorExit::ClientGone;
    }

    bind_quiet_and_pipe(stream, session).await
}

/// The BIND door's quiet wait + reply#2 + the data plane: await the
/// inbound accept through the session (`await_inbound` — the one
/// normative reply#2 reader), re-encode reply#2's peer to the vanilla
/// client, then pipe the pass-through plane raw (the plane owns the
/// in-band stream — `BindDataPlane` impls `AsyncRead + AsyncWrite`,
/// the pass-through to the accepted peer; the session stays the
/// teardown owner, dropped at the end).
async fn bind_quiet_and_pipe(
    mut stream: TcpStream,
    mut session: crate::consumer::BindSession,
) -> DoorExit {
    let plane = match session.await_inbound().await {
        Ok(Some(plane)) => plane,
        Ok(None) => return DoorExit::PlaneEnded,
        // An accept-stage refusal (reply#2's mapped code was written
        // in-band; the producer's pump ended the session): the door
        // closes the vanilla connection — the conversation ends where
        // the in-band one did.
        Err(_) => {
            session.close().await;
            return DoorExit::PlaneEnded;
        }
    };

    // Reply#2 (the vanilla form): the accepted peer re-encoded.
    let reply = match plane.peer {
        TargetAddr::Ip(sa) => new_reply(&ReplyError::Succeeded, sa),
        TargetAddr::Domain(_, port) => new_reply(
            &ReplyError::Succeeded,
            SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), port),
        ),
    };
    if write_vanilla_reply(&mut stream, &reply).await.is_err() {
        session.close().await;
        return DoorExit::ClientGone;
    }

    let _ = tokio::io::copy_bidirectional(&mut stream, &mut { plane }).await;
    session.close().await;
    DoorExit::PlaneEnded
}

/// Write one vanilla reply frame (the door's RFC codec at the
/// boundary — ADR-003): write + flush; a failed write is the client
/// gone.
async fn write_vanilla_reply(stream: &mut TcpStream, reply: &[u8]) -> std::io::Result<()> {
    stream.write_all(reply).await?;
    stream.flush().await
}