feat(review 007 Unit 3): pump_bidi two-pump helper (R-03, ADR-050)

Extracts the two-pump data-plane helper alknet ADR-078 deferred until
the shapes converged (they have: alktunnels POC pump_halves + alktty's
channels session). Purely additive.

- channels::pump::pump_bidi(channel, peer_read, peer_write) -> (u64, u64):
  two joined pumps, shutdown-on-completion per direction; copy counts
  for observability. The channel side is a single AsyncRead +
  AsyncWrite value (the accept_bi BiStream); the peer side takes split
  halves — the establisher's natural dial result (into_split).
- Return (u64, u64), not the review sketch's io::Result<(u64, u64)>:
  both pumps swallow copy errors by contract (mid-stream error =
  abrupt close, no error channel mid-stream per ADR-049 §6), so an
  Err state would be dead code. Deviation recorded in ADR-050.
- alktty's three-pump session does not fit (exit future as a third
  signal) and stays as-is, per the review's scope.
- Tests reproduce the POC's two-pump semantics through the helper:
  bidirectional flow with exact counts, EOF-from-one-side completes
  the other's shutdown (clean EOF at the far end), dead-source =
  EOF-shaped teardown.
- ADR-050 records the decision, deviations, and two-way door type.

Verification: cargo test (625 passed, +2), clippy -D warnings, fmt
--check, doc clean, test --all-features clean, wasm32-unknown-unknown
check clean.

src/channels/mod.rs

@@ -25,6 +25,8 @@
 //!   ALPN crates via `ChannelCore` (ADR-047 §3).
 //! - [`policy`]: `ChannelLifecyclePolicy` + `PerIdentityChannelPolicy`
 //!   (ADR-041, amended by ADR-047 §7 — opener ledger).
+//! - [`pump`]: `pump_bidi` — the two-pump data-plane helper (alknet
+//!   ADR-078, pinned upstream by ADR-050).
 //! - [`client`]: `ChannelClient` — transport-agnostic
 //!   `from_connection` (ADR-043).
 //! - [`self::env`]: `ChannelOperationEnv` extension trait (ADR-047 §4 —
@@ -39,6 +41,7 @@ pub mod manager;
 pub mod mux;
 pub mod operations;
 pub mod policy;
+pub mod pump;
 pub mod reassembly;
 pub mod source;
 pub mod wire;

src/channels/pump.rs

@@ -0,0 +1,204 @@
+//! `pump_bidi` — the two-pump data-plane helper (alknet ADR-078, as
+//! pinned upstream by ADR-050 — review 007 R-03).
+//!
+//! A forwarding channel's data plane is two unidirectional pumps: one
+//! copies channel→peer, the other peer→channel. The contract is
+//! **shutdown-on-completion**: when one pump's copy source EOFs, it
+//! shuts down the opposite sink so the peer sees a clean half-close;
+//! the helper completes when both pumps finish. Copy errors are
+//! EOF-shaped by design (the POC and ADR-078 semantics — there is no
+//! error channel mid-stream; a pump error is an abrupt-close signal,
+//! and shutdown-of-the-opposite-sink is the same either way). The
+//! helper returns the two copy counts for observability.
+//!
+//! The shapes converged across three consumers (the extraction test
+//! ADR-078 set): the alktunnels POC's `pump_halves`, alktty's channels
+//! session, and the assembly-layer copies. This helper pins the
+//! two-pump shape in one place; alktty's three-pump session (an exit
+//! future as a third signal) does not fit and stays as-is.
+
+use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
+
+/// Pump a channel stream against a peer's split halves (ADR-078, as
+/// pinned by ADR-050). Two pumps, joined:
+///
+/// - `channel → peer_write`: copy until the channel EOFs, then
+///   `shutdown()` the peer write half.
+/// - `peer_read → channel`: copy until the peer EOFs, then
+///   `shutdown()` the channel.
+///
+/// Returns `(channel_to_peer, peer_to_channel)` copy counts.
+/// Copy errors are treated as EOF (shutdown still runs) — the
+/// EOF-shaped teardown the ADR-078 contract specifies; there is no
+/// `Err` state because mid-stream errors are indistinguishable from
+/// abrupt closes by design.
+///
+/// The generic bounds follow the POC: the channel side is a single
+/// `AsyncRead + AsyncWrite` value (the `BiStream` the pump handler
+/// received), the peer side arrives as split read/write halves (a
+/// dialed socket, a TTY pty pair — `into_split` is the natural
+/// establisher result).
+///
+/// Await the returned future inline inside the `OpenHandler`'s task —
+/// the handler's returned `JoinHandle` must track the data-plane
+/// lifetime (ADR-049 amendment 2, R-02).
+pub async fn pump_bidi<C, R, W>(channel: C, peer_read: R, mut peer_write: W) -> (u64, u64)
+where
+    C: AsyncRead + AsyncWrite + Unpin,
+    R: AsyncRead + Unpin,
+    W: AsyncWrite + Unpin,
+{
+    let (mut c_read, mut c_write) = tokio::io::split(channel);
+    let mut peer_read = peer_read;
+
+    let c2p = async {
+        let n = tokio::io::copy(&mut c_read, &mut peer_write)
+            .await
+            .unwrap_or(0);
+        // Shutdown-on-completion: the channel side is done writing;
+        // the peer must see the half-close.
+        let _ = peer_write.shutdown().await;
+        n
+    };
+    let p2c = async {
+        let n = tokio::io::copy(&mut peer_read, &mut c_write)
+            .await
+            .unwrap_or(0);
+        let _ = c_write.shutdown().await;
+        n
+    };
+    tokio::join!(c2p, p2c)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt, DuplexStream};
+
+    /// Model the topology precisely: the helper gets the channel
+    /// (`BiStream`-shaped) and the peer\'s split halves. The test
+    /// drives the *other* ends: `channel_peer` (what the channel\'s
+    /// remote writes/reads) and the substrate far ends. `duplex`
+    /// pairs: a write on one half arrives on its counterpart.
+    struct Topology {
+        /// The channel value handed to the helper.
+        channel: DuplexStream,
+        /// The channel\'s remote end (test-driven).
+        channel_peer: DuplexStream,
+        /// The peer read half handed to the helper (helper reads it).
+        peer_read: DuplexStream,
+        /// The peer write half handed to the helper (helper writes it).
+        peer_write: DuplexStream,
+        /// The peer write half\'s remote (test reads what the helper
+        /// copied).
+        peer_write_far: DuplexStream,
+        /// The peer read half\'s remote (test writes what the helper
+        /// will copy).
+        peer_read_far: DuplexStream,
+    }
+
+    fn topology() -> Topology {
+        // The channel\'s two ends.
+        let (channel, channel_peer) = tokio::io::duplex(64);
+        // The substrate: peer_read\'s far end, peer_write\'s far end.
+        let (peer_read, peer_read_far) = tokio::io::duplex(64);
+        let (peer_write_far, peer_write) = tokio::io::duplex(64);
+        Topology {
+            channel,
+            channel_peer,
+            peer_read,
+            peer_write,
+            peer_write_far,
+            peer_read_far,
+        }
+    }
+
+    /// The two-pump semantics test (review 007 Unit 3 gate): the
+    /// POC\'s `pump_halves` behavior through the helper — data flows
+    /// both directions, shutdown-on-completion on EOF, exact counts.
+    #[tokio::test]
+    async fn pump_bidi_moves_data_both_ways_and_shuts_down_on_completion() {
+        let mut topo = topology();
+
+        let pumped = tokio::spawn(pump_bidi(topo.channel, topo.peer_read, topo.peer_write));
+
+        // channel -> substrate: write at the channel\'s remote, read
+        // at the substrate\'s far end.
+        topo.channel_peer
+            .write_all(b"outbound")
+            .await
+            .expect("write");
+        let mut buf = [0u8; 8];
+        topo.peer_write_far
+            .read_exact(&mut buf)
+            .await
+            .expect("helper copied channel->peer");
+        assert_eq!(&buf, b"outbound");
+
+        // substrate -> channel: write at the substrate\'s far end,
+        // read at the channel\'s remote.
+        topo.peer_read_far
+            .write_all(b"inbound!")
+            .await
+            .expect("write");
+        let mut buf2 = [0u8; 8];
+        topo.channel_peer
+            .read_exact(&mut buf2)
+            .await
+            .expect("helper copied peer->channel");
+        assert_eq!(&buf2, b"inbound!");
+
+        // EOF from the substrate\'s far end: the helper shuts the
+        // channel down (shutdown-on-completion).
+        topo.peer_read_far.shutdown().await.expect("shutdown");
+        let mut eof_buf = Vec::new();
+        let n = topo
+            .channel_peer
+            .read_to_end(&mut eof_buf)
+            .await
+            .expect("clean EOF at the channel remote");
+        assert_eq!(n, 0, "the channel sees EOF after the substrate finished");
+
+        // Finish the other pump (channel remote drops) and collect
+        // the counts.
+        drop(topo.channel_peer);
+        let (c2p, p2c) = pumped
+            .await
+            .expect("helper completes when both pumps finish");
+        assert_eq!(c2p, 8, "channel->peer count");
+        assert_eq!(p2c, 8, "peer->channel count");
+
+        // The substrate\'s far write end observes the helper\'s
+        // shutdown: read returns 0 (EOF), not an error.
+        let mut eof_buf2 = Vec::new();
+        let n2 = topo
+            .peer_write_far
+            .read_to_end(&mut eof_buf2)
+            .await
+            .expect("eof read");
+        assert_eq!(n2, 0, "peer sees clean EOF after the channel side finished");
+    }
+
+    /// Copy errors are EOF-shaped: a pump whose source dies at birth
+    /// still lets the helper complete and shut the opposite sink down
+    /// (no Err state — ADR-078\'s abrupt-close semantics).
+    #[tokio::test]
+    async fn pump_bidi_treats_errors_as_eof_shaped_teardown() {
+        let topo = topology();
+
+        // Drop the substrate\'s far write end: the helper\'s peer_read
+        // source EOFs immediately; the channel side must still be shut
+        // down cleanly once the channel itself finishes.
+        drop(topo.peer_read_far);
+
+        let pumped = tokio::spawn(pump_bidi(topo.channel, topo.peer_read, topo.peer_write));
+
+        // The channel remote first receives what nothing sends — EOF
+        // after the p2c pump finishes; drop the remote to finish the
+        // c2p pump too.
+        drop(topo.channel_peer);
+        let _counts = pumped
+            .await
+            .expect("helper completes despite the dead source");
+    }
+}