diff --git a/src/engine.rs b/src/engine.rs
index 057fc4f..621660e 100644
--- a/src/engine.rs
+++ b/src/engine.rs
@@ -4,6 +4,7 @@ mod reading;
 mod selection;
 
 use std::{
+    collections::VecDeque,
     fs::File,
     path::PathBuf,
     sync::{
@@ -312,18 +313,24 @@ impl Engine {
         self.app.as_mut().context("The review has no viewer state")
     }
 
-    /// Clear the host wake token before calling this. Each drain is bounded so input can run.
+    /// Clear the host wake token before calling this. Each drain is bounded so input can run:
+    /// the input worker queues a bounded number of files, and this takes all of them at once.
     pub fn poll(&mut self) -> Result<()> {
         if self.input.is_none() {
             return Ok(());
         }
         if (self.loading || self.full_sources) && self.error.is_none() {
-            for _ in 0..64 {
-                let Some(input) = &self.input else {
-                    break;
-                };
-                match input.try_reply() {
-                    Ok(Ok(input::Message::File(file))) => {
+            let replies = match self.input.as_ref().map(input::Worker::try_replies) {
+                Some(Ok(replies)) => replies,
+                Some(Err(mpsc::TryRecvError::Disconnected)) => {
+                    self.fail_input("The input worker stopped before completing its work.".into());
+                    VecDeque::new()
+                }
+                Some(Err(mpsc::TryRecvError::Empty)) | None => VecDeque::new(),
+            };
+            for reply in replies {
+                match reply {
+                    Ok(input::Message::File(file)) => {
                         let app = self.app()?;
                         let exposes_more = app
                             .visible
@@ -337,7 +344,7 @@ impl Engine {
                         self.dirty |= exposes_more;
                         self.progress = true;
                     }
-                    Ok(Ok(input::Message::Complete { files })) => {
+                    Ok(input::Message::Complete { files }) => {
                         crate::delivery_event!("indexed", 0, 0);
                         ensure!(
                             self.app()?.doc.files.len() == files,
@@ -345,9 +352,8 @@ impl Engine {
                         );
                         self.loading = false;
                         self.dirty = true;
-                        break;
                     }
-                    Ok(Ok(input::Message::Source { file, state })) => {
+                    Ok(input::Message::Source { file, state }) => {
                         self.source_failed |= matches!(state, SourceState::Unavailable(_));
                         self.app
                             .as_mut()
@@ -356,17 +362,10 @@ impl Engine {
                         self.request = true;
                         self.dirty = true;
                     }
-                    Ok(Err(error)) => {
+                    Err(error) => {
                         self.fail_input(format!("{error:#}"));
                         break;
                     }
-                    Err(mpsc::TryRecvError::Disconnected) => {
-                        self.fail_input(
-                            "The input worker stopped before completing its work.".into(),
-                        );
-                        break;
-                    }
-                    Err(mpsc::TryRecvError::Empty) => break,
                 }
             }
         }
diff --git a/src/input/worker.rs b/src/input/worker.rs
index 7a9e04a..ce02936 100644
--- a/src/input/worker.rs
+++ b/src/input/worker.rs
@@ -1,8 +1,10 @@
 use super::*;
 use std::{
+    collections::VecDeque,
     sync::{
+        Condvar, Mutex, MutexGuard,
         atomic::Ordering,
-        mpsc::{self, Receiver, SyncSender},
+        mpsc::{self, SyncSender},
     },
     thread::{self, JoinHandle},
     time::Duration,
@@ -14,11 +16,194 @@ pub enum Message {
     Source { file: usize, state: SourceState },
 }
 
+/// Published files waiting for the screen are bounded by count and index bytes. A full
+/// queue pauses ingestion until the screen drains it; it never delays keyboard input.
+/// The count also bounds the screen's per-turn drain.
+const QUEUED_FILES: usize = 1024;
+const QUEUED_FILE_BYTES: usize = 8 << 20;
+/// Prepared sources retain both complete sides, so only two may wait.
+const QUEUED_SOURCES: usize = 2;
+
+#[derive(Default)]
+struct Queue {
+    replies: VecDeque<Result<Message>>,
+    files: usize,
+    file_bytes: usize,
+    sources: usize,
+    /// The worker was closed; a waiting producer must stop.
+    closed: bool,
+    /// The producer exited; nothing more will arrive.
+    finished: bool,
+}
+
+#[derive(Clone, Copy)]
+enum Cost {
+    File(usize),
+    Source,
+    Final,
+}
+
+impl Cost {
+    fn of(reply: &Result<Message>) -> Self {
+        match reply {
+            Ok(Message::File(file)) => Self::File(
+                size_of::<FileIndex>()
+                    + file.source_index_bytes()
+                    + file.projection_bytes()
+                    + file.metadata_bytes(),
+            ),
+            Ok(Message::Source { .. }) => Self::Source,
+            Ok(Message::Complete { .. }) | Err(_) => Self::Final,
+        }
+    }
+}
+
+impl Queue {
+    fn admits(&self, cost: Cost) -> bool {
+        match cost {
+            // An oversized file still enters an empty queue; it is accounted for in peak memory.
+            Cost::File(bytes) => {
+                self.files == 0
+                    || (self.files < QUEUED_FILES
+                        && self.file_bytes.saturating_add(bytes) <= QUEUED_FILE_BYTES)
+            }
+            Cost::Source => self.sources < QUEUED_SOURCES,
+            // At most one completion and one error are ever published.
+            Cost::Final => true,
+        }
+    }
+
+    fn add(&mut self, cost: Cost) {
+        match cost {
+            Cost::File(bytes) => {
+                self.files += 1;
+                self.file_bytes = self.file_bytes.saturating_add(bytes);
+            }
+            Cost::Source => self.sources += 1,
+            Cost::Final => {}
+        }
+    }
+
+    fn remove(&mut self, cost: Cost) {
+        match cost {
+            Cost::File(bytes) => {
+                self.files = self.files.saturating_sub(1);
+                self.file_bytes = self.file_bytes.saturating_sub(bytes);
+            }
+            Cost::Source => self.sources = self.sources.saturating_sub(1),
+            Cost::Final => {}
+        }
+    }
+}
+
+/// Single-producer handoff from the input thread to the screen. The producer blocks on a
+/// condition variable when the queue is full; the screen takes everything queued at once.
+#[derive(Default)]
+struct Mailbox {
+    queue: Mutex<Queue>,
+    space: Condvar,
+}
+
+impl Mailbox {
+    fn lock(&self) -> Result<MutexGuard<'_, Queue>> {
+        self.queue
+            .lock()
+            .map_err(|_| anyhow::anyhow!("The input queue lock failed"))
+    }
+
+    fn deliver(&self, reply: Result<Message>, stop: &Cancellation) -> Result<()> {
+        let cost = Cost::of(&reply);
+        let mut queue = self.lock()?;
+        loop {
+            if queue.closed {
+                return Err(Cancelled.into());
+            }
+            stop.check()?;
+            if queue.admits(cost) {
+                break;
+            }
+            queue = self
+                .space
+                .wait(queue)
+                .map_err(|_| anyhow::anyhow!("The input queue lock failed"))?;
+        }
+        queue.add(cost);
+        queue.replies.push_back(reply);
+        Ok(())
+    }
+
+    fn finish(&self) {
+        if let Ok(mut queue) = self.queue.lock() {
+            queue.finished = true;
+        }
+    }
+
+    fn close(&self) {
+        let replies = self.queue.lock().map(|mut queue| {
+            queue.closed = true;
+            queue.files = 0;
+            queue.file_bytes = 0;
+            queue.sources = 0;
+            std::mem::take(&mut queue.replies)
+        });
+        self.space.notify_all();
+        // Release queued files and sources outside the lock.
+        drop(replies);
+    }
+
+    fn try_reply(&self) -> std::result::Result<Result<Message>, mpsc::TryRecvError> {
+        let mut queue = self
+            .queue
+            .lock()
+            .map_err(|_| mpsc::TryRecvError::Disconnected)?;
+        match queue.replies.pop_front() {
+            Some(reply) => {
+                queue.remove(Cost::of(&reply));
+                drop(queue);
+                self.space.notify_one();
+                Ok(reply)
+            }
+            None if queue.finished => Err(mpsc::TryRecvError::Disconnected),
+            None => Err(mpsc::TryRecvError::Empty),
+        }
+    }
+
+    fn try_replies(&self) -> std::result::Result<VecDeque<Result<Message>>, mpsc::TryRecvError> {
+        let mut queue = self
+            .queue
+            .lock()
+            .map_err(|_| mpsc::TryRecvError::Disconnected)?;
+        if queue.replies.is_empty() {
+            return Err(if queue.finished {
+                mpsc::TryRecvError::Disconnected
+            } else {
+                mpsc::TryRecvError::Empty
+            });
+        }
+        queue.files = 0;
+        queue.file_bytes = 0;
+        queue.sources = 0;
+        let replies = std::mem::take(&mut queue.replies);
+        drop(queue);
+        self.space.notify_one();
+        Ok(replies)
+    }
+}
+
+/// Marks the producer finished before the exit notification, so the screen observes disconnection.
+struct Finish(Arc<Mailbox>);
+
+impl Drop for Finish {
+    fn drop(&mut self) {
+        self.0.finish();
+    }
+}
+
 pub struct Worker {
     cancel: Cancellation,
     destination: Arc<AtomicUsize>,
     wake: SyncSender<()>,
-    replies: Receiver<Result<Message>>,
+    replies: Arc<Mailbox>,
     join: Option<JoinHandle<Result<()>>>,
 }
 
@@ -40,25 +225,17 @@ impl Worker {
         let cancel = Cancellation::default();
         let destination = Arc::new(AtomicUsize::new(usize::MAX));
         let (wake, requests) = mpsc::sync_channel(1);
-        let (send, replies) = mpsc::sync_channel(2);
+        let replies = Arc::new(Mailbox::default());
+        let mailbox = replies.clone();
         let stop = cancel.clone();
         let wanted = destination.clone();
         let join = thread::spawn(move || {
             let _notify_exit = crate::wake::NotifyOnExit(notify.clone());
-            let send = send;
-            let deliver = |mut value| -> Result<()> {
-                loop {
-                    stop.check()?;
-                    match send.try_send(value) {
-                        Ok(()) => {
-                            let _ = notify.try_send(());
-                            return Ok(());
-                        }
-                        Err(mpsc::TrySendError::Disconnected(_)) => return Err(Cancelled.into()),
-                        Err(mpsc::TrySendError::Full(v)) => value = v,
-                    }
-                    thread::sleep(Duration::from_millis(5));
-                }
+            let _finish = Finish(mailbox.clone());
+            let deliver = |value| -> Result<()> {
+                mailbox.deliver(value, &stop)?;
+                let _ = notify.try_send(());
+                Ok(())
             };
             let result = (|| -> Result<()> {
                 let input = match input {
@@ -120,13 +297,23 @@ impl Worker {
             let _ = self.wake.try_send(());
         }
     }
+    /// Take the oldest reply. `Disconnected` means the worker exited and nothing remains.
     pub fn try_reply(&self) -> std::result::Result<Result<Message>, mpsc::TryRecvError> {
-        self.replies.try_recv()
+        self.replies.try_reply()
+    }
+
+    /// Take every queued reply in order, releasing a producer that waits for space.
+    /// `Disconnected` means the worker exited and nothing remains.
+    pub fn try_replies(
+        &self,
+    ) -> std::result::Result<VecDeque<Result<Message>>, mpsc::TryRecvError> {
+        self.replies.try_replies()
     }
 
     /// Cancel and join. Input errors arrive through replies; this reports cleanup failure or panic.
     pub fn close(&mut self) -> Result<()> {
         self.cancel.stop();
+        self.replies.close();
         let _ = self.wake.try_send(());
         let result = self.join.take().map_or(Ok(()), |join| {
             join.join()
@@ -136,7 +323,7 @@ impl Worker {
                     result => result,
                 })
         });
-        while self.replies.try_recv().is_ok() {}
+        self.replies.close();
         result
     }
 }
@@ -146,3 +333,102 @@ impl Drop for Worker {
         let _ = self.close();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn source(file: usize) -> Result<Message> {
+        Ok(Message::Source {
+            file,
+            state: SourceState::Pending,
+        })
+    }
+
+    #[test]
+    fn queued_files_are_bounded_by_count_and_bytes_but_one_always_fits() {
+        let mut queue = Queue::default();
+        assert!(queue.admits(Cost::File(QUEUED_FILE_BYTES * 2)));
+        for _ in 0..QUEUED_FILES {
+            assert!(queue.admits(Cost::File(1)));
+            queue.add(Cost::File(1));
+        }
+        assert!(!queue.admits(Cost::File(1)));
+        assert!(queue.admits(Cost::Final));
+        queue.remove(Cost::File(1));
+        assert!(queue.admits(Cost::File(1)));
+
+        let mut queue = Queue::default();
+        queue.add(Cost::File(QUEUED_FILE_BYTES - 10));
+        assert!(queue.admits(Cost::File(10)));
+        assert!(!queue.admits(Cost::File(11)));
+        assert!(queue.admits(Cost::Source));
+    }
+
+    #[test]
+    fn a_full_queue_blocks_the_producer_until_the_screen_drains_everything() {
+        let mailbox = Arc::new(Mailbox::default());
+        let stop = Cancellation::default();
+        for file in 0..QUEUED_SOURCES {
+            mailbox.deliver(source(file), &stop).unwrap();
+        }
+        let (done, delivered) = mpsc::sync_channel(1);
+        let producer = {
+            let mailbox = mailbox.clone();
+            let stop = stop.clone();
+            thread::spawn(move || {
+                let result = mailbox.deliver(source(QUEUED_SOURCES), &stop);
+                done.send(()).unwrap();
+                result
+            })
+        };
+        assert_eq!(
+            delivered.recv_timeout(Duration::from_millis(50)),
+            Err(mpsc::RecvTimeoutError::Timeout)
+        );
+        let replies = mailbox.try_replies().unwrap();
+        assert_eq!(replies.len(), QUEUED_SOURCES);
+        delivered.recv_timeout(Duration::from_secs(10)).unwrap();
+        producer.join().unwrap().unwrap();
+        let replies = mailbox.try_replies().unwrap();
+        assert!(matches!(
+            replies.front(),
+            Some(Ok(Message::Source { file, .. })) if *file == QUEUED_SOURCES
+        ));
+        assert!(matches!(
+            mailbox.try_replies(),
+            Err(mpsc::TryRecvError::Empty)
+        ));
+        mailbox.finish();
+        assert!(matches!(
+            mailbox.try_reply(),
+            Err(mpsc::TryRecvError::Disconnected)
+        ));
+    }
+
+    #[test]
+    fn closing_releases_a_blocked_producer_and_drops_queued_replies() {
+        let mailbox = Arc::new(Mailbox::default());
+        let stop = Cancellation::default();
+        for file in 0..QUEUED_SOURCES {
+            mailbox.deliver(source(file), &stop).unwrap();
+        }
+        let producer = {
+            let mailbox = mailbox.clone();
+            let stop = stop.clone();
+            thread::spawn(move || mailbox.deliver(source(QUEUED_SOURCES), &stop))
+        };
+        thread::sleep(Duration::from_millis(20));
+        stop.stop();
+        mailbox.close();
+        let error = producer.join().unwrap().unwrap_err();
+        assert!(error.is::<Cancelled>());
+        assert!(mailbox.lock().unwrap().replies.is_empty());
+        assert!(
+            mailbox
+                .deliver(source(0), &Cancellation::default())
+                .unwrap_err()
+                .is::<Cancelled>()
+        );
+    }
+}
diff --git a/tests/input_worker.rs b/tests/input_worker.rs
index b73e619..7b87e78 100644
--- a/tests/input_worker.rs
+++ b/tests/input_worker.rs
@@ -52,17 +52,17 @@ fn start(repo: &std::path::Path) -> (Worker, mpsc::Receiver<()>) {
 }
 
 #[test]
-fn closing_a_full_patch_queue_releases_the_producer_but_keeps_published_source_valid() {
+fn closing_queued_patch_files_releases_the_producer_but_keeps_published_source_valid() {
     let repo = readiness::repository(false);
     let (worker, notified) = start(repo.path());
-    for _ in 0..2 {
+    // All twelve files and the completion are queued without the screen draining any.
+    for _ in 0..13 {
         notified.recv_timeout(Duration::from_secs(10)).unwrap();
     }
     let Message::File(file) = worker.try_reply().unwrap().unwrap() else {
         panic!("expected the first file");
     };
     assert_eq!(file.id, 0);
-    notified.recv_timeout(Duration::from_secs(10)).unwrap();
     assert_eq!(
         notified.recv_timeout(Duration::from_millis(20)),
         Err(mpsc::RecvTimeoutError::Timeout)
@@ -152,7 +152,7 @@ fn closing_patch_readers_releases_blocked_input_and_queues_before_a_new_review()
         unistd::pipe,
     };
     use std::{fs::File, io::Write, os::fd::AsRawFd};
-    for mode in ["idle", "partial", "full-queue"] {
+    for mode in ["idle", "partial", "queued"] {
         let (read_end, write_end) = pipe().unwrap();
         let observed = File::from(read_end);
         let original = fcntl(observed.as_raw_fd(), FcntlArg::F_GETFL).unwrap();
@@ -166,17 +166,21 @@ fn closing_patch_readers_releases_blocked_input_and_queues_before_a_new_review()
             assert!(Instant::now() < deadline, "patch reader did not start");
             thread::sleep(Duration::from_millis(1));
         }
-        let retained = if mode == "full-queue" {
+        let retained = if mode == "queued" {
             for n in 0..8 {
                 writeln!(producer, "diff --git a/{n}.txt b/{n}.txt\n--- a/{n}.txt\n+++ b/{n}.txt\n@@ -1 +1 @@\n-old\n+new").unwrap();
             }
-            for _ in 0..2 {
+            // The eighth file stays pending until its successor's header or EOF frames it.
+            for _ in 0..7 {
                 notified.recv_timeout(Duration::from_secs(5)).unwrap();
             }
             let Message::File(file) = worker.try_reply().unwrap().unwrap() else {
                 panic!("expected file");
             };
-            notified.recv_timeout(Duration::from_secs(5)).unwrap();
+            assert!(matches!(
+                notified.recv_timeout(Duration::from_millis(20)),
+                Err(mpsc::RecvTimeoutError::Timeout)
+            ));
             Some(file)
         } else {
             if mode == "partial" {