fix: pump_session deadlock when backend emits >64 chunks before exit (P1)

The stdout drainer started only after the pumps+exit join completed,
but the pumps share a bounded (64-slot) channel with it. Once the
channel filled, pump_stdout parked on send, the join never completed,
and the session hung with the client actively reading (review #003 P1;
reproduced empirically at N=62 ok / N=63 hang). Any real session (cat,
ls -R) exceeds 64 chunks instantly.

pump_session now spawns drain_chunks (owning writer_rx + client_write)
before the join; the exit chunk is still enqueued after the join and
writer_tx is dropped after it, so the single FIFO preserves the
exit-chunk-is-last invariant (ADR-005).

Tests (review #003 P13):
- backpressure regression: 80 chunks through a real drive_session,
  in-order delivery + sentinel + exit asserted, 10s per-read timeout
  so a regression fails instead of hanging (validated against the bug:
  fails with the old order reinstated)
- wire: MAX_CHUNK_LEN exact-boundary round trip (limit is inclusive)
- session: stream_type 0/3 from the server ignored mid-stream without
  desynchronizing framing or losing stdout/exit

Verification: cargo test 98 lib (default) / 141 --all-features,
clippy all-targets + wasm32 -D warnings, fmt --check, wasm check, doc
--no-deps all clean.

src/adapter.rs

@@ -422,9 +422,11 @@ where
 
 /// Phase 3: the bidirectional pump. Three concurrent tasks plus a drainer.
 ///
-/// Enforces the exit-chunk-is-last invariant (ADR-055): the adapter waits for
-/// BOTH the stdout/stderr pumps to complete AND `exit_code` to resolve
-/// before enqueueing the exit chunk.
+/// Enforces the exit-chunk-is-last invariant (ADR-055): the exit chunk
+/// is enqueued only after the stdout/stderr pumps have finished sending
+/// AND `exit_code` has resolved. The drainer runs concurrently with the
+/// pumps so the shared writer channel is always being consumed — see
+/// `drain_chunks`.
 async fn pump_session<W, R>(
     client_write: W,
     client_read: R,
@@ -434,7 +436,7 @@ where
     W: AsyncWrite + Send + Unpin + 'static,
     R: AsyncRead + Send + Unpin + 'static,
 {
-    let (writer_tx, mut writer_rx) = mpsc::channel::<Chunk>(64);
+    let (writer_tx, writer_rx) = mpsc::channel::<Chunk>(64);
 
     let TtyHandle {
         stdin,
@@ -444,6 +446,14 @@ where
         control,
     } = handle;
 
+    // The drainer MUST be running while the pumps send: it is the only
+    // consumer of the writer channel, so starting it only after the
+    // pumps+exit join would let a backend producing more chunks than the
+    // channel capacity fill the channel, park the pumps' `send` calls,
+    // and deadlock the session before the join could complete (review
+    // #003 P1).
+    let drainer = tokio::spawn(drain_chunks(client_write, writer_rx));
+
     let writer_tx_out = writer_tx.clone();
     let stdout_pump = tokio::spawn(pump_stdout(stdout, writer_tx_out));
 
@@ -477,6 +487,22 @@ where
     drop(writer_tx);
     drop(input_pump);
 
+    let _ = drainer.await;
+    debug!("tty: session complete");
+    Ok(())
+}
+
+/// Drain the writer channel to the client in arrival order. Spawned
+/// before the pumps start so chunk producers are never parked on a full
+/// channel without an active consumer (review #003 P1). The single FIFO
+/// preserves the exit-chunk-is-last invariant (ADR-055): the exit chunk
+/// is enqueued only after every pump has finished sending. Ends when all
+/// senders are dropped (normal close) or on a client write error (the
+/// pumps' subsequent sends then fail and the pumps wind down).
+async fn drain_chunks<W>(client_write: W, mut writer_rx: mpsc::Receiver<Chunk>)
+where
+    W: AsyncWrite + Send + Unpin + 'static,
+{
     let mut chunk_writer = ChunkWriter::new(client_write);
     while let Some(chunk) = writer_rx.recv().await {
         if let Err(e) = chunk_writer.write_chunk(&chunk).await {
@@ -485,8 +511,6 @@ where
         }
     }
     let _ = chunk_writer.into_inner().shutdown().await;
-    debug!("tty: session complete");
-    Ok(())
 }
 
 async fn send_exit_chunk(writer_tx: &mpsc::Sender<Chunk>, code: i32) {
@@ -1050,6 +1074,88 @@ mod tests {
         let _ = session.await;
     }
 
+    /// Backpressure regression (review #003 P1): a backend producing
+    /// more stdout chunks than the writer-channel capacity (64) must
+    /// not deadlock the session. Before the fix, the drainer started
+    /// only after the pumps+exit join, so the 65th chunk parked the
+    /// stdout pump's send, the join never completed, and the client
+    /// (actively reading) hung with nothing delivered. Empirically: 62
+    /// chunks passed, 63 hung. The drainer now runs concurrently, so
+    /// 80 chunks flow, the sentinel follows the last data chunk, and
+    /// the exit chunk is still last (ADR-055).
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn backpressure_more_chunks_than_channel_capacity() {
+        const N: usize = 80;
+        let (backend, _control, _cancel) = TestBackend::builder().build();
+        let backends = make_backends(backend.clone());
+        let (mut client, server) = make_client_and_server();
+
+        let identity = identity_with_scope(TTY_OPEN_SCOPE);
+        let session = tokio::spawn(async move {
+            drive_session_server(server, backends, None, identity).await;
+        });
+
+        client.write_negotiation(TEST_NEG).await;
+
+        let stdout_tx = backend.take_stdout_tx().await.expect("stdout tx");
+        let _ = backend.take_stderr_tx().await;
+        // Produce from a spawned task: with the pre-fix code the 65th
+        // send would park here forever (channel full, drainer not yet
+        // running), and the test's own awaits below would hang instead
+        // of observing the deadlock via the bounded client reads.
+        let producer = tokio::spawn(async move {
+            for i in 0..N {
+                let payload = format!("chunk-{i:03}");
+                if stdout_tx
+                    .send(Bytes::copy_from_slice(payload.as_bytes()))
+                    .await
+                    .is_err()
+                {
+                    break;
+                }
+            }
+        });
+
+        let exit_tx = backend.take_exit_tx().await.expect("exit tx");
+        exit_tx.send(Ok(0)).unwrap();
+
+        // Bounded: a regression must fail this test, not hang it.
+        let deadline = std::time::Duration::from_secs(10);
+        let mut data_chunks = 0usize;
+        loop {
+            let read = tokio::time::timeout(deadline, client.read_chunk())
+                .await
+                .expect("chunk read must not stall (P1 regression: deadlock)");
+            let (st, bytes) = read;
+            match st {
+                STREAM_STDOUT if !bytes.is_empty() => {
+                    assert_eq!(
+                        bytes.as_ref(),
+                        format!("chunk-{data_chunks:03}").as_bytes(),
+                        "chunks must arrive in order"
+                    );
+                    data_chunks += 1;
+                }
+                STREAM_STDOUT => {
+                    assert_eq!(
+                        data_chunks, N,
+                        "sentinel must follow exactly {N} data chunks"
+                    );
+                }
+                crate::wire::STREAM_CTRL_OUT => {
+                    let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+                    assert_eq!(v["type"], "exit");
+                    assert_eq!(v["code"], 0);
+                    break;
+                }
+                other => panic!("unexpected stream_type {other}"),
+            }
+        }
+        assert_eq!(data_chunks, N, "all {N} chunks must be delivered");
+        let _ = producer.await;
+        let _ = session.await;
+    }
+
     #[tokio::test]
     async fn stdin_eof_zero_length_chunk_closes_backend_stdin() {
         let (backend, _control, _cancel) = TestBackend::builder().build();

src/session.rs

@@ -975,6 +975,80 @@ mod tests {
         let _ = server_handle.await;
     }
 
+    /// Client→server stream types (0 and 3) arriving from the server are
+    /// protocol violations the read pump ignores — they must be skipped
+    /// without ending the pump, desynchronizing framing, or disturbing
+    /// the stdout routing. The server interleaves them with real stdout
+    /// chunks, then sends a well-formed exit; the session should deliver
+    /// exactly the stdout chunks and the exit code. (The first chunk
+    /// must be a legitimate server→client type: a leading `0x00` byte
+    /// is the negotiation-error disambiguation marker — ADR-052 §5 —
+    /// and would surface `NegotiationRejected` before the pump starts.)
+    #[tokio::test]
+    async fn read_pump_ignores_client_to_server_stream_types_from_server() {
+        let (client, mut server) = duplex(64 * 1024);
+        let server_handle = tokio::spawn(async move {
+            use tokio::io::{AsyncReadExt, AsyncWriteExt};
+            let mut len_buf = [0u8; 4];
+            let _ = server.read_exact(&mut len_buf).await;
+            let len = u32::from_be_bytes(len_buf) as usize;
+            let mut body = vec![0u8; len];
+            let _ = server.read_exact(&mut body).await;
+
+            async fn write_chunk(
+                server: &mut tokio::io::DuplexStream,
+                stream_type: u8,
+                payload: &[u8],
+            ) {
+                let mut header = [0u8; 5];
+                header[0] = stream_type;
+                header[1..].copy_from_slice(&(payload.len() as u32).to_be_bytes());
+                server.write_all(&header).await.unwrap();
+                if !payload.is_empty() {
+                    server.write_all(payload).await.unwrap();
+                }
+            }
+
+            write_chunk(&mut server, crate::wire::STREAM_STDOUT, b"out1").await;
+            write_chunk(
+                &mut server,
+                crate::wire::STREAM_STDIN,
+                b"server-must-not-send-stdin",
+            )
+            .await;
+            write_chunk(
+                &mut server,
+                crate::wire::STREAM_CTRL_IN,
+                br#"{"type":"resize"}"#,
+            )
+            .await;
+            write_chunk(&mut server, crate::wire::STREAM_STDOUT, b"out2").await;
+            let exit = br#"{"type":"exit","code":3}"#;
+            write_chunk(&mut server, crate::wire::STREAM_CTRL_OUT, exit).await;
+            let _ = server.flush().await;
+        });
+        let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None);
+        let session = TtySession::connect_direct(client_conn, test_negotiate("mock"))
+            .await
+            .expect("connect_direct");
+
+        let stdout = session.recv_stdout().await;
+        let collected: Vec<Bytes> = stdout.collect().await;
+        let data: Vec<Bytes> = collected.into_iter().filter(|b| !b.is_empty()).collect();
+        assert_eq!(
+            data,
+            vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")],
+            "stdout routing must be unaffected by the ignored chunks"
+        );
+
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait())
+            .await
+            .expect("wait didn't time out")
+            .expect("wait returns exit code");
+        assert_eq!(code, 3);
+        let _ = server_handle.await;
+    }
+
     /// `connect_direct` returns `NegotiationRejected` when the server
     /// rejects the negotiation with an error frame (M1). The server
     /// reads the negotiation frame and writes back a length-prefixed

src/wire.rs

@@ -422,6 +422,31 @@ mod tests {
         assert!(matches!(err, RawError::ChunkTooLarge(v) if v == over));
     }
 
+    /// Exact-boundary: a payload of exactly `MAX_CHUNK_LEN` is valid
+    /// (the limit is inclusive). Round-trips through a full writer→
+    /// reader pass over a duplex pair large enough to hold the whole
+    /// payload plus the header.
+    #[tokio::test]
+    async fn chunk_at_exact_max_len_boundary() {
+        let (a, b) = duplex(MAX_CHUNK_LEN as usize + CHUNK_HEADER_LEN + 64);
+        let mut writer = ChunkWriter::new(a);
+        let mut reader = ChunkReader::new(b);
+
+        let payload = Bytes::from(vec![0xA5u8; MAX_CHUNK_LEN as usize]);
+        writer
+            .write_chunk(&Chunk {
+                stream_type: STREAM_STDOUT,
+                bytes: payload.clone(),
+            })
+            .await
+            .expect("MAX_CHUNK_LEN payload must be accepted");
+
+        let read = reader.read_chunk().await.expect("boundary chunk readable");
+        assert_eq!(read.stream_type, STREAM_STDOUT);
+        assert_eq!(read.bytes.len() as u32, MAX_CHUNK_LEN);
+        assert_eq!(read.bytes, payload);
+    }
+
     #[tokio::test]
     async fn connection_closed_truncated_header() {
         let (mut a, mut b) = duplex(8 * 1024);