alktunnels/tasks/tunnels/consumer-session.md

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

id	name	status	depends_on	scope	risk	impact	level	tags
tunnels/consumer-session	Consumer half — TunnelSession (open/adopt, data planes, teardown)	completed	tunnels/params tunnels/wire-codec tunnels/producer-open-op	broad	high	phase	implementation	consumer session teardown

Description

Implement src/consumer.rs per consumer.md + ADR-005: TunnelSession — the typed client, both construction paths, the substrate-shaped data plane, and teardown ownership. The reverse POC's consumer (/workspace/alktunnels-reverse-poc/src/consumer.rs — ReverseTunnel) is the seed; this task generalizes it to the spec's session type.

Construction

• Forward path: TunnelSession::open(client: &ChannelClient, params: TunnelParams) -> Result<Self, TunnelOpenError> — client.open_channel(OP_TUNNEL_OPEN, params, TUNNEL_ALPN) → adopt → split the channel BiStream → data plane by substrate. Typed errors (ADR-049 §4; never a phantom session).
• Reverse path: TunnelSession::adopt(manager: &ChannelManager, channel_id: u32, alpn: impl Into<String>) -> Result<Self, TunnelOpenError> — adopt a worker-allocated ID (early arrivals parked), install the data plane from the adopted halves. open_reverse_channel(hub_call, params, auth_token) -> Result<u32, ReverseOpenError> as the free function the assembly layer calls before adopt (the POC's shape — the hub's call surface is a CallConnection, not a ChannelClient, so the two-step is the honest API).

Data plane

• Stream variant: stream_halves() → (&mut dyn AsyncRead, &mut dyn AsyncWrite) (borrowed access); take_halves(self) → owned boxed halves (the session's halves ARE the tunnel — raw pass-through).
• Datagram variant: send_datagram(&[u8]) -> Result<(), TunnelIoError> (frame → write → flush; Oversize >65535); recv_datagram() -> Result<Option<Bytes>, TunnelIoError> — Some(bytes) per datagram (possibly empty, len=0 legal), None only on stream EOF (the F-2 layering; the POC's read_one_datagram incremental loop).
• Wrong-substrate operations are WrongSubstrate errors (the POC's shape).

Pump ownership + teardown (ADR-005 — the point)

• pump_against(self, accepted: impl AsyncRead + AsyncWrite + ...) — for the reverse path: spawn pump_bidi(channel_bistream, accepted_read, accepted_write), hold the returned handle. Returns the session (builder-style) or takes self and returns the handle — pick the shape that makes holding easy; document it.
• close(self) -> bool — abort the pump (if session-owned) + teardown_channel (ungraceful path).
• join(self) -> (u64, u64, bool) — await pump completion, then reap; copy counts for observability. Pump-less sessions (after take_halves): completes immediately, reaps only, (0, 0, reaped) (consumer.md's pinned semantics).
• Drop — abort + sync teardown_channel (best-effort; never leak the entry). No Clone.

Tests

Extend the producer task's integration suite: forward open (session halves drive a duplex), reverse open+adopt+pump_against (the POC's ReverseTunnel tests: round-trip, half-close W4, join_and_reap copy counts, out-of-band close + self-reaping, pump-less join), datagram variant (round-trip incl. empty datagram via the codec), teardown matrix (close/join/Drop paths — no leaked channel entries asserted via channel_ids()).

Acceptance Criteria

• open + adopt construction both present; typed error surfaces exact
• stream_halves/take_halves/send_datagram/recv_datagram per consumer.md
• Teardown matrix: close (ungraceful), join (graceful + copy counts), pump-less join (0, 0, reaped), Drop (no leak)
• Half-close semantics test (W4's shape) passes
• No Clone on TunnelSession (compile-asserted)
• Clippy/fmt clean; wasm32 check passes; cargo test green

References

• docs/architecture/consumer.md (the normative API)
• docs/architecture/decisions/005-consumer-session-owns-teardown.md
• POC reference: /workspace/alktunnels-reverse-poc/src/consumer.rs (the seed shape to generalize)

Notes

Agent fills during implementation.

• adopt takes substrate: Substrate (a deviation from the task sketch's adopt(manager, channel_id, alpn)): the substrate shapes the data plane (raw halves vs the UDP codec) and cannot be inferred from the channel ID. The ALPN is observability-only (alkcall manager semantics); the substrate rode the open call that produced the ID — passing it to adopt keeps the two calls consistent.
• pump_against takes the whole accepted stream (S: AsyncRead + AsyncWrite) and splits internally — tokio::io::split at the pump boundary, matching the POC's pump_against(accepted). Returns the session (builder-style); the pump handle is session-owned and private (close/join/Drop are the only handles needed).
• take_halves(self) returns TakenHalves { session, read, write } (not bare halves): the session must stay alive as the channel-entry owner (reap via close/join/Drop); returning bare halves would strand the entry. join on the held session completes immediately ((0, 0, reaped) — pump-less).
• open_reverse_channel's hub_call is &CallConnection (not &Arc<CallConnection>) — borrowing is the honest shape; callers hold the Arc.
• Fields are Option + idempotent Drop: a Drop-impl type cannot be destructured, so take_halves/pump_against/close/ join take fields out via .take(); an emptied session drops as a no-op. Teardown is idempotent (teardown_channel on a reaped entry is Err(UnknownChannel), ignored).
• Bogus-ID adopt succeeds by design (the manager parks early arrivals for any not-yet-seen ID — the adoption-race cover, alkcall ADR-047 §5); the honest AdoptFailed probe is the ID-collision path (same ID adopted twice), pinned in the tests.
• recv_datagram keeps a pending queue: one chunk may batch several frames; DatagramReader::feed decodes all completions per feed, so frames beyond the first queue for subsequent recv_datagram calls (dropped frames were a bug the tests caught).
• Harness gained wire_forward (ForwardTopology): the forward path's consumer is the connect side (from_connection, pure consumer) and the producer accepts via the adapter; the accept-side transport-identity posture attaches the dialer's identity in the install hook (modeled; the transport resolves it out-of-band for real). all_substrate_dial = stream echo + framed UDP echo.

Summary

Agent fills this on completion.

Implemented TunnelSession (src/consumer.rs, ~415 lines) per consumer.md + ADR-005: both construction paths (open via ChannelClient::open_channel, adopt via ChannelManager:: adopt_channel + open_reverse_channel on the CallConnection), the substrate-shaped data planes (raw halves for stream; the ADR-003 codec with a pending frame queue for UDP), pump ownership (pump_against spawns pump_bidi and holds the handle), and the full teardown matrix — close (abort + ungraceful reap), join (await + reap + copy counts; pump-less (0, 0, reaped)), Drop (abort + sync reap, never leaks). No Clone (compile_fail doc-test).

Tests: tests/consumer_session.rs — 12 integration tests over the two harness topologies: forward round-trip (session halves), take- halves + pump-less join, datagram round-trip incl. the F-2 empty datagram, wrong-substrate typed errors, oversize-at-frame-time, reverse open+adopt+pump round-trip, W4 half-close, out-of-band close

• self-reap (the W3 split), Drop-no-leak (pump + pump-less), AdoptFailed-on-collision, failed-open-typed (no phantom session), concurrent reverse sessions. Harness additions: wire_forward (ForwardTopology) + all_substrate_dial/framed_udp_echo_dial.

Verified: cargo test green (38 total), clippy -D warnings clean (native + wasm32), fmt clean, wasm32 check passes (default crate stays wasm-clean).