---
id: implementation/core-consumer-connect
name: "Core — consumer half: Socks5ChannelSession (CONNECT) + e2e tests"
status: completed
depends_on: [implementation/core-producer-connect]
scope: narrow
risk: low
impact: component
level: implementation
---

## Description

Implement the CONNECT consumer session in `src/consumer.rs` per
consumer.md §The Channels-Native Sessions: `Socks5ChannelSession` —
the typed client that opens `channels/socks5/sub` and drives the
client side of the RFC conversation in-stream (POC #1's
`use_stream` shape).

Deliverables (port from `/workspace/alksocks-connect-poc/src/
consumer.rs`, adapted to the crate's params/error surface):

1. **`Socks5ChannelSession::connect(client, target)`**: opens the
   channel via `ChannelClient::open_channel(OP_SOCKS5_OPEN, params,
   SOCKS5_ALPN)` (trivial params — ADR-001), then drives the forked
   client typestate's `use_stream` path over the channel's
   `BiStream` (generic over the backing stream — the POC #1 fit:
   `Socks5Stream` impls `AsyncRead + AsyncWrite` itself, so the
   session over a `BiStream` is the intended use).
2. **Target construction** via `ToTargetAddr` (IP-string forms parse
   to IP targets; domain form passes through unresolved — ADR-012).
3. **Post-reply payload pipe**: after the reply, the session is a
   raw byte pipe — expose `split()` (read/write halves) and
   `into_inner()` (the underlying `Socks5Stream<BiStream>` for
   advanced use). 0 B framing (CONNECT).
4. **Teardown ownership** (alktunnels ADR-005 posture, inherited):
   the session owns the channel; on Drop it closes/joins. No
   `Clone` (compile-asserted — aliasing a session would alias its
   teardown; the alktunnels `compile_fail` doctest shape).
5. **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 the session type (consumer.md §Error surface).
6. **Tests** (end-to-end through the harness): connect → echo
   round-trip (payload bytes both directions); refused dial → the
   RFC reply code surfaced as a typed session error; FORBIDDEN open
   → typed open error, no session; half-close semantics (dropping
   the write half EOFs the producer's pump cleanly); domain-form
   target passes through unresolved (assert the wire bytes, the
   echo dial ignores the target).

## Acceptance Criteria

- [ ] `Socks5ChannelSession` with `connect`/`split`/`into_inner`
      per the POC shape
- [ ] Teardown ownership on Drop; no `Clone` (compile-asserted)
- [ ] Typed error surface (open errors vs reply-code refusals)
- [ ] E2E tests through the harness (round-trip, refusals,
      half-close, domain pass-through)
- [ ] `cargo test` green; clippy/fmt clean; both wasm checks clean

## References

- docs/architecture/consumer.md §The Channels-Native Sessions
- docs/architecture/decisions/003-udp-handle-swap.md (the framing
  mirror — CONNECT is the 0 B case)
- POC source: `/workspace/alksocks-connect-poc/src/consumer.rs`
- alktunnels template: `/workspace/@alkdev/alktunnels/src/consumer.rs`
  (the teardown-ownership + no-Clone posture)

## Notes

> 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

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.