diff --git a/frontend/src-tauri/Cargo.toml b/frontend/src-tauri/Cargo.toml
index 324f9e5..5072da6 100644
--- a/frontend/src-tauri/Cargo.toml
+++ b/frontend/src-tauri/Cargo.toml
@@ -48,7 +48,7 @@ cap-fs-ext = "4.0.2"
 jsonschema = { version = "0.55", default-features = false }
 
 [target.'cfg(windows)'.dependencies]
-windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Security", "Win32_Security_Isolation", "Win32_Security_Authorization", "Win32_System_Com", "Win32_System_JobObjects", "Win32_System_Threading", "Win32_System_Pipes", "Win32_System_SystemInformation", "Win32_Storage_FileSystem", "Win32_System_RemoteDesktop", "Win32_UI_WindowsAndMessaging", "Win32_Graphics_Gdi", "Win32_System_LibraryLoader"] }
+windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Security", "Win32_Security_Isolation", "Win32_Security_Authorization", "Win32_System_Com", "Win32_System_JobObjects", "Win32_System_Threading", "Win32_System_Pipes", "Win32_System_IO", "Win32_System_SystemInformation", "Win32_Storage_FileSystem", "Win32_System_RemoteDesktop", "Win32_UI_WindowsAndMessaging", "Win32_Graphics_Gdi", "Win32_System_LibraryLoader"] }
 
 [build-dependencies]
 tauri-build = { version = "2", optional = true , features = [] }
diff --git a/frontend/src-tauri/src/extension_container.rs b/frontend/src-tauri/src/extension_container.rs
index 2029a61..342645e 100644
--- a/frontend/src-tauri/src/extension_container.rs
+++ b/frontend/src-tauri/src/extension_container.rs
@@ -777,10 +777,7 @@ mod tests {
             // Actual native RPC: the child cannot name an identity or connect to
             // a shared endpoint; only its own stdio pipe reaches this broker.
             {
-                use std::{
-                    io::{BufReader, Read},
-                    os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle},
-                };
+                use std::os::windows::io::{AsRawHandle, FromRawHandle, OwnedHandle};
                 let sentinel = unsafe {
                     windows_sys::Win32::System::Threading::CreateEventW(
                         std::ptr::null(),
@@ -830,34 +827,86 @@ mod tests {
                     .create_suspended_with_stdio(&profile, &bound_entry)
                     .unwrap();
                 let running = unsafe { suspended.resume().unwrap() };
-                let crate::extension_stdio::HostIo {
-                    mut input,
-                    output,
-                    mut error,
-                } = io;
-                let mut output = crate::extension_stdio::Frames::new(BufReader::new(output));
-                let request = output.read().unwrap().unwrap();
+                let mut pump = running.start_io(io).unwrap();
+                let crate::extension_io::Event::Frame(request) =
+                    pump.receive(std::time::Duration::from_secs(5)).unwrap()
+                else {
+                    panic!("missing RPC request")
+                };
                 let response = files.dispatch(&mut workspace, &request).unwrap();
-                crate::extension_stdio::write_frame(
-                    &mut input,
-                    &serde_json::to_vec(&response).unwrap(),
-                )
-                .unwrap();
-                drop(input);
+                pump.send(serde_json::to_vec(&response).unwrap()).unwrap();
+                pump.close_input();
+                let crate::extension_io::Event::Frame(reply) =
+                    pump.receive(std::time::Duration::from_secs(5)).unwrap()
+                else {
+                    panic!("missing RPC reply")
+                };
+                assert_eq!(reply, b"{\"ok\":true}");
+                assert!(matches!(
+                    pump.receive(std::time::Duration::from_secs(5)).unwrap(),
+                    crate::extension_io::Event::Closed
+                ));
                 assert_eq!(
                     running.wait(std::time::Duration::from_secs(5)).unwrap(),
                     Some(0)
                 );
-                assert_eq!(output.read().unwrap().unwrap(), b"{\"ok\":true}");
-                assert!(output.read().unwrap().is_none());
-                let mut diagnostic = String::new();
-                error.read_to_string(&mut diagnostic).unwrap();
-                assert_eq!(diagnostic.trim(), "fixture diagnostic");
+                pump.shutdown().unwrap();
                 assert_eq!(
                     workspace.read("fixture.md").unwrap().content,
                     "from host broker"
                 );
             }
+            for mode in ["wait_tree", "stderr_flood", "stdout_flood"] {
+                let mut io_claims = claims.clone();
+                io_claims.arguments = vec![mode.into()];
+                io_claims.expires_at_ms = 120_000;
+                let permit = authority.issue(&io_claims, 1).unwrap();
+                let prepared = context
+                    .prepare(&authority, &permit, &io_claims, &bound_entry, &broker, 2)
+                    .unwrap();
+                let (suspended, io) = prepared
+                    .create_suspended_with_stdio(&profile, &bound_entry)
+                    .unwrap();
+                let running = unsafe { suspended.resume().unwrap() };
+                let pump = running.start_io(io).unwrap();
+                if mode == "wait_tree" {
+                    pump.send(vec![b'x'; crate::extension_stdio::MAX_FRAME_BYTES])
+                        .unwrap();
+                    let started = std::time::Instant::now();
+                    while running.active_test_processes().unwrap() < 2 {
+                        assert!(started.elapsed() < std::time::Duration::from_secs(5));
+                        std::thread::sleep(std::time::Duration::from_millis(5));
+                    }
+                } else {
+                    let started = std::time::Instant::now();
+                    while pump.check().is_ok() {
+                        assert!(started.elapsed() < std::time::Duration::from_secs(5));
+                        std::thread::sleep(std::time::Duration::from_millis(5));
+                    }
+                    assert_eq!(
+                        pump.check().unwrap_err().code,
+                        if mode == "stderr_flood" {
+                            "EXTENSION_STDERR_LIMIT_EXCEEDED"
+                        } else {
+                            "EXTENSION_BROKER_REQUEST_TOO_LARGE"
+                        }
+                    );
+                }
+                let started = std::time::Instant::now();
+                pump.shutdown().unwrap();
+                assert!(running
+                    .wait(std::time::Duration::from_secs(5))
+                    .unwrap()
+                    .is_some());
+                while running.active_test_processes().unwrap() != 0 {
+                    assert!(started.elapsed() < std::time::Duration::from_secs(5));
+                    std::thread::sleep(std::time::Duration::from_millis(5));
+                }
+                eprintln!(
+                    "native IO {mode}: shutdown/tree empty in {:?}",
+                    started.elapsed()
+                );
+            }
             for cause in [
                 "before_create",
                 "before_resume",
diff --git a/frontend/src-tauri/src/extension_io.rs b/frontend/src-tauri/src/extension_io.rs
new file mode 100644
index 0000000..2387b1d
--- /dev/null
+++ b/frontend/src-tauri/src/extension_io.rs
@@ -0,0 +1,379 @@
+//! Bounded IO for Host-created anonymous pipes. Never run shutdown on the UI
+//! thread: cancellation waits for the native pipe operations to acknowledge it.
+use crate::{
+    extension_job::Job,
+    extension_stdio::{write_frame, Frames, HostIo, MAX_FRAME_BYTES},
+    workspace::{HostError, Result},
+};
+use std::{
+    io::{BufReader, Read},
+    os::windows::io::AsRawHandle,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        mpsc::{self, Receiver, RecvTimeoutError, SyncSender},
+        Arc, Mutex,
+    },
+    thread::JoinHandle,
+    time::{Duration, Instant},
+};
+use windows_sys::Win32::System::IO::CancelSynchronousIo;
+const QUEUED_FRAMES: usize = 4;
+const STDERR_BYTES: usize = 1024 * 1024;
+const OUTPUT_BYTES_PER_SECOND: usize = 8 * 1024 * 1024;
+const OUTPUT_FRAMES_PER_SECOND: usize = 128;
+#[derive(Debug)]
+pub enum Event {
+    Frame(Vec<u8>),
+    Closed,
+}
+struct State {
+    stopped: AtomicBool,
+    error: Mutex<Option<String>>,
+    job: Job,
+}
+impl State {
+    fn fail(&self, error: &str) {
+        let mut saved = self.error.lock().unwrap_or_else(|e| e.into_inner());
+        if saved.is_none() {
+            *saved = Some(error.to_owned());
+        }
+        self.stopped.store(true, Ordering::Release);
+        let _ = self.job.terminate();
+    }
+    fn check(&self) -> Result<()> {
+        if let Some(error) = &*self.error.lock().unwrap_or_else(|e| e.into_inner()) {
+            return Err(HostError::new(error));
+        }
+        if self.stopped.load(Ordering::Acquire) {
+            return Err(HostError::new("EXTENSION_IO_CLOSED"));
+        }
+        Ok(())
+    }
+}
+struct Worker {
+    state: Arc<State>,
+    thread: Option<JoinHandle<()>>,
+}
+impl Worker {
+    fn spawn(
+        state: &Arc<State>,
+        name: &str,
+        task: impl FnOnce(&State) -> Result<()> + Send + 'static,
+    ) -> Result<Self> {
+        let local = Arc::clone(state);
+        let thread = std::thread::Builder::new()
+            .name(name.into())
+            .spawn(move || {
+                match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| task(&local))) {
+                    Ok(Err(error)) if !local.stopped.load(Ordering::Acquire) => {
+                        local.fail(&error.code)
+                    }
+                    Err(_) => local.fail("EXTENSION_IO_WORKER_FAILED"),
+                    _ => {}
+                }
+            })
+            .map_err(|_| HostError::new("EXTENSION_IO_UNAVAILABLE"))?;
+        Ok(Self {
+            state: Arc::clone(state),
+            thread: Some(thread),
+        })
+    }
+}
+impl Drop for Worker {
+    fn drop(&mut self) {
+        self.state.stopped.store(true, Ordering::Release);
+        if let Some(thread) = self.thread.take() {
+            // Cancellation is not sticky: retry to cover the interval between
+            // the worker checking stopped and actually entering Read/WriteFile.
+            // These workers only issue anonymous-pipe IO, never arbitrary device IO.
+            while !thread.is_finished() {
+                unsafe {
+                    CancelSynchronousIo(thread.as_raw_handle());
+                }
+                std::thread::sleep(Duration::from_millis(2));
+            }
+            let _ = thread.join();
+        }
+    }
+}
+pub struct Pump {
+    state: Arc<State>,
+    input: Option<SyncSender<Vec<u8>>>,
+    output: Receiver<Event>,
+    workers: Vec<Worker>,
+}
+impl Pump {
+    pub(crate) fn start(io: HostIo, job: Job) -> Result<Self> {
+        let state = Arc::new(State {
+            stopped: AtomicBool::new(false),
+            error: Mutex::new(None),
+            job,
+        });
+        let (input, writes) = mpsc::sync_channel::<Vec<u8>>(1);
+        let (events, output) = mpsc::sync_channel(QUEUED_FRAMES);
+        let mut pump = Self {
+            state,
+            input: Some(input),
+            output,
+            workers: Vec::with_capacity(3),
+        };
+        let HostIo {
+            mut input,
+            output,
+            mut error,
+        } = io;
+        pump.workers.push(Worker::spawn(
+            &pump.state,
+            "extension-stdin",
+            move |state| {
+                while !state.stopped.load(Ordering::Acquire) {
+                    match writes.recv_timeout(Duration::from_millis(20)) {
+                        Ok(frame) => {
+                            state.check()?;
+                            write_frame(&mut input, &frame)?;
+                        }
+                        Err(RecvTimeoutError::Timeout) => {}
+                        Err(RecvTimeoutError::Disconnected) => return Ok(()),
+                    }
+                }
+                Ok(())
+            },
+        )?);
+        pump.workers.push(Worker::spawn(
+            &pump.state,
+            "extension-stdout",
+            move |state| {
+                let mut frames = Frames::new(BufReader::new(output));
+                let mut window = Instant::now();
+                let mut bytes = 0;
+                let mut count = 0;
+                while !state.stopped.load(Ordering::Acquire) {
+                    let frame = frames.read()?;
+                    state.check()?;
+                    if window.elapsed() >= Duration::from_secs(1) {
+                        window = Instant::now();
+                        bytes = 0;
+                        count = 0;
+                    }
+                    if let Some(frame) = frame {
+                        bytes += frame.len();
+                        count += 1;
+                        if bytes > OUTPUT_BYTES_PER_SECOND || count > OUTPUT_FRAMES_PER_SECOND {
+                            return Err(HostError::new("EXTENSION_IO_RATE_LIMITED"));
+                        }
+                        events
+                            .try_send(Event::Frame(frame))
+                            .map_err(|_| HostError::new("EXTENSION_IO_OUTPUT_BACKPRESSURE"))?;
+                    } else {
+                        events
+                            .try_send(Event::Closed)
+                            .map_err(|_| HostError::new("EXTENSION_IO_OUTPUT_BACKPRESSURE"))?;
+                        return Ok(());
+                    }
+                }
+                Ok(())
+            },
+        )?);
+        pump.workers.push(Worker::spawn(
+            &pump.state,
+            "extension-stderr",
+            move |state| {
+                // Drain without persisting possible secrets. Diagnostic retention
+                // needs an explicit redaction policy before it can be enabled.
+                let mut buffer = [0; 4096];
+                let mut total = 0;
+                while !state.stopped.load(Ordering::Acquire) {
+                    let count = error
+                        .read(&mut buffer)
+                        .map_err(|_| HostError::new("EXTENSION_PIPE_READ_FAILED"))?;
+                    if count == 0 {
+                        return Ok(());
+                    }
+                    total += count;
+                    if total > STDERR_BYTES {
+                        return Err(HostError::new("EXTENSION_STDERR_LIMIT_EXCEEDED"));
+                    }
+                }
+                Ok(())
+            },
+        )?);
+        Ok(pump)
+    }
+    /// Nonblocking admission; at most one pending write plus one in progress.
+    pub fn send(&self, frame: Vec<u8>) -> Result<()> {
+        self.state.check()?;
+        if frame.is_empty()
+            || frame.len() > MAX_FRAME_BYTES
+            || frame.contains(&b'\n')
+            || frame.contains(&b'\r')
+        {
+            return Err(HostError::new("EXTENSION_PIPE_INVALID_FRAME"));
+        }
+        self.input
+            .as_ref()
+            .ok_or_else(|| HostError::new("EXTENSION_IO_INPUT_CLOSED"))?
+            .try_send(frame)
+            .map_err(|_| HostError::new("EXTENSION_IO_INPUT_BACKPRESSURE"))
+    }
+    pub fn close_input(&mut self) {
+        self.input.take();
+    }
+    pub fn receive(&self, timeout: Duration) -> Result<Event> {
+        self.state.check()?;
+        if timeout > Duration::from_secs(60) {
+            return Err(HostError::new("EXTENSION_IO_TIMEOUT_INVALID"));
+        }
+        let started = Instant::now();
+        loop {
+            let remaining = timeout.saturating_sub(started.elapsed());
+            let result = self
+                .output
+                .recv_timeout(remaining.min(Duration::from_millis(20)));
+            self.state.check()?;
+            match result {
+                Ok(event) => return Ok(event),
+                Err(RecvTimeoutError::Disconnected) => {
+                    return Err(HostError::new("EXTENSION_IO_CLOSED"))
+                }
+                Err(RecvTimeoutError::Timeout) if started.elapsed() >= timeout => {
+                    return Err(HostError::new("EXTENSION_IO_TIMEOUT"))
+                }
+                Err(RecvTimeoutError::Timeout) => {}
+            }
+        }
+    }
+
+    pub fn check(&self) -> Result<()> {
+        self.state.check()
+    }
+    /// Stop the process group first, then cancel and join every pipe worker.
+    pub fn shutdown(mut self) -> Result<()> {
+        let result = self.state.job.terminate();
+        self.state.stopped.store(true, Ordering::Release);
+        self.input.take();
+        self.workers.clear();
+        result
+    }
+}
+impl Drop for Pump {
+    fn drop(&mut self) {
+        self.state.stopped.store(true, Ordering::Release);
+        let _ = self.state.job.terminate();
+        self.input.take();
+        self.workers.clear();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{fs::File, io::Write, os::windows::io::FromRawHandle};
+    fn pipes() -> (HostIo, [File; 3]) {
+        fn pair() -> (File, File) {
+            let mut read = std::ptr::null_mut();
+            let mut write = std::ptr::null_mut();
+            assert_ne!(
+                unsafe {
+                    windows_sys::Win32::System::Pipes::CreatePipe(
+                        &mut read,
+                        &mut write,
+                        std::ptr::null(),
+                        4096,
+                    )
+                },
+                0
+            );
+            unsafe { (File::from_raw_handle(read), File::from_raw_handle(write)) }
+        }
+        let (child_input, input) = pair();
+        let (output, child_output) = pair();
+        let (error, child_error) = pair();
+        (
+            HostIo {
+                input,
+                output,
+                error,
+            },
+            [child_input, child_output, child_error],
+        )
+    }
+    #[test]
+    fn cancellation_joins_blocked_pipe_workers_even_when_peers_stay_open() {
+        for _ in 0..20 {
+            let (io, peers) = pipes();
+            let pump = Pump::start(io, Job::new().unwrap()).unwrap();
+            pump.send(vec![b'x'; MAX_FRAME_BYTES]).unwrap();
+            let started = Instant::now();
+            loop {
+                let pending = pump.workers.iter().all(|worker| {
+                    let mut pending = 0;
+                    assert_ne!(
+                        unsafe {
+                            windows_sys::Win32::System::Threading::GetThreadIOPendingFlag(
+                                worker.thread.as_ref().unwrap().as_raw_handle(),
+                                &mut pending,
+                            )
+                        },
+                        0
+                    );
+                    pending != 0
+                });
+                if pending {
+                    break;
+                }
+                assert!(
+                    started.elapsed() < Duration::from_secs(5),
+                    "workers did not enter native IO"
+                );
+                std::thread::sleep(Duration::from_millis(2));
+            }
+            pump.send(b"{}".to_vec()).unwrap();
+            assert_eq!(
+                pump.send(b"{}".to_vec()).unwrap_err().code,
+                "EXTENSION_IO_INPUT_BACKPRESSURE"
+            );
+            let started = Instant::now();
+            pump.shutdown().unwrap();
+            assert!(started.elapsed() < Duration::from_secs(2));
+            // The peer handles remained open throughout shutdown. No process
+            // exit or peer EOF is available to mask broken IO cancellation.
+            drop(peers);
+        }
+    }
+    #[test]
+    fn output_overflow_and_stderr_flood_fail_without_waiting_for_receive_timeout() {
+        for stderr in [false, true] {
+            let (io, [input, mut output, mut error]) = pipes();
+            let pump = Pump::start(io, Job::new().unwrap()).unwrap();
+            let sender = std::thread::spawn(move || {
+                if stderr {
+                    let _ = error.write_all(&vec![b'x'; STDERR_BYTES + 8192]);
+                } else {
+                    output
+                        .write_all(&b"{}\n".repeat(QUEUED_FRAMES + 2))
+                        .unwrap();
+                }
+                (input, output, error)
+            });
+            let start = Instant::now();
+            while pump.check().is_ok() {
+                assert!(start.elapsed() < Duration::from_secs(5));
+                std::thread::sleep(Duration::from_millis(2));
+            }
+            let expected = if stderr {
+                "EXTENSION_STDERR_LIMIT_EXCEEDED"
+            } else {
+                "EXTENSION_IO_OUTPUT_BACKPRESSURE"
+            };
+            assert_eq!(pump.check().unwrap_err().code, expected);
+            let start = Instant::now();
+            assert_eq!(
+                pump.receive(Duration::from_secs(60)).unwrap_err().code,
+                expected
+            );
+            assert!(start.elapsed() < Duration::from_secs(1));
+            pump.shutdown().unwrap();
+            drop(sender.join().unwrap());
+        }
+    }
+}
diff --git a/frontend/src-tauri/src/extension_process.rs b/frontend/src-tauri/src/extension_process.rs
index eb2b61b..b3f318b 100644
--- a/frontend/src-tauri/src/extension_process.rs
+++ b/frontend/src-tauri/src/extension_process.rs
@@ -268,6 +268,15 @@ impl<'a> Suspended<'a> {
     }
 }
 impl Running<'_> {
+    #[cfg(feature = "desktop")]
+    pub fn start_io(
+        &self,
+        io: crate::extension_stdio::HostIo,
+    ) -> Result<crate::extension_io::Pump> {
+        self.check_authorization()?;
+        crate::extension_io::Pump::start(io, self.process.job.clone_for_deadline()?)
+    }
+
     pub fn check_authorization(&self) -> Result<()> {
         #[cfg(feature = "desktop")]
         if let Some(watch) = &self.revocation {
diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs
index 726f47f..e17e130 100644
--- a/frontend/src-tauri/src/lib.rs
+++ b/frontend/src-tauri/src/lib.rs
@@ -81,3 +81,6 @@ pub mod extension_file_broker;
 
 #[cfg(windows)]
 pub mod extension_stdio;
+
+#[cfg(all(windows, feature = "desktop"))]
+pub mod extension_io;
diff --git a/frontend/src-tauri/tests/fixtures/sandbox_network_probe.rs b/frontend/src-tauri/tests/fixtures/sandbox_network_probe.rs
index fb2003b..2eb8fa9 100644
--- a/frontend/src-tauri/tests/fixtures/sandbox_network_probe.rs
+++ b/frontend/src-tauri/tests/fixtures/sandbox_network_probe.rs
@@ -3,6 +3,14 @@ use std::net::{SocketAddr, TcpStream, UdpSocket};
 use std::time::Duration;
 fn main() {
     let args: Vec<_> = std::env::args().collect();
+    if args.get(1).is_some_and(|s| s == "stderr_flood" || s == "stdout_flood") {
+        use std::io::Write;
+        let _child = std::process::Command::new(std::env::current_exe().unwrap()).arg("wait").spawn().unwrap();
+        if args[1] == "stderr_flood" { std::io::stderr().write_all(&vec![b'x'; 2 * 1024 * 1024]).unwrap(); }
+        else { std::io::stdout().write_all(&vec![b'x'; 2 * 1024 * 1024 + 1]).unwrap(); }
+        std::thread::sleep(Duration::from_secs(120));
+        return;
+    }
     if args.get(1).is_some_and(|s| s == "file_rpc") {
         use std::io::{Read, Write};
         #[link(name = "kernel32")]