diff --git a/benches/support/latency.rs b/benches/support/latency.rs
index 622cdbb6..a873173f 100644
--- a/benches/support/latency.rs
+++ b/benches/support/latency.rs
@@ -81,7 +81,7 @@ use std::time::Duration;
 
 use borsh::{BorshDeserialize, BorshSerialize};
 use rumors::Rumors;
-use rumors::link::{Acceptor, Connector, Link, STREAM_COUNT};
+use rumors::link::{Acceptor, Connector, Done, Link, STREAM_COUNT};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::mpsc;
 use tokio::time::{Instant, Sleep};
@@ -303,13 +303,13 @@ pub struct DelayedConnector {
 impl Connector for DelayedConnector {
     type Tx = DelayedWriter;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         let (tx, rx) = delayed_pipe(self.capacity, self.delay);
         self.announce
             .send(rx)
             .await
             .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "peer link is gone"))?;
-        Ok(tx)
+        Ok((tx, Done::discard()))
     }
 }
 
@@ -322,10 +322,11 @@ pub struct DelayedAcceptor {
 impl Acceptor for DelayedAcceptor {
     type Rx = DelayedReader;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         self.streams
             .recv()
             .await
+            .map(|rx| (rx, Done::discard()))
             .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "peer link is gone"))
     }
 }
diff --git a/design/routed-link.md b/design/routed-link.md
index e2a15406..8518b9bd 100644
--- a/design/routed-link.md
+++ b/design/routed-link.md
@@ -318,6 +318,24 @@ Accepted failure modes, stated rather than hidden:
   the cross-stream wait the concurrency clause forbids (and qorb's
   default `max_slots = 16` is below one session's worst-case 18).
   Recorded here so the future evaluation starts from these facts.
+- **DECIDED (2026-08-13): completed streams recover their
+  connections.** Supersedes the single-use decision above; the future
+  case it named arrived (an attested-handshake transport whose RoT
+  serializes handshakes at roughly one per second). The door was not
+  the header after all: no new kind, and no layer-A framing either.
+  The protocol already ends every stream with an in-band end control,
+  so the receiver stopped demanding transport EOF behind it, and the
+  link contract now pairs every stream half with a completion handle
+  (`Done`) invoked exactly at that boundary. The write half goes back
+  to the `Dial` (`recycle`, defaulting to today's drop), the read half
+  back to the router to await its next connect header; a dropped half
+  remains the abort, observed as EOF. Trailing-byte ambiguity does not
+  arise: the codec's reads are exact, so a completed stream leaves the
+  next header untouched, and bytes past an end control belong to the
+  transport, never the session. The half-close rows in the mapping
+  table above describe the abort path only. The qorb facts in the
+  previous entry stand for the pooling `Dial` a deployment builds on
+  `recycle`.
 - **DECIDED (2026-07-29): no peer discovery.** `link()` takes an
   explicit `Addr`. Discovery composes above the adapter (any resolver
   feeding addresses in the caller's namespace) and below it (a `Dial`
diff --git a/examples/swarm.rs b/examples/swarm.rs
index 26df9c02..4cc20207 100644
--- a/examples/swarm.rs
+++ b/examples/swarm.rs
@@ -148,7 +148,7 @@ use ratatui::layout::{Constraint, Layout, Rect};
 use ratatui::style::{Color, Modifier, Style};
 use ratatui::text::{Line, Span};
 use ratatui::widgets::{Block, Borders, Paragraph, Sparkline};
-use rumors::link::{Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink};
+use rumors::link::{Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink};
 use rumors::{Key, Peer, Retire, Rumors, UnorderedMessages};
 use tokio::io::{AsyncRead, AsyncWrite, DuplexStream, ReadBuf};
 
@@ -1159,13 +1159,16 @@ struct CountConnector {
 impl Connector for CountConnector {
     type Tx = CountWrite<DuplexStream>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
-        let tx = self.inner.connect().await?;
-        Ok(CountWrite {
-            inner: tx,
-            wire_bytes: Arc::clone(&self.wire_bytes),
-            rounds: None,
-        })
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
+        let (tx, _) = self.inner.connect().await?;
+        Ok((
+            CountWrite {
+                inner: tx,
+                wire_bytes: Arc::clone(&self.wire_bytes),
+                rounds: None,
+            },
+            Done::discard(),
+        ))
     }
 }
 
diff --git a/src/conformance/link.rs b/src/conformance/link.rs
index bb32a73d..e351fde8 100644
--- a/src/conformance/link.rs
+++ b/src/conformance/link.rs
@@ -63,7 +63,7 @@ use std::task::{Context, Poll};
 use futures::future::{Either, join, join_all, select};
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
-use crate::link::{Acceptor, Connector, Link, LinkParts, STREAM_COUNT};
+use crate::link::{Acceptor, Connector, Done, Link, LinkParts, STREAM_COUNT};
 use crate::{Peer, Rumors};
 
 /// Bytes used to probe stream delivery without assuming any capacity.
@@ -322,11 +322,13 @@ fn duplex_fill(tag: u8) -> Vec<u8> {
     (0..CONTROL_DUPLEX_FILL).map(|i| (i as u8) ^ tag).collect()
 }
 
-/// An opened stream delivers its exact bytes to the peer's acceptor, and the
-/// writer's drop surfaces as end-of-stream after the final byte.
+/// An opened stream delivers its exact bytes to the peer's acceptor,
+/// ended either way the contract allows.
 ///
-/// Probed in both directions: each side's connector against the other
-/// side's acceptor.
+/// A dropped stream's abort surfaces as end-of-stream after the final
+/// byte. A completed stream leaves the acceptor's next streams still
+/// arriving. Probed in both directions: each side's connector against
+/// the other side's acceptor.
 pub async fn check_streams<CRa, CWa, Ca, Aa, CRb, CWb, Cb, Ab>(
     a: Link<CRa, CWa, Ca, Aa>,
     b: Link<CRb, CWb, Cb, Ab>,
@@ -344,36 +346,79 @@ pub async fn check_streams<CRa, CWa, Ca, Aa, CRb, CWb, Cb, Ab>(
     let mut b = b.into_parts();
     probe_stream(&a.connector, &mut b.acceptor).await;
     probe_stream(&b.connector, &mut a.acceptor).await;
+    probe_completed_streams(&a.connector, &mut b.acceptor).await;
+    probe_completed_streams(&b.connector, &mut a.acceptor).await;
 }
 
-/// One direction of [`check_streams`]: a single stream, delivered exactly.
+/// One direction of [`check_streams`]: a single stream, delivered exactly
+/// and ended by the abort (the halves dropped, their handles unused).
 async fn probe_stream<C: Connector, A: Acceptor>(connector: &C, acceptor: &mut A) {
     let send = async {
-        let mut tx = connector
+        let (mut tx, done) = connector
             .connect()
             .await
             .expect("contract: connect succeeds while the peer link lives");
         tx.write_all(PROBE).await.expect("contract: stream write");
         tx.flush().await.expect("contract: stream flush");
-        drop(tx);
+        drop((tx, done));
     };
     let receive = async {
-        let mut rx = acceptor
+        let (mut rx, done) = acceptor
             .accept()
             .await
             .expect("contract: an opened stream is accepted");
         let mut bytes = Vec::new();
         rx.read_to_end(&mut bytes)
             .await
-            .expect("contract: half-close surfaces as end-of-stream");
+            .expect("contract: an abort surfaces as end-of-stream");
         assert_eq!(
             bytes, PROBE,
             "contract: a stream delivers its exact bytes in order",
         );
+        drop((rx, done));
     };
     join(send, receive).await;
 }
 
+/// The completion leg of [`check_streams`]: two consecutive streams,
+/// each ended by completing the write half at its final byte.
+///
+/// The receiver reads exactly the probe and completes its half there,
+/// never probing for end-of-stream: the contract's completion clause
+/// leaves what follows the data transport-defined (nothing at all, on
+/// an instantiation that recovers the connection). The second stream
+/// proves the supply survives the first one's completion, wherever its
+/// connection went.
+async fn probe_completed_streams<C: Connector, A: Acceptor>(connector: &C, acceptor: &mut A) {
+    for _ in 0..2 {
+        let send = async {
+            let (mut tx, done) = connector
+                .connect()
+                .await
+                .expect("contract: connect succeeds while the peer link lives");
+            tx.write_all(PROBE).await.expect("contract: stream write");
+            tx.flush().await.expect("contract: stream flush");
+            done.complete(tx);
+        };
+        let receive = async {
+            let (mut rx, done) = acceptor
+                .accept()
+                .await
+                .expect("contract: an opened stream is accepted");
+            let mut bytes = vec![0u8; PROBE.len()];
+            rx.read_exact(&mut bytes)
+                .await
+                .expect("contract: a completed stream delivers its bytes");
+            assert_eq!(
+                bytes, PROBE,
+                "contract: a stream delivers its exact bytes in order",
+            );
+            done.complete(rx);
+        };
+        join(send, receive).await;
+    }
+}
+
 /// Streams are independent: a stream whose receiver never drains blocks
 /// nothing but itself.
 ///
@@ -421,7 +466,7 @@ async fn probe_independence<C: Connector, A: Acceptor>(connector: &C, acceptor:
     let send = async {
         // The stalled stream: tagged so the receiver can hold it unread
         // wherever it lands in arrival order.
-        let mut stalled = connector
+        let (mut stalled, _) = connector
             .connect()
             .await
             .expect("contract: connect succeeds");
@@ -460,7 +505,7 @@ async fn probe_independence<C: Connector, A: Acceptor>(connector: &C, acceptor:
         // writing to one stream may block only on that stream's receiver.
         let live = async {
             for _ in 1..STREAM_COUNT {
-                let mut tx = connector.connect().await.expect("contract: connect");
+                let (mut tx, _) = connector.connect().await.expect("contract: connect");
                 tx.write_all(&[LIVE_TAG])
                     .await
                     .expect("contract: stream write");
@@ -482,7 +527,7 @@ async fn probe_independence<C: Connector, A: Acceptor>(connector: &C, acceptor:
         let mut stalled = None;
         let mut live_seen = 0usize;
         for _ in 0..STREAM_COUNT {
-            let mut rx = acceptor
+            let (mut rx, _) = acceptor
                 .accept()
                 .await
                 .expect("contract: later streams are accepted beside a stalled one");
@@ -544,7 +589,7 @@ async fn probe_independence_pooled<C: Connector, A: Acceptor>(connector: &C, acc
         // unread wherever it lands in arrival order.
         let mut stalled = Vec::with_capacity(STALLED_COMPLEMENT);
         for _ in 0..STALLED_COMPLEMENT {
-            let mut tx = connector
+            let (mut tx, _) = connector
                 .connect()
                 .await
                 .expect("contract: connect succeeds");
@@ -571,7 +616,7 @@ async fn probe_independence_pooled<C: Connector, A: Acceptor>(connector: &C, acc
         }));
         // The one live stream must flow beside the pressured complement.
         let live = async {
-            let mut tx = connector.connect().await.expect("contract: connect");
+            let (mut tx, _) = connector.connect().await.expect("contract: connect");
             tx.write_all(&[LIVE_TAG])
                 .await
                 .expect("contract: stream write");
@@ -594,7 +639,7 @@ async fn probe_independence_pooled<C: Connector, A: Acceptor>(connector: &C, acc
         let mut held = Vec::with_capacity(STALLED_COMPLEMENT);
         let mut live_seen = 0usize;
         for _ in 0..STREAM_COUNT {
-            let mut rx = acceptor
+            let (mut rx, _) = acceptor
                 .accept()
                 .await
                 .expect("contract: later streams are accepted beside stalled ones");
@@ -693,7 +738,7 @@ async fn probe_concurrency<C: Connector, A: Acceptor>(connector: &C, acceptor: &
         // a capped or open-serializing supply hangs right here.
         let mut held = Vec::with_capacity(STREAM_COUNT);
         for index in 0..STREAM_COUNT {
-            let mut tx = connector
+            let (mut tx, _) = connector
                 .connect()
                 .await
                 .expect("contract: a full complement of opens succeeds");
@@ -721,7 +766,7 @@ async fn probe_concurrency<C: Connector, A: Acceptor>(connector: &C, acceptor: &
         let mut held: Vec<Option<A::Rx>> =
             std::iter::repeat_with(|| None).take(STREAM_COUNT).collect();
         for _ in 0..STREAM_COUNT {
-            let mut rx = acceptor
+            let (mut rx, _) = acceptor
                 .accept()
                 .await
                 .expect("contract: accept succeeds while the peer link lives");
@@ -821,7 +866,8 @@ async fn probe_cancellation<C: Connector, A: Acceptor>(connector: &C, acceptor:
         // would deadlock the probe itself.
         let mut streams = Vec::with_capacity(CANCELLED_DELIVERIES);
         for _ in 0..CANCELLED_DELIVERIES {
-            streams.push(connector.connect().await.expect("contract: connect"));
+            let (tx, _) = connector.connect().await.expect("contract: connect");
+            streams.push(tx);
         }
         let _ = connected.send(());
         // The writes run concurrently: the receiver drains streams in
@@ -853,7 +899,7 @@ async fn probe_cancellation<C: Connector, A: Acceptor>(connector: &C, acceptor:
         // sender does not imply local acceptability (an RTT may separate
         // them), so the poll-drop cycles below start only once a delivery
         // has genuinely surfaced on this side.
-        let first = acceptor
+        let (first, _) = acceptor
             .accept()
             .await
             .expect("contract: accept succeeds while the peer link lives");
@@ -877,7 +923,8 @@ async fn probe_cancellation<C: Connector, A: Acceptor>(connector: &C, acceptor:
             .await;
             match polled_once {
                 Some(rx) => {
-                    drain(rx.expect("contract: accept succeeds while the peer link lives")).await;
+                    let (rx, _) = rx.expect("contract: accept succeeds while the peer link lives");
+                    drain(rx).await;
                     delivered += 1;
                 }
                 None => {
@@ -892,7 +939,7 @@ async fn probe_cancellation<C: Connector, A: Acceptor>(connector: &C, acceptor:
         // Every delivery must now surface from real accepts, however many
         // waits were dropped above.
         while delivered < CANCELLED_DELIVERIES {
-            let rx = acceptor
+            let (rx, _) = acceptor
                 .accept()
                 .await
                 .expect("contract: a delivery in flight across a dropped accept still arrives");
@@ -926,10 +973,10 @@ impl<C: Clone> Clone for CountingConnector<C> {
 impl<C: Connector> Connector for CountingConnector<C> {
     type Tx = C::Tx;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
-        let tx = self.inner.connect().await?;
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
+        let pair = self.inner.connect().await?;
         self.opened.fetch_add(1, Ordering::Relaxed);
-        Ok(tx)
+        Ok(pair)
     }
 }
 
diff --git a/src/conformance/link/tests.rs b/src/conformance/link/tests.rs
index 471d54f8..a960fa26 100644
--- a/src/conformance/link/tests.rs
+++ b/src/conformance/link/tests.rs
@@ -18,7 +18,7 @@ use tokio::sync::mpsc;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 
 use crate::link::{
-    Acceptor, Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink,
+    Acceptor, Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink,
     STREAM_COUNT, memory, memory_with_capacity,
 };
 use crate::testing::{Quiescence, run_to_quiescence};
@@ -49,7 +49,7 @@ fn one_byte_windows_conform() {
 /// loudly instead of silently.
 struct ReversingAcceptor<A: Acceptor> {
     inner: A,
-    held: VecDeque<A::Rx>,
+    held: VecDeque<(A::Rx, Done<A::Rx>)>,
     /// Arrivals buffered before each reversed release.
     batch: usize,
     /// Batches of two or more released: genuine inversions.
@@ -59,7 +59,7 @@ struct ReversingAcceptor<A: Acceptor> {
 impl<A: Acceptor> Acceptor for ReversingAcceptor<A> {
     type Rx = A::Rx;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         if let Some(held) = self.held.pop_front() {
             return Ok(held);
         }
@@ -165,7 +165,7 @@ struct LossyAcceptor<A: Acceptor> {
 impl<A: Acceptor> Acceptor for LossyAcceptor<A> {
     type Rx = A::Rx;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         let rx = self.inner.accept().await?;
         // The loss window: one self-waking yield with the dequeued stream
         // held only in this future's state.
@@ -269,7 +269,7 @@ struct MuxConnector {
 impl Connector for MuxConnector {
     type Tx = MuxTx;
 
-    async fn connect(&self) -> io::Result<MuxTx> {
+    async fn connect(&self) -> io::Result<(MuxTx, Done<MuxTx>)> {
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         // Reserve the close frame's FIFO slot up front, so the drop-time
         // close can be sent synchronously and is never lost to a full FIFO.
@@ -283,13 +283,16 @@ impl Connector for MuxConnector {
             .send(MuxFrame::Open(id))
             .await
             .map_err(|_| mux_gone())?;
-        Ok(MuxTx {
-            id,
-            wire: self.wire.clone(),
-            close: Some(close),
-            in_flight: None,
-            claimed: 0,
-        })
+        Ok((
+            MuxTx {
+                id,
+                wire: self.wire.clone(),
+                close: Some(close),
+                in_flight: None,
+                claimed: 0,
+            },
+            Done::discard(),
+        ))
     }
 }
 
@@ -431,19 +434,22 @@ struct MuxAcceptor {
 impl Acceptor for MuxAcceptor {
     type Rx = MuxRx;
 
-    async fn accept(&mut self) -> io::Result<MuxRx> {
+    async fn accept(&mut self) -> io::Result<(MuxRx, Done<MuxRx>)> {
         loop {
             {
                 let mut state = self.demux.lock().await;
                 if let Some(queue) = state.announced.pop_front() {
-                    return Ok(MuxRx {
-                        queue,
-                        demux: self.demux.clone(),
-                        buffer: Vec::new(),
-                        cursor: 0,
-                        pump: None,
-                        ended: false,
-                    });
+                    return Ok((
+                        MuxRx {
+                            queue,
+                            demux: self.demux.clone(),
+                            buffer: Vec::new(),
+                            cursor: 0,
+                            pump: None,
+                            ended: false,
+                        },
+                        Done::discard(),
+                    ));
                 }
             }
             mux_pump(&self.demux).await?;
@@ -601,18 +607,21 @@ impl<T: AsyncWrite + Unpin> AsyncWrite for CappedTx<T> {
 impl<C: Connector> Connector for CappedConnector<C> {
     type Tx = CappedTx<C::Tx>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         let permit = self
             .permits
             .clone()
             .acquire_owned()
             .await
             .expect("the fixture semaphore is never closed");
-        let inner = self.inner.connect().await?;
-        Ok(CappedTx {
-            inner,
-            _permit: permit,
-        })
+        let (inner, _) = self.inner.connect().await?;
+        Ok((
+            CappedTx {
+                inner,
+                _permit: permit,
+            },
+            Done::discard(),
+        ))
     }
 }
 
@@ -743,12 +752,15 @@ struct WindowedConnector {
 impl Connector for WindowedConnector {
     type Tx = WindowedTx<DuplexStream>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
-        let inner = self.inner.connect().await?;
-        Ok(WindowedTx {
-            inner,
-            window: self.window.clone(),
-        })
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
+        let (inner, _) = self.inner.connect().await?;
+        Ok((
+            WindowedTx {
+                inner,
+                window: self.window.clone(),
+            },
+            Done::discard(),
+        ))
     }
 }
 
@@ -762,12 +774,15 @@ struct WindowedAcceptor {
 impl Acceptor for WindowedAcceptor {
     type Rx = WindowedRx<DuplexStream>;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
-        let inner = self.inner.accept().await?;
-        Ok(WindowedRx {
-            inner,
-            window: self.window.clone(),
-        })
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
+        let (inner, _) = self.inner.accept().await?;
+        Ok((
+            WindowedRx {
+                inner,
+                window: self.window.clone(),
+            },
+            Done::discard(),
+        ))
     }
 }
 
diff --git a/src/link.rs b/src/link.rs
index 12e591dc..ef2e742c 100644
--- a/src/link.rs
+++ b/src/link.rs
@@ -17,7 +17,7 @@
 //!   order the transport delivers them.
 //!
 //! Data streams are session-scoped and cheap: a session opens them lazily
-//! and sparsely (up to [`STREAM_COUNT`], typically far fewer) and closes
+//! and sparsely (up to [`STREAM_COUNT`], typically far fewer) and ends
 //! every one before it completes. Only the control stream survives into
 //! the next session.
 //!
@@ -52,9 +52,15 @@
 //!   open at once, while [`Connector::connect`] calls arrive sparsely and
 //!   mid-session. An instantiation must not require a full complement of
 //!   streams, nor serialize an open behind unrelated stream progress.
-//! - **Half-close.** Dropping a [`Connector::Tx`] ends that stream; the
-//!   peer's [`Acceptor::Rx`] then observes end-of-stream after the final
-//!   bytes. The control stream outlives all data streams of its session.
+//! - **Completion.** A producer ends its stream by handing the
+//!   [`Connector::Tx`] to its [`Done`] after the final bytes, or
+//!   aborts by dropping it. Every written byte reaches the peer either
+//!   way. When the peer aborts, it surfaces at its counterparty's
+//!   [`Acceptor::Rx`] as end-of-stream. A completion may surface as
+//!   nothing at all (a reusing instantiation recovers the connection),
+//!   so a consumer finds the end of the data in the bytes themselves,
+//!   hands its `Rx` back there, and never reads past it. The control
+//!   stream outlives all data streams of its session.
 //! - **Cancellation.** A pending [`Acceptor::accept`] future may be dropped
 //!   at any moment (session teardown is the common source, and the
 //!   conformance suite drops them mid-session); instantiations must tolerate
@@ -158,6 +164,42 @@ use tokio::sync::mpsc;
 /// test, so it cannot drift silently.
 pub const STREAM_COUNT: usize = 17;
 
+/// Where a data-stream half goes at its stream's clean end.
+///
+/// Every [`Connector::connect`] and [`Acceptor::accept`] pairs the
+/// half with one of these. The session invokes it exactly at the
+/// protocol's end of the data, handing the half back; a half dropped
+/// anywhere else (its handle unused) is an abort. A transport that
+/// reuses connections recovers them here; the rest pair every stream
+/// with [`discard`](Self::discard).
+pub struct Done<Half>(Box<dyn FnOnce(Half) + Send>);
+
+impl<Half> Done<Half> {
+    /// For single-use streams, whose clean end is the drop.
+    pub fn discard() -> Self
+    where
+        Half: 'static,
+    {
+        Done(Box::new(drop))
+    }
+
+    /// Hand the completed half to `f`.
+    pub fn new(f: impl FnOnce(Half) + Send + 'static) -> Self {
+        Done(Box::new(f))
+    }
+
+    /// Release `half` at its stream's clean end.
+    pub fn complete(self, half: Half) {
+        (self.0)(half)
+    }
+}
+
+impl<Half> std::fmt::Debug for Done<Half> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("Done(..)")
+    }
+}
+
 /// Opens outgoing unidirectional data streams for one link.
 ///
 /// Opens may arrive concurrently, through clones and through shared
@@ -171,13 +213,14 @@ pub trait Connector: Clone + Send + Sync + 'static {
     /// The write half of one outgoing data stream.
     type Tx: AsyncWrite + Unpin + Send + 'static;
 
-    /// Open one outgoing unidirectional stream.
+    /// Open one outgoing unidirectional stream, paired with where the
+    /// half goes at its clean end.
     ///
     /// # Errors
     ///
     /// Fails only for transport reasons (the link is gone); the session
     /// treats any error as fatal to the session, never retries.
-    fn connect(&self) -> impl Future<Output = io::Result<Self::Tx>> + Send;
+    fn connect(&self) -> impl Future<Output = io::Result<(Self::Tx, Done<Self::Tx>)>> + Send;
 }
 
 /// Accepts incoming unidirectional data streams for one link.
@@ -190,7 +233,8 @@ pub trait Acceptor: Send {
     /// The read half of one incoming data stream.
     type Rx: AsyncRead + Unpin + Send + 'static;
 
-    /// Accept one incoming unidirectional stream.
+    /// Accept one incoming unidirectional stream, paired with where
+    /// the half goes at its clean end.
     ///
     /// Order across streams is the transport's own: the session pairs
     /// streams by the label written as each stream's first bytes, never by
@@ -207,13 +251,13 @@ pub trait Acceptor: Send {
     /// teardown is the common source). A stream that was mid-delivery must
     /// surface from a later `accept` call; it must not be lost while the
     /// link stays healthy.
-    fn accept(&mut self) -> impl Future<Output = io::Result<Self::Rx>> + Send;
+    fn accept(&mut self) -> impl Future<Output = io::Result<(Self::Rx, Done<Self::Rx>)>> + Send;
 }
 
 impl<A: Acceptor> Acceptor for &mut A {
     type Rx = A::Rx;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         A::accept(self).await
     }
 }
@@ -549,13 +593,13 @@ pub struct MemoryConnector {
 impl Connector for MemoryConnector {
     type Tx = DuplexStream;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         let (tx, rx) = tokio::io::duplex(self.capacity);
         self.announce
             .send(rx)
             .await
             .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "peer link is gone"))?;
-        Ok(tx)
+        Ok((tx, Done::discard()))
     }
 }
 
@@ -568,10 +612,11 @@ pub struct MemoryAcceptor {
 impl Acceptor for MemoryAcceptor {
     type Rx = DuplexStream;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         self.streams
             .recv()
             .await
+            .map(|rx| (rx, Done::discard()))
             .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "peer link is gone"))
     }
 }
diff --git a/src/link/erased.rs b/src/link/erased.rs
index 965bb5a2..cc88c13a 100644
--- a/src/link/erased.rs
+++ b/src/link/erased.rs
@@ -16,27 +16,94 @@
 //! yields.
 
 use std::io;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
 
 use futures::future::BoxFuture;
-use std::sync::Arc;
-use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+
+use super::{Acceptor, Connector, Done};
+
+/// A half boxed together with its [`Done`], so completion has the
+/// concrete types.
+struct Bundle<H> {
+    half: H,
+    done: Done<H>,
+}
+
+impl<H: AsyncWrite + Unpin> AsyncWrite for Bundle<H> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        Pin::new(&mut self.half).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.half).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.half).poll_shutdown(cx)
+    }
+}
 
-use super::{Acceptor, Connector};
+impl<H: AsyncRead + Unpin> AsyncRead for Bundle<H> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        Pin::new(&mut self.half).poll_read(cx, buf)
+    }
+}
+
+/// An erased outgoing half, completable through the box.
+pub(crate) trait TxDyn: AsyncWrite + Unpin + Send {
+    /// Release the bundled half at its stream's clean end.
+    fn complete(self: Box<Self>);
+}
+
+impl<H: AsyncWrite + Unpin + Send> TxDyn for Bundle<H> {
+    fn complete(self: Box<Self>) {
+        let Bundle { half, done } = *self;
+        done.complete(half);
+    }
+}
+
+/// An erased incoming half, completable through the box.
+pub(crate) trait RxDyn: AsyncRead + Unpin + Send {
+    /// Release the bundled half at its stream's clean end.
+    fn complete(self: Box<Self>);
+}
+
+impl<H: AsyncRead + Unpin + Send> RxDyn for Bundle<H> {
+    fn complete(self: Box<Self>) {
+        let Bundle { half, done } = *self;
+        done.complete(half);
+    }
+}
 
 /// An owned outgoing stream half with its concrete type erased.
-pub(crate) type DynTx = Box<dyn AsyncWrite + Unpin + Send>;
+pub(crate) type DynTx = Box<dyn TxDyn>;
 
 /// An owned incoming stream half with its concrete type erased.
-pub(crate) type DynRx = Box<dyn AsyncRead + Unpin + Send>;
+pub(crate) type DynRx = Box<dyn RxDyn>;
 
 /// Object-safe [`Connector`], for erasure behind an [`Arc`].
 trait ConnectDyn: Send + Sync {
-    fn connect_dyn(&self) -> BoxFuture<'_, io::Result<DynTx>>;
+    fn connect_dyn(&self) -> BoxFuture<'_, io::Result<(DynTx, Done<DynTx>)>>;
 }
 
 impl<C: Connector> ConnectDyn for C {
-    fn connect_dyn(&self) -> BoxFuture<'_, io::Result<DynTx>> {
-        Box::pin(async { self.connect().await.map(|tx| Box::new(tx) as DynTx) })
+    fn connect_dyn(&self) -> BoxFuture<'_, io::Result<(DynTx, Done<DynTx>)>> {
+        Box::pin(async {
+            let (half, done) = self.connect().await?;
+            let erased: DynTx = Box::new(Bundle { half, done });
+            Ok((erased, Done::new(TxDyn::complete)))
+        })
     }
 }
 
@@ -57,19 +124,23 @@ impl DynConnector {
 impl Connector for DynConnector {
     type Tx = DynTx;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(DynTx, Done<DynTx>)> {
         self.0.connect_dyn().await
     }
 }
 
 /// Object-safe [`Acceptor`], for erasure behind a `&mut` borrow.
 pub(crate) trait AcceptDyn: Send {
-    fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result<DynRx>>;
+    fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result<(DynRx, Done<DynRx>)>>;
 }
 
 impl<A: Acceptor> AcceptDyn for A {
-    fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result<DynRx>> {
-        Box::pin(async { self.accept().await.map(|rx| Box::new(rx) as DynRx) })
+    fn accept_dyn(&mut self) -> BoxFuture<'_, io::Result<(DynRx, Done<DynRx>)>> {
+        Box::pin(async {
+            let (half, done) = self.accept().await?;
+            let erased: DynRx = Box::new(Bundle { half, done });
+            Ok((erased, Done::new(RxDyn::complete)))
+        })
     }
 }
 
@@ -83,7 +154,7 @@ pub(crate) type DynAcceptor<'a> = &'a mut (dyn AcceptDyn + 'a);
 impl<'a, 'd> Acceptor for &'a mut (dyn AcceptDyn + 'd) {
     type Rx = DynRx;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         // Dispatch through the object explicitly: plain method syntax would
         // resolve to the blanket `AcceptDyn` impl for `&mut dyn AcceptDyn`
         // (this very impl's `Acceptor`), recursing instead of erasing.
diff --git a/src/link/routed.rs b/src/link/routed.rs
index e065a357..6cccad04 100644
--- a/src/link/routed.rs
+++ b/src/link/routed.rs
@@ -19,8 +19,16 @@
 //! header and hands the connection — whole, never its bytes — to that
 //! link's bounded queue, where the link's [`Acceptor`] collects it. A
 //! link's first connection carries the control stream; each later one
-//! is a single unidirectional data stream, closed by dropping its write
-//! end.
+//! carries a single unidirectional data stream at a time.
+//!
+//! A data stream ends one of two ways, the link contract's completion
+//! clause. Completed at its clean end, the connection outlives the
+//! stream: the write half goes back to its [`Dial`] (see
+//! [`Dial::recycle`]), the read half back to the router, which reads
+//! the connection's next header there — so a dial that pools recycled
+//! connections carries stream after stream over one connection, paying
+//! connection setup once. Dropped instead, the connection closes with
+//! the stream, which is how the peer observes the abort.
 //!
 //! # Driving the router
 //!
@@ -66,10 +74,14 @@
 //! - A silently dead path hangs a stream open or write until the
 //!   caller's session timeout cancels the session, the same backstop
 //!   every transport relies on. Liveness probing (keepalive) belongs in
-//!   the caller's [`Dial`].
-//! - Every lazy stream open pays one dial. This adapter is the
-//!   compatibility shape, not the latency shape; a transport with
-//!   native substreams avoids the per-stream dial entirely.
+//!   the caller's [`Dial`] — as does discovering that a pooled
+//!   connection died while idle, which surfaces as the recycled
+//!   stream's transport failure and heals at session granularity.
+//! - Every lazy stream open pays one dial. Where the dial itself is
+//!   the expensive step (an authenticating transport, say), a [`Dial`]
+//!   that pools recycled connections pays it once per connection
+//!   rather than once per stream; a transport with native substreams
+//!   avoids the per-stream open entirely.
 //!
 //! # What the transport must provide
 //!
@@ -158,7 +170,7 @@ use std::io;
 
 use tokio::io::{AsyncRead, AsyncWrite};
 
-use super::{Acceptor, Connector, Link};
+use super::{Acceptor, Connector, Done, Link};
 
 mod endpoint;
 mod header;
@@ -184,10 +196,12 @@ pub use stream::{StreamAcceptor, StreamConnector};
 ///   unflushed writes. A connection wrapped in a write buffer that
 ///   holds bytes until it fills stalls the first such exchange.
 /// - **Drop is half-close.** Dropping the connection must deliver all
-///   already-written bytes and then end-of-stream to the peer: the
-///   session ends every data stream by dropping its write end, and the
+///   already-written bytes and then end-of-stream to the peer: an
+///   aborted data stream ends by dropping its connection, and the
 ///   peer reads to end-of-stream. A drop that discards queued bytes,
-///   or never signals the peer, breaks stream teardown.
+///   or never signals the peer, breaks stream teardown. (A *completed*
+///   stream never drops its connection — the adapter recovers it; see
+///   [`Dial::recycle`].)
 ///
 /// `tokio::net::TcpStream` satisfies both, as does anything else whose
 /// writes land in the transport as they are accepted.
@@ -216,6 +230,36 @@ pub trait Dial: Clone + Send + Sync + 'static {
 
     /// Open one connection to the router reachable at `addr`.
     fn dial(&self, addr: &Self::Addr) -> impl Future<Output = io::Result<Self::Conn>> + Send;
+
+    /// Take back a connection to `peer` whose stream completed cleanly.
+    ///
+    /// The connection rests exactly where a fresh dial's would: the
+    /// peer's router is reading for its next connect header, so a dial
+    /// that hands it out again reuses the connection for another stream
+    /// in place of a new connection's setup. The default drops it,
+    /// which suits transports whose connections are cheap; implement it
+    /// (a pool keyed by peer, typically) where connection setup is the
+    /// latency that matters. A connection whose stream failed or was
+    /// abandoned never comes back through here: the adapter drops it,
+    /// so a recycled connection is never mid-stream.
+    ///
+    /// Two cautions. Recycle runs on the session's task at the
+    /// stream's completion, so it must not block. And recycling
+    /// certifies nothing about liveness: the peer's router evicts idle
+    /// connections by count, so a pooled connection may be dead and
+    /// discovered only by the stream that draws it.
+    ///
+    /// The peer's router writes one byte on the connection once it is
+    /// ready for the next stream. A stream sent earlier is not
+    /// delivered until the previous stream's consumer lets the
+    /// connection go. Consume the byte off the dialing path and reuse
+    /// only connections whose byte has arrived, dialing fresh
+    /// otherwise: a `dial` that waits for it serializes the open
+    /// behind another stream's progress, which the link contract
+    /// forbids, and deadlocks a session whose streams cross.
+    fn recycle(&self, _peer: &Self::Addr, conn: Self::Conn) {
+        drop(conn);
+    }
 }
 
 /// Yields inbound connections to an endpoint's router.
diff --git a/src/link/routed/endpoint.rs b/src/link/routed/endpoint.rs
index 35249840..0ab5c777 100644
--- a/src/link/routed/endpoint.rs
+++ b/src/link/routed/endpoint.rs
@@ -45,8 +45,14 @@ pub struct Config {
     /// The bound is hygiene against connections that stall inside
     /// their connect header (the router has no clock, so it evicts by
     /// count, oldest first); wall-clock deadlines belong in the
-    /// caller's [`Listen`] wrapper. Anything past the burst of
-    /// simultaneous dials the deployment expects is enough.
+    /// caller's [`Listen`] wrapper. Connections recovered from
+    /// completed streams wait here for their next header too, so a
+    /// deployment whose [`Dial`] pools connections sizes this past its
+    /// pooled idle count plus the burst of simultaneous dials it
+    /// expects. Eviction of an idle recovered connection is silent: no
+    /// invalidation reaches the dialer's pool, and the next stream
+    /// drawn on the dead entry fails, or hangs to the caller's session
+    /// timeout. Size generously.
     pub pending_headers: usize,
 }
 
@@ -148,9 +154,11 @@ impl<D: Dial> Clone for Endpoint<D> {
 struct Inner<D: Dial> {
     table: Table<D::Conn>,
     dial: D,
-    /// The endpoint's advertised name, pre-encoded (validated at
-    /// construction) for the `LINK` headers this endpoint writes.
-    advertised: Vec<u8>,
+    /// The endpoint's advertised name, as given at construction.
+    local_addr: D::Addr,
+    /// The advertised name, pre-encoded (validated at construction) for
+    /// the `LINK` headers this endpoint writes.
+    encoded: Vec<u8>,
 }
 
 impl<D: Dial> Endpoint<D> {
@@ -209,13 +217,20 @@ impl<D: Dial> Endpoint<D> {
             inner: Arc::new(Inner {
                 table: table.clone(),
                 dial: dial.clone(),
-                advertised: encoded,
+                local_addr: advertised,
+                encoded,
             }),
         };
         let router = router::drive(listen, dial, table, arrivals, config.pending_headers);
         Ok((endpoint, Incoming { links: incoming }, router))
     }
 
+    /// The name peers dial this endpoint at: `advertised`, as given at
+    /// construction.
+    pub fn local_addr(&self) -> &D::Addr {
+        &self.inner.local_addr
+    }
+
     /// Establish one link to the peer reachable at `peer`.
     ///
     /// One round trip: dial the peer's router, announce the link (its
@@ -241,7 +256,7 @@ impl<D: Dial> Endpoint<D> {
         // find the token routable.
         let (token, registration, streams) = router::register(&self.inner.table);
         let mut conn = self.inner.dial.dial(&peer).await?;
-        conn.write_all(&header::link_header(&token, &self.inner.advertised))
+        conn.write_all(&header::link_header(&token, &self.inner.encoded))
             .await?;
         let mut ack = [0; 1];
         match conn.read_exact(&mut ack).await {
diff --git a/src/link/routed/header.rs b/src/link/routed/header.rs
index d5d83c03..3b232559 100644
--- a/src/link/routed/header.rs
+++ b/src/link/routed/header.rs
@@ -63,6 +63,17 @@ pub(super) const PREFIX_LEN: usize = MAGIC.len() + 2 + TOKEN_LEN;
 /// link is on its way to the application.
 pub(super) const ACK: u8 = 1;
 
+/// The byte a router writes back on a recovered connection when it is
+/// reading for the next header.
+///
+/// Until it arrives, the previous stream's consumer still holds the
+/// connection, and a stream sent on it would wait on that consumer's
+/// progress. A reusing [`Dial`] consumes the byte off the dialing
+/// path before the connection is reused.
+///
+/// [`Dial`]: super::Dial
+pub(super) const READY: u8 = 2;
+
 /// Bytes in a [`Token`].
 const TOKEN_LEN: usize = 16;
 
diff --git a/src/link/routed/router.rs b/src/link/routed/router.rs
index bbc6e779..75864a33 100644
--- a/src/link/routed/router.rs
+++ b/src/link/routed/router.rs
@@ -40,17 +40,18 @@ use tokio::sync::mpsc::error::TrySendError;
 use super::endpoint::{Arrival, LinkInfo};
 use super::header::{self, Header, Token};
 use super::stream::{StreamAcceptor, StreamConnector};
-use super::{Dial, Link, Listen};
+use super::{Dial, Done, Link, Listen};
 use crate::link::STREAM_COUNT;
 
 /// The routing table: each live link's token, mapped to the bounded
-/// queue its acceptor drains.
+/// queue its acceptor drains. Each delivery carries the [`Done`] that
+/// returns a completed stream's connection to the router.
 ///
 /// Shared between the router (inserts on inbound links, removes on
 /// eviction), the endpoint (inserts on outbound links), and every
 /// link's [`Registration`] (removes on drop). The mutex is never held
 /// across an await.
-pub(super) type Table<C> = Arc<Mutex<HashMap<Token, mpsc::Sender<C>>>>;
+pub(super) type Table<C> = Arc<Mutex<HashMap<Token, mpsc::Sender<(C, Done<C>)>>>>;
 
 /// Lock the table, riding through a poisoning panic.
 ///
@@ -58,9 +59,10 @@ pub(super) type Table<C> = Arc<Mutex<HashMap<Token, mpsc::Sender<C>>>>;
 /// elsewhere cannot leave the map torn; continuing lets the surviving
 /// side of a panicked test observe eviction rather than a poison
 /// cascade.
+#[allow(clippy::type_complexity)]
 fn entries<C>(
-    table: &Mutex<HashMap<Token, mpsc::Sender<C>>>,
-) -> MutexGuard<'_, HashMap<Token, mpsc::Sender<C>>> {
+    table: &Mutex<HashMap<Token, mpsc::Sender<(C, Done<C>)>>>,
+) -> MutexGuard<'_, HashMap<Token, mpsc::Sender<(C, Done<C>)>>> {
     table
         .lock()
         .unwrap_or_else(|poisoned| poisoned.into_inner())
@@ -96,7 +98,10 @@ impl<C> Drop for Registration<C> {
 ///
 /// Collisions are astronomically unlikely at the token's width; the
 /// loop makes the vacancy structural rather than probabilistic.
-pub(super) fn register<C>(table: &Table<C>) -> (Token, Registration<C>, mpsc::Receiver<C>) {
+#[allow(clippy::type_complexity)]
+pub(super) fn register<C>(
+    table: &Table<C>,
+) -> (Token, Registration<C>, mpsc::Receiver<(C, Done<C>)>) {
     let (sender, receiver) = mpsc::channel(STREAM_COUNT);
     let mut sender = Some(sender);
     let token = loop {
@@ -127,6 +132,13 @@ where
     D: Dial,
     L: Listen<Conn = D::Conn>,
 {
+    // Connections coming back from completed streams, to await their
+    // next header beside fresh arrivals. The queue is bounded by the
+    // same budget as the pending reads it feeds, and its send never
+    // blocks: a return finding it full is dropped, the eviction the
+    // pending bound would deal it anyway. (Declared before `pending`,
+    // whose futures borrow the sender.)
+    let (returns, mut returned) = mpsc::channel::<D::Conn>(pending_headers);
     let mut pending = FuturesUnordered::new();
     // Insertion-ordered abort handles for the pending reads, so the
     // count bound evicts oldest-first; ids reconcile the two
@@ -134,29 +146,37 @@ where
     let mut order: VecDeque<(u64, AbortHandle)> = VecDeque::new();
     let mut next_id: u64 = 0;
     loop {
-        tokio::select! {
-            accepted = listen.accept() => {
-                let conn = accepted?;
-                if order.len() >= pending_headers
-                    && let Some((_, oldest)) = order.pop_front()
-                {
-                    oldest.abort();
-                }
-                let (abort, registration) = AbortHandle::new_pair();
-                let id = next_id;
-                next_id += 1;
-                order.push_back((id, abort));
-                pending.push(Abortable::new(
-                    route(conn, dial.clone(), &table, &incoming, id),
-                    registration,
-                ));
-            }
+        let (conn, recovered) = tokio::select! {
+            accepted = listen.accept() => (accepted?, false),
+            Some(conn) = returned.recv() => (conn, true),
             Some(finished) = pending.next() => {
                 if let Ok(id) = finished {
                     order.retain(|(pending_id, _)| *pending_id != id);
                 }
+                continue;
             }
+        };
+        if order.len() >= pending_headers
+            && let Some((_, oldest)) = order.pop_front()
+        {
+            oldest.abort();
         }
+        let (abort, registration) = AbortHandle::new_pair();
+        let id = next_id;
+        next_id += 1;
+        order.push_back((id, abort));
+        pending.push(Abortable::new(
+            route(
+                conn,
+                recovered,
+                dial.clone(),
+                &table,
+                &incoming,
+                &returns,
+                id,
+            ),
+            registration,
+        ));
     }
 }
 
@@ -167,25 +187,32 @@ where
 /// handle.
 async fn route<D: Dial>(
     conn: D::Conn,
+    recovered: bool,
     dial: D,
     table: &Table<D::Conn>,
     incoming: &mpsc::Sender<Arrival<D>>,
+    returns: &mpsc::Sender<D::Conn>,
     id: u64,
 ) -> u64 {
     // A failure here is a connection that never became anyone's
     // stream: the dialer observes the drop as transport failure, and
     // there is no one else to tell.
-    let _ = deliver(conn, dial, table, incoming).await;
+    let _ = deliver(conn, recovered, dial, table, incoming, returns).await;
     id
 }
 
 /// The routing step behind [`route`]: parse, then attach or establish.
 async fn deliver<D: Dial>(
     mut conn: D::Conn,
+    recovered: bool,
     dial: D,
     table: &Table<D::Conn>,
     incoming: &mpsc::Sender<Arrival<D>>,
+    returns: &mpsc::Sender<D::Conn>,
 ) -> io::Result<()> {
+    if recovered {
+        conn.write_all(&[header::READY]).await?;
+    }
     match header::read::<D::Addr, _>(&mut conn).await? {
         Header::Stream { token } => {
             let Some(queue) = entries(table).get(&token).cloned() else {
@@ -194,7 +221,15 @@ async fn deliver<D: Dial>(
                 // on the dialing side, which owns the retry.
                 return Ok(());
             };
-            match queue.try_send(conn) {
+            let returns = returns.clone();
+            // A completed stream hands its connection back through this
+            // sender, which never blocks (the module's no-await
+            // discipline). A return dropped on a full queue or stopped
+            // router degrades to an abort, which the dialer heals.
+            let done = Done::new(move |conn| {
+                let _ = returns.try_send(conn);
+            });
+            match queue.try_send((conn, done)) {
                 Ok(()) => {}
                 // A full queue proves peer misbehavior (an honest peer
                 // never exceeds a session's complement, and the queue
diff --git a/src/link/routed/stream.rs b/src/link/routed/stream.rs
index 677e17d7..a38d38e4 100644
--- a/src/link/routed/stream.rs
+++ b/src/link/routed/stream.rs
@@ -1,4 +1,10 @@
 //! One link's stream supply: dial-per-open out, routed queue in.
+//!
+//! Completion recovers the connection on both sides. The write half
+//! goes back to its [`Dial`] through [`Dial::recycle`], and the read
+//! half returns to the router, which reads its next connect header
+//! there. A dropped half drops its connection instead, whose close is
+//! the transport half-close the peer observes as an abort.
 
 use std::io;
 
@@ -7,16 +13,14 @@ use tokio::sync::mpsc;
 
 use super::header::{self, Token};
 use super::router::Registration;
-use super::{Acceptor, Conn, Connector, Dial};
+use super::{Acceptor, Conn, Connector, Dial, Done};
 
-/// A routed link's [`Connector`]: every open dials one fresh
-/// connection to the peer's router and labels it with the link's
-/// token.
+/// A routed link's [`Connector`]: every open dials one connection to
+/// the peer's router and labels it with the link's token.
 ///
-/// The returned stream *is* the connection, so dropping it is the
-/// transport half-close (the peer reads the final bytes, then
-/// end-of-stream), and no open ever waits on another stream's
-/// progress: the opens share nothing but the dialer.
+/// Whether "dials" means a fresh connection or a recovered one is the
+/// [`Dial`]'s policy: a dial that pools what [`Dial::recycle`] hands
+/// back pays no new connection setup for the next stream.
 pub struct StreamConnector<D: Dial> {
     dial: D,
     peer: D::Addr,
@@ -43,10 +47,12 @@ impl<D: Dial> Clone for StreamConnector<D> {
 impl<D: Dial> Connector for StreamConnector<D> {
     type Tx = D::Conn;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         let mut conn = self.dial.dial(&self.peer).await?;
         conn.write_all(&header::stream_header(&self.token)).await?;
-        Ok(conn)
+        let dial = self.dial.clone();
+        let peer = self.peer.clone();
+        Ok((conn, Done::new(move |conn| dial.recycle(&peer, conn))))
     }
 }
 
@@ -58,7 +64,7 @@ impl<D: Dial> Connector for StreamConnector<D> {
 /// link's claim on its routing token, tying the routing to the link's
 /// own lifetime: dropping the link revokes its token at that moment.
 pub struct StreamAcceptor<C> {
-    streams: mpsc::Receiver<C>,
+    streams: mpsc::Receiver<(C, Done<C>)>,
     /// Revokes this link's token when the acceptor drops.
     _registration: Registration<C>,
 }
@@ -66,7 +72,10 @@ pub struct StreamAcceptor<C> {
 impl<C> StreamAcceptor<C> {
     /// Bundle a link's incoming supply around its routed queue and
     /// token claim.
-    pub(super) fn new(streams: mpsc::Receiver<C>, registration: Registration<C>) -> Self {
+    pub(super) fn new(
+        streams: mpsc::Receiver<(C, Done<C>)>,
+        registration: Registration<C>,
+    ) -> Self {
         StreamAcceptor {
             streams,
             _registration: registration,
@@ -77,7 +86,7 @@ impl<C> StreamAcceptor<C> {
 impl<C: Conn> Acceptor for StreamAcceptor<C> {
     type Rx = C;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         // The router holds this queue's only sender, and removes it
         // exactly when it evicts the link (a queue overflow, which
         // proves peer misbehavior). Queued deliveries drain first, so
diff --git a/src/link/routed/tests.rs b/src/link/routed/tests.rs
index dd7e6c8d..92acc755 100644
--- a/src/link/routed/tests.rs
+++ b/src/link/routed/tests.rs
@@ -1,10 +1,16 @@
-use std::future::Future;
+use std::collections::HashMap;
+use std::future::{Future, poll_fn};
 use std::io;
-use std::pin::pin;
+use std::mem::take;
+use std::pin::{Pin, pin};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
 
 use futures::FutureExt;
 use futures::future::{Either, select, try_join};
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use futures::task::noop_waker;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt, DuplexStream, ReadBuf};
 
 use super::header::{self, Token};
 use super::{Config, Dial, Endpoint, EndpointError, Incoming, LinkError, LinkInfo, RoutedLink};
@@ -78,11 +84,11 @@ async fn transfer<CRa, CWa, Ca, Aa, CRb, CWb, Cb, Ab>(
     Ab: Acceptor,
 {
     let open = async {
-        let mut tx = opener.connector.connect().await.expect("stream opens");
+        let (mut tx, _) = opener.connector.connect().await.expect("stream opens");
         tx.write_all(payload).await.expect("payload writes");
     };
     let read = async {
-        let mut rx = acceptor.acceptor.accept().await.expect("stream arrives");
+        let (mut rx, _) = acceptor.acceptor.accept().await.expect("stream arrives");
         let mut received = Vec::new();
         rx.read_to_end(&mut received)
             .await
@@ -141,6 +147,16 @@ fn establishment_connects_control_and_streams() {
     });
 }
 
+/// `local_addr` is the advertised name given at construction, the name
+/// peers dial this endpoint at; callers need it back for policies like
+/// dial tiebreaks.
+#[test]
+fn local_addr_is_the_constructed_name() {
+    let net = MemoryNet::new();
+    let (a, _incoming, _router) = endpoint(&net, "a", Config::default());
+    assert_eq!(*a.local_addr(), MemoryName::new("a"));
+}
+
 /// A stream connection quoting a token no live link owns is dropped:
 /// the dialer observes end-of-stream, and no link's queue sees the
 /// connection.
@@ -419,14 +435,14 @@ fn cancelled_opens_leave_the_link_usable() {
             // a connection that closes unwritten; drain any such
             // orphans until the genuine payload arrives.
             let open = async {
-                let mut tx = at_b.connector.connect().await.expect("stream opens");
+                let (mut tx, _) = at_b.connector.connect().await.expect("stream opens");
                 tx.write_all(b"after cancellations")
                     .await
                     .expect("payload writes");
             };
             let read = async {
                 loop {
-                    let mut rx = at_a.acceptor.accept().await.expect("stream arrives");
+                    let (mut rx, _) = at_a.acceptor.accept().await.expect("stream arrives");
                     let mut received = Vec::new();
                     rx.read_to_end(&mut received)
                         .await
@@ -444,6 +460,187 @@ fn cancelled_opens_leave_the_link_usable() {
     });
 }
 
+/// Open one stream from `opener` to `acceptor`, move `payload` across
+/// it, and end it by completion on both halves: the receiver reads
+/// exactly the payload and completes there, never probing for
+/// end-of-stream.
+async fn transfer_completed<CRa, CWa, Ca, Aa, CRb, CWb, Cb, Ab>(
+    opener: &Link<CRa, CWa, Ca, Aa>,
+    acceptor: &mut Link<CRb, CWb, Cb, Ab>,
+    payload: &[u8],
+) where
+    Ca: Connector,
+    Ab: Acceptor,
+{
+    let open = async {
+        let (mut tx, done) = opener.connector.connect().await.expect("stream opens");
+        tx.write_all(payload).await.expect("payload writes");
+        tx.flush().await.expect("payload flushes");
+        done.complete(tx);
+    };
+    let read = async {
+        let (mut rx, done) = acceptor.acceptor.accept().await.expect("stream arrives");
+        let mut received = vec![0u8; payload.len()];
+        rx.read_exact(&mut received)
+            .await
+            .expect("the completed stream delivers its bytes");
+        done.complete(rx);
+        received
+    };
+    let ((), received) = futures::join!(open, read);
+    assert_eq!(received, payload);
+}
+
+/// A dialer that pools recycled connections per peer and counts the
+/// fresh dials it performs: how the reuse tests observe which streams
+/// paid for a connection.
+#[derive(Clone)]
+struct PoolingDial {
+    inner: MemoryDial,
+    /// Recycled, awaiting the router's ready byte.
+    pending: Arc<Mutex<HashMap<String, Vec<DuplexStream>>>>,
+    /// Ready for reuse.
+    pool: Arc<Mutex<HashMap<String, Vec<DuplexStream>>>>,
+    fresh: Arc<AtomicUsize>,
+}
+
+impl PoolingDial {
+    fn new(net: &MemoryNet) -> Self {
+        PoolingDial {
+            inner: net.dial(),
+            pending: Arc::default(),
+            pool: Arc::default(),
+            fresh: Arc::default(),
+        }
+    }
+
+    fn fresh_dials(&self) -> usize {
+        self.fresh.load(Ordering::Relaxed)
+    }
+
+    /// Admit pending connections whose ready byte has arrived, polling
+    /// each read exactly once: the byte is consumed off the dialing
+    /// path, never awaited.
+    fn admit(&self, addr: &MemoryName) {
+        let mut pending = self.pending.lock().expect("pending lock");
+        let Some(conns) = pending.get_mut(&addr.0) else {
+            return;
+        };
+        let waker = noop_waker();
+        let mut cx = Context::from_waker(&waker);
+        for mut conn in take(conns) {
+            let mut byte = [0u8; 1];
+            let mut buf = ReadBuf::new(&mut byte);
+            match Pin::new(&mut conn).poll_read(&mut cx, &mut buf) {
+                Poll::Ready(Ok(())) if buf.filled().len() == 1 => {
+                    self.pool
+                        .lock()
+                        .expect("pool lock")
+                        .entry(addr.0.clone())
+                        .or_default()
+                        .push(conn);
+                }
+                Poll::Pending => conns.push(conn),
+                // EOF or error: the connection is dead.
+                _ => {}
+            }
+        }
+    }
+}
+
+impl Dial for PoolingDial {
+    type Addr = MemoryName;
+    type Conn = DuplexStream;
+
+    async fn dial(&self, addr: &MemoryName) -> io::Result<DuplexStream> {
+        self.admit(addr);
+        let pooled = self
+            .pool
+            .lock()
+            .expect("pool lock")
+            .get_mut(&addr.0)
+            .and_then(Vec::pop);
+        if let Some(conn) = pooled {
+            return Ok(conn);
+        }
+        self.fresh.fetch_add(1, Ordering::Relaxed);
+        self.inner.dial(addr).await
+    }
+
+    fn recycle(&self, peer: &MemoryName, conn: DuplexStream) {
+        self.pending
+            .lock()
+            .expect("pending lock")
+            .entry(peer.0.clone())
+            .or_default()
+            .push(conn);
+    }
+}
+
+/// Give the routers a few polls, so ready bytes land before the next
+/// dial's single-poll admission.
+async fn settle() {
+    for _ in 0..8 {
+        let mut yielded = false;
+        poll_fn(|cx| {
+            if yielded {
+                Poll::Ready(())
+            } else {
+                yielded = true;
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            }
+        })
+        .await;
+    }
+}
+
+/// A completed stream's connection carries the next stream instead of
+/// a fresh dial.
+///
+/// The write half hands it back through [`Dial::recycle`], the read
+/// half returns it to the router for its next connect header, and the
+/// recycled connection routes exactly as a dialed one would — in both
+/// directions of the link.
+#[test]
+fn completed_streams_reuse_their_connection() {
+    pollster::block_on(async {
+        let net = MemoryNet::new();
+        let dial = PoolingDial::new(&net);
+        let a_name = MemoryName::new("a");
+        let (a, mut a_incoming, a_router) =
+            Endpoint::new(net.listen(&a_name), a_name, dial.clone(), Config::default())
+                .expect("a valid construction");
+        let b_name = MemoryName::new("b");
+        let (b, _b_incoming, b_router) =
+            Endpoint::new(net.listen(&b_name), b_name, dial.clone(), Config::default())
+                .expect("a valid construction");
+        drive(routers(a_router, b_router), async {
+            let (linked, arrival) =
+                futures::join!(b.link(MemoryName::new("a")), a_incoming.accept());
+            let at_b = linked.expect("establishment succeeds");
+            let (_info, mut at_a) = arrival.expect("the router delivers the link");
+            let established = dial.fresh_dials();
+
+            // Forward: the first stream dials, the second reuses its
+            // recycled connection.
+            transfer_completed(&at_b, &mut at_a, b"paid by a dial").await;
+            settle().await;
+            transfer_completed(&at_b, &mut at_a, b"reuses recycled").await;
+            assert_eq!(dial.fresh_dials(), established + 1);
+
+            // Reverse: the accepting side's dial-back pools the same way.
+            let mut at_b = at_b;
+            transfer_completed(&at_a, &mut at_b, b"paid by a dial").await;
+            settle().await;
+            transfer_completed(&at_a, &mut at_b, b"reuses recycled").await;
+            assert_eq!(dial.fresh_dials(), established + 2);
+            drop((a, b));
+        })
+        .await;
+    });
+}
+
 /// A dialer over socket addresses for construction tests; endpoint
 /// construction never dials, so its `dial` is unreachable.
 #[derive(Clone)]
diff --git a/src/link/tests.rs b/src/link/tests.rs
index 7cdccf9f..b7ef057b 100644
--- a/src/link/tests.rs
+++ b/src/link/tests.rs
@@ -48,12 +48,12 @@ fn connect_delivers_an_ordered_half_closing_stream() {
     let (a, mut b) = memory();
     run_to_quiescence(async {
         let send = async {
-            let mut tx = a.connector.connect().await.unwrap();
+            let (mut tx, _) = a.connector.connect().await.unwrap();
             tx.write_all(b"hello").await.unwrap();
             drop(tx);
         };
         let receive = async {
-            let mut rx = b.acceptor.accept().await.unwrap();
+            let (mut rx, _) = b.acceptor.accept().await.unwrap();
             let mut bytes = Vec::new();
             rx.read_to_end(&mut bytes).await.unwrap();
             assert_eq!(bytes, b"hello");
@@ -70,18 +70,18 @@ fn a_stalled_stream_does_not_couple_its_siblings() {
     let (a, mut b) = memory_with_capacity(4);
     run_to_quiescence(async {
         let send = async {
-            let mut stalled = a.connector.connect().await.unwrap();
+            let (mut stalled, _) = a.connector.connect().await.unwrap();
             // Fill the stalled stream's bounded buffer to its brim; its
             // reader never drains it, so more writes would block.
             stalled.write_all(&[0u8; 4]).await.unwrap();
-            let mut live = a.connector.connect().await.unwrap();
+            let (mut live, _) = a.connector.connect().await.unwrap();
             live.write_all(b"live").await.unwrap();
             drop(live);
             stalled
         };
         let receive = async {
             let _stalled = b.acceptor.accept().await.unwrap();
-            let mut rx = b.acceptor.accept().await.unwrap();
+            let (mut rx, _) = b.acceptor.accept().await.unwrap();
             let mut bytes = Vec::new();
             rx.read_to_end(&mut bytes).await.unwrap();
             assert_eq!(bytes, b"live");
@@ -98,12 +98,12 @@ fn stream_writes_block_on_their_own_reader() {
     let (a, mut b) = memory_with_capacity(2);
     run_to_quiescence(async {
         let send = async {
-            let mut tx = a.connector.connect().await.unwrap();
+            let (mut tx, _) = a.connector.connect().await.unwrap();
             tx.write_all(b"abcdef").await.unwrap();
             drop(tx);
         };
         let receive = async {
-            let mut rx = b.acceptor.accept().await.unwrap();
+            let (mut rx, _) = b.acceptor.accept().await.unwrap();
             let mut bytes = Vec::new();
             rx.read_to_end(&mut bytes).await.unwrap();
             assert_eq!(bytes, b"abcdef");
diff --git a/src/testing/transport.rs b/src/testing/transport.rs
index ada5418b..b24799a8 100644
--- a/src/testing/transport.rs
+++ b/src/testing/transport.rs
@@ -13,6 +13,8 @@ use std::{
 
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 
+use crate::link::Done;
+
 /// Which endpoint owns an observed transport operation.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum Side {
@@ -51,8 +53,8 @@ pub enum FaultUnit {
 /// A read fault fires in place of the first *payload-bearing* read beyond
 /// the prefix: end-of-stream probes pass through untouched. This keeps
 /// "would the fault fire?" a function of the clean run's successful-read
-/// counts alone — the link world checks for end-of-stream after every
-/// stream's end control, so attempts-after-last-success are structural and
+/// counts alone — the control stream reads for a clean goodbye at every
+/// session boundary, so attempts-after-last-success are structural and
 /// must not trip an operations-counted fault.
 ///
 /// Stream-supply faults follow the same discipline, counted in operations
@@ -561,7 +563,7 @@ impl<C: Clone> Clone for AdversarialConnector<C> {
 impl<C: crate::link::Connector> crate::link::Connector for AdversarialConnector<C> {
     type Tx = AdversarialWrite<C::Tx>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         // The fault fires in place of the call: a healthy supply's connects
         // always succeed, so the clean run's success count is also its call
         // count and "would the fault fire?" remains a function of it.
@@ -573,13 +575,17 @@ impl<C: crate::link::Connector> crate::link::Connector for AdversarialConnector<
         {
             return Err(error);
         }
-        let tx = self.inner.connect().await?;
+        let (tx, done) = self.inner.connect().await?;
         self.state
             .lock()
             .expect("transport state lock")
             .report
             .connects += 1;
-        Ok(wrap_write(tx, self.state.clone()))
+        // Completion unwraps the adversity and passes the half through.
+        Ok((
+            wrap_write(tx, self.state.clone()),
+            Done::new(move |wrapped: AdversarialWrite<C::Tx>| done.complete(wrapped.inner)),
+        ))
     }
 }
 
@@ -593,7 +599,7 @@ pub struct AdversarialAcceptor<A> {
 impl<A: crate::link::Acceptor> crate::link::Acceptor for AdversarialAcceptor<A> {
     type Rx = AdversarialRead<A::Rx>;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         // An already-injected accept fault keeps failing without consuming
         // further arrivals.
         {
@@ -604,7 +610,7 @@ impl<A: crate::link::Acceptor> crate::link::Acceptor for AdversarialAcceptor<A>
                 return Err(error);
             }
         }
-        let rx = self.inner.accept().await?;
+        let (rx, done) = self.inner.accept().await?;
         let mut state = self.state.lock().expect("transport state lock");
         // The fault fires in place of a successful accept, so the final
         // forever-pending accept a session parks on against an honest peer
@@ -617,11 +623,15 @@ impl<A: crate::link::Acceptor> crate::link::Acceptor for AdversarialAcceptor<A>
         }
         state.report.accepts += 1;
         drop(state);
-        Ok(AdversarialRead {
-            inner: rx,
-            state: self.state.clone(),
-            delay: None,
-        })
+        // Completion unwraps the adversity and passes the half through.
+        Ok((
+            AdversarialRead {
+                inner: rx,
+                state: self.state.clone(),
+                delay: None,
+            },
+            Done::new(move |wrapped: AdversarialRead<A::Rx>| done.complete(wrapped.inner)),
+        ))
     }
 }
 
@@ -659,7 +669,7 @@ const REORDER_PATIENCE: u8 = 32;
 /// does not depend on the public `conformance` feature.
 pub struct ReorderingAcceptor<A: crate::link::Acceptor> {
     inner: A,
-    held: VecDeque<A::Rx>,
+    held: VecDeque<(A::Rx, Done<A::Rx>)>,
     /// Arrivals buffered before each reversed release.
     batch: usize,
     /// Batches of two or more released: genuine inversions.
@@ -669,7 +679,7 @@ pub struct ReorderingAcceptor<A: crate::link::Acceptor> {
 impl<A: crate::link::Acceptor> crate::link::Acceptor for ReorderingAcceptor<A> {
     type Rx = A::Rx;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         if let Some(held) = self.held.pop_front() {
             return Ok(held);
         }
diff --git a/src/tests.rs b/src/tests.rs
index 052203fc..9e6d33ce 100644
--- a/src/tests.rs
+++ b/src/tests.rs
@@ -297,12 +297,15 @@ struct FusedConnector {
 impl Connector for FusedConnector {
     type Tx = Fuse<tokio::io::DuplexStream>;
 
-    async fn connect(&self) -> std::io::Result<Self::Tx> {
-        let inner = self.inner.connect().await?;
-        Ok(Fuse {
-            inner,
-            remaining: Arc::clone(&self.remaining),
-        })
+    async fn connect(&self) -> std::io::Result<(Self::Tx, crate::link::Done<Self::Tx>)> {
+        let (inner, _) = self.inner.connect().await?;
+        Ok((
+            Fuse {
+                inner,
+                remaining: Arc::clone(&self.remaining),
+            },
+            crate::link::Done::discard(),
+        ))
     }
 }
 
diff --git a/src/tree/mirror/streaming/remote/codec/decode/async_io.rs b/src/tree/mirror/streaming/remote/codec/decode/async_io.rs
index b4cb5f7c..4a0a80f0 100644
--- a/src/tree/mirror/streaming/remote/codec/decode/async_io.rs
+++ b/src/tree/mirror/streaming/remote/codec/decode/async_io.rs
@@ -28,6 +28,13 @@ impl<R> FrameRead<R> {
     pub fn new(speaker: Speaker, read: R) -> Self {
         Self { speaker, read }
     }
+
+    /// Recover the transport half. The reader buffers nothing (every
+    /// read is exact), so between frames the half rests exactly at a
+    /// frame boundary.
+    pub fn into_inner(self) -> R {
+        self.read
+    }
 }
 
 impl<R: AsyncRead + Unpin> FrameRead<R> {
diff --git a/src/tree/mirror/streaming/remote/codec/encode/async_io.rs b/src/tree/mirror/streaming/remote/codec/encode/async_io.rs
index b7ed223e..f8c5dc4d 100644
--- a/src/tree/mirror/streaming/remote/codec/encode/async_io.rs
+++ b/src/tree/mirror/streaming/remote/codec/encode/async_io.rs
@@ -25,8 +25,9 @@ impl<W> FrameWrite<W> {
         Self { speaker, write }
     }
 
-    /// Recover the transport writer without buffered frame state.
-    #[cfg(test)]
+    /// Recover the transport writer without buffered frame state. Every
+    /// frame is flushed as it is written, so between frames the writer
+    /// rests exactly at a frame boundary.
     pub fn into_inner(self) -> W {
         self.write
     }
diff --git a/src/tree/mirror/streaming/remote/proxy/tests/harness.rs b/src/tree/mirror/streaming/remote/proxy/tests/harness.rs
index cadbdc8d..4f633a1c 100644
--- a/src/tree/mirror/streaming/remote/proxy/tests/harness.rs
+++ b/src/tree/mirror/streaming/remote/proxy/tests/harness.rs
@@ -14,7 +14,7 @@ use tokio::io::{AsyncRead, AsyncWrite};
 
 use tokio::io::ReadBuf;
 
-use crate::link::{Acceptor, Connector, Link, MemoryLink, memory_with_capacity};
+use crate::link::{Acceptor, Connector, Done, Link, MemoryLink, memory_with_capacity};
 use crate::testing::{IoPlan, IoReportHandle, IoSide, wrap_link};
 use crate::tree::mirror::framing::{GREETING_WORD_LEN, LENGTH_HEADER_LEN};
 use crate::tree::mirror::streaming::window::WindowConfig;
@@ -244,9 +244,9 @@ pub struct ScriptedConnector<C> {
 impl<C: Connector> Connector for ScriptedConnector<C> {
     type Tx = ScriptedWrite<C::Tx>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
-        let tx = self.inner.connect().await?;
-        Ok(ScriptedWrite::new(tx, self.script.clone()))
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
+        let (tx, _) = self.inner.connect().await?;
+        Ok((ScriptedWrite::new(tx, self.script.clone()), Done::discard()))
     }
 }
 
diff --git a/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs b/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs
index e961056e..68dd500c 100644
--- a/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs
+++ b/src/tree/mirror/streaming/remote/proxy/tests/malformed.rs
@@ -182,10 +182,16 @@ fn duplicated_reply_is_rejected_as_unasked() {
     assert!(right_result.is_err());
 }
 
-/// Duplicating a stream-end frame is rejected as traffic after closure rather
-/// than being mistaken for a second clean end.
+/// Bytes past a stream's end control belong to the transport, not the
+/// session: a duplicated stream-end frame is never read, and the session
+/// completes as if it were absent.
+///
+/// The receiver completes its transport half exactly at the end control
+/// (the link contract's completion clause), so trailing bytes are left
+/// where they lie — on a reusing link they would be the next stream's
+/// connect header — rather than parsed as protocol.
 #[test]
-fn duplicate_stream_end_is_rejected_by_the_session() {
+fn bytes_past_the_stream_end_are_never_read() {
     const STREAM_END_STATE: u8 = 9;
 
     for corrupt_left in [false, true] {
@@ -200,13 +206,12 @@ fn duplicate_stream_end_is_rejected_by_the_session() {
             corrupt_left.then(|| script.clone()),
             (!corrupt_left).then(|| script.clone()),
         ))
-        .expect("duplicate stream end must terminate both sessions");
+        .expect("sessions complete despite the trailing frame");
         assert!(script.fired(), "no stream-end frame reached the mutator");
-        assert!(matches!(
-            receiving_error(corrupt_left, &left_result, &right_result),
-            RemoteError::Stream(StreamError::AfterEnd { .. })
-        ));
-        assert!(left_result.is_err());
-        assert!(right_result.is_err());
+        assert!(left_result.is_ok(), "left session failed: {left_result:?}");
+        assert!(
+            right_result.is_ok(),
+            "right session failed: {right_result:?}"
+        );
     }
 }
diff --git a/src/tree/mirror/streaming/remote/proxy/work/tests.rs b/src/tree/mirror/streaming/remote/proxy/work/tests.rs
index e2972a14..c83bf35c 100644
--- a/src/tree/mirror/streaming/remote/proxy/work/tests.rs
+++ b/src/tree/mirror/streaming/remote/proxy/work/tests.rs
@@ -231,7 +231,8 @@ fn queued_supply_closed_outranks_a_selected_consequence_at_stream_granularity()
     // A reporter that needed the supply: publishes SupplyClosed at stream
     // granularity, then parks. Spawned first so it publishes in the same
     // wave the consequence resolves.
-    let (claim_send, claim_receive) = oneshot::channel::<DuplexStream>();
+    let (claim_send, claim_receive) =
+        oneshot::channel::<(DuplexStream, crate::link::Done<DuplexStream>)>();
     drop(claim_send);
     let mut receiver: StreamReceiver<DuplexStream, ()> = StreamReceiver::new(
         claim_receive,
@@ -295,7 +296,8 @@ fn published_stream_error_preempts_a_parked_protocol() {
     // A receiver whose claim sender is already gone: its first poll reports
     // `SupplyClosed` to the error route and parks, never resolving — the
     // pump driving it is the parked reporter the select must not wait for.
-    let (claim_send, claim_receive) = oneshot::channel::<DuplexStream>();
+    let (claim_send, claim_receive) =
+        oneshot::channel::<(DuplexStream, crate::link::Done<DuplexStream>)>();
     drop(claim_send);
     let mut receiver: StreamReceiver<DuplexStream, ()> = StreamReceiver::new(
         claim_receive,
diff --git a/src/tree/mirror/streaming/remote/streams.rs b/src/tree/mirror/streaming/remote/streams.rs
index 3bb35001..1b2019a0 100644
--- a/src/tree/mirror/streaming/remote/streams.rs
+++ b/src/tree/mirror/streaming/remote/streams.rs
@@ -47,7 +47,7 @@ use tokio::{
     sync::{mpsc, oneshot},
 };
 
-use crate::link::{Acceptor, Connector};
+use crate::link::{Acceptor, Connector, Done};
 use crate::tree::mirror::streaming::stats::{CountedRead, CountedWrite, Recorder};
 use crate::tree::mirror::streaming::tasks::cancelled;
 
@@ -126,7 +126,7 @@ pub struct StreamSender<C: Connector, T> {
 
 enum SendState<Tx> {
     Unopened,
-    Open(FrameWrite<CountedWrite<Tx>>),
+    Open(FrameWrite<CountedWrite<Tx>>, Done<Tx>),
 }
 
 impl<C: Connector, T> StreamSender<C, T> {
@@ -161,13 +161,23 @@ impl<C: Connector, T> StreamSender<C, T> {
 
     /// End this logical stream after all of its replies, if it ever opened.
     ///
-    /// Dropping the transport half afterward is the transport-level
-    /// half-close; the explicit end control before it distinguishes a
-    /// completed stream from one truncated mid-reply.
+    /// The explicit end control distinguishes a completed stream from one
+    /// truncated mid-reply. The transport half is handed to its [`Done`]
+    /// right behind it, resting at the frame boundary; failure paths drop
+    /// the half instead, the contract's abort.
     pub async fn finish(mut self) -> Result<(), SendError> {
         match self.state {
             SendState::Unopened => Ok(()),
-            SendState::Open(_) => self.write(Frame::End(End::Stream)).await,
+            SendState::Open(..) => {
+                self.write(Frame::End(End::Stream)).await?;
+                let SendState::Open(write, done) =
+                    std::mem::replace(&mut self.state, SendState::Unopened)
+                else {
+                    unreachable!("the open state was just written through");
+                };
+                done.complete(write.into_inner().into_inner());
+                Ok(())
+            }
         }
     }
 
@@ -175,9 +185,9 @@ impl<C: Connector, T> StreamSender<C, T> {
     async fn write(&mut self, frame: Frame<T>) -> Result<(), SendError> {
         let stream = self.stream;
         let write = match &mut self.state {
-            SendState::Open(write) => write,
+            SendState::Open(write, _) => write,
             state @ SendState::Unopened => {
-                let mut tx =
+                let (mut tx, done) =
                     self.connector
                         .connect()
                         .await
@@ -191,11 +201,11 @@ impl<C: Connector, T> StreamSender<C, T> {
                         origin: Origin::stream(self.speaker, stream),
                         source,
                     })?;
-                *state = SendState::Open(FrameWrite::new(
-                    self.speaker,
-                    CountedWrite::new(tx, self.stats.clone()),
-                ));
-                let SendState::Open(write) = state else {
+                *state = SendState::Open(
+                    FrameWrite::new(self.speaker, CountedWrite::new(tx, self.stats.clone())),
+                    done,
+                );
+                let SendState::Open(write, _) = state else {
                     unreachable!("the open state was just stored");
                 };
                 write
@@ -254,9 +264,6 @@ pub enum StreamError {
     /// The transport stream ended before its explicit end control.
     #[error("{origin}: transport stream ended before its end control")]
     Truncated { origin: Origin },
-    /// The peer sent more frames after ending its logical stream.
-    #[error("{origin}: peer sent a frame after ending the logical stream")]
-    AfterEnd { origin: Origin },
     /// The stream supply failed before an awaited stream was delivered.
     ///
     /// `source` carries the supply's own transport failure when the session
@@ -287,7 +294,7 @@ pub struct StreamReceiver<Rx, T> {
 }
 
 struct ReceiverStart<Rx> {
-    claim: oneshot::Receiver<Rx>,
+    claim: oneshot::Receiver<(Rx, Done<Rx>)>,
     /// The remote role whose direction this stream carries.
     speaker: Speaker,
     stream: Stream,
@@ -306,7 +313,7 @@ where
 {
     /// Bind one incoming logical stream to its claim slot.
     pub fn new(
-        claim: oneshot::Receiver<Rx>,
+        claim: oneshot::Receiver<(Rx, Done<Rx>)>,
         speaker: Speaker,
         stream: Stream,
         route: ErrorRoute,
@@ -384,7 +391,7 @@ where
 /// Every failure path publishes to the session error route and parks: the
 /// consumer never observes a truncated stream as a clean end.
 fn read_frames<Rx, T>(
-    claim: oneshot::Receiver<Rx>,
+    claim: oneshot::Receiver<(Rx, Done<Rx>)>,
     speaker: Speaker,
     stream: Stream,
     route: ErrorRoute,
@@ -395,7 +402,7 @@ where
     T: BorshDeserialize + Send + Sync + 'static,
 {
     stream! {
-        let Ok(rx) = claim.await else {
+        let Ok((rx, done)) = claim.await else {
             // The claim slot is gone: the link's stream supply failed before
             // the peer's stream for this level arrived. This is the one
             // consumer that provably needed it, so the report comes from
@@ -436,26 +443,19 @@ where
             };
             if matches!(frame, Frame::End(End::Stream)) {
                 // The lifecycle control is consumed here; the consumer sees
-                // only complete replies followed by a clean end. The sender
-                // half-closes immediately after this control, so requiring
-                // end-of-stream costs no waiting against an honest peer and
-                // catches one that keeps talking past its own end.
-                match read.frame::<T>().await {
-                    Ok(None) => break,
-                    Ok(Some(_)) => {
-                        route.report(StreamError::AfterEnd {
-                            origin: Origin::stream(speaker, stream),
-                        });
-                        cancelled().await
-                    }
-                    Err(error) => {
-                        route.report(StreamError::Decode(error));
-                        cancelled().await
-                    }
-                }
+                // only complete replies followed by a clean end.
+                break;
             }
             yield frame;
         }
+        // The end control is exactly where the data ends, so the transport
+        // half rests at the link contract's clean boundary: hand it back
+        // there, never reading past it. The hand-back is a framing
+        // judgment, not a protocol one: a stream the consumer goes on to
+        // rule invalid has still completed here. Every failure path above
+        // parks instead, dropping the half at teardown, which is the
+        // contract's abort.
+        done.complete(read.into_inner().into_inner());
     }
 }
 
@@ -565,17 +565,17 @@ pub fn error_route() -> (ErrorRoute, FirstStreamError) {
 
 /// The claim slots the accept driver delivers incoming streams into.
 pub struct ClaimSlots<Rx> {
-    slots: [Option<oneshot::Sender<Rx>>; STREAM_COUNT],
+    slots: [Option<oneshot::Sender<(Rx, Done<Rx>)>>; STREAM_COUNT],
 }
 
 /// The claim receivers the session's typed states take streams from.
 pub struct Claims<Rx> {
-    slots: [Option<oneshot::Receiver<Rx>>; STREAM_COUNT],
+    slots: [Option<oneshot::Receiver<(Rx, Done<Rx>)>>; STREAM_COUNT],
 }
 
 impl<Rx> Claims<Rx> {
     /// Take the sole claim for `stream`.
-    pub fn take(&mut self, stream: Stream) -> oneshot::Receiver<Rx> {
+    pub fn take(&mut self, stream: Stream) -> oneshot::Receiver<(Rx, Done<Rx>)> {
         self.slots[usize::from(stream.index())]
             .take()
             .expect("each incoming logical stream is claimed exactly once")
@@ -677,7 +677,7 @@ impl<A: Acceptor> AcceptDriver<A> {
     /// driver serves belongs to the same peer, so there is no bystander to
     /// protect with a concurrent label read.
     async fn accept_one(&mut self) -> Result<(), AcceptFate> {
-        let mut rx = self
+        let (mut rx, done) = self
             .acceptor
             .accept()
             .await
@@ -705,7 +705,7 @@ impl<A: Acceptor> AcceptDriver<A> {
                 .ok_or(AcceptError::Duplicate {
                     origin: Origin::stream(self.speaker, stream),
                 })?;
-        slot.send(rx).map_err(|_| {
+        slot.send((rx, done)).map_err(|_| {
             // The claim's consumer already finished without asking anything
             // at this level, so whatever this stream carries was never
             // asked for.
diff --git a/src/tree/mirror/streaming/remote/streams/tests.rs b/src/tree/mirror/streaming/remote/streams/tests.rs
index 5453cd4d..f9c94698 100644
--- a/src/tree/mirror/streaming/remote/streams/tests.rs
+++ b/src/tree/mirror/streaming/remote/streams/tests.rs
@@ -222,7 +222,7 @@ async fn raw_labeled(
     connector: &crate::link::MemoryConnector,
     stream: Stream,
 ) -> FrameWrite<tokio::io::DuplexStream> {
-    let mut tx = connector.connect().await.expect("stream opens");
+    let (mut tx, _) = connector.connect().await.expect("stream opens");
     tx.write_all(&label(EPOCH, stream))
         .await
         .expect("label writes");
@@ -362,44 +362,6 @@ fn truncated_stream_is_reported_not_ended() {
     );
 }
 
-/// A frame arriving after the explicit end control is reported as
-/// `AfterEnd`: a peer that keeps talking past its own end is caught.
-///
-/// After `End(Stream)` the receiver requires transport end-of-stream
-/// before ending cleanly: the double-checked stream end, so trailing
-/// frames are a reportable protocol violation, never silently dropped.
-#[test]
-fn frames_after_the_end_control_are_reported() {
-    let (a, mut b) = memory();
-    let stream = Stream::new(5).expect("stream 5 exists");
-    let error = run_to_quiescence(async {
-        let send = async {
-            // An honest `StreamSender` half-closes right after its end
-            // control, so the frame beyond it comes from the raw writer.
-            let mut write = raw_labeled(&a.connector, stream).await;
-            write
-                .frame(&(stream, Frame::<Unit>::End(End::Stream)))
-                .await
-                .expect("the end control writes");
-            write
-                .frame(&(stream, Frame::<Unit>::End(End::Reply)))
-                .await
-                .expect("the frame beyond the end writes");
-        };
-        let receive = first_reported_error(&mut b.acceptor, stream, &[]);
-        join(send, receive).await.1.0
-    })
-    .expect("after-end detection resolves");
-    // Pinned in full, origin included, like the truncation test above.
-    assert!(
-        matches!(
-            error,
-            StreamError::AfterEnd { origin } if origin == Origin::stream(Speaker::Initiator, stream)
-        ),
-        "unexpected stream error: {error:?}",
-    );
-}
-
 /// A second transport stream bearing an already-delivered label is
 /// rejected by the accept driver as `Duplicate`.
 ///
@@ -448,7 +410,7 @@ fn accept_driver_rejects_unknown_stream_index() {
     let (a, mut b) = memory();
     run_to_quiescence(async {
         let send = async {
-            let mut tx = a.connector.connect().await.expect("stream opens");
+            let (mut tx, _) = a.connector.connect().await.expect("stream opens");
             // One past the last logical stream: no claim slot can exist.
             tx.write_all(&[EPOCH, Stream::COUNT])
                 .await
diff --git a/src/tree/mirror/streaming/stats.rs b/src/tree/mirror/streaming/stats.rs
index 8e2cd71c..b28cf2a2 100644
--- a/src/tree/mirror/streaming/stats.rs
+++ b/src/tree/mirror/streaming/stats.rs
@@ -247,6 +247,11 @@ impl<W> CountedWrite<W> {
     pub fn new(inner: W, recorder: Recorder) -> Self {
         Self { inner, recorder }
     }
+
+    /// Recover the transport half; the counter holds no bytes.
+    pub fn into_inner(self) -> W {
+        self.inner
+    }
 }
 
 impl<W: AsyncWrite + Unpin> AsyncWrite for CountedWrite<W> {
@@ -291,6 +296,11 @@ impl<R> CountedRead<R> {
     pub fn new(inner: R, recorder: Recorder) -> Self {
         Self { inner, recorder }
     }
+
+    /// Recover the transport half; the counter holds no bytes.
+    pub fn into_inner(self) -> R {
+        self.inner
+    }
 }
 
 impl<R: AsyncRead + Unpin> AsyncRead for CountedRead<R> {
diff --git a/tests/common/fault.rs b/tests/common/fault.rs
index 5dfd9cdd..5ab74306 100644
--- a/tests/common/fault.rs
+++ b/tests/common/fault.rs
@@ -31,7 +31,7 @@ use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
 
 use rumors::link::{
-    Acceptor, Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink,
+    Acceptor, Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector, MemoryLink,
 };
 use tokio::io::{AsyncRead, AsyncWrite, DuplexStream, ReadBuf};
 
@@ -149,15 +149,19 @@ impl<C: Clone> Clone for FaultConnector<C> {
 impl<C: Connector> Connector for FaultConnector<C> {
     type Tx = Fuse<C::Tx>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         // A dead write direction cannot open new streams either; this is
         // what lets a cut exercise `SendError::Connect` deterministically
         // instead of only through real-transport races.
         if *self.budget.lock().expect("write budget lock") == 0 {
             return Err(write_severed());
         }
-        let tx = self.inner.connect().await?;
-        Ok(Fuse::new(tx, self.budget.clone()))
+        let (tx, done) = self.inner.connect().await?;
+        // Completion unwraps the fuse and passes the half through.
+        Ok((
+            Fuse::new(tx, self.budget.clone()),
+            Done::new(move |fuse: Fuse<C::Tx>| done.complete(fuse.inner)),
+        ))
     }
 }
 
@@ -171,15 +175,19 @@ pub struct FaultAcceptor<A> {
 impl<A: Acceptor> Acceptor for FaultAcceptor<A> {
     type Rx = Cut<A::Rx>;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         // A dead read direction cannot deliver new streams either; this
         // reaches the session's deferred supply-failure path (the parked
         // accept driver) deterministically rather than only via races.
         if *self.budget.lock().expect("read budget lock") == 0 {
             return Err(read_severed());
         }
-        let rx = self.inner.accept().await?;
-        Ok(Cut::new(rx, self.budget.clone()))
+        let (rx, done) = self.inner.accept().await?;
+        // Completion unwraps the cut and passes the half through.
+        Ok((
+            Cut::new(rx, self.budget.clone()),
+            Done::new(move |cut: Cut<A::Rx>| done.complete(cut.inner)),
+        ))
     }
 }
 
diff --git a/tests/common/gossip_snapshot.rs b/tests/common/gossip_snapshot.rs
index 186f50d6..9825c1af 100644
--- a/tests/common/gossip_snapshot.rs
+++ b/tests/common/gossip_snapshot.rs
@@ -42,7 +42,7 @@ use std::sync::{Arc, Mutex};
 use std::task::{Context, Poll};
 
 use borsh::{BorshDeserialize, BorshSerialize};
-use rumors::link::{Connector, Link, LinkParts, MemoryAcceptor, MemoryConnector};
+use rumors::link::{Connector, Done, Link, LinkParts, MemoryAcceptor, MemoryConnector};
 use rumors::{
     Rumors,
     testing::{LinkCapture, render_v2_capture},
@@ -155,11 +155,11 @@ pub struct CaptureConnector {
 impl Connector for CaptureConnector {
     type Tx = CaptureWrite;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
-        let tx = self.inner.connect().await?;
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
+        let (tx, _) = self.inner.connect().await?;
         let buffer = StreamBuf::default();
         self.streams.lock().unwrap().push(buffer.clone());
-        Ok(CaptureWrite { inner: tx, buffer })
+        Ok((CaptureWrite { inner: tx, buffer }, Done::discard()))
     }
 }
 
diff --git a/tests/common/routed_tcp.rs b/tests/common/routed_tcp.rs
index c448cb7e..70f65868 100644
--- a/tests/common/routed_tcp.rs
+++ b/tests/common/routed_tcp.rs
@@ -6,10 +6,13 @@
 //! the OS floor. Socket policy stops here: the adapter above sees only
 //! byte streams.
 
+use std::collections::HashMap;
 use std::io;
 use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
 
 use rumors::link::routed::{Dial, Listen};
+use tokio::io::AsyncReadExt;
 use tokio::net::{TcpListener, TcpSocket, TcpStream};
 
 /// Listener backlog: a full session complement of simultaneous stream
@@ -40,6 +43,48 @@ impl Dial for TcpDial {
     }
 }
 
+/// A TCP dialer pooling recycled connections per peer, so completed
+/// streams ride recycled connections instead of fresh dials.
+#[derive(Clone, Default)]
+pub struct PoolingTcpDial {
+    pool: Arc<Mutex<HashMap<SocketAddr, Vec<TcpStream>>>>,
+}
+
+impl Dial for PoolingTcpDial {
+    type Addr = SocketAddr;
+    type Conn = TcpStream;
+
+    async fn dial(&self, addr: &SocketAddr) -> io::Result<TcpStream> {
+        let pooled = self
+            .pool
+            .lock()
+            .expect("pool lock")
+            .get_mut(addr)
+            .and_then(Vec::pop);
+        match pooled {
+            Some(conn) => Ok(conn),
+            None => TcpStream::connect(*addr).await,
+        }
+    }
+
+    fn recycle(&self, peer: &SocketAddr, mut conn: TcpStream) {
+        // Pooled only once the router's ready byte arrives, read off
+        // the dialing path.
+        let pool = Arc::clone(&self.pool);
+        let peer = *peer;
+        tokio::spawn(async move {
+            let mut ready = [0u8; 1];
+            if conn.read_exact(&mut ready).await.is_ok() {
+                pool.lock()
+                    .expect("pool lock")
+                    .entry(peer)
+                    .or_default()
+                    .push(conn);
+            }
+        });
+    }
+}
+
 /// Accepts one process's inbound routed-link connections.
 pub struct TcpListen(TcpListener);
 
diff --git a/tests/common/tcp.rs b/tests/common/tcp.rs
index 7ef3cfc4..84756209 100644
--- a/tests/common/tcp.rs
+++ b/tests/common/tcp.rs
@@ -24,7 +24,7 @@ use std::io;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use rumors::link::{Acceptor, Connector, Link, STREAM_COUNT};
+use rumors::link::{Acceptor, Connector, Done, Link, STREAM_COUNT};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{
     TcpListener, TcpSocket, TcpStream,
@@ -108,7 +108,7 @@ pub struct TcpConnector(Arc<Target>);
 impl Connector for TcpConnector {
     type Tx = OwnedWriteHalf;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         let stream = match self.0.send_buffer {
             None => TcpStream::connect(self.0.peer).await?,
             Some(send) => {
@@ -122,7 +122,7 @@ impl Connector for TcpConnector {
         // dropping it later half-closes toward the peer. The unread half is
         // dropped now; the peer never writes on this socket.
         drop(read);
-        Ok(write)
+        Ok((write, Done::discard()))
     }
 }
 
@@ -132,12 +132,12 @@ pub struct TcpAcceptor(TcpListener);
 impl Acceptor for TcpAcceptor {
     type Rx = OwnedReadHalf;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         let (stream, _) = self.0.accept().await?;
         let (read, write) = stream.into_split();
         // Half-close our unused direction immediately; the peer never reads
         // this socket, so the shutdown is invisible to it.
         drop(write);
-        Ok(read)
+        Ok((read, Done::discard()))
     }
 }
diff --git a/tests/dispute_wire.rs b/tests/dispute_wire.rs
index 6f8106fb..91744ea8 100644
--- a/tests/dispute_wire.rs
+++ b/tests/dispute_wire.rs
@@ -32,7 +32,7 @@ use std::task::{Context, Poll};
 use borsh::{BorshDeserialize, BorshSerialize};
 use rand::rngs::SmallRng;
 use rand::{RngCore, SeedableRng};
-use rumors::link::{Connector, Link, LinkParts, MemoryLink};
+use rumors::link::{Connector, Done, Link, LinkParts, MemoryLink};
 use rumors::testing::{dispute_overhead_bytes, envelope_and_wire_bytes};
 use rumors::{Peer, Rumors};
 use tokio::io::AsyncWrite;
@@ -123,11 +123,15 @@ struct CountingConnector<C> {
 impl<C: Connector> Connector for CountingConnector<C> {
     type Tx = CountingWrite<C::Tx>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
-        Ok(CountingWrite {
-            inner: self.inner.connect().await?,
-            written: self.written.clone(),
-        })
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
+        let (inner, _) = self.inner.connect().await?;
+        Ok((
+            CountingWrite {
+                inner,
+                written: self.written.clone(),
+            },
+            Done::discard(),
+        ))
     }
 }
 
diff --git a/tests/hop_trace.rs b/tests/hop_trace.rs
index 168bfee1..38b50d4a 100644
--- a/tests/hop_trace.rs
+++ b/tests/hop_trace.rs
@@ -39,7 +39,7 @@ use std::time::Duration;
 use rand::rngs::SmallRng;
 use rand::seq::SliceRandom;
 use rand::{RngCore, SeedableRng};
-use rumors::link::{Acceptor, Connector, Link, STREAM_COUNT};
+use rumors::link::{Acceptor, Connector, Done, Link, STREAM_COUNT};
 use rumors::{DEFAULT_SYNC_MEMORY_BUDGET, Key, Peer, Protocol, Rumors};
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::sync::mpsc;
@@ -234,7 +234,7 @@ struct TracedConnector {
 impl Connector for TracedConnector {
     type Tx = TracedWriter;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         let (tx, rx) = delayed_pipe(CAPACITY, DELAY);
         let pipe = PipeId {
             side: self.side,
@@ -248,11 +248,14 @@ impl Connector for TracedConnector {
             })
             .await
             .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "peer link is gone"))?;
-        Ok(TracedWriter {
-            inner: tx,
-            pipe,
-            trace: self.trace.clone(),
-        })
+        Ok((
+            TracedWriter {
+                inner: tx,
+                pipe,
+                trace: self.trace.clone(),
+            },
+            Done::discard(),
+        ))
     }
 }
 
@@ -264,10 +267,11 @@ struct TracedAcceptor {
 impl Acceptor for TracedAcceptor {
     type Rx = TracedReader;
 
-    async fn accept(&mut self) -> io::Result<Self::Rx> {
+    async fn accept(&mut self) -> io::Result<(Self::Rx, Done<Self::Rx>)> {
         self.streams
             .recv()
             .await
+            .map(|rx| (rx, Done::discard()))
             .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "peer link is gone"))
     }
 }
diff --git a/tests/routed_link.rs b/tests/routed_link.rs
index b248f0bb..985d79d3 100644
--- a/tests/routed_link.rs
+++ b/tests/routed_link.rs
@@ -28,7 +28,7 @@ use tokio::io::AsyncWriteExt;
 use tokio::net::TcpStream;
 use tokio::time::timeout;
 
-use crate::common::routed_tcp::{TcpDial, TcpListen};
+use crate::common::routed_tcp::{PoolingTcpDial, TcpDial, TcpListen};
 use crate::common::wire::bootstrap_fork_async;
 
 /// Bound on one whole suite run; loopback and in-memory checks finish
@@ -163,6 +163,102 @@ async fn conforms_over_tcp_at_minimal_buffers_swapped() {
     tcp_conformance(Some(MINIMAL_BUFFER_REQUEST), false).await;
 }
 
+/// Mint a routed-link pair whose shared dialer pools recycled
+/// connections, so the suite's completed streams ride recycled
+/// connections wherever a pooled one is available.
+async fn pooled_tcp_pair(
+    dialer_first: bool,
+) -> (RoutedLink<PoolingTcpDial>, RoutedLink<PoolingTcpDial>) {
+    let dial = PoolingTcpDial::default();
+    let pooled_endpoint = async |dial: PoolingTcpDial| {
+        let (listen, addr) = TcpListen::bind(None)
+            .await
+            .expect("bind a loopback listener");
+        let (endpoint, incoming, router) = Endpoint::new(listen, addr, dial, Config::default())
+            .expect("an unscoped loopback name is routable");
+        tokio::spawn(router);
+        (endpoint, incoming, addr)
+    };
+    let (_a, mut a_incoming, a_addr) = pooled_endpoint(dial.clone()).await;
+    let (b, _b_incoming, _b_addr) = pooled_endpoint(dial).await;
+    let (linked, arrival) = tokio::join!(b.link(a_addr), a_incoming.accept());
+    let dialed = linked.expect("establishment succeeds");
+    let (_info, accepted) = arrival.expect("the router delivers the link");
+    if dialer_first {
+        (dialed, accepted)
+    } else {
+        (accepted, dialed)
+    }
+}
+
+/// The suite holds when completed streams ride recycled connections:
+/// pooling is invisible to every clause the contract states.
+#[tokio::test]
+async fn conforms_over_tcp_with_a_pooling_dial() {
+    timeout(
+        SUITE_TIMEOUT,
+        rumors::conformance::link::check(async || pooled_tcp_pair(true).await),
+    )
+    .await
+    .expect("conformance suite ran past its liveness bound");
+}
+
+/// The pooling suite with the pair's seats swapped, as for the plain
+/// TCP variants.
+#[tokio::test]
+async fn conforms_over_tcp_with_a_pooling_dial_swapped() {
+    timeout(
+        SUITE_TIMEOUT,
+        rumors::conformance::link::check(async || pooled_tcp_pair(false).await),
+    )
+    .await
+    .expect("conformance suite ran past its liveness bound");
+}
+
+/// Mutual gossip sessions over a pooling dialer converge under a
+/// multi-thread scheduler.
+///
+/// The regression this pins: a pool that hands out a recycled
+/// connection before the peer router's ready byte couples the next
+/// stream's delivery to the previous stream's consumer, and mutual
+/// sessions then deadlock. Single-thread schedules rarely close the
+/// cycle, so the flavor here is load-bearing.
+#[tokio::test(flavor = "multi_thread")]
+async fn pooled_mutual_sessions_converge() {
+    timeout(SUITE_TIMEOUT, async {
+        let (mut a, mut b) = pooled_tcp_pair(true).await;
+        let seed: Rumors<u64> = Peer::seed().into_rumors();
+        let (served, joined) =
+            tokio::join!(seed.gossip(&mut a), Peer::<u64>::bootstrap().join(&mut b));
+        served.expect("the bootstrap-serving session completes");
+        let newcomer = joined
+            .expect("the bootstrap session completes")
+            .expect("the seed serves the bootstrap")
+            .into_rumors();
+        {
+            let mut batch = seed.batch();
+            for payload in 0..48u64 {
+                batch.send(payload);
+            }
+        }
+        {
+            let mut batch = newcomer.batch();
+            for payload in 48..96u64 {
+                batch.send(payload);
+            }
+        }
+        for _ in 0..2 {
+            let (near, far) = tokio::join!(seed.gossip(&mut a), newcomer.gossip(&mut b));
+            near.expect("gossip completes over the link");
+            far.expect("gossip completes over the link");
+        }
+        assert_eq!(seed.snapshot().len(), 96);
+        assert_eq!(seed.snapshot(), newcomer.snapshot());
+    })
+    .await
+    .expect("pooled mutual sessions timed out");
+}
+
 /// The adapter conforms over the in-memory network too: nothing in
 /// the contract mapping leans on socket semantics, and the string
 /// names prove the address seam carries non-IP namespaces.
diff --git a/tests/session_stats.rs b/tests/session_stats.rs
index 8585141a..ae3dbed1 100644
--- a/tests/session_stats.rs
+++ b/tests/session_stats.rs
@@ -19,7 +19,7 @@ use std::task::{Context, Poll};
 
 use futures::StreamExt;
 use proptest::prelude::*;
-use rumors::link::{Connector, Link, LinkParts, MemoryLink};
+use rumors::link::{Connector, Done, Link, LinkParts, MemoryLink};
 use rumors::{Gossiped, Led, Peer, Rumors, SessionStats};
 use tokio::io::AsyncWrite;
 
@@ -195,12 +195,16 @@ struct CountingConnector<C> {
 impl<C: Connector> Connector for CountingConnector<C> {
     type Tx = CountingWrite<C::Tx>;
 
-    async fn connect(&self) -> io::Result<Self::Tx> {
+    async fn connect(&self) -> io::Result<(Self::Tx, Done<Self::Tx>)> {
         self.opens.fetch_add(1, Ordering::Relaxed);
-        Ok(CountingWrite {
-            inner: self.inner.connect().await?,
-            written: self.written.clone(),
-        })
+        let (inner, _) = self.inner.connect().await?;
+        Ok((
+            CountingWrite {
+                inner,
+                written: self.written.clone(),
+            },
+            Done::discard(),
+        ))
     }
 }