--- id: implementation/core-producer-connect name: "Core: producer half, establisher + dispatch + registration + CONNECT" status: completed depends_on: [implementation/core-seam-types, implementation/core-params-errors, implementation/core-test-harness] scope: moderate risk: medium impact: phase level: implementation --- ## Description Implement the producer half of `src/producer.rs` per producer.md: the session-gate establisher, the command-dispatch pump handler, the registration surface, and the CONNECT data plane (POC #1's validated scope). This is the crate's core task: the open op all three commands ride. Deliverables (port from `/workspace/alksocks-connect-poc/src/ producer.rs` + `/workspace/alksocks-udp-poc/src/producer.rs`, adapted to the crate's seam types and alkcall 0.8.0): 1. **The session-gate establisher** (`socks5_session_establisher`): validates/records the opener identity (the CF-005/CF-006 seam in ADR-005) and applies session-level policy. Runs `Establishment::default()`-shaped (no plan payload; the dial happens post-establishment, so there is nothing to thread establisher→handler; POC #1's wire note). Optional `identity_witness` probe seam (the alktunnels posture) for tests. 2. **The pump handler** (`make_socks5_pump_handler(dial, egress, accept)`): `accept_bi()` → drive the forked typestate over the `BiStream` (`accept_no_auth → read_command`) → dispatch: - **CONNECT** → invoke `DialFn(target)` at the command-read interception point; success: `reply_success` then `pump_bidi` **awaited inline** (R-02: the returned `JoinHandle` tracks the data-plane lifetime; early return = teardown-at-birth). Refusal: `reply_error` with the mapped RFC code (dial-failure → `ConnectionRefused`; scope-class → `ConnectionNotAllowed`), no phantom data plane. - **UDP ASSOCIATE / TCP BIND** → this task lands the dispatch arms as RFC-faithful refusals (`CommandNotSupported`); the associate/bind producer tasks replace them. (The dispatch skeleton exists from day one so the regression guard (one registration, all three commands) is testable early.) - The handler is substrate-agnostic by construction: it sees boxed halves, never a socket type. 3. **`register_socks5_openable`**: spec + establisher + pump via `register_openable_with_establisher` (`timeout: None` = the 10s default), scope-gated `socks5:open` (the registry runs the ACL before the establisher). Missing callbacks = those commands refuse RFC-faithfully (`CommandNotSupported`-class), never a panic (ADR-008). 4. **`serve_socks5_over`**: the substrate-free protocol entry point (the POC shape): generic over `T: AsyncRead + AsyncWrite + Unpin`, driving the typestate and dispatching. This is what makes the producer testable without channels (the differential-suite pattern) and reusable by the batch-2 local backend. 5. **Tests** (via the harness + the raw RFC client helper): CONNECT echo round-trip through `pump_bidi`; dial refusal → 0x05 in the reply, no phantom data plane; FORBIDDEN for a scope-less identity (fails closed, no phantom channel); per-call opener identity witnessed (CF-006); ASSOCIATE/BIND refuse `CommandNotSupported` (the pre-vertical posture); half-close teardown (client EOF ends the session cleanly). ## Acceptance Criteria - [x] Establisher: session-gate shape (`Establishment::default()`, identity recorded; witness seam optional) - [x] Pump handler: typestate-driven dispatch; `pump_bidi` inline for CONNECT; no substrate types leak - [x] `register_socks5_openable` wires spec + establisher + pump + injected seams; missing-callback commands refuse RFC-faithfully - [x] `serve_socks5_over` is substrate-free and directly testable - [x] Integration tests: round-trip, refusal mapping (0x05), FORBIDDEN fails closed, identity witness, refusal arms for ASSOCIATE/BIND, half-close teardown - [x] `cargo test` green; clippy/fmt clean; both wasm checks clean ## References - docs/architecture/producer.md (the normative shapes; the wiring requirement) - docs/architecture/decisions/001/004/005/006 (op, seams, auth, gates) - POC sources: `/workspace/alksocks-connect-poc/src/producer.rs` (CONNECT + establisher), `/workspace/alksocks-udp-poc/src/producer.rs` (the dispatch skeleton + registration) - alkcall ADR-049 (establishment; R-02 inline-await contract), `/workspace/@alkdev/alktunnels/src/producer.rs` (the 0.8.0 registration template) ## Notes > Implementation notes (2026-09-23): > > - `Socks5ProducerBuilder` was added beyond the deliverable list: > it is the spec'd assembly shape (producer.md §Wiring, the > `Socks5Producer::builder()` sketch); `register_socks5_openable` > stays as the explicit per-registration form. Both ride the same > `register_openable_with_establisher` call; the builder's > `register`/`serve_socks5_over` delegate to them verbatim. > - The `establisher` accepts `Option` (the > alktunnels `*_with_witness` posture) rather than a required > witness, because the witness is a test seam, not a protocol surface; > `register_socks5_openable` threads it through. > - The ASSOCIATE arm carries the `EgressFactory` parameter (and the > BIND arm the `BindAcceptFactory`) in the registration signature > from day one, so the associate/bind tasks replace the refusal arms > without touching the registration shape (no churn for the > callers). Until those tasks land, the seams are accepted and > their commands refuse `CommandNotSupported` (the pre-vertical > posture, ADR-008's missing-callback class), tested. > - CONNECT's refusal error shape: dial-seam refusal returns > `SocksServerError::Bug("dial refused in-stream")` after > `reply_error` (the refusal was already RFC-reported; the error > return is the session-end signal, not a second wire message). > A missing dial callback refuses `CommandNotSupported` via > `SocksServerError::UnknownCommand(cmd)`. ## Summary > Completed 2026-09-23 (commit on `main`). `src/producer.rs` filled: > the session-gate establisher ([`socks5_session_establisher`]; > CF-005/CF-006 witness seam, `Establishment::default()` per POC #1's > no-plan wire note), the command-dispatch pump handler > ([`make_socks5_pump_handler`]: `accept_bi` → typestate > (`accept_no_auth → read_command`) → dispatch), the CONNECT data > plane (`DialFn` at the command-read interception point → > `reply_success(sentinel)` → `pump_bidi` awaited inline, R-02), > `register_socks5_openable` (spec + establisher + pump via > `register_openable_with_establisher`, `timeout: None` = the 10s > default, `socks5:open`-gated), `serve_socks5_over` (the > substrate-free entry), and the builder assembly shape. > > Dispatch arms: CONNECT serves; ASSOCIATE/BIND refuse > `CommandNotSupported` (the pre-vertical posture; the skeleton > exists so the regression guard is testable early; the associate/ > bind tasks replace the arms). > > Tests: 11 in-module (substrate-free, duplex-driven: round-trip + > half-close teardown, refusal mapping per variant 0x05/0x03/0x04/ > 0x02/0x01, missing-dial refusal, ASSOCIATE/BIND arms, dispatch > guard, establisher witness + no-plan, builder shape, sentinel > shape) + 5 harness integration (end-to-end CONNECT echo through > channels with the identity witness, dial refusal 0x05 with no > phantom data plane, scope-class refusal 0x02, ASSOCIATE/BIND > refusals with CONNECT still serving (the one-registration > regression guard, repeated CONNECTs on one registration). The > harness's `RegisterMode` gained the real-producer mode; the > minimal CONNECT stand-in is retained only as a harness-internal > helper. > > Verification: `cargo test` 87+19+37 green; clippy `--all-targets` > clean; fmt clean; both wasm checks clean; `cargo doc` clean; > `--all-features` green.