alksocks 0.3.2

//! The `composed` feature's egress + accept providers (ADR-008's
//! composed variants, ADR-009's feature gate): [`direct_egress`], the
//! UDP ASSOCIATE egress that dials each destination through an
//! alktunnels 0.2.0 direct-op channel
//! ([`alktunnels::TunnelSession::open_direct`], `Substrate::Udp`);
//! and [`bindfirst_accept`], the TCP BIND accept whose listener lives
//! on a bind-first listen resource (ADR-007 variant 2 / ADR-008's
//! composed accept).
//!
//! **Per-target ACL for free** (ADR-008's composed benefit): each
//! destination rides its own `channels/tunnel/direct`-governed channel,
//! so the producer's `tunnel:direct` scope gate and dial policy apply
//! per destination, with no SOCKS5-layer allowlist needed
//! (ADR-006/ADR-012: whatever ACL governs the tunnel capability
//! governs everything reachable through it).
//!
//! The shape is POC #2's `ComposedEgress`
//! (`/workspace/alksocks-udp-poc/src/egress.rs`), ported to the landed
//! seam types: a lazy destination→tunnel table, where each new
//! destination opens one UDP-tunnel channel, a relay task per tunnel
//! drains inbound datagrams into a shared queue, the write half stays
//! in the table (LRU-capped; dropping a write half EOFs the tunnel
//! producer-side; the entry's session drop reaps the channel, which
//! is alktunnels ADR-005's teardown contract).
//!
//! Contracts (the landed seam types' normative docs, honored here):
//!
//! - **The readiness contract** (load-bearing, `Egress::next_reply`):
//!   parks on the shared inbound queue (`recv().await`, hence
//!   `Pending` until a relay task delivers); a closed queue (all
//!   relays ended) returns `Err`, never a hot `Ok(None)`. The stage's
//!   zero-yield defense (review 002 R-1) is the backstop, not the
//!   design.
//! - **Framing-once**: replies carry *inner* bytes (`addr + data`;
//!   the reply address is the entry's recorded destination, the RFC's
//!   per-datagram source semantics); the length prefix is the wire
//!   layer's, applied exactly once downstream. `send` takes the
//!   payload only; the tunnel-side framing (`[len: u16 BE]`, the
//!   alktunnels ADR-003 codec) is the tunnel wire's, not the RFC
//!   wire's.
//! - **Per-datagram `send` is awaited inline** (the stage
//!   interleaves sends with reply drains): a slow egress applies
//!   backpressure to the association, not to other channels, and the
//!   batch-2 composed-egress author inherits this bound knowingly
//!   (wire.md's note). The table's open cost rides the send path
//!   (lazy open on first datagram per destination; the POC's
//!   measured ~0.8 ms).
//! - **The resource/target mapping**: `open_direct` takes a
//!   [`SubstrateAddr`], which the destination's [`TargetAddr`] renders
//!   to (IP forms direct; domain form passes through unresolved,
//!   producer-side resolution, the ssh `-D` semantic; ADR-012).
//!
//! **The Sync-bound wrinkle** (the batch-1 review's seam contract vs
//! alktunnels' surface): the landed `TargetHandle`/`BindTargetHandle`
//! halves are `Box<dyn AsyncRead + Send + Sync + Unpin>`, but
//! alktunnels' `TakenHalves` yields `Send + Unpin` only; and
//! `alkcall::BiStream` is `Send + Unpin` but NOT `Sync` (its inner is
//! a `dyn AsyncReadWrite` trait object). The owner-wrapper posture
//! (egress.rs's documented contract: "keep-alive state a substrate
//! needs rides inside the halves, a provider-side owner wrapper")
//! takes two shapes here: [`SessionOwner`] (the generic
//! half-wraps-session wrapper, the egress's design point, exported as
//! the wrapper-aware deployment's surface) and the accept's own
//! [`SyncHalves`]/[`OwnedWrite`] adapter pair (a `std::sync::Mutex`
//! around the half; `Mutex<T: Send>` is `Sync`, no unsafe; with
//! the channel session owned inside the payload, [`ForwardedOwner`]:
//! dropping the handle drops the session AND the halves together).
//! For the egress the wrapper is not the handle payload's burden,
//! because the egress owns the session in the table entry (the POC's
//! shape);

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

use alkcall::channels::client::ChannelClient;
use alkcall::channels::operations::ChannelCore;
use alkcall::channels::pump::pump_bidi;
use alkcall::core::auth::AuthContext;
use alkcall::core::types::BiStream;
use alktunnels::params::{Substrate, SubstrateAddr, TunnelForwardedParams};
use alktunnels::{TunnelListener, TunnelSession};
use bytes::Bytes;
use futures::future::BoxFuture;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::sync::oneshot;

use crate::egress::{
    BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError, BindTargetHandle,
    Egress, EgressError, EgressFactory,
};
pub use crate::rfc1928::util::target_addr::TargetAddr;
use crate::wire::MAX_INNER_LEN;

/// The default LRU cap (the 256-channel default cap bounds fan-out,
/// ADR-010's posture; the cap is config, not wire).
pub const DEFAULT_MAX_TUNNELS: usize = 256;

/// The destination's render into the direct op's [`SubstrateAddr`] (the
/// resource/target mapping): IP forms direct; domain form passes
/// through unresolved (producer-side resolution, the ssh `-D`
/// semantic, per ADR-012).
pub fn substrate_addr_of(addr: &TargetAddr) -> SubstrateAddr {
    match addr {
        TargetAddr::Ip(sa) => SubstrateAddr::HostPort {
            host: sa.ip().to_string(),
            port: sa.port(),
        },
        TargetAddr::Domain(domain, port) => SubstrateAddr::HostPort {
            host: domain.clone(),
            port: *port,
        },
    }
}

/// Render the destination to the table's key form (the string the LRU
/// orders by): `host:port` per [`SubstrateAddr`]'s render (IP literal
/// or unresolved domain; the domain stays unresolved, so domain and
/// IP destinations key distinctly, as the client addressed them).
pub fn key_of(addr: &TargetAddr) -> String {
    match substrate_addr_of(addr) {
        SubstrateAddr::HostPort { host, port } => format!("{host}:{port}"),
        SubstrateAddr::Path { path } => path,
    }
}

/// The owner wrapper (the batch-1 review's seam contract vs
/// alktunnels' surface, providing the owner-wrapper posture's generic
/// shape;
/// the composed accept meets the `+ Sync` bound with its own
/// [`SyncHalves`]/[`OwnedWrite`] adapter pair instead): an alktunnels
/// half wrapped with its [`TunnelSession`] inside. The landed handle
/// payloads carry `+ Sync` (the
/// `ChannelPlan` payload bound), alktunnels' `TakenHalves` yields
/// `Send + Unpin` only; the wrapper owns the session beside the half,
/// so dropping the wrapper drops the half and the session together
/// (the session's drop reaps the channel entry, per alktunnels
/// ADR-005). Whether the wrapper is `Sync` (the accept's need) is the
/// half's own `Sync`ness, asserted by the compile-shape test against
/// the alktunnels 0.2.0 surface; the egress needs only `Send` (the
/// relay task), which the wrapper always has.
pub struct SessionOwner<H> {
    half: H,
    session: TunnelSession,
}

impl<H: AsyncRead + Send + Unpin> AsyncRead for SessionOwner<H> {
    fn poll_read(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        let this = std::pin::Pin::into_inner(self);
        std::pin::Pin::new(&mut this.half).poll_read(cx, buf)
    }
}

impl<H> SessionOwner<H> {
    /// Wrap the half with its session (the owner).
    pub fn new(half: H, session: TunnelSession) -> Self {
        Self { half, session }
    }

    /// The wrapped half.
    pub fn half(&self) -> &H {
        &self.half
    }

    /// The wrapped half, mutably.
    pub fn half_mut(&mut self) -> &mut H {
        &mut self.half
    }

    /// Unwrap: the half plus the session (the owner), for a caller
    /// that needs them separated again.
    pub fn into_parts(self) -> (H, TunnelSession) {
        (self.half, self.session)
    }
}

/// The composed egress: the lazy destination→tunnel table (the
/// per-association ASSOCIATE seam object). Constructed via
/// [`direct_egress`] (the `EgressFactory` the producer registration
/// consumes).
pub struct ComposedEgress {
    client: Arc<ChannelClient>,
    max_tunnels: usize,
    entries: HashMap<String, TableEntry>,
    order: std::collections::VecDeque<(String, TargetAddr)>,
    inbound: tokio::sync::mpsc::Receiver<Inbound>,
    inbound_tx: tokio::sync::mpsc::Sender<Inbound>,
    /// Live relay tasks, hence the readiness contract's closed-queue
    /// condition: when every relay has ended (each reported via
    /// [`Inbound::Ended`]) and the queue is drained, `next_reply`
    /// returns `Err` (the association end), never a hot `Ok(None)`.
    /// The factory `Sender` alone does not close the queue (the
    /// egress holds it), so the count is the design.
    live_relays: usize,
    /// Whether any tunnel has ever been opened. The lazy table starts
    /// empty: a fresh association's `next_reply` parks until the first
    /// `send` opens a tunnel (the producer's stage selects the channel
    /// branch first on a quiescent association, and the association
    /// serves until a datagram arrives or the channel EOFs). The
    /// ended-relay `Err` fires only once at least one tunnel existed
    /// and every relay has ended (the closed-queue condition).
    ever_opened: bool,
}

/// The shared inbound queue's message: a tunnel's datagram (keyed by
/// the table key, the reply address riding the entry) or a relay's
/// end marker (the tunnel's EOF drained).
#[derive(Debug)]
enum Inbound {
    Datagram(String, Bytes),
    Ended,
}

/// One table entry: the tunnel's write half (the udp substrate's
/// framed half, `[len: u16 BE]` frames on the tunnel wire) + the
/// reply address to report for datagrams coming back through it + the
/// session (the channel-entry owner; the egress owns it, not a
/// handle payload; dropping the entry EOFs the write half and reaps
/// the channel, per alktunnels ADR-005).
struct TableEntry {
    _session: TunnelSession,
    write: Box<dyn AsyncWrite + Send + Unpin>,
    reply_addr: TargetAddr,
}

/// The per-tunnel relay task: drain the tunnel's framed read half
/// into the egress's shared inbound queue. Exits on EOF (the tunnel
/// ended); the write half stays in the table until eviction or stage
/// exit. Framing decode is the tunnel codec's (the alktunnels
/// ADR-003), so the datagrams here are inner payloads (no RFC address
/// on the tunnel wire; the reply address is the entry's recorded
/// destination).
async fn relay_read(
    mut read: impl AsyncRead + Send + Unpin,
    tx: tokio::sync::mpsc::Sender<Inbound>,
    key: String,
) {
    use tokio::io::AsyncReadExt;
    let mut reader = alktunnels::wire::DatagramReader::new();
    let mut chunk = vec![0u8; 16 * 1024];
    loop {
        match read.read(&mut chunk).await {
            Ok(0) | Err(_) => break,
            Ok(n) => {
                for dg in reader.feed(&chunk[..n]) {
                    if tx.send(Inbound::Datagram(key.clone(), dg)).await.is_err() {
                        return;
                    }
                }
            }
        }
    }
    let _ = tx.send(Inbound::Ended).await;
}

/// Assemble the reply's inner bytes (`addr + data`, the entry's
/// recorded destination + the payload; framing-once: no length prefix
/// here, the wire layer's `frame_inner` adds it downstream). The
/// oversize guard mirrors the wire layer's frame bound.
fn inner_bytes(addr: &TargetAddr, payload: &[u8]) -> Result<Bytes, EgressError> {
    let mut inner = addr
        .to_be_bytes()
        .map_err(|e| EgressError::General(format!("addr encode: {e}")))?;
    inner.extend_from_slice(payload);
    if inner.len() > MAX_INNER_LEN {
        return Err(EgressError::General(format!(
            "reply oversize: {} > {MAX_INNER_LEN}",
            inner.len()
        )));
    }
    Ok(Bytes::from(inner))
}

impl ComposedEgress {
    /// Build the egress (the per-association ASSOCIATE seam object).
    /// `client` is the channels client the direct ops ride; the
    /// constructor's config carries the LRU cap (`max_tunnels`, the
    /// fan-out bound, config not wire; the producer's `tunnel:direct`
    /// ACL governs which destinations the channels accept, which is
    /// the per-target-ACL-for-free benefit).
    pub fn new(client: Arc<ChannelClient>, max_tunnels: usize) -> Self {
        let (tx, rx) = tokio::sync::mpsc::channel(64);
        Self {
            client,
            max_tunnels,
            entries: HashMap::new(),
            order: std::collections::VecDeque::new(),
            inbound: rx,
            inbound_tx: tx,
            live_relays: 0,
            ever_opened: false,
        }
    }

    /// The entry for `addr`, opened lazily on first datagram (the
    /// open cost rides the send path, per-datagram awaited inline).
    async fn entry_for(&mut self, addr: &TargetAddr) -> Result<&mut TableEntry, EgressError> {
        let key = key_of(addr);
        if !self.entries.contains_key(&key) {
            while self.entries.len() >= self.max_tunnels {
                let Some((evict, _)) = self.order.pop_front() else {
                    break;
                };
                self.entries.remove(&evict);
            }
            let target = substrate_addr_of(addr);
            let session = TunnelSession::open_direct(self.client.as_ref(), target, Substrate::Udp)
                .await
                .map_err(|e| EgressError::Unreachable(format!("tunnel open `{key}`: {e}")))?;
            let taken = session.take_halves();
            // The session stays INSIDE the entry: dropping it reaps
            // the channel (ADR-005's teardown contract), which would
            // EOF the relay task instantly (the POC's insert-time
            // bug). The relay task owns the bare read half.
            tokio::spawn(relay_read(taken.read, self.inbound_tx.clone(), key.clone()));
            self.live_relays += 1;
            self.ever_opened = true;
            self.entries.insert(
                key.clone(),
                TableEntry {
                    _session: taken.session,
                    write: taken.write,
                    reply_addr: addr.clone(),
                },
            );
        }
        self.order.retain(|(k, _)| k != &key);
        self.order.push_back((key.clone(), addr.clone()));
        // The entry was inserted above (or the key was already
        // present): an absent entry here is a structural invariant
        // break, not a runtime condition; it is surfaced as an
        // egress error instead of a panic (convention 2).
        self.entries.get_mut(&key).ok_or_else(|| {
            EgressError::General(format!(
                "tunnel table invariant broken: `{key}` absent after insert"
            ))
        })
    }
}

impl Egress for ComposedEgress {
    fn send(&mut self, addr: TargetAddr, payload: Bytes) -> BoxFuture<'_, Result<(), EgressError>> {
        Box::pin(async move {
            let entry = self.entry_for(&addr).await?;
            use tokio::io::AsyncWriteExt;
            let framed = alktunnels::wire::frame_datagram(&payload)
                .map_err(|e| EgressError::General(format!("tunnel frame: {e}")))?;
            let write: &mut (dyn AsyncWrite + Send + Unpin) = &mut *entry.write;
            write
                .write_all(&framed)
                .await
                .map_err(|e| EgressError::General(format!("tunnel write: {e}")))?;
            write
                .flush()
                .await
                .map_err(|e| EgressError::General(format!("tunnel flush: {e}")))?;
            Ok(())
        })
    }

    fn next_reply(&mut self) -> BoxFuture<'_, Result<Option<Bytes>, EgressError>> {
        Box::pin(async move {
            loop {
                // A non-blocking first pass over the shared queue (the
                // relay tasks may have queued replies already), then a
                // real async wait on the queue (`Pending` until a
                // relay task delivers; this is the readiness contract).
                loop {
                    match self.inbound.try_recv() {
                        Ok(Inbound::Datagram(key, dg)) => {
                            let addr = match self.entries.get(&key) {
                                Some(entry) => entry.reply_addr.clone(),
                                None => continue,
                            };
                            return Ok(Some(inner_bytes(&addr, &dg)?));
                        }
                        // A relay ended (its tunnel's EOF drained);
                        // count it, and the closed-queue condition is
                        // every relay ended.
                        Ok(Inbound::Ended) => {
                            self.live_relays = self.live_relays.saturating_sub(1);
                            continue;
                        }
                        Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break,
                        Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => break,
                    }
                }
                if self.ever_opened && self.live_relays == 0 {
                    // Every relay ended (the tunnels ended): a closed
                    // relay set must NOT spin the stage's select (an
                    // instantly-ready None would starve the channel
                    // branch), so the stage ends the association
                    // instead. A fresh association (no tunnel ever
                    // opened) parks here, and the channel branch
                    // governs until the first
                    // datagram arrives.
                    return Err(EgressError::General("all tunnel relays ended".into()));
                }
                match self.inbound.recv().await {
                    Some(Inbound::Datagram(key, dg)) => {
                        let addr = match self.entries.get(&key) {
                            Some(entry) => entry.reply_addr.clone(),
                            None => continue,
                        };
                        return Ok(Some(inner_bytes(&addr, &dg)?));
                    }
                    // A relay ended while parked; loop back, since the
                    // count check above decides whether the
                    // association ends (never a hot `Ok(None)`).
                    Some(Inbound::Ended) => {
                        self.live_relays = self.live_relays.saturating_sub(1);
                        continue;
                    }
                    None => {
                        return Err(EgressError::General("all tunnel relays ended".into()));
                    }
                }
            }
        })
    }
}

/// The composed egress constructor (ADR-008's composed variant,
/// producer.md §Wiring): the `EgressFactory` the producer
/// registration consumes. One per ASSOCIATE; the stage owns the
/// returned egress; stage exit drops it (write halves dropped = EOF
/// to the tunnel producers; the entries' sessions reap the channels,
/// per alktunnels ADR-005's teardown contract).
///
/// # Backpressure (inherited knowingly)
///
/// Per-datagram `send` is awaited inline by the producer stage (it
/// interleaves sends with reply drains): a slow egress applies
/// backpressure to the association, not to other channels. The lazy
/// open's cost (~0.8 ms per new destination in the POC's
/// measurement) rides the send path.
pub fn direct_egress(client: Arc<ChannelClient>, max_tunnels: usize) -> EgressFactory {
    Arc::new(move || {
        let client = Arc::clone(&client);
        Box::pin(async move {
            Ok(Box::new(ComposedEgress::new(client, max_tunnels)) as Box<dyn Egress + Send>)
        })
    })
}
// --- The composed accept (ADR-007 variant 2, the sibling task's shape) ------

/// The correlation registry (the assembly-side bookkeeping): `listen`
/// token → the oneshot delivering the forwarded session (the adopted
/// channel's `BiStream` + the forwarded op's `peer` param). One entry
/// per live BIND session; the forwarded handler (the provider's shared
/// opener half) removes and completes the entry on correlation. A
/// dropped receiver (client-gone teardown) makes completion fail, so
/// the handler drops the channel stream (teardown-at-birth).
pub type CorrelationRegistry =
    Arc<tokio::sync::Mutex<HashMap<u32, oneshot::Sender<ForwardedSession>>>>;

/// What a correlated forwarded session delivers: the forwarded
/// channel's `BiStream` (raw pass-through, since the SOCKS5 BIND data
/// plane is raw on both ends) + the forwarded
/// op's `peer` param (reply#2's real peer, the graduation's win).
/// Public because the
/// [`CorrelationRegistry`] it travels through is public API surface
/// (the assembly holds the handle [`register_forwarded_serving`]
/// returns); constructing one is the forwarded handler's job alone.
pub struct ForwardedSession {
    pub read: tokio::io::ReadHalf<tokio::io::DuplexStream>,
    pub write: tokio::io::WriteHalf<tokio::io::DuplexStream>,
    pub peer: SubstrateAddr,
}

/// The forwarded-op serving handler (the provider's shared opener
/// half): correlates the arriving forwarded open by `listen` token and
/// OWNS the data-plane pump (R-02: the `register_openable` wrapper
/// awaits this task inline, then tears the forwarded channel down; a
/// handler that returns early reaps the channel at birth). The pump
/// bridges the forwarded channel's `BiStream` to a local duplex pair:
/// the near end pumps here (this task lives exactly as long as the
/// data plane, so either end's EOF ends it), the far end travels
/// through
/// the token's oneshot into the accept future's [`BindTargetHandle`]
/// payloads (dropping the handle drops the far end → the pump sees
/// EOF → the task returns → the wrapper reaps the forwarded channel,
/// i.e. teardown-at-birth, structural). A correlation miss (no
/// waiting
/// session (client gone, or a token nobody minted) drops the stream:
/// the task returns immediately, the same reaping shape. A dead
/// oneshot receiver makes the send fail, and the task drops the stream
/// and returns (same reaping).
fn make_forwarded_handler(registry: CorrelationRegistry) -> alktunnels::ForwardedHandler {
    Arc::new(move |forwarded: TunnelForwardedParams, bidi: BiStream| {
        let registry = Arc::clone(&registry);
        tracing::debug!(
            listen = forwarded.listen,
            "composed accept: forwarded open arrived"
        );
        tokio::spawn(async move {
            let tx = registry.lock().await.remove(&forwarded.listen);
            let Some(tx) = tx else {
                tracing::warn!(
                    listen = forwarded.listen,
                    "composed accept: forwarded open with no waiting session; dropping"
                );
                return;
            };
            let (near, far) = tokio::io::duplex(64 * 1024);
            let (far_read, far_write) = tokio::io::split(far);
            if tx
                .send(ForwardedSession {
                    read: far_read,
                    write: far_write,
                    peer: forwarded.peer,
                })
                .is_err()
            {
                tracing::warn!(
                    listen = forwarded.listen,
                    "composed accept: waiting session vanished; dropping the forwarded channel"
                );
                return;
            }
            // The middle pump (three-party half-close would cycle:
            // pump_bidi's per-direction shutdown-on-completion
            // deadlocks a bridge: the SOCKS5 pump's far.write shutdown
            // needs near.read EOF; the wire's bidi.read EOF needs the
            // producer pump's far.read EOF; far.read needs near.write
            // shutdown; near.write needs bidi.read EOF). The bridge
            // therefore completes on EITHER direction's EOF and drops
            // everything: the far halves (the SOCKS5 pump sees EOF),
            // the near halves, and the BiStream (the forwarded channel
            // tears down; the wrapper's inline await returns, per
            // R-02, and reaps the entry).
            let (near_read, near_write) = tokio::io::split(near);
            pump_bidi(bidi, near_read, near_write).await;
        })
    })
}

/// The forwarded channel's `BiStream` reassembled after the
/// payload-split: the read half rides the [`SyncHalves`] adapter (the
/// mutex-joined form), the write half rides [`OwnedWrite`], and the
/// owner rides inside the write payload so dropping the handle drops
/// everything: the teardown-at-birth discipline, compiler-enforced.
/// The owner is the [`TunnelListener`] (NOT an adopted session, since
/// the forwarded channel's entry is the producer's, adopted inside the
/// producer's `ForwardedOpener`): the listener's keep-alive holds the
/// far accept loop (and every per-accept pump it spawned; R-02)
/// alive; dropping the handle drops the listener → the listen channel
/// EOFs → the far teardown cascades. The edition-2021 disjoint-capture
/// trap (an earlier draft took `listener.channel_id`, a `u32`, and
/// dropped the listener at the factory's return, EOF-ing the listen
/// channel at birth) is the reason the listener is forced into the
/// future with a whole-struct rebinding.
struct ForwardedOwner {
    _listener: TunnelListener,
}

/// The `BindTargetHandle` read-half adapter over the forwarded
/// channel's read half: `BiStream` is `Send + Unpin` but NOT `Sync`
/// (its inner is a `dyn AsyncReadWrite` trait object, with no `Sync`
/// supertrait), so the raw halves cannot meet the handle payloads'
/// `+ Sync` bound directly. A `std::sync::Mutex` guard is the
/// no-unsafe Sync proof (`Mutex<T: Send>` is `Sync`); the RFC 1928
/// BIND data plane is a raw pass-through pumped by ONE `pump_bidi`
/// task at a time, so contention never occurs at runtime; the mutex
/// is the Sync-proof shape, not a concurrency mechanism. A poisoned
/// lock (a panic in one poll) unwraps via `into_inner`, so a panic in
/// one operation does not cascade (AGENTS.md convention 2).
struct SyncHalves {
    inner: std::sync::Mutex<tokio::io::ReadHalf<tokio::io::DuplexStream>>,
}

impl AsyncRead for SyncHalves {
    fn poll_read(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        let this = self.get_mut();
        let mut inner = match this.inner.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        std::pin::Pin::new(&mut *inner).poll_read(cx, buf)
    }
}

/// The write-half payload: the owner ([`ForwardedOwner`], the
/// adopted channel session) and the write half ride inside one
/// `std::sync::Mutex` (the same no-unsafe Sync proof as
/// [`SyncHalves`]: `TunnelSession` and `WriteHalf<BiStream>` are
/// `Send` but not `Sync`). Contention never occurs at runtime (one
/// pump task); a poisoned lock unwraps via `into_inner`.
struct OwnedWrite {
    inner: std::sync::Mutex<OwnedInner>,
}

struct OwnedInner {
    _owner: ForwardedOwner,
    write: tokio::io::WriteHalf<tokio::io::DuplexStream>,
}

impl AsyncWrite for OwnedWrite {
    fn poll_write(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        let this = self.get_mut();
        let mut inner = match this.inner.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        std::pin::Pin::new(&mut inner.write).poll_write(cx, buf)
    }

    fn poll_flush(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        let this = self.get_mut();
        let mut inner = match this.inner.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        std::pin::Pin::new(&mut inner.write).poll_flush(cx)
    }

    fn poll_shutdown(
        self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        let this = self.get_mut();
        let mut inner = match this.inner.lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        };
        std::pin::Pin::new(&mut inner.write).poll_shutdown(cx)
    }
}

/// The forwarded op's serving registration (the opener half, ADR-008
/// §3, `register_tunnel_forwarded_openable` + this provider's
/// correlation handler): register on the SOCKS5 producer's serving
/// registry so its connection serves the forwarded op. Call once per
/// channels connection at assembly time, since the registry is shared
/// across all BIND sessions every [`bindfirst_accept`] factory over
/// the same client fabricates (one token space per producer
/// connection).
///
/// The scope gate is `tunnel:forwarded` on the PRODUCER's per-call
/// identity (the ingress grant, ADR-006 Amendment 1; never implied by
/// `tunnel:open`). The returned registry handle feeds the factories;
/// keep it alongside the client they ride.
///
/// # Errors
///
/// A duplicate `channels/tunnel/forwarded` op id on the registry (an
/// assembly bug, not a runtime condition), and it surfaces the
/// underlying registration's `String` diagnostic (convention 2: no
/// panic path; the caller decides).
pub fn register_forwarded_serving(
    core: &ChannelCore,
    on_registry: &Arc<alkcall::registry::registration::OperationRegistry>,
    auth: AuthContext,
) -> Result<CorrelationRegistry, String> {
    let registry: CorrelationRegistry = Arc::new(tokio::sync::Mutex::new(HashMap::new()));
    alktunnels::producer::register_tunnel_forwarded_openable(
        core,
        on_registry,
        auth,
        make_forwarded_handler(Arc::clone(&registry)),
    )?;
    Ok(registry)
}

/// The composed accept factory (ADR-007 variant 2, producer.md
/// §Wiring `composed::bindfirst_accept`): the `BindAcceptFactory` the
/// producer registration consumes, the BIND accept whose listener
/// lives on an alktunnels 0.2.0 bind-first listen resource
/// (`register_tunnel_bindfirst_openable` + `BindFirstInjections`,
/// giving per-listener ACL for free, hub-relayable).
///
/// Per session the factory:
///
/// 1. Opens the listen channel (`TunnelListener::open`; the
///    establishment bound covers only this open-op round trip,
///    ADR-007); a failed open is a factory failure (reply#1's mapped
///    code, so no phantom listener).
/// 2. Advertises reply#1's BND.ADDR as the REAL `bound` (the open-op
///    reply's `bound` field, the alkcall 0.8.0 reply projection; the
///    graduation's vanilla-fidelity win; no expected-addr stand-in).
///    A producer that supplied no `bound`/`listen` (not a bind-first
///    resource) fails the factory, with no stand-in on this variant.
/// 3. Inserts the minted `listen` token → oneshot into the
///    correlation registry and returns the accept future parked on
///    it.
///
/// The accept future waits for the ONE forwarded session matching the
/// token (the correlation registry is assembly-side bookkeeping; the
/// listener does not own the receipt queue), then builds the
/// [`BindTargetHandle`]: reply#2's peer is the forwarded op's `peer`
/// param (the graduation's win, with no sentinel), and the data-plane
/// halves are the [`SyncHalves`]/[`OwnedWrite`] adapters over the
/// adopted channel's `BiStream` with the [`ForwardedOwner`] (the
/// adopted channel session) riding inside the write payload, so
/// dropping the handle drops the session AND the halves together.
///
/// **The client-gone teardown** (the seam contract, end-to-end): the
/// listener and the oneshot receiver ride the accept future, so the
/// quiet wait's `select!` drop of that future drops the
/// [`TunnelListener`] (its held write half EOFs the listen channel,
/// so the far accept dispatcher tears down) AND drops the oneshot
/// receiver (a later forwarded open for that token finds a dead
/// oneshot and drops the channel stream), leaving no leaked
/// listener and no leaked channel.
///
/// Only TCP-stream listen resources are supported (the SOCKS5 BIND
/// data plane is raw on both ends; a UDP listen substrate would
/// frame, and a datagram BIND variant is a different task). A
/// non-TCP `bound`
/// fails the factory (`NotAllowed`).
///
/// **The 0x02 layering** (review 003 N-2): on the channels path the
/// ADR-006 command gate precedes this factory, so an identity
/// without
/// `socks5:bind` never reaches it. A factory-side `NotAllowed` (0x02,
/// the same code the gate uses) therefore means the *tunnel* layer's
/// per-listener ACL (this variant's real policy source), not the
/// command gate (the two are semantically distinct, wire-identical).
pub fn bindfirst_accept(
    client: Arc<ChannelClient>,
    registry: CorrelationRegistry,
    resource: impl Into<String>,
) -> BindAcceptFactory {
    let resource: Arc<str> = resource.into().into();
    Arc::new(move |_target: TargetAddr| {
        let client = Arc::clone(&client);
        let registry = Arc::clone(&registry);
        let resource = Arc::clone(&resource);
        Box::pin(async move {
            let params = alktunnels::TunnelParams {
                resource: resource.to_string(),
                substrate: Substrate::Tcp,
            };
            let listener = TunnelListener::open(client.as_ref(), params)
                .await
                .map_err(|e| BindAcceptFactoryError::Refused(format!("listen open: {e}")))?;
            let bind_addr = match listener.bound() {
                Some(SubstrateAddr::HostPort { host, port }) => resolve_bound(host, *port)
                    .map_err(|e| BindAcceptFactoryError::General(format!("bound render: {e}")))?,
                Some(SubstrateAddr::Path { path }) => {
                    return Err(BindAcceptFactoryError::NotAllowed(format!(
                        "non-tcp bound address `{path}`: this provider serves TCP listeners only"
                    )));
                }
                None => {
                    return Err(BindAcceptFactoryError::Shortage(
                        "the producer supplied no `bound` reply field, so this is not a bind-first \
                         resource; no stand-in on this variant"
                            .into(),
                    ));
                }
            };
            let token = listener.listen().ok_or_else(|| {
                BindAcceptFactoryError::Shortage(
                    "the producer supplied no `listen` correlation token, so this is not a \
                     bind-first resource; no stand-in on this variant"
                        .into(),
                )
            })?;
            let (tx, rx) = oneshot::channel();
            registry.lock().await.insert(token, tx);
            tracing::debug!(token, "composed accept: registry entry inserted");

            let fut = async move {
                // Force the whole-struct capture: edition-2021 disjoint
                // capture would take `listener.channel_id` (a `u32`)
                // and drop the listener itself at the factory's return,
                // EOF-ing the listen channel at birth. The listener
                // rides the future end-to-end (the seam contract).
                let listener = listener;
                let ForwardedSession { read, write, peer } = match rx.await {
                    Ok(session) => session,
                    Err(_) => {
                        tracing::warn!(
                            token,
                            "composed accept: oneshot sender dropped; listener closing"
                        );
                        // The registry entry was removed without
                        // delivery (the forwarded serving half tore
                        // down; the whole provider is going away).
                        // The listener's drop tears the listen
                        // channel down (the held write half EOFs).
                        return Err(BindAcceptError::General(
                            "the forwarded-op serving half went away; the listener closed".into(),
                        ));
                    }
                };
                let peer_addr = peer_to_socket_addr(&peer);
                Ok(BindTargetHandle {
                    peer: TargetAddr::Ip(peer_addr),
                    read: Box::new(SyncHalves {
                        inner: std::sync::Mutex::new(read),
                    }) as Box<dyn AsyncRead + Send + Sync + Unpin>,
                    write: Box::new(OwnedWrite {
                        inner: std::sync::Mutex::new(OwnedInner {
                            // The listener rides INSIDE the payload:
                            // its keep-alive holds the far accept loop
                            // (and every per-accept pump it spawned;
                            // R-02) alive for the data plane's
                            // lifetime; dropping the handle drops the
                            // listener → the listen channel EOFs →
                            // the teardown cascades (teardown
                            // structural, compiler-enforced). The
                            // edition-2021 disjoint-capture trap (an
                            // earlier draft took `listener.channel_id`
                            // (a `u32`) and dropped the listener at
                            // the factory's return, EOF-ing the listen
                            // channel at birth) is why the whole
                            // struct is rebound into the future above.
                            _owner: ForwardedOwner {
                                _listener: listener,
                            },
                            write,
                        }),
                    }),
                })
            };
            Ok(BindAccept::new(bind_addr, fut))
        })
    })
}

/// Render the `bound` host (an IP literal or an unresolved host name)
/// + port into reply#1's `SocketAddr`:
///
/// - IP literals resolve directly.
/// - A non-IP host fails (reply#1's address is a real socket address
///   on this variant, but a hostname bound is the producer's
///   resolution
///   surface, not the advertised address's).
///
/// The parse error is a `String` diagnostic (the factory maps it to
/// `General`).
fn resolve_bound(host: &str, port: u16) -> Result<SocketAddr, String> {
    host.parse::<std::net::IpAddr>()
        .map(|ip| SocketAddr::new(ip, port))
        .map_err(|e| format!("`{host}`: {e}"))
}

/// The forwarded op's `peer` param rendered to reply#2's socket
/// address (the graduation's peer fidelity; the param is the real
/// peer). A `Path` peer (a unix-socket accept) has no RFC-shaped
/// address; the RFC sentinel `0.0.0.0:0` carries it (the data plane
/// is unaffected either way).
fn peer_to_socket_addr(peer: &SubstrateAddr) -> SocketAddr {
    match peer {
        SubstrateAddr::HostPort { host, port } => host
            .parse::<std::net::IpAddr>()
            .map(|ip| SocketAddr::new(ip, *port))
            .unwrap_or_else(|_| sentinel_peer()),
        SubstrateAddr::Path { .. } => sentinel_peer(),
    }
}

fn sentinel_peer() -> SocketAddr {
    SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0)
}

#[cfg(test)]
mod egress_unit_tests {
    use super::*;
    use crate::rfc1928::util::target_addr::ToTargetAddr;
    use tokio::io::AsyncWriteExt;

    /// The reply-oversize guard fires (review 003 D-2): a reply whose
    /// inner bytes exceed the joint bound is a `General` egress error,
    /// though the guard exists in case the tunnel codec's u16 datagram
    /// bound upstream fails to keep every real reply under it.
    #[test]
    fn inner_bytes_oversize_guard_fires() {
        let addr = ("10.0.0.1", 53u16).to_target_addr().unwrap();
        let payload = vec![0u8; MAX_INNER_LEN];
        let err = inner_bytes(&addr, &payload).expect_err("the joint bound trips");
        assert!(
            matches!(err, EgressError::General(ref m) if m.contains("oversize")),
            "the oversize guard: {err}"
        );
        let ok = inner_bytes(
            &addr,
            &payload[..MAX_INNER_LEN - addr.to_be_bytes().unwrap().len()],
        );
        assert!(ok.is_ok(), "the bound-edge reply passes");
    }

    /// The relay task's sender-closed exit (review 003 D-2): the
    /// shared queue closed with the egress (the stage exited
    /// mid-relay), so it ends the relay's datagram pump silently (the
    /// `tx.send(..).is_err()` early return), distinct from the EOF
    /// exit that still reports `Ended`.
    #[tokio::test]
    async fn relay_read_exits_silently_when_the_queue_is_closed() {
        let (tx, rx) = tokio::sync::mpsc::channel::<Inbound>(4);
        drop(rx);
        let (client_side, mut server_side) = tokio::io::duplex(1024);
        server_side
            .write_all(&alktunnels::wire::frame_datagram(b"queued").unwrap())
            .await
            .unwrap();
        server_side.shutdown().await.unwrap();
        drop(server_side);
        let handle = tokio::spawn(relay_read(client_side, tx, "k".into()));
        // The relay exits on the closed-queue send (silently), and it
        // joins promptly with no panic.
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("the relay exits (the closed queue ends the pump)")
            .expect("no panic");
    }

    /// The relay task's EOF exit still reports `Ended` (the count's
    /// decrement path, the shape the integration tests observe
    /// indirectly, pinned here directly).
    #[tokio::test]
    async fn relay_read_reports_ended_on_eof() {
        let (tx, mut rx) = tokio::sync::mpsc::channel::<Inbound>(4);
        let (client_side, mut server_side) = tokio::io::duplex(1024);
        server_side
            .write_all(&alktunnels::wire::frame_datagram(b"one").unwrap())
            .await
            .unwrap();
        server_side
            .write_all(&alktunnels::wire::frame_datagram(b"two").unwrap())
            .await
            .unwrap();
        server_side.shutdown().await.unwrap();
        drop(server_side);
        let handle = tokio::spawn(relay_read(client_side, tx, "k".into()));
        let first = rx.recv().await.expect("a datagram arrives");
        assert!(
            matches!(first, Inbound::Datagram(ref key, ref dg) if key == "k" && (&dg[..] == b"one" || &dg[..] == b"two")),
            "a queued datagram: {first:?}"
        );
        // The second datagram (the batched frame) then the EOF
        // marker; drain until the `Ended` marker arrives.
        let mut saw_end = false;
        for _ in 0..3 {
            match rx.recv().await.expect("the queue delivers") {
                Inbound::Datagram(key, dg) => {
                    assert_eq!(key, "k");
                    assert!(
                        &dg[..] == b"one" || &dg[..] == b"two",
                        "a queued datagram: {dg:?}"
                    );
                }
                Inbound::Ended => {
                    saw_end = true;
                    break;
                }
            }
        }
        assert!(saw_end, "the EOF exit reports Ended");
        tokio::time::timeout(std::time::Duration::from_secs(5), handle)
            .await
            .expect("the relay exits after the EOF")
            .expect("no panic");
    }
}