diff --git a/rust/tests/src/mock_arrow_flight.rs b/rust/tests/src/mock_arrow_flight.rs
index 306a0299..eaa02dab 100644
--- a/rust/tests/src/mock_arrow_flight.rs
+++ b/rust/tests/src/mock_arrow_flight.rs
@@ -1,6 +1,6 @@
 //! Mock Arrow Flight server for testing the Arrow Flight SDK functionality.
 
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::pin::Pin;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
@@ -152,6 +152,8 @@ pub enum MockFlightResponse {
 pub struct MockFlightServer {
     /// Responses to inject for each table
     responses: Arc<Mutex<HashMap<String, Vec<MockFlightResponse>>>>,
+    /// Optional scripts consumed once per DoPut, for independent mux lanes/reconnects.
+    connection_scripts: Arc<Mutex<HashMap<String, VecDeque<Vec<MockFlightResponse>>>>>,
     connection_count: Arc<AtomicU64>,
     /// Track the maximum offset received from clients
     max_offset_received: Arc<Mutex<i64>>,
@@ -188,6 +190,7 @@ impl MockFlightServer {
     pub fn new() -> Self {
         Self {
             responses: Arc::new(Mutex::new(HashMap::new())),
+            connection_scripts: Arc::new(Mutex::new(HashMap::new())),
             connection_count: Arc::new(AtomicU64::new(0)),
             max_offset_received: Arc::new(Mutex::new(-1)),
             batch_count: Arc::new(Mutex::new(0)),
@@ -227,6 +230,17 @@ impl MockFlightServer {
         indices.insert(table_name.to_string(), 0);
     }
 
+    pub async fn inject_connection_scripts(
+        &self,
+        table: &str,
+        scripts: Vec<Vec<MockFlightResponse>>,
+    ) {
+        self.connection_scripts
+            .lock()
+            .await
+            .insert(table.into(), scripts.into());
+    }
+
     pub fn connection_count(&self) -> u64 {
         self.connection_count.load(Ordering::Relaxed)
     }
@@ -378,7 +392,18 @@ impl FlightService for MockFlightServer {
         let mut stream = request.into_inner();
         let (tx, rx) = mpsc::channel(100);
 
-        self.connection_count.fetch_add(1, Ordering::Relaxed);
+        let connection_id = self.connection_count.fetch_add(1, Ordering::Relaxed);
+        let connection_script = self
+            .connection_scripts
+            .lock()
+            .await
+            .get_mut(&table_name)
+            .and_then(VecDeque::pop_front);
+        let response_key = if connection_script.is_some() {
+            format!("{table_name}#connection-{connection_id}")
+        } else {
+            table_name.clone()
+        };
         let responses = Arc::clone(&self.responses);
         let max_offset_received = Arc::clone(&self.max_offset_received);
         let batch_count = Arc::clone(&self.batch_count);
@@ -419,9 +444,12 @@ impl FlightService for MockFlightServer {
                 }
             }
 
+            if let Some(script) = connection_script {
+                stream_responses = script;
+            }
             let mut response_index = {
                 let indices = response_indices.lock().await;
-                *indices.get(&table_name).unwrap_or(&0)
+                *indices.get(&response_key).unwrap_or(&0)
             };
 
             let clean_request_eof = loop {
@@ -483,7 +511,7 @@ impl FlightService for MockFlightServer {
                             response_index += 1;
                             {
                                 let mut indices = response_indices.lock().await;
-                                indices.insert(table_name.clone(), response_index);
+                                indices.insert(response_key.clone(), response_index);
                             }
                             if delay_ms > 0 {
                                 delayed_setup_armed.notify_one();
@@ -634,7 +662,7 @@ impl FlightService for MockFlightServer {
                                 // Update response index
                                 {
                                     let mut indices = response_indices.lock().await;
-                                    indices.insert(table_name.clone(), response_index);
+                                    indices.insert(response_key.clone(), response_index);
                                 }
                             }
                         }
@@ -653,7 +681,7 @@ impl FlightService for MockFlightServer {
                             // Save response index before returning so next connection continues from here
                             {
                                 let mut indices = response_indices.lock().await;
-                                indices.insert(table_name.clone(), response_index);
+                                indices.insert(response_key.clone(), response_index);
                             }
                             let _ = tx.send(Err(status.clone())).await;
                             return;
@@ -665,7 +693,7 @@ impl FlightService for MockFlightServer {
                             response_index += 1;
                             {
                                 let mut indices = response_indices.lock().await;
-                                indices.insert(table_name.clone(), response_index);
+                                indices.insert(response_key.clone(), response_index);
                             }
                             let _ = tx.send(Err(status)).await;
                             return;
@@ -679,7 +707,7 @@ impl FlightService for MockFlightServer {
                             response_index += 1;
                             {
                                 let mut indices = response_indices.lock().await;
-                                indices.insert(table_name.clone(), response_index);
+                                indices.insert(response_key.clone(), response_index);
                             }
                             let _ = tx.send(Err(status)).await;
                             return;
@@ -694,7 +722,7 @@ impl FlightService for MockFlightServer {
                             // Save response index before returning so next connection continues from here
                             {
                                 let mut indices = response_indices.lock().await;
-                                indices.insert(table_name.clone(), response_index);
+                                indices.insert(response_key.clone(), response_index);
                             }
                             return;
                         }
@@ -731,7 +759,7 @@ impl FlightService for MockFlightServer {
                             response_index += 1;
                             {
                                 let mut indices = response_indices.lock().await;
-                                indices.insert(table_name.clone(), response_index);
+                                indices.insert(response_key.clone(), response_index);
                             }
                             // Continue processing - the main loop waits for more batches.
                             // During grace period the client won't send new batches,
@@ -865,6 +893,7 @@ async fn start_mock_flight_server_inner(
     let mock_server = MockFlightServer::new();
     let server_clone = MockFlightServer {
         responses: Arc::clone(&mock_server.responses),
+        connection_scripts: Arc::clone(&mock_server.connection_scripts),
         connection_count: Arc::clone(&mock_server.connection_count),
         max_offset_received: Arc::clone(&mock_server.max_offset_received),
         batch_count: Arc::clone(&mock_server.batch_count),
diff --git a/rust/tests/src/multiplexed_arrow_tests.rs b/rust/tests/src/multiplexed_arrow_tests.rs
index aaa78084..9347f40f 100644
--- a/rust/tests/src/multiplexed_arrow_tests.rs
+++ b/rust/tests/src/multiplexed_arrow_tests.rs
@@ -1,14 +1,17 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use arrow_array::{Int64Array, RecordBatch};
 use databricks_zerobus_ingest_sdk::{
     channel_exporter, MessageId, NoTlsConfig, StreamStat, ZerobusError, ZerobusSdk,
 };
+use futures::poll;
+use tonic::Status;
 
-use crate::mock_arrow_flight::start_mock_flight_server;
+use crate::mock_arrow_flight::{start_mock_flight_server, MockFlightResponse};
 use crate::utils::{
     create_test_arrow_schema, create_test_record_batch, record_batch_to_ipc_bytes,
-    TestHeadersProvider,
+    run_with_paused_time_watchdog, TestHeadersProvider,
 };
 
 const TABLE: &str = "catalog.schema.arrow_mux";
@@ -30,6 +33,16 @@ fn batch(ids: &[i64]) -> RecordBatch {
     )
 }
 
+fn ids(batch: &RecordBatch) -> Vec<i64> {
+    batch
+        .column(0)
+        .as_any()
+        .downcast_ref::<Int64Array>()
+        .unwrap()
+        .values()
+        .to_vec()
+}
+
 fn builder(sdk: &ZerobusSdk) -> databricks_zerobus_ingest_sdk::StreamBuilder<'_> {
     sdk.stream_builder()
         .table(TABLE)
@@ -186,3 +199,441 @@ async fn standalone_capacity_wait_rejects_when_close_starts() {
     assert_eq!(retained.len(), 1);
     assert_eq!(retained[0].num_rows(), 1);
 }
+
+#[tokio::test]
+async fn terminal_poison_preserves_healthy_wait_and_flush_and_partial_suffixes() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_connection_scripts(
+            TABLE,
+            vec![
+                vec![
+                    MockFlightResponse::BatchAck {
+                        ack_up_to_offset: 0,
+                        ack_up_to_records: 1,
+                        delay_ms: 0,
+                    },
+                    MockFlightResponse::Error {
+                        status: Status::permission_denied("mux terminal"),
+                        delay_ms: 0,
+                    },
+                ],
+                vec![],
+            ],
+        )
+        .await;
+    let sdk = sdk(endpoint);
+    let mut mux = builder(&sdk)
+        .recovery(false)
+        .multiplexed(2)
+        .build_arrow()
+        .await
+        .unwrap();
+    let mut messages = Vec::new();
+    for data in [
+        batch(&[1, 2, 3]),
+        batch(&[11, 12, 13]),
+        batch(&[4]),
+        batch(&[14]),
+    ] {
+        messages.push(mux.ingest_batch(data).await.unwrap());
+    }
+    let result = mux.flush().await.unwrap_err();
+    assert!(result.to_string().contains("mux terminal"), "{result:?}");
+    assert!(mux.is_closed());
+    let mut healthy = 0;
+    for id in messages {
+        if mux.wait_for_message_id(id).await.is_ok() {
+            healthy += 1;
+        }
+    }
+    assert_eq!(
+        healthy, 2,
+        "the healthy lane's messages remain acknowledged"
+    );
+    let ingest_error = mux.ingest_batch(batch(&[99])).await.unwrap_err();
+    assert_eq!(ingest_error.to_string(), result.to_string());
+    assert_eq!(
+        mux.close().await.unwrap_err().to_string(),
+        result.to_string()
+    );
+    let failed = mux.get_unacked_batches().await.unwrap();
+    let rows: Vec<_> = failed.iter().map(ids).collect();
+    assert!(
+        rows == [vec![2, 3], vec![4]] || rows == [vec![12, 13], vec![14]],
+        "{rows:?}"
+    );
+    assert_eq!(
+        mux.get_unacked_batches()
+            .await
+            .unwrap()
+            .iter()
+            .map(ids)
+            .collect::<Vec<_>>(),
+        rows
+    );
+}
+
+#[tokio::test]
+async fn retryable_lane_failure_replays_without_poisoning_mux() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_connection_scripts(
+            TABLE,
+            vec![
+                vec![
+                    MockFlightResponse::BatchAck {
+                        ack_up_to_offset: 0,
+                        ack_up_to_records: 1,
+                        delay_ms: 0,
+                    },
+                    MockFlightResponse::Error {
+                        status: Status::unavailable("retry me"),
+                        delay_ms: 0,
+                    },
+                ],
+                vec![],
+            ],
+        )
+        .await;
+    let sdk = sdk(endpoint);
+    let mut mux = builder(&sdk)
+        .recovery_backoff_ms(1)
+        .multiplexed(2)
+        .build_arrow()
+        .await
+        .unwrap();
+    for data in [
+        batch(&[1, 2, 3]),
+        batch(&[11, 12, 13]),
+        batch(&[4]),
+        batch(&[14]),
+    ] {
+        mux.ingest_batch(data).await.unwrap();
+    }
+    mux.flush().await.unwrap();
+    assert!(!mux.is_closed());
+    assert_eq!(server.connection_count(), 3);
+    // Eight original rows plus only the failed lane's three unacknowledged rows.
+    assert_eq!(server.get_total_records_received().await, 11);
+    mux.close().await.unwrap();
+    assert!(mux.get_unacked_batches().await.unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn capacity_is_mux_wide_waits_on_selected_lane_and_releases_on_cancellation() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_responses(
+            TABLE,
+            vec![MockFlightResponse::BatchAck {
+                ack_up_to_offset: 0,
+                ack_up_to_records: 1,
+                delay_ms: 10_000,
+            }],
+        )
+        .await;
+    let sdk = sdk(endpoint);
+    // floor(3 / 2) = one pending batch per lane; one budget slot stays unused.
+    let mut mux = builder(&sdk)
+        .max_inflight_batches(3)
+        .multiplexed(2)
+        .build_arrow()
+        .await
+        .unwrap();
+    tokio::time::pause();
+    run_with_paused_time_watchdog(async {
+        mux.ingest_batch(batch(&[0])).await.unwrap();
+        server.delayed_ack_armed().notified().await;
+        mux.ingest_batch(batch(&[1])).await.unwrap();
+        server.delayed_ack_armed().notified().await;
+        // The third batch cannot be queued despite its encoder channel draining.
+        {
+            let blocked = mux.ingest_batch(batch(&[2]));
+            tokio::pin!(blocked);
+            assert!(poll!(&mut blocked).is_pending());
+        }
+        let next = mux.ingest_batch(batch(&[3]));
+        tokio::pin!(next);
+        assert!(poll!(&mut next).is_pending());
+        tokio::time::advance(Duration::from_millis(10_001)).await;
+        let id = next.await.unwrap();
+        assert_eq!((id.stream_index(), id.sub_offset()), (1, 1));
+        mux.flush().await.unwrap();
+    })
+    .await;
+    mux.close().await.unwrap();
+    assert_eq!(
+        server.get_total_records_received().await,
+        3,
+        "cancelled input must never be retained"
+    );
+}
+
+#[tokio::test]
+async fn capacity_timeout_is_nonterminal_and_does_not_admit_input() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_responses(
+            TABLE,
+            vec![MockFlightResponse::BatchAck {
+                ack_up_to_offset: 0,
+                ack_up_to_records: 1,
+                delay_ms: 40_000,
+            }],
+        )
+        .await;
+    let sdk = sdk(endpoint);
+    let mut mux = builder(&sdk)
+        .max_inflight_batches(2)
+        .server_lack_of_ack_timeout_ms(120_000)
+        .multiplexed(2)
+        .build_arrow()
+        .await
+        .unwrap();
+    tokio::time::pause();
+    run_with_paused_time_watchdog(async {
+        mux.ingest_batch(batch(&[0])).await.unwrap();
+        server.delayed_ack_armed().notified().await;
+        mux.ingest_batch(batch(&[1])).await.unwrap();
+        server.delayed_ack_armed().notified().await;
+        let blocked = mux.ingest_batch(batch(&[2]));
+        tokio::pin!(blocked);
+        assert!(poll!(&mut blocked).is_pending());
+        tokio::time::advance(Duration::from_millis(30_001)).await;
+        assert!(matches!(
+            blocked.await,
+            Err(ZerobusError::ConnectionTimeout(_))
+        ));
+        assert!(!mux.is_closed());
+        tokio::time::advance(Duration::from_millis(10_001)).await;
+        mux.flush().await.unwrap();
+    })
+    .await;
+    mux.close().await.unwrap();
+    assert_eq!(server.get_total_records_received().await, 2);
+}
+
+#[tokio::test]
+async fn cancelled_close_keeps_original_deadline_and_unacked_batches() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_responses(TABLE, vec![MockFlightResponse::HoldResponseAfterRequestEof])
+        .await;
+    let sdk = sdk(endpoint);
+    let mut mux = builder(&sdk).multiplexed(2).build_arrow().await.unwrap();
+    mux.ingest_batch(batch(&[0])).await.unwrap();
+    mux.ingest_batch(batch(&[1])).await.unwrap();
+    tokio::time::pause();
+    {
+        let close = mux.close();
+        tokio::pin!(close);
+        assert!(poll!(&mut close).is_pending());
+    }
+    tokio::time::advance(Duration::from_millis(2_001)).await;
+    let result = mux.close().await.unwrap_err();
+    assert!(result.to_string().contains("timed out"), "{result:?}");
+    let failed = mux.get_unacked_batches().await.unwrap();
+    assert_eq!(
+        failed.iter().map(ids).collect::<Vec<_>>(),
+        [vec![0], vec![1]]
+    );
+    assert_eq!(
+        mux.close().await.unwrap_err().to_string(),
+        result.to_string()
+    );
+}
+
+#[tokio::test]
+async fn failed_construction_closes_successfully_opened_arrow_lanes() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_connection_scripts(
+            TABLE,
+            vec![
+                vec![],
+                vec![MockFlightResponse::FailSetup {
+                    status: Status::permission_denied("construction failure"),
+                }],
+            ],
+        )
+        .await;
+    let sdk = sdk(endpoint);
+    let error = builder(&sdk)
+        .recovery(false)
+        .multiplexed(2)
+        .build_arrow()
+        .await
+        .err()
+        .unwrap();
+    assert!(
+        error.to_string().contains("construction failure"),
+        "{error:?}"
+    );
+    assert_eq!(server.connection_count(), 2);
+    assert_eq!(server.get_request_half_close_count(), 1);
+}
+
+#[tokio::test]
+async fn poison_wakes_capacity_waiters_on_both_failed_and_healthy_lanes() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_connection_scripts(
+            TABLE,
+            vec![
+                vec![MockFlightResponse::Error {
+                    status: Status::permission_denied("poison waiters"),
+                    delay_ms: 0,
+                }],
+                vec![MockFlightResponse::BatchAck {
+                    ack_up_to_offset: 0,
+                    ack_up_to_records: 1,
+                    delay_ms: 10_000,
+                }],
+            ],
+        )
+        .await;
+    let sdk = sdk(endpoint);
+    let mut mux = builder(&sdk)
+        .recovery(false)
+        .max_inflight_batches(2)
+        .multiplexed(2)
+        .build_arrow()
+        .await
+        .unwrap();
+    tokio::time::pause();
+    run_with_paused_time_watchdog(async {
+        let a = mux.ingest_batch(batch(&[0])).await.unwrap();
+        let b = mux.ingest_batch(batch(&[1])).await.unwrap();
+        let blocked_a = mux.ingest_batch(batch(&[2]));
+        let blocked_b = mux.ingest_batch(batch(&[3]));
+        tokio::pin!(blocked_a, blocked_b);
+        assert!(poll!(&mut blocked_a).is_pending());
+        assert!(poll!(&mut blocked_b).is_pending());
+        let failure = futures::future::select(
+            Box::pin(mux.wait_for_message_id(a)),
+            Box::pin(mux.wait_for_message_id(b)),
+        )
+        .await;
+        let error = match failure {
+            futures::future::Either::Left((result, _))
+            | futures::future::Either::Right((result, _)) => result.unwrap_err(),
+        };
+        assert!(error.to_string().contains("poison waiters"));
+        for result in [blocked_a.await, blocked_b.await] {
+            assert_eq!(result.unwrap_err().to_string(), error.to_string());
+        }
+        server.delayed_ack_armed().notified().await;
+        tokio::time::advance(Duration::from_millis(10_001)).await;
+        assert_eq!(
+            mux.flush().await.unwrap_err().to_string(),
+            error.to_string()
+        );
+    })
+    .await;
+    assert!(mux.close().await.is_err());
+    assert_eq!(server.get_total_records_received().await, 2);
+    assert_eq!(mux.get_unacked_batches().await.unwrap().len(), 1);
+}
+
+struct GatedHeaders {
+    calls: std::sync::atomic::AtomicUsize,
+    allowed: usize,
+    reached: tokio::sync::Notify,
+}
+
+#[async_trait::async_trait]
+impl databricks_zerobus_ingest_sdk::HeadersProvider for GatedHeaders {
+    async fn get_headers(
+        &self,
+    ) -> databricks_zerobus_ingest_sdk::ZerobusResult<std::collections::HashMap<&'static str, String>>
+    {
+        if self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) >= self.allowed {
+            self.reached.notify_one();
+            std::future::pending::<()>().await;
+        }
+        Ok(std::collections::HashMap::new())
+    }
+}
+
+#[tokio::test]
+async fn cancelled_construction_releases_open_lane_and_pending_headers() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    let sdk = sdk(endpoint);
+    let provider = Arc::new(GatedHeaders {
+        calls: std::sync::atomic::AtomicUsize::new(0),
+        allowed: 1,
+        reached: tokio::sync::Notify::new(),
+    });
+    let build = {
+        let provider = provider.clone();
+        tokio::spawn(async move {
+            builder(&sdk)
+                .headers_provider(provider)
+                .multiplexed(2)
+                .build_arrow()
+                .await
+        })
+    };
+    provider.reached.notified().await;
+    run_with_paused_time_watchdog(async {
+        while server.connection_count() == 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await;
+    build.abort();
+    assert!(matches!(build.await, Err(e) if e.is_cancelled()));
+    run_with_paused_time_watchdog(async {
+        while Arc::strong_count(&provider) != 1 {
+            tokio::task::yield_now().await;
+        }
+        while server.get_request_half_close_count() + server.get_request_reset_count() == 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await;
+}
+
+#[tokio::test]
+async fn close_interrupts_lane_recovery_and_retains_its_trigger_and_batches() {
+    let (server, endpoint) = start_mock_flight_server().await.unwrap();
+    server
+        .inject_connection_scripts(
+            TABLE,
+            vec![
+                vec![MockFlightResponse::Error {
+                    status: Status::unavailable("interrupted recovery"),
+                    delay_ms: 0,
+                }],
+                vec![],
+            ],
+        )
+        .await;
+    let sdk = sdk(endpoint);
+    let provider = Arc::new(GatedHeaders {
+        calls: std::sync::atomic::AtomicUsize::new(0),
+        allowed: 2,
+        reached: tokio::sync::Notify::new(),
+    });
+    let mut mux = builder(&sdk)
+        .headers_provider(provider.clone())
+        .recovery_backoff_ms(1)
+        .multiplexed(2)
+        .build_arrow()
+        .await
+        .unwrap();
+    mux.ingest_batch(batch(&[0])).await.unwrap();
+    mux.ingest_batch(batch(&[1])).await.unwrap();
+    provider.reached.notified().await;
+    let result = run_with_paused_time_watchdog(mux.close())
+        .await
+        .unwrap_err();
+    assert!(
+        result.to_string().contains("interrupted recovery"),
+        "{result:?}"
+    );
+    let failed = mux.get_unacked_batches().await.unwrap();
+    assert_eq!(failed.len(), 1);
+    assert_eq!(failed[0].num_rows(), 1);
+}