consumer: Socks5ChannelSession (CONNECT) — teardown ownership + e2e suite

- Socks5ChannelSession::connect: open_channel(OP_SOCKS5_OPEN, {},
  SOCKS5_ALPN) then the forked client typestate's use_stream path
  over the channel BiStream (POC #1's fit); post-reply the session
  is the raw byte pipe (0 B framing)
- Target construction via ToTargetAddr: IP-string forms parse to IP
  targets, domain form passes through unresolved (ADR-012)
- split() borrowed halves (SplitPipe) / into_inner() owned pipe
  (TakenInner — the session rides along as the teardown owner);
  channel_id() for correlation
- Teardown ownership (alktunnels ADR-005 posture): Drop sync-reaps
  the adopted entry (teardown_channel is sync; Drop cannot await) +
  the pipe drop EOFs the producer pump structurally; close() is the
  graceful path (write-half shutdown, then reap); the in-stream
  refusal tears the channel down in connect (teardown-at-birth —
  no phantom channel); no Clone (compile_fail doctest)
- Error surface: open failures are the typed SocksOpenError reasons
  (ADR-049 §4); a refused target is the RFC reply code through
  SocksSessionError::Rfc (consumer.md §Error surface)
- Harness: the consumer-side topology (wire_consumer_side[_with] —
  the consumer dials with a real ChannelClient; the producer accepts
  with the real registration) + witnessing_echo_dial; the
  allow(dead_code) attr rides the alktunnels harness posture
- tests/consumer-connect.rs: 9 e2e tests — echo round-trip + CF-006
  witness, target forms at the dial seam (ADR-012), refused dial →
  typed RFC error + no phantom channel, FORBIDDEN open → typed open
  error + no session, drop teardown + connection health, half-close
  EOF semantics, into_inner pairing, Send shape

Verified: cargo test (219 green incl. the compile_fail doctest),
clippy --all-targets -D warnings, fmt --check,
check/clippy --target wasm32-unknown-unknown

src/consumer.rs

@@ -1,5 +1,238 @@
-//! The consumer half: the channels-native session types (CONNECT /
-//! ASSOCIATE / the BIND reply#2 reader) and teardown ownership.
+//! The consumer half: the channels-native session types and teardown
+//! ownership (consumer.md §The Channels-Native Sessions).
 //!
-//! Stub: filled by the `implementation/core-consumer-*` tasks
-//! (CONNECT first).
+//! [`Socks5ChannelSession`] is the CONNECT session: it opens
+//! `channels/socks5/sub` via `ChannelClient::open_channel` (trivial
+//! params — ADR-001), drives the forked client typestate's
+//! `use_stream` path over the channel's `BiStream` (POC #1's fit —
+//! `Socks5Stream` impls `AsyncRead + AsyncWrite` itself, so a session
+//! over a `BiStream` is the intended use, not a hack), and after the
+//! reply presents the raw byte pipe (0 B framing — the session
+//! reads/writes the target's bytes directly). The ASSOCIATE and BIND
+//! sessions land with their tasks (`core-consumer-associate` /
+//! `core-consumer-bind`).
+//!
+//! Target construction rides [`ToTargetAddr`]: IP-string forms parse
+//! to IP targets; the domain form passes through unresolved (ADR-012
+//! — resolution is the egress's job, never the protocol's).
+//!
+//! Teardown ownership (alktunnels ADR-005 posture, inherited via
+//! ADR-004): the session owns the adopted channel entry. On Drop it
+//! reaps it (`ChannelManager::teardown_channel` — sync, best-effort;
+//! `Drop` cannot await) and the data plane EOFs structurally (the
+//! mux's implicit-EOF sentinel on the dropped write half). The
+//! teardown-at-birth discipline covers the in-stream refusal path:
+//! a refused CONNECT tears the adopted channel down before the
+//! error returns — a refused session never leaves a phantom channel.
+//! The session does **not** implement `Clone` (compile-asserted on
+//! the struct): aliasing a session would alias its teardown.
+//!
+//! Error surface (consumer.md §Error surface): open failures are the
+//! typed [`SocksOpenError`] reasons (ADR-049 §4 — a refused session
+//! never yields a phantom channel); a refused target is an RFC reply
+//! code surfaced through [`SocksSessionError::Rfc`].
+
+use alkcall::channels::client::ChannelClient;
+use alkcall::channels::manager::ChannelManager;
+use alkcall::core::types::BiStream;
+use tokio::io::{AsyncWriteExt, ReadHalf, WriteHalf};
+
+use crate::error::{SocksOpenError, SocksSessionError};
+use crate::params::{Socks5Params, OP_SOCKS5_OPEN, SOCKS5_ALPN};
+use crate::rfc1928::client::{Config, Socks5Stream};
+use crate::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
+use crate::rfc1928::Socks5Command;
+
+/// A completed SOCKS5 CONNECT session on the consumer side: the
+/// adopted channel + the post-reply byte pipe to the target.
+///
+/// The session is a handle to one channel (one session = one channel
+/// = one CONNECT, convention 10). It does **not** implement `Clone`:
+///
+/// ```compile_fail
+/// use alksocks::consumer::Socks5ChannelSession;
+/// fn assert_clone<T: Clone>() {}
+/// assert_clone::<Socks5ChannelSession>();
+/// ```
+///
+/// The inner stream is `Option` so `Drop`/`close` can empty the
+/// type (the alktunnels `Drop`-impl posture); a spent session drops
+/// as a teardown-only no-op.
+pub struct Socks5ChannelSession {
+    channel_id: u32,
+    manager: Option<ChannelManager>,
+    inner: Option<Socks5Stream<BiStream>>,
+}
+
+impl std::fmt::Debug for Socks5ChannelSession {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Socks5ChannelSession")
+            .field("channel_id", &self.channel_id)
+            .field("connected", &self.inner.is_some())
+            .finish()
+    }
+}
+
+impl Socks5ChannelSession {
+    /// CONNECT to `target` through a produced SOCKS5 resource: opens
+    /// the channel (`open_channel(OP_SOCKS5_OPEN, {}, SOCKS5_ALPN)`
+    /// — the trivial params, ADR-001), drives the no-auth handshake
+    /// and the CONNECT request in-stream over the channel's
+    /// `BiStream`, and returns the session owning both.
+    ///
+    /// Error surface: an open failure is [`SocksSessionError::Open`]
+    /// (the typed ADR-049 reasons; no channel was ever returned); a
+    /// refused target surfaces the RFC reply code through
+    /// [`SocksSessionError::Rfc`] — and the adopted channel is torn
+    /// down before the error returns (no phantom channel).
+    pub async fn connect(
+        client: &ChannelClient,
+        target: impl ToTargetAddr,
+    ) -> Result<Self, SocksSessionError> {
+        let target: TargetAddr = target.to_target_addr()?;
+        let (channel_id, send, recv) = client
+            .open_channel(OP_SOCKS5_OPEN, Socks5Params::as_json(), SOCKS5_ALPN)
+            .await
+            .map_err(SocksOpenError::from)?;
+
+        let bidi = BiStream::from_joined(recv, send);
+        let negotiated = async {
+            let mut inner = Socks5Stream::use_stream(bidi, None, Config::default()).await?;
+            inner.request(Socks5Command::TCPConnect, target).await?;
+            Result::<Socks5Stream<BiStream>, SocksSessionError>::Ok(inner)
+        };
+
+        match negotiated.await {
+            Ok(inner) => Ok(Self {
+                channel_id,
+                manager: Some(client.manager().clone()),
+                inner: Some(inner),
+            }),
+            Err(err) => {
+                let _ = client.manager().teardown_channel(channel_id);
+                Err(err)
+            }
+        }
+    }
+
+    /// The adopted channel's id (observability + correlation; one
+    /// session = one channel).
+    pub fn channel_id(&self) -> u32 {
+        self.channel_id
+    }
+
+    /// The post-CONNECT pipe as split read/write halves (0 B framing
+    /// — raw bytes to/from the target). The halves borrow the
+    /// session: teardown stays owned (`close`/`Drop`). `None` after
+    /// [`Self::into_inner`] took the pipe.
+    pub fn split(&mut self) -> Option<SplitPipe<'_>> {
+        let inner = self.inner.as_mut()?;
+        Some(tokio::io::split(inner))
+    }
+
+    /// Take the underlying [`Socks5Stream<BiStream>`] (advanced use —
+    /// the forked client type's socket accessors, owned-half pumps).
+    /// The session rides along still owning the channel's teardown:
+    /// drop the pair (or `close` the session) when done — dropping
+    /// `inner` EOFs the data plane structurally.
+    pub fn into_inner(mut self) -> TakenInner {
+        let inner = self.inner.take();
+        TakenInner {
+            session: self,
+            inner: inner.expect("into_inner on a spent session"),
+        }
+    }
+
+    /// Tear the session down gracefully: half-close the pipe (the
+    /// EOF sentinel reaches the producer's pump), then reap the
+    /// adopted channel entry. Returns whether the entry existed.
+    pub async fn close(mut self) -> bool {
+        if let Some(inner) = self.inner.as_mut() {
+            let _ = inner.shutdown().await;
+        }
+        self.inner.take();
+        self.reap()
+    }
+
+    fn reap(&mut self) -> bool {
+        self.manager
+            .take()
+            .map(|m| m.teardown_channel(self.channel_id).is_ok())
+            .unwrap_or(false)
+    }
+}
+
+impl Drop for Socks5ChannelSession {
+    /// Sync-reap the adopted channel entry (`teardown_channel` is
+    /// sync — best-effort; `Drop` cannot await) and drop the pipe
+    /// (the mux's implicit-EOF sentinel EOFs the producer's pump).
+    /// Dropping without `close` never leaks the channel entry.
+    fn drop(&mut self) {
+        self.inner.take();
+        let _ = self.reap();
+    }
+}
+
+/// The owned pipe [`Socks5ChannelSession::into_inner`] yields, paired
+/// with the session that still owns the channel's teardown.
+pub struct TakenInner {
+    /// The teardown owner — reap via `close`/`Drop`.
+    pub session: Socks5ChannelSession,
+    /// The post-CONNECT pipe (the forked client type over the
+    /// channel's `BiStream`).
+    pub inner: Socks5Stream<BiStream>,
+}
+
+/// The borrowed halves [`Socks5ChannelSession::split`] yields (the
+/// `tokio::io::split` pair over the pipe). A half-close rides the
+/// write half's [`AsyncWriteExt::shutdown`].
+pub type SplitPipe<'s> = (
+    ReadHalf<&'s mut Socks5Stream<BiStream>>,
+    WriteHalf<&'s mut Socks5Stream<BiStream>>,
+);
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alkcall::channels::client::ChannelOpenError;
+
+    fn assert_send<T: Send>() {}
+
+    #[test]
+    fn session_shape_is_send_and_not_clone() {
+        assert_send::<Socks5ChannelSession>();
+        assert_send::<TakenInner>();
+    }
+
+    #[test]
+    fn debug_impl_names_the_channel_and_liveness() {
+        let session = Socks5ChannelSession {
+            channel_id: 7,
+            manager: None,
+            inner: None,
+        };
+        let rendered = format!("{session:?}");
+        assert!(rendered.contains("channel_id: 7"), "{rendered}");
+        assert!(rendered.contains("connected: false"), "{rendered}");
+    }
+
+    #[test]
+    fn spent_session_reap_is_a_no_op() {
+        let mut session = Socks5ChannelSession {
+            channel_id: 9,
+            manager: None,
+            inner: None,
+        };
+        assert!(!session.reap(), "no manager — nothing reaped");
+        assert!(session.split().is_none(), "no pipe after teardown");
+    }
+
+    #[test]
+    fn open_errors_carry_the_typed_surface_through_connects_mapping() {
+        // The mapping `connect` applies to an open failure: the typed
+        // `SocksOpenError` vocabulary rides `SocksSessionError::Open`.
+        let open = SocksOpenError::Open(ChannelOpenError::MissingChannelId);
+        let session: SocksSessionError = open.into();
+        assert!(matches!(session, SocksSessionError::Open(_)));
+    }
+}

src/lib.rs

@@ -29,6 +29,7 @@ pub mod params;
 pub mod producer;
 pub mod wire;
 
+pub use consumer::Socks5ChannelSession;
 pub use egress::{
     egress_factory, BindAccept, BindAcceptError, BindAcceptFactory, BindAcceptFactoryError,
     BindTargetHandle, DialFn, Egress, EgressError, EgressFactory, SocksDialError, TargetHandle,

tasks/implementation/core-consumer-connect.md

@@ -1,7 +1,7 @@
 ---
 id: implementation/core-consumer-connect
 name: "Core — consumer half: Socks5ChannelSession (CONNECT) + e2e tests"
-status: pending
+status: completed
 depends_on: [implementation/core-producer-connect]
 scope: narrow
 risk: low
@@ -70,8 +70,51 @@ consumer.rs`, adapted to the crate's params/error surface):
 
 ## Notes
 
-> To be filled by implementation agent
+> Filled by the implementation agent (2026-09-23):
+
+- The session shape is exactly the POC's, adapted to the crate
+  surface: `Socks5Stream::use_stream(bidi, None, Config::default())`
+  over `BiStream::from_joined(recv, send)`, then `request(TCPConnect,
+  target)`. `Config::default()` is the no-auth posture (the
+  `skip_auth` knob stays internal to the fork; the identity seam is
+  the open op's, CF-005/CF-006 — no RFC-facing auth on this path).
+- `split()` returns borrowed halves (`tokio::io::split` over a
+  `&mut Socks5Stream<BiStream>`; the `SplitPipe` type alias), not
+  owned ones: the session keeps teardown ownership for every shape,
+  including the split-driven one. The owned path is `into_inner()`
+  (`TakenInner { session, inner }` — the session rides along as the
+  teardown owner).
+- Refusal teardown: an in-stream refusal (Rfc error) tears the
+  adopted channel down inside `connect` before the error returns
+  (teardown-at-birth; the phantom-channel property, tested both
+  sides). The producer's own handler-exit teardown races the
+  consumer's — the leak-check probe polls past that race.
+- Drop is the sync-reap posture (`teardown_channel` is sync;
+  `Drop` cannot await) + the structural EOF of the dropped pipe; the
+  graceful path is `close()` (write-half shutdown first, then reap).
+- Harness: added the consumer-side topology (`wire_consumer_side[_with]`
+  — the consumer dials with a real `ChannelClient`; the producer
+  accepts with the real registration) plus `witnessing_echo_dial`
+  (the dial-witness target-form probe). The `allow(dead_code)` attr
+  rides the alktunnels harness posture (each test binary compiles the
+  harness and uses a different subset).
+- No `Clone` is compile-asserted by the doctest; a runtime probe
+  pins `Send` (the `Clone` bound itself is the doctest's job).
 
 ## Summary
 
-> To be filled on completion
+Implemented `Socks5ChannelSession` (CONNECT) in `src/consumer.rs` —
+`connect`/`channel_id`/`split`/`into_inner`/`close` + `Drop` teardown
+ownership and the `SplitPipe`/`TakenInner` companions — with the
+typed error surface (`SocksOpenError` for opens, the RFC reply code
+through `SocksSessionError::Rfc` for refused targets) and
+teardown-at-birth on the in-stream refusal path. E2E suite
+(`tests/consumer-connect.rs`, 9 tests over the new consumer-side
+harness topology): echo round-trip + CF-006 witness, target forms
+(IP-string parses / domain passes through — ADR-012, wire-observed
+at the dial), refused dial → typed RFC error + no phantom channel,
+FORBIDDEN open → typed open error + no session, drop teardown +
+connection health, half-close EOF semantics, `into_inner` pairing,
+and the no-`Clone`/`Send` compile shape (the `compile_fail` doctest
+pins `!Clone`). Verification: `cargo test` (219 incl. the doctest),
+clippy `-D warnings` (all targets), fmt, and both wasm checks clean.

tests/consumer-connect.rs

@@ -0,0 +1,432 @@
+//! The consumer-half integration tests — `Socks5ChannelSession`
+//! (CONNECT) end to end through the harness's consumer-side topology
+//! (the producer's real command-dispatch path over channels; the
+//! alkcall 0.8.0 establishment surface consumed from the consumer's
+//! own `ChannelClient`).
+//!
+//! Covered: connect → echo round-trip (payload bytes both
+//! directions); the dial-witness target forms (IP-string parses to an
+//! IP target, domain passes through unresolved — ADR-012); refused
+//! dial → the RFC reply code surfaced as a typed session error; the
+//! scope-less FORBIDDEN open → typed open error, no session; drop
+//! teardown (no leaked channel entries, the connection stays
+//! healthy); half-close semantics (the write-half shutdown EOFs the
+//! producer's pump cleanly); the compile-asserted no-`Clone` posture.
+
+mod harness;
+
+use std::sync::Arc;
+
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+use alksocks::consumer::{Socks5ChannelSession, TakenInner};
+use alksocks::egress::SocksDialError;
+use alksocks::error::SocksSessionError;
+use alksocks::rfc1928::consts;
+use alksocks::rfc1928::util::target_addr::{TargetAddr, ToTargetAddr};
+use alksocks::rfc1928::SocksError as SocksErrorKind;
+
+use harness::{
+    echo_dial, refusing_dial, wire_consumer_side, wire_consumer_side_with, ConsumerSideTopology,
+};
+
+/// A `SocksOpenError`-carrying open failure (the FORBIDDEN probe's
+/// expectation shape).
+fn expect_forbidden(err: &SocksSessionError) -> bool {
+    match err {
+        SocksSessionError::Open(open) => match open.open_ref() {
+            alkcall::channels::client::ChannelOpenError::CallFailed { error } => {
+                error.code == "FORBIDDEN"
+            }
+            _ => false,
+        },
+        _ => false,
+    }
+}
+
+/// A dial that always refuses (the refusal-riding-in-stream probe's
+/// dial — the RFC reply code must surface through the session type,
+/// not as an open failure).
+fn refusing_session_dial() -> alksocks::egress::DialFn {
+    refusing_dial(SocksDialError::Refused("target down".into()))
+}
+
+/// Both managers hold only channel 0 (the phantom-channel property,
+/// asserted from both sides). The producer's entry is reaped by its
+/// open-op wrapper when the pump handler exits — an await-able
+/// teardown that races the consumer-side reap; this probe polls past
+/// the race instead of asserting instantly.
+async fn no_leaked_channels(topo: &ConsumerSideTopology) {
+    let deadline = std::time::Duration::from_secs(5);
+    let consumer_ids = topo.consumer.manager().channel_ids();
+    assert!(
+        consumer_ids.iter().all(|&id| id == 0),
+        "consumer leaked channel entries: {consumer_ids:?}"
+    );
+    tokio::time::timeout(deadline, async {
+        loop {
+            let producer_ids = topo.producer_manager.channel_ids();
+            if producer_ids.iter().all(|&id| id == 0) {
+                return;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("the producer's handler-exit teardown reaps the entry");
+}
+
+/// The happy path: `connect` → the session is a raw byte pipe → echo
+/// round-trip both directions; the dial seam saw the parsed target;
+/// the establisher witnessed the opener identity (CF-006).
+#[tokio::test]
+async fn connect_echo_round_trip_and_dial_witness() {
+    let topo = wire_consumer_side(echo_dial()).await;
+
+    let mut session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("echo.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session connects");
+
+    // The `split()`-returned halves are the last use of the session;
+    // the tail test's alias import lives here.
+    session
+        .split()
+        .expect("post-reply pipe")
+        .1
+        .write_all(b"ping")
+        .await
+        .expect("write the probe");
+    let (mut read, _) = session.split().expect("post-reply pipe");
+    let mut buf = [0u8; 4];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        read.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the echo");
+    assert_eq!(&buf, b"ping");
+
+    // The establisher witnessed the per-call opener identity.
+    assert_eq!(
+        topo.identity_witness.lock().await.as_deref(),
+        Some("consumer")
+    );
+
+    session.close().await;
+    no_leaked_channels(&topo).await;
+}
+
+/// Target construction (ADR-012): an IP-string form parses to an IP
+/// target (the dial seam receives `TargetAddr::Ip`); the domain form
+/// passes through unresolved (`TargetAddr::Domain`). Both ride the
+/// same happy path.
+#[tokio::test]
+async fn target_construction_ip_strings_parse_domains_pass_through() {
+    let targets = Arc::new(tokio::sync::Mutex::new(Vec::new()));
+    let topo = wire_consumer_side(harness::witnessing_echo_dial(Arc::clone(&targets))).await;
+
+    // The domain form: passes through unresolved (the dial's job).
+    let session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("echo.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("domain connect");
+    session.close().await;
+
+    // The IP-string form: parsed to an IP target by `ToTargetAddr`.
+    let session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("10.0.0.1", 8443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("ip-string connect");
+    session.close().await;
+
+    let dialed = targets.lock().await.clone();
+    assert_eq!(
+        dialed,
+        vec![
+            TargetAddr::Domain("echo.test".to_owned(), 443),
+            TargetAddr::Ip(std::net::SocketAddr::from(([10, 0, 0, 1], 8443)))
+        ],
+        "the dial seam received the parsed target forms (ADR-012)"
+    );
+    no_leaked_channels(&topo).await;
+}
+
+/// A refused dial: the RFC reply code surfaces through the session
+/// type (`SocksSessionError::Rfc` — the in-stream refusal, never an
+/// open failure), and no phantom channel remains on either side.
+#[tokio::test]
+async fn refused_dial_surfaces_rfc_reply_code_no_phantom_channel() {
+    let topo = wire_consumer_side(refusing_session_dial()).await;
+
+    let err = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("down.test", 80u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect_err("the dial refused — the reply code rides the session error");
+
+    match &err {
+        SocksSessionError::Rfc(SocksErrorKind::ReplyError(code)) => {
+            assert_eq!(code.as_u8(), consts::SOCKS5_REPLY_CONNECTION_REFUSED);
+        }
+        other => panic!("expected an Rfc reply-code refusal, got {other:?}"),
+    }
+
+    // Teardown-at-birth: the adopted channel is reaped before the
+    // error returns (no phantom channel, either side).
+    no_leaked_channels(&topo).await;
+
+    // The connection is healthy: a fresh connect through a working
+    // dial succeeds immediately.
+    let topo2 = wire_consumer_side(echo_dial()).await;
+    let session = Socks5ChannelSession::connect(
+        topo2.consumer.as_ref(),
+        ("echo.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("fresh connect after the refusal");
+    session.close().await;
+}
+
+/// A scope-less identity: the registry's FORBIDDEN open failure is
+/// the typed `SocksOpenError` through `SocksSessionError::Open` —
+/// no session, no phantom channel (the open gate is the boundary).
+#[tokio::test]
+async fn forbidden_open_is_a_typed_open_error_with_no_session() {
+    let bare = alkcall::core::auth::Identity {
+        id: "scopeless".to_string(),
+        scopes: vec![],
+        resources: Default::default(),
+    };
+    let topo = wire_consumer_side_with(echo_dial(), None, None, Some(bare)).await;
+
+    let err = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("gated.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect_err("a scope-less identity fails closed");
+
+    assert!(
+        expect_forbidden(&err),
+        "FORBIDDEN is the open error: {err:?}"
+    );
+    no_leaked_channels(&topo).await;
+}
+
+/// Drop teardown: dropping a live session mid-pipe reaps the adopted
+/// channel entry (no leak) and the connection stays healthy — a
+/// fresh connect succeeds immediately after.
+#[tokio::test]
+async fn drop_teardown_reaps_and_connection_stays_healthy() {
+    let topo = wire_consumer_side(echo_dial()).await;
+
+    let mut session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("dropped.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session connects");
+    session
+        .split()
+        .expect("pipe")
+        .1
+        .write_all(b"mid")
+        .await
+        .expect("write before the drop");
+    let (mut read, _) = session.split().expect("pipe");
+    let mut buf = [0u8; 3];
+    read.read_exact(&mut buf).await.expect("echo before drop");
+    assert_eq!(&buf, b"mid");
+
+    // Drop mid-stream (the teardown-ownership posture).
+    drop(session);
+    no_leaked_channels(&topo).await;
+
+    // The connection is healthy: a fresh session round-trips.
+    let mut fresh = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("fresh.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("fresh connect after the drop");
+    fresh
+        .split()
+        .expect("pipe")
+        .1
+        .write_all(b"again")
+        .await
+        .expect("write");
+    let (mut read, _) = fresh.split().expect("pipe");
+    let mut buf = [0u8; 5];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        read.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the fresh echo");
+    assert_eq!(&buf, b"again");
+    fresh.close().await;
+    no_leaked_channels(&topo).await;
+}
+
+/// Half-close semantics: shutting the write half EOFs the producer's
+/// pump cleanly (the EOF sentinel), while the read half stays alive
+/// until the target's data flows. The pump's completion is the
+/// producer-side evidence (the handler tracks the data plane — R-02).
+#[tokio::test]
+async fn half_close_write_eof_ends_the_producer_pump() {
+    let topo = wire_consumer_side(echo_dial()).await;
+
+    let session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("halfclose.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session connects");
+    let channel_id = session.channel_id();
+    assert!(channel_id >= 1, "a producer-allocated id");
+    assert!(topo.consumer.manager().has_channel(channel_id));
+
+    // Split into owned halves via `into_inner` (the advanced-use
+    // accessor), then half-close the write half.
+    let TakenInner {
+        session: _owner,
+        inner,
+    } = session.into_inner();
+    let (mut read, mut write) = tokio::io::split(inner);
+    write.write_all(b"pre").await.expect("write");
+    let mut buf = [0u8; 3];
+    read.read_exact(&mut buf).await.expect("echo");
+    assert_eq!(&buf, b"pre");
+    write.shutdown().await.expect("half-close the pipe");
+
+    // The producer's pump observes the EOF sentinel and ends (the
+    // two-pump shape's shutdown-on-completion; the echo target's
+    // pump exits with it). The session-side observable: the read
+    // half drains to EOF.
+    drop(write);
+    let mut eof = [0u8; 1];
+    tokio::time::timeout(std::time::Duration::from_secs(10), async {
+        loop {
+            match read.read(&mut eof).await {
+                Ok(0) | Err(_) => break,
+                Ok(_) => continue,
+            }
+        }
+    })
+    .await
+    .expect("the read resolves after the half-close");
+    drop(read);
+    drop(_owner);
+    no_leaked_channels(&topo).await;
+}
+
+/// `into_inner` yields the forked client type paired with the
+/// teardown-owning session; the pair's `close` reaps the channel.
+#[tokio::test]
+async fn into_inner_pairs_the_pipe_with_the_teardown_owner() {
+    let topo = wire_consumer_side(echo_dial()).await;
+
+    let session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("inner.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session connects");
+    let channel_id = session.channel_id();
+    let TakenInner { session, inner } = session.into_inner();
+    assert!(topo.consumer.manager().has_channel(channel_id));
+
+    let (mut read, write) = tokio::io::split(inner);
+    let mut buf = [0u8; 1];
+    // The pipe is idle; the close below reaps the entry.
+    let _ = tokio::time::timeout(std::time::Duration::from_millis(50), read.read(&mut buf)).await;
+    drop(write);
+    drop(read);
+
+    let reaped = session.close().await;
+    assert!(reaped, "the adopted channel entry existed");
+    no_leaked_channels(&topo).await;
+}
+
+/// The no-`Clone` posture: the type is `!Clone` because its fields
+/// are (the `compile_fail` doctest pins the type-level posture).
+#[test]
+fn session_type_is_not_clone() {
+    // A `Clone`-bound assertion on a non-`Clone` type cannot be a
+    // runtime test; the doctest is the compile-assert. This probe
+    // pins the Send shape the type promises.
+    fn assert_send<T: Send>() {}
+    assert_send::<Socks5ChannelSession>();
+}
+
+/// The read half stays alive across a half-close (the echo flows
+/// after the write half's EOF was already seen by the producer): a
+/// payload written before the shutdown round-trips, and the read
+/// half observes the target's own EOF after it (the two-pump
+/// half-close semantics, both directions).
+#[tokio::test]
+async fn both_directions_survive_across_halves() {
+    let topo = wire_consumer_side(echo_dial()).await;
+
+    let mut session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("halves.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session connects");
+
+    // Write half: two writes, flush each (the chunk stream carries
+    // them in order).
+    let (_, mut write) = session.split().expect("pipe");
+    write.write_all(b"one").await.expect("write one");
+    write.flush().await.expect("flush");
+    write.write_all(b"two").await.expect("write two");
+    write.flush().await.expect("flush");
+    drop(write);
+
+    // Read half: the echo arrives in order.
+    let (mut read, _) = session.split().expect("pipe");
+    let mut buf = [0u8; 6];
+    tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        read.read_exact(&mut buf),
+    )
+    .await
+    .expect("echo timed out")
+    .expect("read the echo");
+    assert_eq!(&buf, b"onetwo");
+
+    session.close().await;
+    no_leaked_channels(&topo).await;
+}
+
+/// The `split()` accessor's `Option` shape: `None` after the pipe
+/// was taken (the spent-session posture; the borrow checker makes
+/// the direct post-`into_inner` call the only reachable shape).
+#[tokio::test]
+async fn close_after_into_inner_reaps() {
+    let topo = wire_consumer_side(echo_dial()).await;
+    let session = Socks5ChannelSession::connect(
+        topo.consumer.as_ref(),
+        ("spent.test", 443u16).to_target_addr().unwrap(),
+    )
+    .await
+    .expect("the session connects");
+    let channel_id = session.channel_id();
+    let TakenInner { session, .. } = session.into_inner();
+    // The inner pipe was taken; `close` still reaps the entry.
+    assert!(session.close().await, "the entry existed before the close");
+    no_leaked_channels(&topo).await;
+    assert!(channel_id >= 1);
+}

tests/harness.rs

@@ -1,5 +1,6 @@
 //! Test harness — the producer↔consumer wiring over a
 //! `tokio::io::duplex` carrying the channels 8-byte chunk format.
+#![allow(dead_code)]
 //!
 //! Topology (the alktunnels harness shape, adapted to this crate's
 //! seams; the alkcall 0.8.0 establishment surface under test — this
@@ -26,6 +27,8 @@
 //!
 //! Everything is in-process: no sockets, no `local` feature. The
 //! harness is dev-only (`tests/`) — it stays out of the wasm build.
+//! The `allow(dead_code)` rides the alktunnels posture: each test
+//! binary compiles the harness and uses a different subset.
 
 use std::collections::HashMap;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
@@ -689,6 +692,163 @@ pub async fn wire(dial: DialFn) -> Topology {
     .await
 }
 
+// --- The consumer-side (forward) topology -------------------------------------
+
+/// The consumer-session topology (`Socks5ChannelSession::connect` is
+/// normative for this shape): the consumer dials the transport
+/// (connect side — `ChannelClient::from_connection`, a pure consumer
+/// serving nothing) and the producer accepts (the adapter + an
+/// install hook registering the SOCKS5 openable over the injected
+/// seams on the producer's serving registry).
+pub struct ConsumerSideTopology {
+    /// The consumer's own client (connect side, pure consumer). Its
+    /// manager adopted the producer-allocated channel IDs; the
+    /// session reaps via it.
+    pub consumer: Arc<ChannelClient>,
+    /// The producer's manager (accept side, serving) — for asserting
+    /// the handler-exit reaping (no leaked channel entries).
+    pub producer_manager: ChannelManager,
+    /// What the establisher's per-call auth carried (the CF-006
+    /// probe — `None` until an open runs).
+    pub identity_witness: IdentityWitness,
+}
+
+/// Wire the consumer-side topology over one duplex transport:
+/// consumer dials (identity attached before `from_connection` — the
+/// CF-005 (b) transport-identity path), producer accepts (the hook
+/// attaches the resolved identity to channel 0 before the
+/// `CallConnection` — the load-bearing POC #1 wiring requirement).
+pub async fn wire_consumer_side_with(
+    dial: DialFn,
+    egress: Option<EgressFactory>,
+    accept: Option<BindAcceptFactory>,
+    identity: Option<Identity>,
+) -> ConsumerSideTopology {
+    let producer_op_registry = Arc::new(OperationRegistry::new());
+    let (producer_manager_tx, mut producer_manager_rx) =
+        tokio::sync::mpsc::channel::<ChannelManager>(1);
+    let identity_witness: IdentityWitness = Arc::new(tokio::sync::Mutex::new(None));
+    let witness_for_hook = Arc::clone(&identity_witness);
+    let resolved = Arc::new(identity.unwrap_or_else(consumer_identity));
+
+    let resolved_for_hook = Arc::clone(&resolved);
+    let dial_for_hook = Arc::clone(&dial);
+    let install_hook: InstallChannelZero = Arc::new(
+        move |manager: ChannelManager, channel0_conn: CoreConnection, _auth: AuthContext| {
+            let registry = Arc::clone(&producer_op_registry);
+            let manager_tx = producer_manager_tx.clone();
+            let witness = Arc::clone(&witness_for_hook);
+            let resolved = Arc::clone(&resolved_for_hook);
+            let dial = Arc::clone(&dial_for_hook);
+            let egress = egress.clone();
+            let accept = accept.clone();
+            tokio::spawn(async move {
+                let channel0_bidi = match channel0_conn.accept_bi().await {
+                    Ok(s) => s,
+                    Err(_) => return,
+                };
+                let (writer, reader) = split_single_stream(channel0_bidi);
+                let _ = channel0_conn.set_identity((*resolved).clone());
+                let call_connection = Arc::new(CallConnection::new_single_stream(
+                    channel0_conn,
+                    Arc::clone(&writer),
+                ));
+                let core =
+                    ChannelCore::new(manager.clone(), alkcall::channels::policy::default_policy());
+                let ops = ChannelOperations::with_default_policy(manager.clone());
+                ops.register_on(&registry)
+                    .expect("register channel ops on producer");
+                alksocks::producer::register_socks5_openable(
+                    &core,
+                    &registry,
+                    AuthContext::anonymous(b"alk/socks5"),
+                    Some(dial),
+                    egress,
+                    accept,
+                    Some(witness),
+                )
+                .expect("register socks5 openable (the consumer-side topology)");
+                let _ = manager_tx.send(manager).await;
+                let dp = Dispatcher::new(
+                    registry,
+                    Arc::new(alkcall::core::auth::NoopIdentityProvider),
+                );
+                dp.serve_single_stream(call_connection, reader, writer)
+                    .await;
+            })
+        },
+    );
+
+    let (consumer_end, producer_end) = tokio::io::duplex(64 * 1024);
+    let consumer_conn = CoreConnection::from_bidi(consumer_end, b"alk/channels".to_vec(), None);
+    let producer_conn = CoreConnection::from_bidi(producer_end, b"alk/channels".to_vec(), None);
+    consumer_conn
+        .set_identity((*resolved).clone())
+        .expect("transport identity set once");
+
+    let adapter = ChannelsAdapter::new(install_hook, Arc::new(alkcall::channels::policy::NoCap));
+    let producer_transport_auth = AuthContext::anonymous(b"alk/channels");
+    let _adapter_task = tokio::spawn(async move {
+        let _ = alkcall::core::types::ProtocolHandler::handle(
+            &adapter,
+            producer_conn,
+            &producer_transport_auth,
+        )
+        .await;
+    });
+
+    let consumer_client = ChannelClient::from_connection(consumer_conn)
+        .await
+        .expect("consumer channel client init");
+    let producer_manager = producer_manager_rx
+        .recv()
+        .await
+        .expect("producer manager captured");
+
+    ConsumerSideTopology {
+        consumer: Arc::new(consumer_client),
+        producer_manager,
+        identity_witness,
+    }
+}
+
+/// The default consumer-side topology: the CONNECT path over the
+/// given dial, the consumer identity on the transport.
+pub async fn wire_consumer_side(dial: DialFn) -> ConsumerSideTopology {
+    wire_consumer_side_with(dial, None, None, Some(consumer_identity())).await
+}
+
+/// An echo dial that records every dialed target (the
+/// interception-point probe: domain form arrives unresolved —
+/// ADR-012; IP-string forms arrive as parsed IP targets).
+pub fn witnessing_echo_dial(targets: Arc<tokio::sync::Mutex<Vec<TargetAddr>>>) -> DialFn {
+    Arc::new(move |target: TargetAddr| {
+        let targets = Arc::clone(&targets);
+        Box::pin(async move {
+            targets.lock().await.push(target);
+            let (mut echo_end, target_end) = tokio::io::duplex(64 * 1024);
+            tokio::spawn(async move {
+                loop {
+                    let mut buf = vec![0u8; 4096];
+                    match echo_end.read(&mut buf).await {
+                        Ok(0) | Err(_) => return,
+                        Ok(n) => {
+                            if echo_end.write_all(&buf[..n]).await.is_err() {
+                                return;
+                            }
+                        }
+                    }
+                }
+            });
+            let (read_half, write_half) = tokio::io::split(target_end);
+            Ok(TargetHandle {
+                read: Box::new(read_half),
+                write: Box::new(write_half),
+            })
+        })
+    })
+}
+
 // --- The raw RFC 1928 client helper -----------------------------------------
 
 /// A hand-rolled minimal RFC 1928 client over any