diff --git a/Cargo.lock b/Cargo.lock
index 64a400e..7891f9a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1530,7 +1530,9 @@ dependencies = [
  "futures-timer",
  "ironposh-client-core",
  "ironposh-psrp",
+ "ironposh-test-support",
  "tracing",
+ "tracing-subscriber",
  "uuid",
  "web-sys",
  "web-time",
@@ -1680,7 +1682,7 @@ dependencies = [
 
 [[package]]
 name = "ironposh-web"
-version = "0.5.1"
+version = "0.6.0"
 dependencies = [
  "anyhow",
  "console_error_panic_hook",
diff --git a/crates/ironposh-async/Cargo.toml b/crates/ironposh-async/Cargo.toml
index 7017330..53afad2 100644
--- a/crates/ironposh-async/Cargo.toml
+++ b/crates/ironposh-async/Cargo.toml
@@ -13,6 +13,7 @@ futures = { version = "0.3.31", default-features = false, features = [
 ] }
 futures-timer = { version = "3.0.3", default-features = false }
 ironposh-client-core = { version = "0.1.0", path = "../ironposh-client-core" }
+ironposh-psrp = { version = "0.1.0", path = "../ironposh-psrp" }
 tracing = "0.1"
 uuid = "1.0"
 
@@ -25,6 +26,9 @@ web-sys = { version = "0.3.81", features = ["console"], optional = true }
 base64 = "0.22.1"
 ironposh-client-core = { path = "../ironposh-client-core", features = ["test-helpers"] }
 ironposh-psrp = { path = "../ironposh-psrp" }
+ironposh-test-support = { path = "../ironposh-test-support" }
+futures = { version = "0.3.31", features = ["executor"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [features]
 default = []
diff --git a/crates/ironposh-async/examples/serial_latency_bench.rs b/crates/ironposh-async/examples/serial_latency_bench.rs
new file mode 100644
index 0000000..ee1aff4
--- /dev/null
+++ b/crates/ironposh-async/examples/serial_latency_bench.rs
@@ -0,0 +1,647 @@
+//! Deterministic latency bench for the serial (single-connection) session loop.
+//!
+//! Drives the real `start_serial_session_loop` (via the public
+//! `open_task_serial` API) against an in-process fake WinRM server that honors
+//! the `OperationTimeout` of each Receive, so the bench measures exactly the
+//! scheduling latency the loop adds on top of the wire:
+//!
+//! - `output_latency_ms` — server-has-data → `PipelineOutput` event delivered
+//! - `input_latency_ms`  — Invoke issued → Command request reaches the server
+//! - `kill_signal_ms`    — Kill issued → Signal request reaches the server
+//! - request counts      — polling efficiency (receives / timeouts)
+//!
+//! Run: `cargo run --release -p ironposh-async --example serial_latency_bench`
+//! One process run = one bench run; repeat N times per the bench doc.
+
+#![allow(clippy::significant_drop_tightening)] // bench: lock scopes are already minimal
+#![allow(
+    clippy::items_after_statements,
+    clippy::large_futures,
+    clippy::large_stack_frames
+)]
+
+use std::collections::{HashMap, VecDeque};
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+use futures::{FutureExt, StreamExt, join};
+use futures_timer::Delay;
+use ironposh_async::{HttpClient, RemoteAsyncPowershellClient};
+use ironposh_client_core::connector::active_session::UserEvent;
+use ironposh_client_core::connector::connection_pool::TrySend;
+use ironposh_client_core::connector::http::HttpResponseTargeted;
+use ironposh_psrp::ps_value::PsObjectWithType;
+use ironposh_psrp::{
+    ApplicationPrivateData, PipelineOutput, PipelineStateMessage, PsPrimitiveValue, PsValue,
+    RunspacePoolStateMessage, RunspacePoolStateValue, SessionCapability,
+};
+use ironposh_test_support::fake_server;
+use uuid::Uuid;
+
+// ── Scenario scripting ──────────────────────────────────────────────────────
+
+/// Timed output plan for one pipeline, offsets relative to Command arrival.
+#[derive(Clone)]
+struct PipelineScript {
+    /// (offset, payload) — payload strings must be globally unique.
+    outputs: Vec<(Duration, String)>,
+    /// When the pipeline reports Completed. `None` = runs until killed.
+    complete_at: Option<Duration>,
+}
+
+struct CommandRun {
+    started_at: Instant,
+    script: PipelineScript,
+    delivered: usize,
+    completed: bool,
+}
+
+#[derive(Default)]
+struct Records {
+    /// payload → instant the server had it ready
+    ready_at: HashMap<String, Instant>,
+    command_arrived_at: Vec<Instant>,
+    signal_arrived_at: Vec<Instant>,
+    receive_count: u64,
+    timeout_count: u64,
+    request_count: u64,
+}
+
+struct ServerState {
+    rpid: Option<Uuid>,
+    handshake_receive_done: bool,
+    scripts: VecDeque<PipelineScript>,
+    runs: HashMap<Uuid, CommandRun>,
+    object_id: u64,
+    records: Records,
+}
+
+struct FakeWinRmServer {
+    state: Mutex<ServerState>,
+}
+
+impl FakeWinRmServer {
+    fn new(scripts: Vec<PipelineScript>) -> Self {
+        Self {
+            state: Mutex::new(ServerState {
+                rpid: None,
+                handshake_receive_done: false,
+                scripts: scripts.into(),
+                runs: HashMap::new(),
+                object_id: 100,
+                records: Records::default(),
+            }),
+        }
+    }
+}
+
+// ── Request parsing helpers (plaintext XML — Basic auth + HttpInsecure) ────
+
+fn extract_between<'a>(haystack: &'a str, prefix: &str, suffix: char) -> Option<&'a str> {
+    let start = haystack.find(prefix)? + prefix.len();
+    let rest = &haystack[start..];
+    let end = rest.find(suffix)?;
+    Some(&rest[..end])
+}
+
+fn extract_action(body: &str) -> &'static str {
+    for (needle, name) in [
+        ("transfer/Create", "create"),
+        ("shell/Command", "command"),
+        ("shell/Receive", "receive"),
+        ("shell/Signal", "signal"),
+        ("transfer/Delete", "delete"),
+    ] {
+        if body.contains(needle) {
+            return name;
+        }
+    }
+    "other"
+}
+
+fn extract_operation_timeout(body: &str) -> Duration {
+    // Anchor on the element content ("...OperationTimeout>PT0.250S<...") —
+    // a bare "PT" search would match WSMAN_CMDSHELL_OPTION_KEEPALIVE first.
+    extract_between(body, "OperationTimeout>PT", 'S')
+        .and_then(|s| s.parse::<f64>().ok())
+        .map_or(Duration::from_millis(250), Duration::from_secs_f64)
+}
+
+fn extract_command_id(body: &str) -> Option<Uuid> {
+    extract_between(body, "CommandId=\"", '"').and_then(|s| s.parse().ok())
+}
+
+fn extract_message_id(body: &str) -> Option<String> {
+    let inner = extract_between(body, "<a:MessageID>", '<')?;
+    Some(inner.trim().to_owned())
+}
+
+// ── Fake server behavior ────────────────────────────────────────────────────
+
+/// Newtype so `HttpClient` can be implemented for a shared server (orphan rule).
+#[derive(Clone)]
+struct SharedServer(std::sync::Arc<FakeWinRmServer>);
+
+impl HttpClient for SharedServer {
+    fn send_request(
+        &self,
+        try_send: TrySend,
+    ) -> impl Future<Output = anyhow::Result<HttpResponseTargeted>> {
+        let server = std::sync::Arc::clone(&self.0);
+        async move {
+            let (request, conn_id) = fake_server::expect_just_send(try_send);
+            let body = request
+                .body
+                .as_ref()
+                .and_then(|b| b.as_str().ok())
+                .unwrap_or_default()
+                .to_owned();
+
+            {
+                let mut st = server.state.lock().unwrap();
+                st.records.request_count += 1;
+            }
+
+            let action = extract_action(&body);
+            let xml = match action {
+                "create" => {
+                    let rpid = fake_server::extract_shell_id(&body);
+                    server.state.lock().unwrap().rpid = Some(rpid);
+                    include_str!("../../ironposh-client-core/tests/resources/resource_created.xml")
+                        .to_owned()
+                }
+                "command" => {
+                    let now = Instant::now();
+                    let command_id =
+                        extract_command_id(&body).expect("Command request must carry a CommandId");
+                    let mut st = server.state.lock().unwrap();
+                    let script = st
+                        .scripts
+                        .pop_front()
+                        .expect("more Commands than scripted pipelines");
+                    for (offset, payload) in &script.outputs {
+                        st.records.ready_at.insert(payload.clone(), now + *offset);
+                    }
+                    st.records.command_arrived_at.push(now);
+                    st.runs.insert(
+                        command_id,
+                        CommandRun {
+                            started_at: now,
+                            script,
+                            delivered: 0,
+                            completed: false,
+                        },
+                    );
+                    fake_server::command_response_xml(command_id)
+                }
+                "signal" => {
+                    let now = Instant::now();
+                    let command_id = extract_command_id(&body);
+                    let relates_to =
+                        extract_message_id(&body).expect("Signal request must carry a MessageID");
+                    let mut st = server.state.lock().unwrap();
+                    st.records.signal_arrived_at.push(now);
+                    // Killing a live pipeline makes it report Completed on the
+                    // next Receive (mock simplification of Stopped).
+                    if let Some(run) = command_id.and_then(|id| st.runs.get_mut(&id))
+                        && run.script.complete_at.is_none()
+                    {
+                        run.script.complete_at = Some(now - run.started_at);
+                    }
+                    fake_server::signal_response_xml(&relates_to)
+                }
+                "receive" => server.handle_receive(&body).await,
+                _ => fake_server::timeout_fault_xml(),
+            };
+
+            Ok(fake_server::xml_response(conn_id, xml))
+        }
+    }
+}
+
+impl FakeWinRmServer {
+    /// Serve a Receive: respond the moment scripted data is due, otherwise
+    /// hold the request until its own OperationTimeout and return a
+    /// `w:TimedOut` fault — exactly like a real WSMan server.
+    async fn handle_receive(&self, body: &str) -> String {
+        let op_timeout = extract_operation_timeout(body);
+        if std::env::var("BENCH_DEBUG").is_ok() {
+            eprintln!(
+                "recv hold={}ms cmd={:?}",
+                op_timeout.as_millis(),
+                extract_command_id(body)
+            );
+        }
+        let deadline = Instant::now() + op_timeout;
+        let command_id = extract_command_id(body);
+
+        {
+            let mut st = self.state.lock().unwrap();
+            st.records.receive_count += 1;
+
+            // Handshake receive: runspace-pool stream before the pool opened.
+            if command_id.is_none() && !st.handshake_receive_done {
+                st.handshake_receive_done = true;
+                let rpid = st.rpid.expect("Create must precede Receive");
+                let session_capability = SessionCapability {
+                    protocol_version: "2.3".to_owned(),
+                    ps_version: "2.0".to_owned(),
+                    serialization_version: "1.1.0.1".to_owned(),
+                    time_zone: None,
+                };
+                let private_data = ApplicationPrivateData::new();
+                let opened = RunspacePoolStateMessage::builder()
+                    .runspace_state(RunspacePoolStateValue::Opened)
+                    .build();
+                return fake_server::receive_response_xml(
+                    rpid,
+                    &[&session_capability, &private_data, &opened],
+                );
+            }
+        }
+
+        let Some(command_id) = command_id else {
+            // Runspace-pool long-poll with nothing to say: hold, then time out.
+            Delay::new(op_timeout).await;
+            let mut st = self.state.lock().unwrap();
+            st.records.timeout_count += 1;
+            return fake_server::timeout_fault_xml();
+        };
+
+        loop {
+            let now = Instant::now();
+            enum Next {
+                Respond(String),
+                WaitUntil(Instant),
+            }
+
+            let next = {
+                let mut st = self.state.lock().unwrap();
+                let rpid = st.rpid.expect("Create must precede Receive");
+                let object_id = st.object_id;
+                let run = st.runs.get_mut(&command_id).expect("unknown CommandId");
+                let elapsed = now - run.started_at;
+
+                let due_outputs: Vec<String> = run.script.outputs[run.delivered..]
+                    .iter()
+                    .take_while(|(offset, _)| *offset <= elapsed)
+                    .map(|(_, payload)| payload.clone())
+                    .collect();
+                let complete_due = !run.completed
+                    && run
+                        .script
+                        .complete_at
+                        .is_some_and(|offset| offset <= elapsed);
+
+                if !due_outputs.is_empty() || complete_due {
+                    run.delivered += due_outputs.len();
+                    run.completed |= complete_due;
+
+                    let outputs: Vec<PipelineOutput> = due_outputs
+                        .iter()
+                        .map(|payload| PipelineOutput {
+                            data: PsValue::Primitive(PsPrimitiveValue::Str(payload.clone())),
+                        })
+                        .collect();
+                    let state_msg = PipelineStateMessage::completed();
+                    let mut messages: Vec<&dyn PsObjectWithType> =
+                        outputs.iter().map(|o| o as &dyn PsObjectWithType).collect();
+                    if complete_due {
+                        messages.push(&state_msg);
+                    }
+
+                    let xml = fake_server::pipeline_receive_response_xml(
+                        rpid,
+                        command_id,
+                        &messages,
+                        complete_due,
+                        object_id,
+                    );
+                    st.object_id += messages.len() as u64;
+                    Next::Respond(xml)
+                } else {
+                    // Nothing due: wake at the earliest of next scripted event
+                    // or this request's OperationTimeout.
+                    let next_event = run.script.outputs[run.delivered..]
+                        .first()
+                        .map(|(offset, _)| run.started_at + *offset);
+                    let next_complete = if run.completed {
+                        None
+                    } else {
+                        run.script.complete_at.map(|offset| run.started_at + offset)
+                    };
+                    let wake_at = [next_event, next_complete]
+                        .into_iter()
+                        .flatten()
+                        .min()
+                        .map_or(deadline, |t| t.min(deadline));
+                    Next::WaitUntil(wake_at)
+                }
+            };
+
+            match next {
+                Next::Respond(xml) => return xml,
+                Next::WaitUntil(wake_at) => {
+                    if wake_at >= deadline {
+                        let sleep = deadline.saturating_duration_since(now);
+                        Delay::new(sleep).await;
+                        let mut st = self.state.lock().unwrap();
+                        st.records.timeout_count += 1;
+                        return fake_server::timeout_fault_xml();
+                    }
+                    Delay::new(wake_at.saturating_duration_since(now)).await;
+                }
+            }
+        }
+    }
+}
+
+// ── Metrics ─────────────────────────────────────────────────────────────────
+
+#[derive(Default)]
+struct Metric {
+    samples_ms: Vec<f64>,
+}
+
+impl Metric {
+    fn push(&mut self, d: Duration) {
+        self.samples_ms.push(d.as_secs_f64() * 1000.0);
+    }
+
+    fn summary(&self) -> String {
+        if self.samples_ms.is_empty() {
+            return "n=0".to_owned();
+        }
+        let mut sorted = self.samples_ms.clone();
+        sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
+        let median = sorted[sorted.len() / 2];
+        let max = sorted[sorted.len() - 1];
+        let min = sorted[0];
+        format!(
+            "n={} min={min:.0}ms median={median:.0}ms max={max:.0}ms",
+            sorted.len()
+        )
+    }
+
+    fn json_fields(&self) -> String {
+        if self.samples_ms.is_empty() {
+            return r#""n":0"#.to_owned();
+        }
+        let mut sorted = self.samples_ms.clone();
+        sorted.sort_by(|a, b| a.partial_cmp(b).unwrap());
+        format!(
+            r#""n":{},"min_ms":{:.1},"median_ms":{:.1},"max_ms":{:.1}"#,
+            sorted.len(),
+            sorted[0],
+            sorted[sorted.len() / 2],
+            sorted[sorted.len() - 1]
+        )
+    }
+}
+
+// ── Scenario driver ─────────────────────────────────────────────────────────
+
+struct ScenarioResult {
+    name: &'static str,
+    output_latency: Metric,
+    input_latency: Metric,
+    kill_signal: Metric,
+    total_requests: u64,
+    receive_timeouts: u64,
+    wall_ms: u64,
+}
+
+fn serial_config() -> ironposh_client_core::connector::WinRmConfig {
+    let mut config = fake_server::test_config();
+    // Match production serial mode (web + tokio client default).
+    config.operation_timeout_secs = Some(0.25);
+    config
+}
+
+/// Drive one scenario: fresh server, fresh session, run `actions`, collect metrics.
+async fn run_scenario(
+    name: &'static str,
+    scripts: Vec<PipelineScript>,
+    kill_after: Option<Duration>,
+) -> ScenarioResult {
+    let server = std::sync::Arc::new(FakeWinRmServer::new(scripts.clone()));
+    let (client, host_io, mut session_events, task) = RemoteAsyncPowershellClient::open_task_serial(
+        serial_config(),
+        SharedServer(server.clone()),
+    );
+
+    let started = Instant::now();
+    let server_for_driver = server.clone();
+
+    let driver = async move {
+        let mut client = client;
+        let mut output_latency = Metric::default();
+        let mut input_latency = Metric::default();
+        let mut kill_signal = Metric::default();
+
+        for _script in &scripts {
+            let invoked_at = Instant::now();
+            let mut events = client
+                .send_script_raw("bench".to_owned())
+                .await
+                .expect("invoke pipeline");
+
+            let mut handle = None;
+            let mut kill_sent_at = None;
+
+            loop {
+                // Arm the kill timer only while the target pipeline is live.
+                let event = if let (Some(kill_after), true, None) =
+                    (kill_after, handle.is_some(), kill_sent_at)
+                {
+                    let due = invoked_at + kill_after;
+                    let now = Instant::now();
+                    if due > now {
+                        futures::select! {
+                            ev = events.next() => ev,
+                            () = Delay::new(due - now).fuse() => {
+                                kill_sent_at = Some(Instant::now());
+                                let h = handle.take().expect("handle checked above");
+                                client.kill_pipeline(h).await.expect("kill pipeline");
+                                continue;
+                            }
+                        }
+                    } else {
+                        kill_sent_at = Some(Instant::now());
+                        let h = handle.take().expect("handle checked above");
+                        client.kill_pipeline(h).await.expect("kill pipeline");
+                        continue;
+                    }
+                } else {
+                    events.next().await
+                };
+
+                let Some(event) = event else { break };
+                match event {
+                    UserEvent::PipelineCreated { pipeline } => handle = Some(pipeline),
+                    UserEvent::PipelineOutput { output, .. } => {
+                        let payload = output
+                            .assume_primitive_string()
+                            .expect("bench outputs are strings")
+                            .clone();
+                        let ready_at = {
+                            let st = server_for_driver.state.lock().unwrap();
+                            st.records.ready_at[&payload]
+                        };
+                        output_latency.push(ready_at.elapsed());
+                    }
+                    UserEvent::PipelineFinished { .. } => break,
+                    _ => {}
+                }
+            }
+
+            // Input latency: Invoke → Command arrival, paired in order.
+            {
+                let st = server_for_driver.state.lock().unwrap();
+                if let Some(arrived) = st.records.command_arrived_at.last() {
+                    input_latency.push(*arrived - invoked_at);
+                }
+                if let (Some(sent), Some(arrived)) =
+                    (kill_sent_at, st.records.signal_arrived_at.last())
+                {
+                    kill_signal.push(*arrived - sent);
+                }
+            }
+        }
+
+        drop(client);
+        drop(host_io);
+        (output_latency, input_latency, kill_signal)
+    };
+
+    let events_drain = async move { while session_events.next().await.is_some() {} };
+
+    let (task_result, (output_latency, input_latency, kill_signal), ()) =
+        join!(task, driver, events_drain);
+    if let Err(e) = task_result {
+        // Channel-closed on shutdown is the expected exit path; anything else
+        // means the bench itself is broken.
+        let msg = format!("{e:#}");
+        assert!(
+            msg.contains("channel closed") || msg.contains("channel disconnected"),
+            "session task failed: {msg}"
+        );
+    }
+
+    let st = server.state.lock().unwrap();
+    ScenarioResult {
+        name,
+        output_latency,
+        input_latency,
+        kill_signal,
+        total_requests: st.records.request_count,
+        receive_timeouts: st.records.timeout_count,
+        wall_ms: started.elapsed().as_millis() as u64,
+    }
+}
+
+fn secs(s: f64) -> Duration {
+    Duration::from_secs_f64(s)
+}
+
+fn main() {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("off")),
+        )
+        .init();
+
+    let results = futures::executor::block_on(async {
+        let mut results = Vec::new();
+
+        // 1. REPL-style roundtrips: pipeline completes immediately.
+        results.push(
+            run_scenario(
+                "prompt_roundtrip",
+                (0..5)
+                    .map(|i| PipelineScript {
+                        outputs: vec![(secs(0.0), format!("pong-{i}"))],
+                        complete_at: Some(secs(0.0)),
+                    })
+                    .collect(),
+                None,
+            )
+            .await,
+        );
+
+        // 2. Steady drip: one line every 2s for 16s.
+        results.push(
+            run_scenario(
+                "drip_2s",
+                vec![PipelineScript {
+                    outputs: (1..=8)
+                        .map(|i| (secs(2.0 * i as f64), format!("drip-{i}")))
+                        .collect(),
+                    complete_at: Some(secs(16.5)),
+                }],
+                None,
+            )
+            .await,
+        );
+
+        // 3. Quiet then burst: silence long enough to reach max backoff, then
+        // one line. Staggered quiet durations sample different phases of the
+        // client-side backoff cycle (the worst case is a full backoff period).
+        results.push(
+            run_scenario(
+                "quiet_burst_9_to_15s",
+                [9.0, 10.5, 12.0, 13.5, 15.0]
+                    .iter()
+                    .enumerate()
+                    .map(|(i, quiet)| PipelineScript {
+                        outputs: vec![(secs(*quiet), format!("burst-{i}"))],
+                        complete_at: Some(secs(*quiet)),
+                    })
+                    .collect(),
+                None,
+            )
+            .await,
+        );
+
+        // 4. Ctrl+C during a quiet pipeline at +4s.
+        results.push(
+            run_scenario(
+                "ctrl_c_at_4s",
+                vec![PipelineScript {
+                    outputs: vec![],
+                    complete_at: None,
+                }],
+                Some(secs(4.0)),
+            )
+            .await,
+        );
+
+        results
+    });
+
+    println!("\n=== serial_latency_bench results ===");
+    for r in &results {
+        println!("\nscenario: {}", r.name);
+        println!("  output_latency: {}", r.output_latency.summary());
+        println!("  input_latency:  {}", r.input_latency.summary());
+        println!("  kill_signal:    {}", r.kill_signal.summary());
+        println!(
+            "  requests={} receive_timeouts={} wall_ms={}",
+            r.total_requests, r.receive_timeouts, r.wall_ms
+        );
+    }
+    println!("\n=== json ===");
+    for r in &results {
+        println!(
+            r#"{{"scenario":"{}","output":{{{}}},"input":{{{}}},"kill":{{{}}},"requests":{},"timeouts":{},"wall_ms":{}}}"#,
+            r.name,
+            r.output_latency.json_fields(),
+            r.input_latency.json_fields(),
+            r.kill_signal.json_fields(),
+            r.total_requests,
+            r.receive_timeouts,
+            r.wall_ms
+        );
+    }
+}
diff --git a/crates/ironposh-async/src/session.rs b/crates/ironposh-async/src/session.rs
index a47a7dd..36f0063 100644
--- a/crates/ironposh-async/src/session.rs
+++ b/crates/ironposh-async/src/session.rs
@@ -29,13 +29,13 @@ fn resolve_deferred_sends(
             then_receive_streams,
         } => {
             let recv = active_session
-                .fire_receive(then_receive_streams)
+                .fire_receive(then_receive_streams, None)
                 .context("Failed to build receive after send-then-receive")?;
             Ok(ActiveSessionOutput::SendBack(vec![send_request, recv]))
         }
         ActiveSessionOutput::PendingReceive { desired_streams } => {
             let recv = active_session
-                .fire_receive(desired_streams)
+                .fire_receive(desired_streams, None)
                 .context("Failed to build receive from PendingReceive")?;
             Ok(ActiveSessionOutput::SendBack(vec![recv]))
         }
diff --git a/crates/ironposh-async/src/session_serial/core.rs b/crates/ironposh-async/src/session_serial/core.rs
index 643e2c8..74ad856 100644
--- a/crates/ironposh-async/src/session_serial/core.rs
+++ b/crates/ironposh-async/src/session_serial/core.rs
@@ -10,12 +10,15 @@ use std::collections::VecDeque;
 
 use anyhow::Context;
 use ironposh_client_core::PwshCoreError;
-use ironposh_client_core::connector::active_session::{ActiveSession, UserEvent};
+use ironposh_client_core::connector::active_session::{
+    ActiveSession, TransportErrorDisposition, UserEvent,
+};
 use ironposh_client_core::connector::http::HttpResponseTargeted;
 use ironposh_client_core::connector::{
-    ActiveSessionOutput, UserOperation, connection_pool::TrySend,
+    ActiveSessionOutput, UserOperation,
+    connection_pool::{ConnectionId, TrySend},
 };
-use ironposh_client_core::host::HostCall;
+use ironposh_client_core::host::{HostCall, HostCallScope};
 use ironposh_client_core::runspace_pool::DesiredStream;
 use tracing::{debug, error, info, trace, warn};
 use uuid::Uuid;
@@ -70,12 +73,30 @@ impl Queues {
     }
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 enum HostCallState {
     Idle,
-    Waiting { call_id: i64 },
+    Waiting {
+        call_id: i64,
+        scope: HostCallScope,
+        /// End of the Receive-gating grace window (ms since the loop epoch).
+        /// The call stays answerable past this — only the gating stops.
+        deadline_ms: u64,
+    },
 }
 
+/// How long Receive promotion is held back for a dispatched host call. Most
+/// answers (auto-replied metadata calls) arrive within milliseconds; a human at
+/// Read-Host can take minutes, and holding Receives that long would wedge the
+/// session — so past this window we poll again while still awaiting the answer.
+const HOST_CALL_GATE_GRACE_MS: u64 = 5_000;
+
+/// Consecutive transport failures on in-flight Receives tolerated before the
+/// serial loop gives up. A long-poll Receive is idempotent (re-issuable), so a
+/// transient gateway/WS drop should not kill the session — but a dead link must
+/// still terminate it.
+const MAX_CONSECUTIVE_RECEIVE_TRANSPORT_FAILURES: u32 = 3;
+
 // ── Backend trait ─────────────────────────────────────────────────────────
 
 /// Abstraction over [`ActiveSession`] so that [`SessionCore`] can be tested
@@ -91,7 +112,15 @@ pub(super) trait SessionBackend {
         resp: HttpResponseTargeted,
     ) -> Result<Vec<ActiveSessionOutput>, PwshCoreError>;
 
-    fn fire_receive(&mut self, streams: Vec<DesiredStream>) -> Result<TrySend, PwshCoreError>;
+    fn fire_receive(
+        &mut self,
+        streams: Vec<DesiredStream>,
+        hold_secs: Option<f64>,
+    ) -> Result<TrySend, PwshCoreError>;
+
+    fn handle_transport_error(&mut self, conn_id: ConnectionId) -> TransportErrorDisposition;
+
+    fn active_desired_streams(&self) -> Vec<DesiredStream>;
 }
 
 impl SessionBackend for ActiveSession {
@@ -109,8 +138,20 @@ impl SessionBackend for ActiveSession {
         Self::accept_server_response(self, resp)
     }
 
-    fn fire_receive(&mut self, streams: Vec<DesiredStream>) -> Result<TrySend, PwshCoreError> {
-        Self::fire_receive(self, streams)
+    fn fire_receive(
+        &mut self,
+        streams: Vec<DesiredStream>,
+        hold_secs: Option<f64>,
+    ) -> Result<TrySend, PwshCoreError> {
+        Self::fire_receive(self, streams, hold_secs)
+    }
+
+    fn handle_transport_error(&mut self, conn_id: ConnectionId) -> TransportErrorDisposition {
+        Self::handle_transport_error(self, conn_id)
+    }
+
+    fn active_desired_streams(&self) -> Vec<DesiredStream> {
+        Self::active_desired_streams(self)
     }
 }
 
@@ -151,6 +192,7 @@ pub(super) struct SessionCore<S: SessionBackend = ActiveSession> {
     in_flight_receive_target: Option<TargetId>,
     queues: Queues,
     host_call_state: HostCallState,
+    consecutive_receive_transport_failures: u32,
 }
 
 impl SessionCore {
@@ -169,6 +211,7 @@ impl<S: SessionBackend> SessionCore<S> {
             in_flight_receive_target: None,
             queues: Queues::new(first_receive),
             host_call_state: HostCallState::Idle,
+            consecutive_receive_transport_failures: 0,
         }
     }
 
@@ -243,10 +286,13 @@ impl<S: SessionBackend> SessionCore<S> {
             return Ok(Some(req));
         }
 
-        // Only build Receives when all HostCalls are resolved. The server blocks
-        // Receives while waiting for our HostCall response, so sending one early
-        // would always time out (adding an unnecessary OperationTimeout delay).
-        if self.is_host_call_active() || !self.queues.host_calls.is_empty() {
+        // Prefer to hold Receives while a host call awaits its answer — the
+        // server blocks them anyway, so polling early just burns round-trips.
+        // But the answer can take arbitrarily long (a human at Read-Host) or
+        // never come (dropped consumer promise), so after a grace period we
+        // resume polling; the session must stay live either way.
+        if let Some(gate_until_ms) = self.receive_gate_deadline(now_ms) {
+            self.next_wakeup_at_ms = Some(gate_until_ms);
             return Ok(None);
         }
 
@@ -296,9 +342,10 @@ impl<S: SessionBackend> SessionCore<S> {
                 speculative_remaining = self.queues.speculative_streams.len(),
                 "promoting demanded stream to Receive"
             );
+            let hold_secs = self.receive_hold_secs(target, now_ms);
             let receive = self
                 .active_session
-                .fire_receive(vec![stream])
+                .fire_receive(vec![stream], Some(hold_secs))
                 .context("Failed to build Receive from demanded stream")?;
             self.in_flight_receive_target = Some(target);
             return Ok(Some(receive));
@@ -335,15 +382,17 @@ impl<S: SessionBackend> SessionCore<S> {
                 "DIAG promote: Receive for pipeline stream ({} speculative remaining)",
                 self.queues.speculative_streams.len()
             );
+            let hold_secs = self.receive_hold_secs(target, now_ms);
             trace!(
                 target: "serial",
                 ?stream,
+                hold_secs,
                 speculative_remaining = self.queues.speculative_streams.len(),
                 "promoting pipeline stream to Receive"
             );
             let receive = self
                 .active_session
-                .fire_receive(vec![stream])
+                .fire_receive(vec![stream], Some(hold_secs))
                 .context("Failed to build Receive from speculative stream")?;
             self.in_flight_receive_target = Some(target);
             return Ok(Some(receive));
@@ -372,6 +421,8 @@ impl<S: SessionBackend> SessionCore<S> {
     /// Process an HTTP response from the server.
     pub(super) fn accept_response(&mut self, resp: HttpResponseTargeted) -> anyhow::Result<()> {
         let now_ms = self.now_ms();
+        // A response came back — the link is alive; reset the transient-failure tally.
+        self.consecutive_receive_transport_failures = 0;
         let in_flight_receive = self.in_flight_receive_target.take();
 
         let outputs = match self.active_session.accept_server_response(resp) {
@@ -444,6 +495,15 @@ impl<S: SessionBackend> SessionCore<S> {
             call_id = hr.call_id,
             "host-call response received while idle"
         );
+        self.scheduler.note_user_activity(self.now_ms());
+        if !self.host_response_matches(&hr) {
+            warn!(
+                target: "serial",
+                call_id = hr.call_id,
+                "dropping stale host-call response (no matching call outstanding)"
+            );
+            return Ok(());
+        }
         self.host_call_state = HostCallState::Idle;
         let output = self
             .active_session
@@ -491,6 +551,15 @@ impl<S: SessionBackend> SessionCore<S> {
             call_id = hr.call_id,
             "buffering host-call response (HTTP in flight)"
         );
+        self.scheduler.note_user_activity(self.now_ms());
+        if !self.host_response_matches(&hr) {
+            warn!(
+                target: "serial",
+                call_id = hr.call_id,
+                "dropping stale host-call response (no matching call outstanding)"
+            );
+            return;
+        }
         self.queues
             .user_ops
             .push_back(UserOperation::SubmitHostResponse {
@@ -524,6 +593,8 @@ impl<S: SessionBackend> SessionCore<S> {
         );
         self.host_call_state = HostCallState::Waiting {
             call_id: hc.call_id(),
+            scope: hc.scope(),
+            deadline_ms: self.now_ms() + HOST_CALL_GATE_GRACE_MS,
         };
         Some(hc)
     }
@@ -538,6 +609,100 @@ impl<S: SessionBackend> SessionCore<S> {
         matches!(self.host_call_state, HostCallState::Waiting { .. })
     }
 
+    /// How long Receive promotion should still be held back for host-call
+    /// handling, if at all.
+    ///
+    /// While a dispatched host call is inside its grace window we wait for the
+    /// answer (returning the window's end as a wake-up time). Past the window
+    /// the call may legitimately still be answered later (slow human input),
+    /// but Receives flow again so the session cannot wedge. Undispatched
+    /// queued host calls gate only until the loop's next dispatch pass.
+    fn receive_gate_deadline(&self, now_ms: u64) -> Option<u64> {
+        match &self.host_call_state {
+            HostCallState::Waiting { deadline_ms, .. } => {
+                (now_ms < *deadline_ms).then_some(*deadline_ms)
+            }
+            HostCallState::Idle => (!self.queues.host_calls.is_empty()).then_some(now_ms),
+        }
+    }
+
+    /// A finished (completed/killed) pipeline can no longer consume host-call
+    /// responses, and its consumer promise may already be gone (Ctrl+C during
+    /// Read-Host). Clearing the waiting state here is what lets the next
+    /// pipeline's host calls dispatch instead of queueing behind a corpse.
+    fn clear_host_call_for_finished_pipeline(&mut self, pipeline_id: Uuid) {
+        let HostCallState::Waiting {
+            call_id,
+            scope: HostCallScope::Pipeline { command_id },
+            ..
+        } = &self.host_call_state
+        else {
+            return;
+        };
+        if *command_id != pipeline_id {
+            return;
+        }
+        info!(
+            target: "serial",
+            call_id = *call_id,
+            pipeline_id = %pipeline_id,
+            "clearing outstanding host call for finished pipeline"
+        );
+        self.host_call_state = HostCallState::Idle;
+    }
+
+    /// Whether this response answers the host call we are actually waiting on.
+    /// A stale response (e.g. the consumer rejecting an input promise after the
+    /// pipeline was already killed and cleaned up) must be dropped, not
+    /// submitted against a dead server-side call.
+    fn host_response_matches(&self, hr: &HostResponse) -> bool {
+        matches!(
+            &self.host_call_state,
+            HostCallState::Waiting { call_id, .. } if *call_id == hr.call_id
+        )
+    }
+
+    /// Whether the request currently in flight is a Receive (vs. a Send).
+    pub(super) fn is_in_flight_receive(&self) -> bool {
+        self.in_flight_receive_target.is_some()
+    }
+
+    /// Tolerate a transport-level failure on the in-flight Receive: a long-poll
+    /// Receive is idempotent, so a transient gateway/WS drop is recovered by
+    /// re-arming polling instead of tearing down the session. Consults
+    /// [`SessionBackend::handle_transport_error`] for its connection bookkeeping,
+    /// caps consecutive failures so a dead link still terminates.
+    pub(super) fn tolerate_receive_transport_error(
+        &mut self,
+        conn_id: ConnectionId,
+    ) -> anyhow::Result<()> {
+        let target = self.in_flight_receive_target.take();
+        let disposition = self.active_session.handle_transport_error(conn_id);
+
+        self.consecutive_receive_transport_failures += 1;
+        let count = self.consecutive_receive_transport_failures;
+        if count > MAX_CONSECUTIVE_RECEIVE_TRANSPORT_FAILURES {
+            return Err(anyhow::anyhow!(
+                "giving up after {count} consecutive Receive transport failures"
+            ));
+        }
+
+        warn!(
+            target: "serial",
+            conn_id = conn_id.inner(),
+            ?disposition,
+            count,
+            "tolerating transport error on in-flight Receive; re-arming polling"
+        );
+
+        if let Some(target) = target {
+            self.scheduler.note_receive_timeout(target, self.now_ms());
+        }
+        let streams = self.active_session.active_desired_streams();
+        merge_speculative_streams(&mut self.queues.speculative_streams, streams);
+        Ok(())
+    }
+
     /// Whether buffered user operations are still waiting to be processed.
     ///
     /// The main loop drains `user_ops` one op per iteration via
@@ -623,6 +788,7 @@ impl<S: SessionBackend> SessionCore<S> {
                 if let UserEvent::PipelineFinished { pipeline } = &event {
                     self.scheduler
                         .note_pipeline_finished(pipeline.id(), self.now_ms());
+                    self.clear_host_call_for_finished_pipeline(pipeline.id());
                 }
                 self.queues.user_events.push(event);
             }
@@ -639,12 +805,14 @@ impl<S: SessionBackend> SessionCore<S> {
     }
 
     fn observe_user_op(&mut self, op: &UserOperation) {
+        let now_ms = self.now_ms();
+        self.scheduler.note_user_activity(now_ms);
+
         let UserOperation::KillPipeline { pipeline } = op else {
             return;
         };
 
         let pipeline_id: Uuid = pipeline.id();
-        let now_ms = self.now_ms();
         self.scheduler.note_cancel_requested(pipeline_id, now_ms);
 
         info!(
@@ -653,6 +821,13 @@ impl<S: SessionBackend> SessionCore<S> {
             "scheduler: cancel requested"
         );
     }
+
+    /// Server-side hold (Receive OperationTimeout) for the next poll of this
+    /// target, in seconds.
+    #[allow(clippy::cast_precision_loss)]
+    fn receive_hold_secs(&self, target: TargetId, now_ms: u64) -> f64 {
+        self.scheduler.receive_hold_ms(target, now_ms) as f64 / 1000.0
+    }
 }
 
 // ── Helpers ──────────────────────────────────────────────────────────────────
@@ -706,6 +881,8 @@ mod tests {
     struct MockBackend {
         op_responses: VecDeque<ActiveSessionOutput>,
         receive_results: VecDeque<TrySend>,
+        transport_error_disposition: TransportErrorDisposition,
+        active_streams: Vec<DesiredStream>,
     }
 
     impl MockBackend {
@@ -713,6 +890,8 @@ mod tests {
             Self {
                 op_responses: VecDeque::new(),
                 receive_results: VecDeque::new(),
+                transport_error_disposition: TransportErrorDisposition::Fatal,
+                active_streams: Vec::new(),
             }
         }
     }
@@ -732,15 +911,27 @@ mod tests {
             &mut self,
             _resp: HttpResponseTargeted,
         ) -> Result<Vec<ActiveSessionOutput>, PwshCoreError> {
-            unimplemented!("accept_server_response not needed for these tests")
+            Ok(Vec::new())
         }
 
-        fn fire_receive(&mut self, _streams: Vec<DesiredStream>) -> Result<TrySend, PwshCoreError> {
+        fn fire_receive(
+            &mut self,
+            _streams: Vec<DesiredStream>,
+            _hold_secs: Option<f64>,
+        ) -> Result<TrySend, PwshCoreError> {
             Ok(self
                 .receive_results
                 .pop_front()
                 .expect("MockBackend: receive_results exhausted"))
         }
+
+        fn handle_transport_error(&mut self, _conn_id: ConnectionId) -> TransportErrorDisposition {
+            self.transport_error_disposition
+        }
+
+        fn active_desired_streams(&self) -> Vec<DesiredStream> {
+            self.active_streams.clone()
+        }
     }
 
     /// Build a dummy `TrySend::JustSend`.
@@ -787,6 +978,15 @@ mod tests {
         PipelineHandle::new(id)
     }
 
+    /// Build a `Waiting` host-call state for testing.
+    fn waiting(call_id: i64, scope: HostCallScope, deadline_ms: u64) -> HostCallState {
+        HostCallState::Waiting {
+            call_id,
+            scope,
+            deadline_ms,
+        }
+    }
+
     // ── Promotion priority (6 tests) ────────────────────────────────────
 
     #[test]
@@ -858,7 +1058,7 @@ mod tests {
         core.queues
             .demanded_streams
             .push_back(pipeline_stream(Uuid::new_v4()));
-        core.host_call_state = HostCallState::Waiting { call_id: 1 };
+        core.host_call_state = waiting(1, HostCallScope::RunspacePool, u64::MAX);
 
         let promoted = core.promote_next_request().unwrap();
         assert!(promoted.is_none());
@@ -1010,7 +1210,7 @@ mod tests {
         let mock = MockBackend::new();
         let mut core = core_idle(mock);
         core.queues.host_calls.push_back(dummy_host_call(1));
-        core.host_call_state = HostCallState::Waiting { call_id: 1 };
+        core.host_call_state = waiting(1, HostCallScope::RunspacePool, u64::MAX);
 
         assert!(core.poll_host_call().is_none());
     }
@@ -1031,7 +1231,7 @@ mod tests {
     fn buffer_host_response_clears_active_flag() {
         let mock = MockBackend::new();
         let mut core = core_idle(mock);
-        core.host_call_state = HostCallState::Waiting { call_id: 1 };
+        core.host_call_state = waiting(1, HostCallScope::RunspacePool, u64::MAX);
 
         core.buffer_host_response(HostResponse {
             call_id: 1,
@@ -1214,29 +1414,35 @@ mod tests {
     }
 
     #[test]
-    fn scheduler_backed_off_demanded_sets_wakeup() {
+    fn empty_polled_demanded_still_promoted_without_wakeup() {
         let mut mock = MockBackend::new();
         mock.receive_results.push_back(dummy_try_send(60));
 
         let mut core = core_idle(mock);
         let id = Uuid::new_v4();
         let target = TargetId::Pipeline(id);
-        // Trigger backoff (not cancellation) — should set a wakeup.
+        // Empty polls only grow the server-side hold; they never park the loop
+        // on a client-side sleep, so the target stays eligible for promotion.
+        core.scheduler.note_receive_timeout(target, 0);
         core.scheduler.note_receive_timeout(target, 0);
         core.queues.demanded_streams.push_back(pipeline_stream(id));
 
         let promoted = core.promote_next_request().unwrap();
-        assert!(promoted.is_none());
-        // A backed-off (not cancelled) target should set a wakeup.
         assert!(
-            core.next_wakeup_at_ms.is_some(),
-            "backed-off stream should schedule a wakeup"
+            promoted.is_some(),
+            "empty-polled target must still be polled"
         );
+        assert!(
+            core.next_wakeup_at_ms.is_none(),
+            "adaptive hold must not schedule a client-side wakeup"
+        );
+        // The grown hold is what parks the Receive server-side.
+        assert_eq!(core.scheduler.receive_hold_ms(target, core.now_ms()), 1_000);
     }
 
     #[test]
-    fn accept_response_timeout_triggers_backoff() {
-        // This test verifies that timeout-like responses trigger scheduler backoff.
+    fn accept_response_timeout_grows_receive_hold() {
+        // Timeout-like responses grow the server-side hold (no client backoff).
         // We need accept_server_response, so we use a specialized mock.
         struct TimeoutMock;
         impl SessionBackend for TimeoutMock {
@@ -1258,9 +1464,19 @@ mod tests {
             fn fire_receive(
                 &mut self,
                 _streams: Vec<DesiredStream>,
+                _hold_secs: Option<f64>,
             ) -> Result<TrySend, PwshCoreError> {
                 Ok(dummy_try_send(70))
             }
+            fn handle_transport_error(
+                &mut self,
+                _conn_id: ConnectionId,
+            ) -> TransportErrorDisposition {
+                TransportErrorDisposition::Fatal
+            }
+            fn active_desired_streams(&self) -> Vec<DesiredStream> {
+                Vec::new()
+            }
         }
 
         let mut core = core_idle(TimeoutMock);
@@ -1269,6 +1485,7 @@ mod tests {
 
         // Simulate an in-flight Receive for this target.
         core.in_flight_receive_target = Some(target);
+        let hold_before = core.scheduler.receive_hold_ms(target, core.now_ms());
 
         // Build a minimal HttpResponseTargeted to pass to accept_response.
         // We need ConnectionId and HttpResponse. Using test_new for ConnectionId.
@@ -1284,10 +1501,224 @@ mod tests {
         );
         core.accept_response(resp).unwrap();
 
-        // After a timeout-like response, the scheduler should have applied backoff.
+        // A timeout-like response grows the server-side hold, and the target
+        // stays eligible (no client-side blocking).
         assert!(
-            !core.scheduler.is_allowed_target(target, core.now_ms()),
-            "scheduler should have applied backoff after timeout"
+            core.scheduler.receive_hold_ms(target, core.now_ms()) > hold_before,
+            "timeout-like response should grow the receive hold"
+        );
+        assert!(
+            core.scheduler.is_allowed_target(target, core.now_ms()),
+            "target must remain eligible after an empty poll"
+        );
+    }
+
+    // ── Host-call gate (4 tests) ────────────────────────────────────────
+
+    /// Within the grace window a pending host call gates Receives (and sets a
+    /// wake-up); past it, the same waiting call no longer blocks promotion, so
+    /// a slow-to-answer (or never-answering) consumer cannot wedge the session.
+    #[test]
+    fn host_call_gate_expires_and_promotion_resumes() {
+        let mut mock = MockBackend::new();
+        mock.receive_results.push_back(dummy_try_send(20));
+
+        let mut core = core_idle(mock);
+        core.queues
+            .demanded_streams
+            .push_back(pipeline_stream(Uuid::new_v4()));
+
+        // Deadline far in the future → gated, with a wake-up at the deadline.
+        core.host_call_state = waiting(7, HostCallScope::RunspacePool, u64::MAX);
+        assert!(core.promote_next_request().unwrap().is_none());
+        assert_eq!(core.next_wakeup_at_ms, Some(u64::MAX));
+
+        // Deadline already elapsed → the Receive promotes while still Waiting.
+        core.host_call_state = waiting(7, HostCallScope::RunspacePool, 0);
+        let promoted = core.promote_next_request().unwrap().unwrap();
+        assert_eq!(promoted.get_connection_id().inner(), 20);
+        assert!(
+            core.is_host_call_active(),
+            "the call must remain answerable after the gate expires"
         );
     }
+
+    /// A response for a call we are no longer waiting on (e.g. the consumer
+    /// rejecting an input promise after the pipeline was killed) is dropped,
+    /// never submitted. MockBackend panics on submit, so reaching the backend
+    /// would fail this test.
+    #[test]
+    fn stale_host_response_is_dropped() {
+        let mock = MockBackend::new();
+        let mut core = core_idle(mock);
+        core.host_call_state = waiting(7, HostCallScope::RunspacePool, 0);
+
+        core.accept_host_response(HostResponse {
+            call_id: 9,
+            scope: HostCallScope::RunspacePool,
+            submission: ironposh_client_core::host::Submission::NoSend,
+        })
+        .unwrap();
+        assert!(
+            core.is_host_call_active(),
+            "a mismatched response must not clear the waiting call"
+        );
+
+        core.buffer_host_response(HostResponse {
+            call_id: 9,
+            scope: HostCallScope::RunspacePool,
+            submission: ironposh_client_core::host::Submission::NoSend,
+        });
+        assert!(core.is_host_call_active());
+        assert!(
+            core.queues.user_ops.is_empty(),
+            "a stale response must not be buffered for submission"
+        );
+    }
+
+    /// PipelineFinished clears an outstanding host call scoped to that pipeline
+    /// (its consumer promise may be gone after Ctrl+C), letting the next
+    /// pipeline's host calls dispatch. Other scopes are untouched.
+    #[test]
+    fn finished_pipeline_clears_its_outstanding_host_call() {
+        let mock = MockBackend::new();
+        let mut core = core_idle(mock);
+        let pipeline_id = Uuid::new_v4();
+
+        // A call scoped to a different pipeline is untouched.
+        core.host_call_state = waiting(
+            5,
+            HostCallScope::Pipeline {
+                command_id: Uuid::new_v4(),
+            },
+            u64::MAX,
+        );
+        core.route_output(
+            ActiveSessionOutput::UserEvent(UserEvent::PipelineFinished {
+                pipeline: pipeline_handle(pipeline_id),
+            }),
+            SendPriority::Normal,
+        )
+        .unwrap();
+        assert!(core.is_host_call_active());
+
+        // A call scoped to the finished pipeline is cleared.
+        core.host_call_state = waiting(
+            6,
+            HostCallScope::Pipeline {
+                command_id: pipeline_id,
+            },
+            u64::MAX,
+        );
+        core.route_output(
+            ActiveSessionOutput::UserEvent(UserEvent::PipelineFinished {
+                pipeline: pipeline_handle(pipeline_id),
+            }),
+            SendPriority::Normal,
+        )
+        .unwrap();
+        assert!(!core.is_host_call_active());
+    }
+
+    /// The matching response still clears the waiting state and submits.
+    #[test]
+    fn matching_host_response_clears_waiting_and_submits() {
+        let mut mock = MockBackend::new();
+        mock.op_responses.push_back(ActiveSessionOutput::Ignore);
+        // The demanded Receive that was gated behind the host call.
+        mock.receive_results.push_back(dummy_try_send(20));
+
+        let mut core = core_idle(mock);
+        core.queues
+            .demanded_streams
+            .push_back(pipeline_stream(Uuid::new_v4()));
+        core.host_call_state = waiting(7, HostCallScope::RunspacePool, u64::MAX);
+        assert!(core.promote_next_request().unwrap().is_none());
+
+        core.accept_host_response(HostResponse {
+            call_id: 7,
+            scope: HostCallScope::RunspacePool,
+            submission: ironposh_client_core::host::Submission::NoSend,
+        })
+        .unwrap();
+        assert!(!core.is_host_call_active());
+
+        // The previously-gated Receive now promotes.
+        let second = core.promote_next_request().unwrap().unwrap();
+        assert_eq!(second.get_connection_id().inner(), 20);
+    }
+
+    // ── Receive transport-error tolerance (2 tests) ─────────────────────
+
+    /// A transient transport drop on an in-flight Receive must not kill the loop:
+    /// it re-arms polling from the active streams instead.
+    #[test]
+    fn tolerated_receive_transport_error_rearms_polling() {
+        let mut mock = MockBackend::new();
+        // Serial mode never disconnects, so ActiveSession classifies an Opened-state
+        // Receive failure as Fatal; the serial loop tolerates it regardless.
+        mock.transport_error_disposition = TransportErrorDisposition::Fatal;
+        let id = Uuid::new_v4();
+        mock.active_streams = vec![pipeline_stream(id)];
+
+        let mut core = core_idle(mock);
+        core.in_flight_receive_target = Some(TargetId::Pipeline(id));
+
+        core.tolerate_receive_transport_error(ConnectionId::test_new(1))
+            .expect("first transport error must be tolerated");
+
+        assert!(core.in_flight_receive_target.is_none());
+        assert_eq!(
+            core.queues.speculative_streams.len(),
+            1,
+            "polling must be re-armed from the active streams"
+        );
+    }
+
+    #[test]
+    fn receive_transport_errors_become_fatal_after_cap() {
+        let mock = MockBackend::new();
+        let mut core = core_idle(mock);
+
+        for _ in 0..MAX_CONSECUTIVE_RECEIVE_TRANSPORT_FAILURES {
+            core.tolerate_receive_transport_error(ConnectionId::test_new(1))
+                .expect("failures under the cap are tolerated");
+        }
+        assert!(
+            core.tolerate_receive_transport_error(ConnectionId::test_new(1))
+                .is_err(),
+            "a dead link must terminate once the cap is exceeded"
+        );
+    }
+
+    #[test]
+    fn successful_response_resets_receive_failure_tally() {
+        let mut mock = MockBackend::new();
+        mock.transport_error_disposition = TransportErrorDisposition::Fatal;
+        let mut core = core_idle(mock);
+
+        core.tolerate_receive_transport_error(ConnectionId::test_new(1))
+            .unwrap();
+        core.tolerate_receive_transport_error(ConnectionId::test_new(1))
+            .unwrap();
+
+        // A good response resets the tally...
+        let resp = HttpResponseTargeted::new(
+            ironposh_client_core::connector::http::HttpResponse {
+                status_code: 200,
+                headers: vec![],
+                body: ironposh_client_core::connector::http::HttpBody::Xml(String::new()),
+                peer_cert_der: None,
+            },
+            ConnectionId::test_new(1),
+            None,
+        );
+        core.accept_response(resp).unwrap();
+
+        // ...so the cap starts fresh and the next failures are tolerated again.
+        for _ in 0..MAX_CONSECUTIVE_RECEIVE_TRANSPORT_FAILURES {
+            core.tolerate_receive_transport_error(ConnectionId::test_new(1))
+                .expect("tally reset means these are tolerated");
+        }
+    }
 }
diff --git a/crates/ironposh-async/src/session_serial/mod.rs b/crates/ironposh-async/src/session_serial/mod.rs
index 07bfa32..ad72f71 100644
--- a/crates/ironposh-async/src/session_serial/mod.rs
+++ b/crates/ironposh-async/src/session_serial/mod.rs
@@ -16,7 +16,7 @@ use ironposh_client_core::connector::active_session::{ActiveSession, UserEvent};
 use ironposh_client_core::connector::connection_pool::TrySend;
 use ironposh_client_core::host::HostCall;
 use std::time::Duration;
-use tracing::{info, instrument, trace};
+use tracing::{info, instrument, trace, warn};
 
 use ironposh_client_core::connector::UserOperation;
 
@@ -65,8 +65,10 @@ pub async fn start_serial_session_loop(
         core.process_one_buffered_op()?;
 
         if let Some(req) = core.promote_next_request()? {
+            let conn_id = req.get_connection_id();
+            let was_receive = core.is_in_flight_receive();
             // HTTP in-flight: send request and buffer incoming ops until response.
-            let resp = send_and_buffer(
+            match send_and_buffer(
                 &client,
                 req,
                 &mut core,
@@ -74,9 +76,25 @@ pub async fn start_serial_session_loop(
                 &mut host_resp_rx,
                 &host_call_tx,
             )
-            .await?;
-
-            core.accept_response(resp)?;
+            .await
+            {
+                Ok(resp) => core.accept_response(resp)?,
+                Err(e) => {
+                    // A Receive is an idempotent long-poll: tolerate a transient
+                    // transport drop and re-arm. A Send's server-side effect is
+                    // unknown, so it stays fatal.
+                    if !was_receive {
+                        return Err(e);
+                    }
+                    warn!(
+                        target: "serial",
+                        conn_id = conn_id.inner(),
+                        error = %e,
+                        "transport error on in-flight Receive; attempting to tolerate"
+                    );
+                    core.tolerate_receive_transport_error(conn_id)?;
+                }
+            }
 
             // Drain any user ops that arrived while HTTP was in flight.
             if drain_channel(&mut core, &mut user_input_rx) {
@@ -110,7 +128,8 @@ pub async fn start_serial_session_loop(
 
             futures::select! {
                 () = wake_guard => {
-                    // Timer wake-up (e.g. receive backoff window elapsed) — loop again to promote work.
+                    // Timer wake-up (receive backoff window or host-call gate
+                    // grace elapsed) — loop again to promote work.
                 }
                 op = user_input_rx.next() => {
                     if let Some(op) = op {
diff --git a/crates/ironposh-async/src/session_serial/scheduler.rs b/crates/ironposh-async/src/session_serial/scheduler.rs
index 144dd5c..7980d3b 100644
--- a/crates/ironposh-async/src/session_serial/scheduler.rs
+++ b/crates/ironposh-async/src/session_serial/scheduler.rs
@@ -18,7 +18,6 @@ impl TargetId {
 
 #[derive(Debug, Default, Clone)]
 struct TargetState {
-    next_eligible_at_ms: u64,
     timeout_streak: u32,
     finished: bool,
     cancel_requested_at_ms: Option<u64>,
@@ -29,32 +28,44 @@ pub trait ReceiveScheduler {
     fn note_pipeline_finished(&mut self, pipeline_id: Uuid, now_ms: u64);
     fn note_receive_timeout(&mut self, target: TargetId, now_ms: u64);
     fn note_receive_progress(&mut self, target: TargetId, now_ms: u64);
+    fn note_user_activity(&mut self, now_ms: u64);
 
     fn is_allowed_target(&self, target: TargetId, now_ms: u64) -> bool;
     /// Returns the earliest time this target may be polled, or `None` if the
     /// target is finished and should never be scheduled again.
     fn next_eligible_at_ms(&self, target: TargetId) -> Option<u64>;
+
+    /// Server-side hold (Receive OperationTimeout) in milliseconds for the next
+    /// poll of this target.
+    fn receive_hold_ms(&self, target: TargetId, now_ms: u64) -> u64;
 }
 
 /// Default serial scheduler policy:
 /// - if a pipeline is finished, never poll it again
-/// - after cancel is requested, keep polling (short slices) until a finish signal arrives
-/// - apply exponential backoff after repeated receive timeouts per target
+/// - after cancel is requested, cap the hold so we observe the finish quickly
+/// - grow the server-side hold on repeated empty polls so the connection always
+///   has a Receive parked instead of idling on a client-side backoff sleep
 #[derive(Debug, Default)]
 pub struct DefaultReceiveScheduler {
     targets: HashMap<TargetId, TargetState>,
-    base_backoff_ms: u64,
-    max_backoff_ms: u64,
-    max_backoff_after_cancel_ms: u64,
+    base_hold_ms: u64,
+    max_hold_ms: u64,
+    max_hold_after_cancel_ms: u64,
+    activity_cap_ms: u64,
+    activity_window_ms: u64,
+    last_user_activity_ms: Option<u64>,
 }
 
 impl DefaultReceiveScheduler {
     pub fn new() -> Self {
         Self {
             targets: HashMap::new(),
-            base_backoff_ms: 200,
-            max_backoff_ms: 5_000,
-            max_backoff_after_cancel_ms: 1_000,
+            base_hold_ms: 250,
+            max_hold_ms: 1_000,
+            max_hold_after_cancel_ms: 500,
+            activity_cap_ms: 250,
+            activity_window_ms: 3_000,
+            last_user_activity_ms: None,
         }
     }
 
@@ -66,75 +77,68 @@ impl DefaultReceiveScheduler {
         self.targets.get(&id)
     }
 
-    fn backoff_for_streak(&self, streak: u32) -> u64 {
-        // backoff = base * 2^(streak-1), capped
-        let exp = streak.saturating_sub(1).min(31);
-        let pow = 1u64 << exp;
-        (self.base_backoff_ms.saturating_mul(pow)).min(self.max_backoff_ms)
+    fn recently_active(&self, now_ms: u64) -> bool {
+        self.last_user_activity_ms
+            .is_some_and(|last| now_ms.saturating_sub(last) <= self.activity_window_ms)
     }
 }
 
 impl ReceiveScheduler for DefaultReceiveScheduler {
     fn note_cancel_requested(&mut self, pipeline_id: Uuid, now_ms: u64) {
         let st = self.state_mut(TargetId::Pipeline(pipeline_id));
-        // Cancellation is cooperative. Keep polling this target (short slices)
+        // Cancellation is cooperative. Keep polling this target (short holds)
         // so we can observe PipelineFinished or a non-fatal InvalidSelectors
         // fault that will be translated into PipelineFinished by the backend.
         st.cancel_requested_at_ms = Some(now_ms);
         st.timeout_streak = 0;
-        st.next_eligible_at_ms = now_ms;
     }
 
-    fn note_pipeline_finished(&mut self, pipeline_id: Uuid, now_ms: u64) {
+    fn note_pipeline_finished(&mut self, pipeline_id: Uuid, _now_ms: u64) {
         let st = self.state_mut(TargetId::Pipeline(pipeline_id));
         st.finished = true;
-        st.next_eligible_at_ms = now_ms;
-    }
-
-    fn note_receive_timeout(&mut self, target: TargetId, now_ms: u64) {
-        let streak = {
-            let st = self.state_mut(target);
-            st.timeout_streak = st.timeout_streak.saturating_add(1);
-            st.timeout_streak
-        };
-
-        let backoff = {
-            let b = self.backoff_for_streak(streak);
-            let st = self.state_mut(target);
-            if st.cancel_requested_at_ms.is_some() {
-                b.min(self.max_backoff_after_cancel_ms)
-            } else {
-                b
-            }
-        };
+    }
+
+    fn note_receive_timeout(&mut self, target: TargetId, _now_ms: u64) {
+        // No client-side sleep: an empty poll just grows the next server-side
+        // hold via the streak. The connection stays parked on a Receive.
         let st = self.state_mut(target);
-        st.next_eligible_at_ms = now_ms.saturating_add(backoff);
+        st.timeout_streak = st.timeout_streak.saturating_add(1);
     }
 
     fn note_receive_progress(&mut self, target: TargetId, _now_ms: u64) {
         let st = self.state_mut(target);
         st.timeout_streak = 0;
-        st.next_eligible_at_ms = 0;
     }
 
-    fn is_allowed_target(&self, target: TargetId, now_ms: u64) -> bool {
-        let Some(st) = self.state(target) else {
-            return true;
-        };
-        if st.finished {
-            return false;
-        }
-        now_ms >= st.next_eligible_at_ms
+    fn note_user_activity(&mut self, now_ms: u64) {
+        self.last_user_activity_ms = Some(now_ms);
+    }
+
+    fn is_allowed_target(&self, target: TargetId, _now_ms: u64) -> bool {
+        self.state(target).is_none_or(|st| !st.finished)
     }
 
     fn next_eligible_at_ms(&self, target: TargetId) -> Option<u64> {
-        let Some(st) = self.state(target) else {
-            return Some(0);
-        };
-        if st.finished {
-            return None;
+        match self.state(target) {
+            Some(st) if st.finished => None,
+            _ => Some(0),
+        }
+    }
+
+    fn receive_hold_ms(&self, target: TargetId, now_ms: u64) -> u64 {
+        let st = self.state(target);
+        let streak = st.map_or(0, |s| s.timeout_streak);
+        let exp = streak.min(31);
+        let pow = 1u64 << exp;
+        let mut hold = self.base_hold_ms.saturating_mul(pow).min(self.max_hold_ms);
+
+        if st.is_some_and(|s| s.cancel_requested_at_ms.is_some()) {
+            hold = hold.min(self.max_hold_after_cancel_ms);
         }
-        Some(st.next_eligible_at_ms)
+        if self.recently_active(now_ms) {
+            hold = hold.min(self.activity_cap_ms);
+        }
+        hold
     }
 }
 
@@ -154,37 +158,82 @@ mod tests {
     }
 
     #[test]
-    fn timeout_backoff_delays_polling_then_allows() {
+    fn hold_doubles_per_empty_poll_and_caps_at_one_second() {
         let mut sched = DefaultReceiveScheduler::new();
         let id = Uuid::new_v4();
         let target = TargetId::Pipeline(id);
 
-        // First timeout applies base backoff (200ms)
-        sched.note_receive_timeout(target, 1_000);
-        assert!(!sched.is_allowed_target(target, 1_050));
-        assert!(sched.is_allowed_target(target, 1_250));
+        // Fresh target: base hold.
+        assert_eq!(sched.receive_hold_ms(target, 0), 250);
+        sched.note_receive_timeout(target, 0);
+        assert_eq!(sched.receive_hold_ms(target, 0), 500);
+        sched.note_receive_timeout(target, 0);
+        assert_eq!(sched.receive_hold_ms(target, 0), 1_000);
+        // Further empty polls stay capped at 1s.
+        sched.note_receive_timeout(target, 0);
+        sched.note_receive_timeout(target, 0);
+        assert_eq!(sched.receive_hold_ms(target, 0), 1_000);
+    }
+
+    #[test]
+    fn progress_resets_hold_to_base() {
+        let mut sched = DefaultReceiveScheduler::new();
+        let id = Uuid::new_v4();
+        let target = TargetId::Pipeline(id);
+
+        sched.note_receive_timeout(target, 0);
+        sched.note_receive_timeout(target, 0);
+        assert_eq!(sched.receive_hold_ms(target, 0), 1_000);
+        sched.note_receive_progress(target, 0);
+        assert_eq!(sched.receive_hold_ms(target, 0), 250);
+    }
+
+    #[test]
+    fn recent_activity_caps_hold_at_quarter_second() {
+        let mut sched = DefaultReceiveScheduler::new();
+        let id = Uuid::new_v4();
+        let target = TargetId::Pipeline(id);
+
+        // Grow the streak so the uncapped hold would be 1s.
+        sched.note_receive_timeout(target, 0);
+        sched.note_receive_timeout(target, 0);
+        assert_eq!(sched.receive_hold_ms(target, 10_000), 1_000);
+
+        // Activity within 3s caps at 250ms; outside the window it lifts again.
+        sched.note_user_activity(10_000);
+        assert_eq!(sched.receive_hold_ms(target, 12_000), 250);
+        assert_eq!(sched.receive_hold_ms(target, 13_001), 1_000);
     }
 
     #[test]
-    fn progress_resets_backoff() {
+    fn cancel_requested_caps_hold_at_half_second() {
         let mut sched = DefaultReceiveScheduler::new();
         let id = Uuid::new_v4();
         let target = TargetId::Pipeline(id);
 
+        // Would otherwise be 1s after two empty polls.
+        sched.note_receive_timeout(target, 0);
+        sched.note_receive_timeout(target, 0);
+        sched.note_cancel_requested(id, 1_000);
+        // Cancel resets the streak, so hold is base (250ms), still under the cap.
+        assert_eq!(sched.receive_hold_ms(target, 1_000), 250);
+        // Grow it back past the cap; the cancel cap holds it at 500ms.
+        sched.note_receive_timeout(target, 1_000);
         sched.note_receive_timeout(target, 1_000);
-        assert!(!sched.is_allowed_target(target, 1_050));
-        sched.note_receive_progress(target, 1_060);
-        assert!(sched.is_allowed_target(target, 1_061));
+        sched.note_receive_timeout(target, 1_000);
+        assert_eq!(sched.receive_hold_ms(target, 1_000), 500);
     }
 
     #[test]
-    fn next_eligible_is_none_for_cancelled_targets() {
+    fn next_eligible_is_none_for_finished_targets() {
         let mut sched = DefaultReceiveScheduler::new();
         let id = Uuid::new_v4();
         let target = TargetId::Pipeline(id);
 
         assert_eq!(sched.next_eligible_at_ms(target), Some(0));
+        assert!(sched.is_allowed_target(target, 0));
         sched.note_pipeline_finished(id, 1_000);
         assert_eq!(sched.next_eligible_at_ms(target), None);
+        assert!(!sched.is_allowed_target(target, 1_000));
     }
 }
diff --git a/crates/ironposh-client-core/src/connector/active_session.rs b/crates/ironposh-client-core/src/connector/active_session.rs
index 595e2fc..2861660 100644
--- a/crates/ironposh-client-core/src/connector/active_session.rs
+++ b/crates/ironposh-client-core/src/connector/active_session.rs
@@ -231,11 +231,18 @@ impl ActiveSession {
 
     /// Generate a Receive TrySend for the given streams.
     /// Used by the serial session loop to issue Receives after processing sends.
+    /// `hold_secs` sets the server-side Receive OperationTimeout — how long the
+    /// server parks the poll waiting for output. `None` uses the configured
+    /// default. The serial loop passes an adaptive value so a Receive is always
+    /// parked while a pipeline runs, instead of the old client-side backoff.
     pub fn fire_receive(
         &mut self,
         desired_streams: Vec<DesiredStream>,
+        hold_secs: Option<f64>,
     ) -> Result<TrySend, PwshCoreError> {
-        let recv_xml = self.runspace_pool.fire_receive(desired_streams)?;
+        let recv_xml = self
+            .runspace_pool
+            .fire_receive(desired_streams, hold_secs)?;
         let ts_send = self.connection_pool.send(&recv_xml)?;
         self.outstanding_receive_conns
             .insert(ts_send.get_connection_id());
@@ -247,7 +254,14 @@ impl ActiveSession {
     /// the pool to Opened, since the pre-disconnect Receive was retired.
     pub fn fire_active_receive(&mut self) -> Result<TrySend, PwshCoreError> {
         let desired = self.runspace_pool.compute_active_desired_streams();
-        self.fire_receive(desired)
+        self.fire_receive(desired, None)
+    }
+
+    /// Desired streams for the currently-active work (running pipelines, else the
+    /// runspace-pool stream). The serial loop re-arms polling with these after
+    /// tolerating a transport error on an in-flight Receive.
+    pub fn active_desired_streams(&self) -> Vec<DesiredStream> {
+        self.runspace_pool.compute_active_desired_streams()
     }
 
     /// Client-initiated operation → produce network work (`TrySend`) or a user-level event.
diff --git a/crates/ironposh-client-core/src/connector/mod.rs b/crates/ironposh-client-core/src/connector/mod.rs
index 08930af..3c02a8d 100644
--- a/crates/ironposh-client-core/src/connector/mod.rs
+++ b/crates/ironposh-client-core/src/connector/mod.rs
@@ -369,8 +369,8 @@ impl Connector {
                         // pool is Opened right away. Fire the initial Receive
                         // and hand off to the ActiveSession like the normal path.
                         let runspace_pool = expect_shell_connected.accept(&xml)?;
-                        let next_receive_xml =
-                            runspace_pool.fire_receive(DesiredStream::runspace_pool_streams())?;
+                        let next_receive_xml = runspace_pool
+                            .fire_receive(DesiredStream::runspace_pool_streams(), None)?;
                         info!(connect_receive_xml = %next_receive_xml, "outgoing unencrypted post-connect receive SOAP");
                         let next_req = connection_pool.send(&next_receive_xml)?;
 
@@ -411,8 +411,8 @@ impl Connector {
                     ConnectionPoolAccept::Body(xml) => {
                         // Advance runspace handshake
                         let runspace_pool = expect_shell_created.accept(&xml)?;
-                        let receive_xml =
-                            runspace_pool.fire_receive(DesiredStream::runspace_pool_streams())?;
+                        let receive_xml = runspace_pool
+                            .fire_receive(DesiredStream::runspace_pool_streams(), None)?;
                         info!(connecting_receive_xml = %receive_xml, "outgoing unencrypted connecting receive SOAP");
                         let try_send = connection_pool.send(&receive_xml)?;
 
@@ -459,7 +459,7 @@ impl Connector {
                         };
 
                         if runspace_pool.state == RunspacePoolState::NegotiationSent {
-                            let receive_xml = runspace_pool.fire_receive(desired_streams)?;
+                            let receive_xml = runspace_pool.fire_receive(desired_streams, None)?;
                             let try_send = connection_pool.send(&receive_xml)?;
                             let new_state = ConnectorState::ConnectReceiveCycle {
                                 runspace_pool,
@@ -468,7 +468,8 @@ impl Connector {
                             (new_state, ConnectorStepResult::SendBack { try_send })
                         } else if runspace_pool.state == RunspacePoolState::Opened {
                             // Hand off to ActiveSession: it should carry the pool forward
-                            let next_receive_xml = runspace_pool.fire_receive(desired_streams)?;
+                            let next_receive_xml =
+                                runspace_pool.fire_receive(desired_streams, None)?;
                             let next_req = connection_pool.send(&next_receive_xml)?;
                             let active_session = ActiveSession::new(runspace_pool, connection_pool);
                             let new_state = ConnectorState::Connected;
diff --git a/crates/ironposh-client-core/src/runspace/win_rs.rs b/crates/ironposh-client-core/src/runspace/win_rs.rs
index 7a839f4..107a76b 100644
--- a/crates/ironposh-client-core/src/runspace/win_rs.rs
+++ b/crates/ironposh-client-core/src/runspace/win_rs.rs
@@ -111,6 +111,7 @@ impl WinRunspace {
         &'a self,
         ws_man: &'a WsMan,
         desired_streams: Vec<crate::runspace_pool::DesiredStream>,
+        hold_secs: Option<f64>,
     ) -> impl Into<Element<'a>> {
         // Group streams by CommandId - streams with the same CommandId go into one DesiredStream element
         let mut grouped_streams: std::collections::BTreeMap<Option<uuid::Uuid>, Vec<String>> =
@@ -156,12 +157,13 @@ impl WinRunspace {
             .as_ref()
             .map(|shell_id| SelectorSetValue::new().add_selector("ShellId", shell_id));
 
-        ws_man.invoke(
+        ws_man.invoke_with_operation_timeout(
             &WsAction::ShellReceive,
             Some(&self.resource_uri),
             SoapBody::builder().receive(receive_tag).build(),
             Some(option_set),
             selector_set,
+            hold_secs,
         )
     }
 
diff --git a/crates/ironposh-client-core/src/runspace_pool/incoming.rs b/crates/ironposh-client-core/src/runspace_pool/incoming.rs
index 730929b..e80d486 100644
--- a/crates/ironposh-client-core/src/runspace_pool/incoming.rs
+++ b/crates/ironposh-client-core/src/runspace_pool/incoming.rs
@@ -326,6 +326,27 @@ impl RunspacePool {
                     result.push(AcceptResponsResult::PipelineFinished(PipelineHandle { id }));
                 }
 
+                let desired_streams = self.compute_active_desired_streams();
+                if !desired_streams.is_empty() {
+                    result.push(AcceptResponsResult::ReceiveResponse { desired_streams });
+                }
+            } else if let Some(stopping) = self.stopping_pipelines_for_fault() {
+                // A non-timeout fault can answer a Receive we issued for a pipeline the
+                // server already tore down after our Ctrl+C. If a pipeline is Stopping,
+                // treat the fault as its completion instead of killing the session.
+                let reason = fault.reason_text().unwrap_or("unknown");
+                warn!(
+                    target: "accept_response",
+                    reason = %reason,
+                    stopping_count = stopping.len(),
+                    "non-timeout SOAP fault while a pipeline is stopping; finishing it and continuing"
+                );
+
+                for id in stopping {
+                    self.pipelines.remove(&id);
+                    result.push(AcceptResponsResult::PipelineFinished(PipelineHandle { id }));
+                }
+
                 let desired_streams = self.compute_active_desired_streams();
                 if !desired_streams.is_empty() {
                     result.push(AcceptResponsResult::ReceiveResponse { desired_streams });
@@ -359,6 +380,20 @@ impl RunspacePool {
         Ok(result)
     }
 
+    /// Pipelines currently in the `Stopping` state, if any. Used to decide
+    /// whether a non-timeout SOAP fault is answering a Receive for a pipeline we
+    /// are already tearing down (Ctrl+C), in which case the fault is expected and
+    /// non-fatal.
+    fn stopping_pipelines_for_fault(&self) -> Option<Vec<Uuid>> {
+        let stopping: Vec<Uuid> = self
+            .pipelines
+            .iter()
+            .filter(|(_, p)| p.state() == PsInvocationState::Stopping)
+            .map(|(id, _)| *id)
+            .collect();
+        (!stopping.is_empty()).then_some(stopping)
+    }
+
     /// Fire create pipeline for a specific pipeline handle (used by service API)
     #[expect(clippy::too_many_lines)]
     #[instrument(skip(self, responses))]
diff --git a/crates/ironposh-client-core/src/runspace_pool/pool.rs b/crates/ironposh-client-core/src/runspace_pool/pool.rs
index 5bfce3c..ca00138 100644
--- a/crates/ironposh-client-core/src/runspace_pool/pool.rs
+++ b/crates/ironposh-client-core/src/runspace_pool/pool.rs
@@ -747,4 +747,38 @@ mod tests {
             "a WSMan fault must surface as SoapFault, got: {result:?}"
         );
     }
+
+    #[test]
+    fn non_timeout_fault_while_pipeline_stopping_finishes_it() {
+        let mut pool = test_pool(RunspacePoolState::Opened);
+        let id = uuid::Uuid::new_v4();
+        let mut pipeline = Pipeline::new();
+        pipeline.set_state(PsInvocationState::Stopping);
+        pool.pipelines.insert(id, pipeline);
+
+        let results = pool
+            .accept_response(FAULT_ENVELOPE)
+            .expect("a fault answering a Stopping pipeline must not kill the session");
+
+        assert!(
+            results
+                .iter()
+                .any(|r| matches!(r, AcceptResponsResult::PipelineFinished(h) if h.id == id)),
+            "the stopping pipeline should be reported finished, got: {results:?}"
+        );
+        assert!(
+            pool.pipelines.is_empty(),
+            "the stopping pipeline should be removed from the pool"
+        );
+    }
+
+    #[test]
+    fn non_timeout_fault_without_stopping_pipeline_stays_fatal() {
+        let mut pool = test_pool(RunspacePoolState::Opened);
+        let result = pool.accept_response(FAULT_ENVELOPE);
+        assert!(
+            matches!(result, Err(PwshCoreError::SoapFault { .. })),
+            "a fault unrelated to a stopping pipeline must still be fatal, got: {result:?}"
+        );
+    }
 }
diff --git a/crates/ironposh-client-core/src/runspace_pool/requests.rs b/crates/ironposh-client-core/src/runspace_pool/requests.rs
index 1f9d938..0e00d07 100644
--- a/crates/ironposh-client-core/src/runspace_pool/requests.rs
+++ b/crates/ironposh-client-core/src/runspace_pool/requests.rs
@@ -27,11 +27,12 @@ impl RunspacePool {
     pub(crate) fn fire_receive(
         &self,
         desired_streams: Vec<DesiredStream>,
+        hold_secs: Option<f64>,
     ) -> Result<String, crate::PwshCoreError> {
         debug_assert!(!desired_streams.is_empty(), "At least one desired stream");
         Ok(self
             .shell
-            .fire_receive(&self.connection, desired_streams)
+            .fire_receive(&self.connection, desired_streams, hold_secs)
             .into()
             .to_xml_string()?)
     }
diff --git a/crates/ironposh-client-sync/src/main.rs b/crates/ironposh-client-sync/src/main.rs
index 115f0bb..a5fde83 100644
--- a/crates/ironposh-client-sync/src/main.rs
+++ b/crates/ironposh-client-sync/src/main.rs
@@ -206,7 +206,7 @@ fn run_event_loop(
                         .send(send_request)
                         .context("Failed to send HTTP request")?;
                     let recv = active_session
-                        .fire_receive(then_receive_streams)
+                        .fire_receive(then_receive_streams, None)
                         .context("Failed to build receive after send-then-receive")?;
                     network_request_tx
                         .send(recv)
@@ -219,7 +219,7 @@ fn run_event_loop(
                         "firing deferred receive"
                     );
                     let recv = active_session
-                        .fire_receive(desired_streams)
+                        .fire_receive(desired_streams, None)
                         .context("Failed to build deferred receive")?;
                     network_request_tx
                         .send(recv)
diff --git a/crates/ironposh-test-support/src/fake_server.rs b/crates/ironposh-test-support/src/fake_server.rs
index 955ece9..35cf7c2 100644
--- a/crates/ironposh-test-support/src/fake_server.rs
+++ b/crates/ironposh-test-support/src/fake_server.rs
@@ -132,6 +132,149 @@ pub fn connect_response_xml(rpid: Uuid, messages: &[&dyn PsObjectWithType]) -> S
     )
 }
 
+/// Build a CommandResponse SOAP envelope acknowledging a pipeline start.
+pub fn command_response_xml(command_id: Uuid) -> String {
+    format!(
+        r#"<s:Envelope xml:lang="en-US"
+    xmlns:s="http://www.w3.org/2003/05/soap-envelope"
+    xmlns:a="http://schemas.xmlsoap.org/ws/2004/08/addressing"
+    xmlns:rsp="http://schemas.microsoft.com/wbem/wsman/1/windows/shell">
+    <s:Header>
+        <a:Action>http://schemas.microsoft.com/wbem/wsman/1/windows/shell/CommandResponse</a:Action>
+        <a:MessageID>uuid:{message_id}</a:MessageID>
+        <a:To>http://schemas.xmlsoap.org/ws/2004/08/addressing/role/anonymous</a:To>
+    </s:Header>
+    <s:Body>
+        <rsp:CommandResponse>
+            <rsp:CommandId>{command_id}</rsp:CommandId>
+        </rsp:CommandResponse>
+    </s:Body>
+</s:Envelope>"#,
+        message_id = Uuid::new_v4(),
+    )
+}
+
+/// Build a `w:TimedOut` WS-Management fault envelope — what a real server
+/// returns when a Receive exhausts its OperationTimeout with no data.
+pub fn timeout_fault_xml() -> String {
+    format!(
+        r#"<s:Envelope xml:lang="en-US"
+    xmlns:s="http://www.w3.org/2003/05/soap-envelope"
+    xmlns:a="http://schemas.xmlsoap.org/ws/2004/08/addressing"
+    xmlns:w="http://schemas.dmtf.org/wbem/wsman/1/wsman.xsd">
+    <s:Header>
+        <a:Action>http://schemas.dmtf.org/wbem/wsman/1/wsman/fault</a:Action>
+        <a:MessageID>uuid:{message_id}</a:MessageID>
+        <a:To>http://schemas.xmlsoap.org/ws/2004/08/addressing/role/anonymous</a:To>
+    </s:Header>
+    <s:Body>
+        <s:Fault>
+            <s:Code>
+                <s:Value>s:Receiver</s:Value>
+                <s:Subcode>
+                    <s:Value>w:TimedOut</s:Value>
+                </s:Subcode>
+            </s:Code>
+            <s:Reason>
+                <s:Text xml:lang="en-US">The WS-Management service cannot complete the operation within the time specified in OperationTimeout.</s:Text>
+            </s:Reason>
+        </s:Fault>
+    </s:Body>
+</s:Envelope>"#,
+        message_id = Uuid::new_v4(),
+    )
+}
+
+/// Build a SignalResponse SOAP envelope. `relates_to` must echo the Signal
+/// request's MessageID (the client correlates signal acks through it).
+pub fn signal_response_xml(relates_to: &str) -> String {
+    format!(
+        r#"<s:Envelope xml:lang="en-US"
+    xmlns:s="http://www.w3.org/2003/05/soap-envelope"
+    xmlns:a="http://schemas.xmlsoap.org/ws/2004/08/addressing"
+    xmlns:rsp="http://schemas.microsoft.com/wbem/wsman/1/windows/shell">
+    <s:Header>
+        <a:Action>http://schemas.microsoft.com/wbem/wsman/1/windows/shell/SignalResponse</a:Action>
+        <a:MessageID>uuid:{message_id}</a:MessageID>
+        <a:RelatesTo>{relates_to}</a:RelatesTo>
+        <a:To>http://schemas.xmlsoap.org/ws/2004/08/addressing/role/anonymous</a:To>
+    </s:Header>
+    <s:Body>
+        <rsp:SignalResponse/>
+    </s:Body>
+</s:Envelope>"#,
+        message_id = Uuid::new_v4(),
+    )
+}
+
+/// Build a ReceiveResponse SOAP envelope carrying pipeline-scoped PSRP messages
+/// (streams tagged with `CommandId`, PSRP `pid` set to the command id).
+///
+/// `object_id_start` seeds the PSRP fragment object ids so ids stay unique
+/// across successive responses in one session. When `done` is set, a
+/// `CommandState Done` element is appended — mirroring how a real server closes
+/// out a finished pipeline.
+pub fn pipeline_receive_response_xml(
+    rpid: Uuid,
+    command_id: Uuid,
+    messages: &[&dyn PsObjectWithType],
+    done: bool,
+    object_id_start: u64,
+) -> String {
+    use std::fmt::Write as _;
+
+    let mut streams = String::new();
+    for (index, message) in messages.iter().enumerate() {
+        let remoting_message = PowerShellRemotingMessage::new(
+            Destination::Client,
+            message.message_type(),
+            rpid,
+            Some(command_id),
+            &message.to_ps_object(),
+        )
+        .expect("serialize PSRP message");
+
+        let fragment = Fragment::new(
+            object_id_start + index as u64,
+            0,
+            remoting_message.pack(),
+            true,
+            true,
+        );
+        let payload = base64::engine::general_purpose::STANDARD.encode(fragment.pack());
+        write!(
+            streams,
+            r#"<rsp:Stream Name="stdout" CommandId="{command_id}">{payload}</rsp:Stream>"#
+        )
+        .expect("write stream XML");
+    }
+
+    let command_state = if done {
+        format!(
+            r#"<rsp:CommandState CommandId="{command_id}" State="http://schemas.microsoft.com/wbem/wsman/1/windows/shell/CommandState/Done"><rsp:ExitCode>0</rsp:ExitCode></rsp:CommandState>"#
+        )
+    } else {
+        String::new()
+    };
+
+    format!(
+        r#"<s:Envelope xml:lang="en-US"
+    xmlns:s="http://www.w3.org/2003/05/soap-envelope"
+    xmlns:a="http://schemas.xmlsoap.org/ws/2004/08/addressing"
+    xmlns:rsp="http://schemas.microsoft.com/wbem/wsman/1/windows/shell">
+    <s:Header>
+        <a:Action>http://schemas.microsoft.com/wbem/wsman/1/windows/shell/ReceiveResponse</a:Action>
+        <a:MessageID>uuid:{message_id}</a:MessageID>
+        <a:To>http://schemas.xmlsoap.org/ws/2004/08/addressing/role/anonymous</a:To>
+    </s:Header>
+    <s:Body>
+        <rsp:ReceiveResponse>{streams}{command_state}</rsp:ReceiveResponse>
+    </s:Body>
+</s:Envelope>"#,
+        message_id = Uuid::new_v4(),
+    )
+}
+
 /// Build a ReceiveResponse SOAP envelope carrying the given server-to-client PSRP
 /// messages as single-fragment `stdout` streams (no command id => runspace pool stream).
 pub fn receive_response_xml(rpid: Uuid, messages: &[&dyn PsObjectWithType]) -> String {
diff --git a/crates/ironposh-web/Cargo.toml b/crates/ironposh-web/Cargo.toml
index 0faecd7..1e8c150 100644
--- a/crates/ironposh-web/Cargo.toml
+++ b/crates/ironposh-web/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "ironposh-web"
-version = "0.5.1"
+version = "0.6.0"
 authors = ["irving ou <jou@devolutions.net>"]
 edition = "2018"
 description = "PowerShell Remoting over WinRM for WebAssembly"
diff --git a/web/powershell-terminal-component/src/hostcall-handler.ts b/web/powershell-terminal-component/src/hostcall-handler.ts
index 12927f2..65d1f0d 100644
--- a/web/powershell-terminal-component/src/hostcall-handler.ts
+++ b/web/powershell-terminal-component/src/hostcall-handler.ts
@@ -60,17 +60,30 @@ export function createHostCallHandler(config: HostCallHandlerConfig) {
   };
 
   const keyQueue: JsKeyInfo[] = [];
-  const keyWaiters: Array<(k: JsKeyInfo) => void> = [];
+  const keyWaiters: Array<{
+    resolve: (k: JsKeyInfo) => void;
+    reject: (e: Error) => void;
+  }> = [];
 
   const enqueueKey = (keyInfo: JsKeyInfo) => {
     const waiter = keyWaiters.shift();
     if (waiter) {
-      waiter(keyInfo);
+      waiter.resolve(keyInfo);
       return;
     }
     keyQueue.push(keyInfo);
   };
 
+  // Reject any input a host call is currently blocked on. Without this, a
+  // Ctrl+C during Read-Host/Prompt leaves the handler promise pending forever,
+  // so the Rust side never submits a response and the session loop wedges.
+  const cancelPendingInput = (reason: string) => {
+    const err = new Error(reason);
+    while (keyWaiters.length > 0) {
+      keyWaiters.shift()!.reject(err);
+    }
+  };
+
   const normalizeChar = (s: string): string => {
     if (s.length === 0) return "\u0000";
     if (s.length === 1) return s;
@@ -304,8 +317,8 @@ export function createHostCallHandler(config: HostCallHandlerConfig) {
   const readKeyAsync = async (): Promise<JsKeyInfo> => {
     const next = keyQueue.shift();
     if (next) return next;
-    return await new Promise<JsKeyInfo>((resolve) => {
-      keyWaiters.push(resolve);
+    return await new Promise<JsKeyInfo>((resolve, reject) => {
+      keyWaiters.push({ resolve, reject });
     });
   };
 
@@ -832,5 +845,5 @@ export function createHostCallHandler(config: HostCallHandlerConfig) {
     return (handler as any)(variant.params);
   }) as TypedHostCallHandler;
 
-  return dispatch;
+  return { handler: dispatch, cancelPendingInput };
 }
diff --git a/web/powershell-terminal-component/src/powershell-terminal.ts b/web/powershell-terminal-component/src/powershell-terminal.ts
index 05dd848..4c33216 100644
--- a/web/powershell-terminal-component/src/powershell-terminal.ts
+++ b/web/powershell-terminal-component/src/powershell-terminal.ts
@@ -74,6 +74,7 @@ export class PowerShellTerminalElement extends HTMLElement {
   private isRunning = false;
   private hostCallInputDepth = 0;
   private isPrompting = false;
+  private cancelHostCallInput: ((reason: string) => void) | null = null;
 
   constructor() {
     super();
@@ -201,19 +202,21 @@ export class PowerShellTerminalElement extends HTMLElement {
       console.log("Connecting with config:", { ...config, password: "*****" });
 
       // Create host call handler with terminal integration
-      const hostCallHandler = createHostCallHandler({
-        terminal: this.terminal,
-        hostName: "PowerShell Terminal",
-        hostVersion: "1.0.0",
-        culture: "en-US",
-        uiCulture: "en-US",
-        beginHostCallInput: () => {
-          this.hostCallInputDepth += 1;
-        },
-        endHostCallInput: () => {
-          this.hostCallInputDepth = Math.max(0, this.hostCallInputDepth - 1);
-        },
-      });
+      const { handler: hostCallHandler, cancelPendingInput } =
+        createHostCallHandler({
+          terminal: this.terminal,
+          hostName: "PowerShell Terminal",
+          hostVersion: "1.0.0",
+          culture: "en-US",
+          uiCulture: "en-US",
+          beginHostCallInput: () => {
+            this.hostCallInputDepth += 1;
+          },
+          endHostCallInput: () => {
+            this.hostCallInputDepth = Math.max(0, this.hostCallInputDepth - 1);
+          },
+        });
+      this.cancelHostCallInput = cancelPendingInput;
 
       // Create PowerShell client
       await new Promise((resolve, reject) => {
@@ -224,11 +227,25 @@ export class PowerShellTerminalElement extends HTMLElement {
             if (event === "ActiveSessionStarted") {
               this.setState("connected");
               resolve(undefined);
+              return;
             }
 
             if (typeof event === "object" && "error" in event) {
-              this.setState("closed");
-              reject(new Error(event.error));
+              if (this.state === "connecting") {
+                this.setState("closed");
+                reject(new Error(event.error));
+              } else {
+                this.handleSessionLost(event.error);
+              }
+              return;
+            }
+
+            // The session ended after it was established (task exited / closed).
+            if (
+              this.state === "connected" &&
+              (event === "ActiveSessionEnded" || event === "Closed")
+            ) {
+              this.handleSessionLost("session ended");
             }
           }
         );
@@ -262,6 +279,9 @@ export class PowerShellTerminalElement extends HTMLElement {
     if (this.hostCallInputDepth > 0) {
       // Still allow Ctrl+C to cancel a running command.
       if (data === "\u0003") {
+        // Reject the host call's pending input so the Rust side submits a
+        // response instead of leaving the session loop wedged on the host call.
+        this.cancelHostCallInput?.("Canceled by user");
         if (this.isRunning && this.runningController) {
           this.terminal.write("^C\r\n");
           this.runningController.abort(new Error("Canceled by user"));
@@ -379,6 +399,22 @@ export class PowerShellTerminalElement extends HTMLElement {
     }
   }
 
+  // A session that dies mid-run (transport reset, task error, remote close)
+  // must be surfaced: otherwise input silently stops working and the terminal
+  // just goes dark. Print why, unstick the running flag, then close.
+  private handleSessionLost(reason: string): void {
+    if (this.state === "closed") return;
+    this.isRunning = false;
+    this.runningController = null;
+    this.cancelHostCallInput?.(`Connection lost: ${reason}`);
+    if (this.terminal) {
+      this.terminal.writeln("");
+      this.terminal.writeln(`\x1b[31mConnection lost: ${reason}\x1b[0m`);
+    }
+    this.setState("closed");
+    this.emitEvent({ type: "error", detail: new Error(reason) });
+  }
+
   disconnect(): void {
     if (this.state === "connected" && this.terminal) {
       this.setState("closed");