---
id: tunnels/consumer-session
name: Consumer half — TunnelSession (open/adopt, data planes, teardown)
status: pending
depends_on: [tunnels/params, tunnels/wire-codec, tunnels/producer-open-op]
scope: broad
risk: high
impact: phase
level: implementation
tags: [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.

## Summary

> Agent fills this on completion.