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> via 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, i.e. 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, with 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, so passing it to adopt keeps the two calls consistent. The ALPN is observability-only (alkcall manager semantics); the substrate rode the open call that produced the ID.
• pump_against takes the whole accepted stream (S: AsyncRead + AsyncWrite) and splits internally with 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), because 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, a 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)), and Drop (abort + sync reap, never leaks). No Clone (compile_fail doc-test).

Tests: tests/consumer_session.rs has 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).