alksocks/tasks/implementation/core-producer-connect.md

id, name, status, depends_on, scope, risk, impact, level

id	name	status	depends_on	scope	risk	impact	level
implementation/core-producer-connect	Core — producer half: establisher + dispatch + registration + CONNECT	completed	implementation/core-seam-types implementation/core-params-errors implementation/core-test-harness	moderate	medium	phase	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 — 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<T> — 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

• Establisher: session-gate shape (Establishment::default(), identity recorded; witness seam optional)
• Pump handler: typestate-driven dispatch; pump_bidi inline for CONNECT; no substrate types leak
• register_socks5_openable wires spec + establisher + pump + injected seams; missing-callback commands refuse RFC-faithfully
• serve_socks5_over<T> is substrate-free and directly testable
• Integration tests: round-trip, refusal mapping (0x05), FORBIDDEN fails closed, identity witness, refusal arms for ASSOCIATE/BIND, half-close teardown
• 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<IdentityWitness> (the alktunnels *_with_witness posture) rather than a required witness — 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 — 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<T> (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.