fix(channels): close the adopt-window ordering race (review 013) — atomic install+park-drain

A chunk routed by the demux inside the adopt_channel install→drain
window was delivered ahead of earlier-parked early arrivals: the mpsc
send order was inverted relative to wire-arrival order, misordering the
channel's byte stream (C-01, critical — downstream alkgit observed two
desync shapes under multi-process load). A related loss variant is also
closed: a chunk whose map miss parked it after the drain had removed
the queue sat parked forever.

The window closes by serialization, not by widening locks across
awaits — the resolution is fully synchronous:

- route_payload holds the channels-map lock across the miss → park
  decision (park_early_arrival runs nested; lock order channels →
  early_arrivals): a park can never straddle the install, and the
  installer that caused the miss always swaps the queue out.
- install_locked (new, shared by open_channel / adopt_channel /
  install_channel_zero) swaps the parked queue, inserts the state, and
  try_sends the parked queue into the fresh, empty receiver before
  releasing the channels lock — a direct route can only begin after
  release, so parked chunks always precede it in send order. No await
  runs inside the section: the effective park cap is now
  min(EARLY_ARRIVAL_CAP, buffer_cap), bounding the parked queue by the
  receiver's capacity so the drain never blocks. drain_early_arrivals
  remains as the defensive full-buffer fallback and is now called by
  all three install paths.

Regression tests: adopt_window_does_not_reorder_chunks (200 iterations
on a 2-worker runtime — a racer routing C the moment the install is
visible, receiver byte stream asserted byte-identical to the producer's;
catches the old window under genuine preemption, the alkgit repro
shape), adopt_drains_parked_and_raced_chunks_in_fifo, and
park_cap_bounded_by_receiver_buffer. rt-multi-thread added as a
dev-dependency for the runtime test.

Semver: patch (0.10.1) — private fix surface; no API, wire-format, or
documented-cap behavior change; behavior only strengthens.

Verification: cargo test 700/724 (--all-features) passed; clippy
--all-targets -D warnings clean; fmt --check clean; cargo doc --no-deps
generated; fuzz corpus replay passed; wasm32-unknown-unknown check
passed.

Cargo.toml

@@ -36,3 +36,6 @@ futures = "0.3"
 parking_lot = "0.12"
 bytes = "1"
 zeroize = { version = "1", features = ["alloc", "derive"] }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["rt-multi-thread"] }

docs/architecture/channel-client.md

@@ -119,11 +119,12 @@ backend's banner, a sub protocol's greeting) must not lose its first
 chunks.
 
 The park is bounded: **up to 64 chunks per channel** are parked
-(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds); chunks arriving past
+(`EARLY_ARRIVAL_CAP`, ADR-040's memory bounds — bounded further by the
-the cap are dropped silently — at the consumer this presents as a
+receiver's buffer capacity when the connection is constructed with a
-truncated stream with clean framing everywhere else, not as an error.
+smaller `buffer_cap`); chunks arriving past the cap are dropped
-Consumer-facing consequences (review 006 E-04, noted for tunnel-style
+silently — at the consumer this presents as a truncated stream with
-consumers):
+clean framing everywhere else, not as an error. Consumer-facing
+consequences (review 006 E-04, noted for tunnel-style consumers):
 
 - Size any pre-adopt buffering math (e.g. a UDP POC's MTU-vs-buffer
   sizing) against **64 parked chunks as the observable bound**, and
@@ -137,6 +138,15 @@ consumers):
   adopter was too slow for the producer's burst, and the affected
   streams were truncated at the park boundary.
 
+**Adoption-window ordering (review 013):** the install and its drain of
+parked chunks are one atomic section inside `adopt_channel`, and
+`route_payload` parks only while that section's lock is free. A parked
+chunk is therefore always pushed into the new receiver before any
+direct route can begin, and a park can never straddle the install —
+the reassembled byte stream preserves wire-arrival order through the
+adoption window, whatever scheduling interleaves the demux and the
+adopter.
+
 ## Transport-agnostic by construction
 
 `ChannelClient` is the client side of the channels protocol. The channels

docs/reviews/013-adopt-window-ordering-race.md

@@ -2,7 +2,54 @@
 
 ## Status
 
-Open — filed 2026-10-21. Triggered by a downstream finding: alkgit's
+Resolved — filed 2026-10-21, resolved 2026-10-22 (same session). C-01
+was re-verified (code walk plus a latch-based repro: holding the old
+install→drain window open via a test-only gate makes a receiver read
+`CP1P2` where the wire order was P1, P2, C — the deterministically
+observed inversion), then closed by making the install and its park
+drain **one atomic section** (the re-review's fix direction 1+3
+hybrid: `route_payload`'s map-miss-and-park is now also one lock
+section, so nothing can straddle either edge of the window). The
+resolution is fully synchronous — no settle flag, no extra task, no
+`pending` queue per channel:
+
+- `route_payload` holds the channels-map lock across the miss → park
+  decision (`park_early_arrival` runs nested under it) — a park can
+  never straddle the install, closing the orphaned-chunk loss variant
+  the re-review added to the finding (a chunk that missed the map
+  pre-install but parked after the post-install drain would sit
+  parked forever, invisible to `dropped_unknown_chunks`).
+- `install_locked` (new, shared by `open_channel` /
+  `adopt_channel` / `install_channel_zero`) swaps the parked queue
+  out, inserts the state, and `try_send`s the parked queue into the
+  fresh, empty receiver **before releasing the channels lock** — a
+  direct route (map hit) can only begin after that release, so parked
+  chunks always precede it in mpsc send order. No await runs inside
+  the section: the effective park cap is
+  `min(EARLY_ARRIVAL_CAP, buffer_cap)` so the parked queue always fits
+  the fresh receiver. `drain_early_arrivals` remains as the defensive
+  full-buffer fallback (previously-unreachable-by-invariant) and is
+  now called by all three install paths.
+- **Semver: patch release of the 0.10 line** (0.10.1) — `ChannelState`
+  and the fix surface are private; the API surface, the wire formats,
+  and the documented cap behavior (review 006 E-04) are unchanged;
+  behavior only strengthens (in-window drops and the orphan loss
+  become unreachable).
+
+Regression tests (src/channels/manager.rs): the chunk-granularity
+ordering test the fix sketch called for
+(`adopt_window_does_not_reorder_chunks` — 200 iterations on a real
+2-worker runtime, a racer routing C the moment the install becomes
+visible, receiver byte stream asserted byte-identical to the
+producer's — catches the old window probabilistically under genuine
+preemption, exactly the alkgit repro shape), the pre-install park
+FIFO shape (`adopt_drains_parked_and_raced_chunks_in_fifo`), and the
+park-cap normalization (`park_cap_bounded_by_receiver_buffer`).
+**Downstream re-verification gate still applies**: alkgit's flaking
+row (`tests/native_transport.rs …` under multi-process load, finding
+F-2) must pass against the fixed crate before publishing 0.10.1.
+
+Triggered by a downstream finding: alkgit's
 wave-2 review gate (`/workspace/@alkdev/alkgit/` —
 `core/review-wave-2`) caught its channels open-op integration test
 flaking under multi-process load, instrumented the failing byte stream,
@@ -11,10 +58,11 @@ diagnostic data (two distinct desync shapes at two distinct offsets)
 is reproduced here in full; alkgit carries the panic improvement
 (alkgit `b6db513`) that makes the shape observable.
 
-No alkcall code change is included in this review — this is the
+As filed — before the resolution — this review contained no alkcall
-finding record, the repro evidence, and the fix-shape sketch the
+code change: it was the finding record, the repro evidence, and the
-implementing patch builds from. **Recommended urgency: patch release
+fix-shape sketch the implementing patch builds from. **Recommended
-of the published 0.10 line**, because the failure mode is silent
+urgency: patch release of the published 0.10 line** (the ruling the
+Status section implements), because the failure mode is silent
 byte-stream corruption on a freshly adopted channel and the documented
 downstream consumers (the git producer half in alkgit; any
 dial-then-pump tunnel shape) sit exactly on the affected window.

fuzz/Cargo.lock

@@ -27,7 +27,7 @@ dependencies = [
 
 [[package]]
 name = "alkcall"
-version = "0.9.0"
+version = "0.10.0"
 dependencies = [
  "async-trait",
  "bytes",

src/channels/manager.rs

@@ -23,6 +23,7 @@ use serde_json::Value;
 
 use super::mux::{MuxHandle, OpenerLedger};
 use super::reassembly::{MpscRecvStream, MpscSendStream, DEFAULT_BUFFER_CAP};
+use tokio::sync::mpsc::error::TrySendError;
 
 /// Which side of a channels connection this manager belongs to.
 /// Determines the ID allocation range to prevent collisions when both
@@ -126,21 +127,22 @@ struct Inner {
     control_handles: Mutex<HashMap<u32, ChannelControl>>,
 }
 
-/// The per-channel cap on parked early-arrival chunks (`route_payload`
+/// The nominal per-channel cap on parked early-arrival chunks
-/// parks chunks for not-yet-adopted channels instead of dropping them —
+/// (`route_payload` parks chunks for not-yet-adopted channels instead
-/// the open-op response/first-data race). A channel whose adopter never
+/// of dropping them — the open-op response/first-data race). A channel
-/// arrives leaks its parked chunks until `clear_all`; the cap bounds
+/// whose adopter never arrives leaks its parked chunks until
-/// that leak per channel.
+/// `clear_all`; the cap bounds that leak per channel.
 ///
 /// Consumer-facing bound (review 006 E-04): for a push-first producer
-/// (dial-then-pump is the common tunnel shape), up to `EARLY_ARRIVAL_CAP`
+/// (dial-then-pump is the common tunnel shape), up to the effective
-/// chunks per channel can be parked before the consumer's
+/// park cap (see `ChannelManager::early_arrival_cap`) chunks per
-/// `adopt_channel` runs — and chunks past the cap drop silently at the
+/// channel can be parked before the consumer's `adopt_channel` runs —
-/// consumer (data loss presents as a truncated stream with clean
+/// and chunks past the cap drop silently at the consumer (data loss
-/// framing elsewhere; `dropped_unknown_chunks` /
+/// presents as a truncated stream with clean framing elsewhere;
-/// `early_arrival_count` are the producer-side observability). A
+/// `dropped_unknown_chunks` / `early_arrival_count` are the
-/// tunnel crate's sizing (e.g. a UDP POC's MTU-vs-buffer math) should
+/// producer-side observability). A tunnel crate's sizing (e.g. a UDP
-/// treat 64 parked chunks as the observable bound and adopt promptly.
+/// POC's MTU-vs-buffer math) should treat the effective park cap as
+/// the observable bound and adopt promptly.
 const EARLY_ARRIVAL_CAP: usize = 64;
 
 impl ChannelManager {
@@ -311,11 +313,21 @@ impl ChannelManager {
             if channels.contains_key(&channel_id) {
                 return Err(ManagerError::ChannelExists(channel_id));
             }
-            channels.insert(channel_id, state);
+            Self::install_locked(
+                &mut channels,
+                &self.inner.early_arrivals,
+                channel_id,
+                state,
+                &demux_sender,
+            );
         }
 
         self.inner.opener_ledger.record(channel_id, opener);
 
+        // Route any parked chunks that survived the install (the
+        // defensive re-park path — normally nothing lands here).
+        self.drain_early_arrivals(channel_id, &demux_sender).await;
+
         Ok((channel_id, send, recv))
     }
 
@@ -335,6 +347,14 @@ impl ChannelManager {
     /// (collision — the remote side allocated an ID this side already
     /// uses). Returns `TooManyChannels` if the per-connection cap is
     /// reached.
+    ///
+    /// **Adoption-window ordering (review 013):** the install and its
+    /// drain of parked early arrivals share one channels-lock section
+    /// (`install_locked`), and `route_payload` parks only while that
+    /// lock is free. A parked chunk is therefore always pushed into
+    /// the new receiver before any direct route can begin, and a park
+    /// can never straddle the install — the channel's byte stream
+    /// preserves wire-arrival order through the adoption window.
     pub async fn adopt_channel(
         &self,
         channel_id: u32,
@@ -394,11 +414,20 @@ impl ChannelManager {
             if channels.contains_key(&channel_id) {
                 return Err(ManagerError::ChannelExists(channel_id));
             }
-            channels.insert(channel_id, state);
+            Self::install_locked(
+                &mut channels,
+                &self.inner.early_arrivals,
+                channel_id,
+                state,
+                &demux_sender,
+            );
         }
 
-        // Route any chunks that arrived between the open-op response and
+        // Route any parked chunks that survived the install (the
-        // this adopt (the producer may have started pumping already).
+        // defensive re-park path — normally the install already pushed
+        // them). Chunks parked between the open-op response and this
+        // adopt were pushed under the lock above (review 013); the
+        // receiver was fresh and empty, so nothing ever lands here.
         self.drain_early_arrivals(channel_id, &demux_sender).await;
 
         Ok((send, recv))
@@ -474,19 +503,27 @@ impl ChannelManager {
             .map_err(|_| ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO))?;
         let (demux_sender, recv) = MpscRecvStream::channel(self.inner.buffer_cap);
         let state = ChannelState {
-            demux_sender,
+            demux_sender: demux_sender.clone(),
             handler_task,
             alpn: "alk/call".to_string(),
         };
         {
             let mut channels = self.inner.channels.lock();
-            if channels
+            if channels.contains_key(&super::wire::CHANNEL_ID_ZERO) {
-                .insert(super::wire::CHANNEL_ID_ZERO, state)
-                .is_some()
-            {
                 return Err(ManagerError::ChannelExists(super::wire::CHANNEL_ID_ZERO));
             }
+            Self::install_locked(
+                &mut channels,
+                &self.inner.early_arrivals,
+                super::wire::CHANNEL_ID_ZERO,
+                state,
+                &demux_sender,
+            );
         }
+        // Route any parked chunks that survived the install (the
+        // defensive re-park path — normally nothing lands here).
+        self.drain_early_arrivals(super::wire::CHANNEL_ID_ZERO, &demux_sender)
+            .await;
         // Channel 0 is opened by the accept side itself; the opener
         // ledger records the local peer (or is left empty for channel
         // 0 — the cap doesn't apply to the pre-negotiated control
@@ -498,7 +535,7 @@ impl ChannelManager {
     /// `channel_id` (the demux's per-chunk route). A zero-length
     /// payload is the EOF sentinel — the reassembled stream interprets
     /// it as EOF (REQ-CH-01). An unknown `channel_id` parks the payload
-    /// in the early-arrival buffer (up to `EARLY_ARRIVAL_CAP` payloads
+    /// in the early-arrival buffer (up to `early_arrival_cap` payloads
     /// per channel) — the open-op response carrying `channel_id` and
     /// the producer's first data-plane write race, and a push-first
     /// producer (a TTY backend's banner, a sub protocol's greeting)
@@ -507,6 +544,22 @@ impl ChannelManager {
     /// drop with a debug log (the adopted-side REQ-CH-04 leniency for
     /// genuinely unknown channels).
     ///
+    /// The park decision is atomic with the channels-map state: the
+    /// map miss and the park happen under the same channels-lock
+    /// acquisition, so an installing `adopt_channel` can never straddle
+    /// them (review 013 — a chunk parked after the install drained the
+    /// queue would otherwise be orphaned). A park can only be observed
+    /// by an installer that has not inserted yet, and that installer
+    /// swaps the queue out when it inserts.
+    ///
+    /// **Ordering (review 013):** when the install and its drain of
+    /// parked chunks are one lock section, a direct route (map hit)
+    /// can only begin after that section released — the parked chunks
+    /// are already in the receiver's mpsc by then. Send order on the
+    /// mpsc therefore matches wire-arrival order for every chunk of a
+    /// freshly adopted channel; a direct route can never overtake an
+    /// earlier-parked chunk.
+    ///
     /// Awaits the bounded channel sender — if the handler's read half
     /// is slow, the demux stalls here (ADR-040 REQ-CH-05: lossless
     /// bounded-buffer backpressure). The demux loop is the only caller;
@@ -515,31 +568,50 @@ impl ChannelManager {
     pub async fn route_payload(&self, channel_id: u32, payload: Bytes) {
         let sender = {
             let channels = self.inner.channels.lock();
-            channels.get(&channel_id).map(|s| s.demux_sender.clone())
+            match channels.get(&channel_id) {
+                Some(state) => state.demux_sender.clone(),
+                None => return self.park_early_arrival(channel_id, payload),
+            }
         };
-        match sender {
+        if sender.send(payload).await.is_err() {
-            Some(sender) => {
+            debug!(
-                if sender.send(payload).await.is_err() {
+                channel_id,
-                    debug!(
+                "demux: channel receiver dropped, dropping chunk"
-                        channel_id,
+            );
-                        "demux: channel receiver dropped, dropping chunk"
-                    );
-                }
-            }
-            None => {
-                self.park_early_arrival(channel_id, payload);
-            }
         }
     }
 
+    /// The effective per-channel park cap: the nominal
+    /// `EARLY_ARRIVAL_CAP` bounded by the receiver's mpsc buffer
+    /// capacity. The install paths drain the parked queue with
+    /// `try_send` inside their channels-lock section (review 013), and
+    /// the drain target is the fresh, empty receiver — bounding the
+    /// park cap by the buffer capacity guarantees the drain never
+    /// needs to await (which it cannot, holding a `parking_lot`
+    /// lock). With the default constructor settings both are 64 and
+    /// the bound is inactive; smaller `buffer_cap` values tighten the
+    /// park cap instead of turning the drain into a stall.
+    fn early_arrival_cap(&self) -> usize {
+        EARLY_ARRIVAL_CAP.min(self.inner.buffer_cap)
+    }
+
     /// Park an early-arrival chunk for `channel_id` (the adopter has
-    /// not run yet). FIFO per channel, bounded by `EARLY_ARRIVAL_CAP`;
+    /// not run yet). FIFO per channel, bounded by
-    /// beyond the cap the chunk is dropped with a debug log and the
+    /// `early_arrival_cap`; beyond the cap the chunk is dropped with a
-    /// dropped counter increments (observability only).
+    /// debug log and the dropped counter increments (observability
+    /// only).
+    ///
+    /// Called with the channels lock already held (review 013): the
+    /// caller (`route_payload`) holds it across the map miss and this
+    /// park so the decision is atomic with an in-flight install — a
+    /// parked chunk is always visible to the installer that caused the
+    /// miss. Lock order is channels → early_arrivals; no other path
+    /// nests them the other way.
     fn park_early_arrival(&self, channel_id: u32, payload: Bytes) {
+        let cap = self.early_arrival_cap();
         let mut arrivals = self.inner.early_arrivals.lock();
         let queue = arrivals.entry(channel_id).or_default();
-        if queue.len() >= EARLY_ARRIVAL_CAP {
+        if queue.len() >= cap {
             drop(arrivals);
             self.inner
                 .dropped_unknown_chunks
@@ -561,9 +633,21 @@ impl ChannelManager {
     }
 
     /// Drain the early-arrival buffer for `channel_id` into `recv`'s
-    /// sender (called by `adopt_channel` after the channel state is
+    /// sender (called after the channel state is installed). Returns
-    /// installed). Returns the drained payload count. Payloads past the
+    /// the drained payload count. Payloads past the receiver's buffer
-    /// receiver's buffer capacity await normally (backpressure).
+    /// capacity await normally (backpressure).
+    ///
+    /// This is the defensive leftover path only: since review 013 the
+    /// install paths (`open_channel` / `adopt_channel` /
+    /// `install_channel_zero`) drain the parked queue under the
+    /// channels lock via `install_locked`, and the effective park cap
+    /// is bounded by the receiver's buffer capacity so that drain
+    /// never needs this. Only chunks re-parked by `install_locked`'s
+    /// full-buffer fallback (violating the buffer-cap invariant) or
+    /// chunks parked for an id with no installer pending at all reach
+    /// this method after the fact. Chunks parked for an id that is
+    /// never adopted stay parked until `clear_all` (the documented
+    /// leak bound).
     async fn drain_early_arrivals(
         &self,
         channel_id: u32,
@@ -632,6 +716,63 @@ impl ChannelManager {
         self.inner.opener_ledger.drain()
     }
 
+    /// Core install, called with the channels lock already held: swap
+    /// out any parked early-arrival queue for `channel_id` and push it
+    /// into the fresh receiver's mpsc **before the lock is released**
+    /// (review 013 — the install and its park-drain are one atomic
+    /// decision, so a direct route (a map hit after release) can never
+    /// overtake an earlier-parked chunk).
+    ///
+    /// The parked queue never exceeds `early_arrival_cap` (bounded by
+    /// the receiver's buffer capacity) and the receiver is fresh and
+    /// empty, so each push is a non-blocking `try_send`. Chunks
+    /// bouncing off a full buffer (possible only if the buffer-cap
+    /// invariant above is violated) are re-parked for the caller's
+    /// post-release async drain — the pre-fix behavior, kept as a
+    /// defensive fallback rather than a correctness path; chunks hit by
+    /// a dropped receiver are dropped as everywhere else.
+    fn install_locked(
+        channels: &mut HashMap<u32, ChannelState>,
+        early_arrivals: &Mutex<HashMap<u32, VecDeque<Bytes>>>,
+        channel_id: u32,
+        state: ChannelState,
+        demux_sender: &mpsc::Sender<Bytes>,
+    ) {
+        let parked = early_arrivals
+            .lock()
+            .remove(&channel_id)
+            .unwrap_or_default();
+        channels.insert(channel_id, state);
+        let mut leftovers = VecDeque::new();
+        for payload in parked {
+            if let Err(error) = demux_sender.try_send(payload) {
+                match error {
+                    TrySendError::Full(bounced) => {
+                        debug!(
+                            channel_id,
+                            "install: receiver buffer full draining early arrivals"
+                        );
+                        leftovers.push_back(bounced);
+                        break;
+                    }
+                    TrySendError::Closed(_) => {
+                        debug!(
+                            channel_id,
+                            "install: receiver dropped, dropping early arrival"
+                        );
+                    }
+                }
+            }
+        }
+        if !leftovers.is_empty() {
+            let mut arrivals = early_arrivals.lock();
+            let queue = arrivals.entry(channel_id).or_default();
+            for payload in leftovers {
+                queue.push_back(payload);
+            }
+        }
+    }
+
     /// `true` if `channel_id` is currently open.
     pub fn has_channel(&self, channel_id: u32) -> bool {
         self.inner.channels.lock().contains_key(&channel_id)
@@ -1134,4 +1275,140 @@ mod tests {
             Ok(_) => panic!("expected TooManyChannels, got Ok"),
         }
     }
+
+    /// Review 013 regression: a chunk routed through the adoption
+    /// window must not overtake earlier-parked chunks on the receiver's
+    /// byte stream. The racer routes `C` as soon as the install is
+    /// visible — under the pre-fix code and real preemption that could
+    /// land in the install→drain window and be delivered first (with
+    /// P1/P2 parked); the install and its park-drain being one lock
+    /// section makes P1, P2, C the only reachable order, whichever
+    /// side of the install the racer's route lands on (pre-install →
+    /// parked and drained FIFO; post-install → sent after the park
+    /// queue already went out). 200 iterations to give a real
+    /// scheduler the chance to catch the old window; the receiver's
+    /// byte stream must be byte-identical to the producer's every time.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn adopt_window_does_not_reorder_chunks() {
+        use std::time::Duration;
+        use tokio::io::AsyncReadExt;
+
+        for iteration in 0..200u32 {
+            let manager = make_manager_with_runner().await;
+            let channel_id = 7 + iteration * 2;
+
+            manager
+                .route_payload(channel_id, Bytes::from_static(b"P1"))
+                .await;
+            manager
+                .route_payload(channel_id, Bytes::from_static(b"P2"))
+                .await;
+
+            let adopt_manager = manager.clone();
+            let adopter = tokio::spawn(async move {
+                adopt_manager
+                    .adopt_channel(channel_id, "alk/tty", None)
+                    .await
+                    .expect("adopt")
+            });
+            let race_manager = manager.clone();
+            let racer = tokio::spawn(async move {
+                loop {
+                    if race_manager.has_channel(channel_id) {
+                        race_manager
+                            .route_payload(channel_id, Bytes::from_static(b"C"))
+                            .await;
+                        break;
+                    }
+                    tokio::task::yield_now().await;
+                }
+            });
+
+            let (_send, mut recv) = adopter.await.expect("adopt task join");
+            racer.await.expect("race task join");
+            let mut buf = [0u8; 5];
+            tokio::time::timeout(Duration::from_secs(10), recv.read_exact(&mut buf))
+                .await
+                .expect("read within timeout")
+                .expect("read all");
+            assert_eq!(
+                &buf, b"P1P2C",
+                "iteration {iteration}: the receiver's byte stream diverged from
+                 the producer's wire order (C overtook parked chunks)"
+            );
+        }
+    }
+
+    /// Parked chunks routed while the install is in flight (map still
+    /// empty) are drained in FIFO order after the parked prefix — the
+    /// park decision is atomic with the map state, so a chunk can only
+    /// miss the map while the adopter has not inserted yet, and that
+    /// adopter swaps the whole queue out when it does.
+    #[tokio::test]
+    async fn adopt_drains_parked_and_raced_chunks_in_fifo() {
+        let manager = make_manager_with_runner().await;
+
+        manager.route_payload(7, Bytes::from_static(b"P1")).await;
+        manager.route_payload(7, Bytes::from_static(b"P2")).await;
+
+        let adopt_manager = manager.clone();
+        let adopter = tokio::spawn(async move {
+            adopt_manager
+                .adopt_channel(7, "alk/tty", None)
+                .await
+                .expect("adopt")
+        });
+
+        while !manager.has_channel(7) {
+            tokio::task::yield_now().await;
+        }
+
+        let (_send, mut recv) = adopter.await.expect("adopt task join");
+        manager.route_payload(7, Bytes::from_static(b"C")).await;
+
+        use tokio::io::AsyncReadExt;
+        let mut buf = [0u8; 5];
+        recv.read_exact(&mut buf).await.expect("read all");
+        assert_eq!(&buf, b"P1P2C", "raced chunk must follow the parked prefix");
+    }
+
+    /// The effective park cap is the nominal EARLY_ARRIVAL_CAP bounded
+    /// by the receiver's buffer capacity (review 013 — the install
+    /// drains the parked queue with non-blocking sends into the fresh
+    /// receiver, which must always fit). With buffer_cap 4 the fifth
+    /// parked chunk drops at park time (the documented E-04 cap
+    /// behavior, tightened) and the first four arrive on adopt.
+    #[tokio::test]
+    async fn park_cap_bounded_by_receiver_buffer() {
+        let (_client, server) = duplex(1024);
+        let (_reader, writer) = tokio::io::split(server);
+        let (handle, runner) = MuxRunner::new(Box::new(writer));
+        tokio::spawn(async move {
+            let _ = runner.run().await;
+        });
+        let manager = ChannelManager::new(handle, 256, 4, None, ChannelSide::Accept);
+
+        for chunk in b"abcde" {
+            manager
+                .route_payload(7, Bytes::copy_from_slice(&[*chunk]))
+                .await;
+        }
+        assert_eq!(
+            manager.dropped_unknown_chunks(),
+            1,
+            "the fifth parked chunk dropped (park cap = min(64, buffer_cap=4))"
+        );
+
+        let (_send, mut recv) = manager
+            .adopt_channel(7, "alk/tty", None)
+            .await
+            .expect("adopt");
+
+        use tokio::io::AsyncReadExt;
+        let mut buf = [0u8; 4];
+        recv.read_exact(&mut buf)
+            .await
+            .expect("read parked quartet");
+        assert_eq!(&buf, b"abcd", "parked chunks arrive in FIFO order");
+    }
 }