diff --git a/codex-rs/message-history/src/batch.rs b/codex-rs/message-history/src/batch.rs
new file mode 100644
index 000000000000..6adf7e73d3b5
--- /dev/null
+++ b/codex-rs/message-history/src/batch.rs
@@ -0,0 +1,412 @@
+use std::collections::VecDeque;
+use std::fs::File;
+use std::fs::OpenOptions;
+use std::io::BufRead;
+use std::io::BufReader;
+use std::io::Read;
+use std::io::Seek;
+use std::io::SeekFrom;
+use std::time::SystemTime;
+
+use super::HISTORY_READ_BUFFER_SIZE;
+use super::HistoryConfig;
+use super::HistoryEntry;
+use super::MAX_RETRIES;
+use super::RETRY_SLEEP;
+use super::history_filepath;
+use super::log_identity;
+
+const MAX_BATCH_ROWS: usize = 128;
+const MAX_BATCH_BYTES: usize = 64 * 1024;
+
+/// Position of the newest record to include in a bounded history lookup.
+///
+/// The initial cursor identifies only an absolute row offset. Continuation cursors also retain a
+/// byte position so older batches can scan backward from the previous batch instead of rescanning
+/// the history prefix.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct HistoryBatchCursor {
+    end_offset: usize,
+    byte_anchor: Option<HistoryByteAnchor>,
+}
+
+impl HistoryBatchCursor {
+    /// Creates an initial cursor ending at the given absolute history offset.
+    pub fn new(end_offset: usize) -> Self {
+        Self {
+            end_offset,
+            byte_anchor: None,
+        }
+    }
+
+    /// Returns the absolute history offset covered first by this cursor.
+    pub fn end_offset(self) -> usize {
+        self.end_offset
+    }
+}
+
+/// Validated row boundary used to continue scanning one unchanged file revision.
+///
+/// Byte positions and file lengths use `u64` to match filesystem and seek APIs, while history row
+/// offsets use `usize` to match collection indices.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+struct HistoryByteAnchor {
+    position: u64,
+    revision: HistoryFileRevision,
+}
+
+/// File metadata that must remain unchanged before a byte position can be reused.
+///
+/// Byte positions are only reused for uncapped histories, which Codex writes append-only. Capped
+/// histories can be rewritten in place when they are trimmed, so their cursors always fall back to
+/// an offset scan. Filesystems without a modification time also fall back to an offset scan.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+struct HistoryFileRevision {
+    len: u64,
+    modified: Option<SystemTime>,
+}
+
+/// One absolute history offset covered by a bounded lookup.
+///
+/// Malformed records retain their offset with `entry` set to `None`, allowing callers to continue
+/// searching older valid records without changing offset semantics.
+#[derive(Clone, Debug, PartialEq)]
+pub struct HistoryBatchEntry {
+    /// Zero-based position in the history file, counted from the oldest record.
+    pub offset: usize,
+    /// Parsed record, or `None` when the row at `offset` is malformed.
+    pub entry: Option<HistoryEntry>,
+}
+
+/// A bounded newest-first suffix ending at a requested absolute history offset.
+///
+/// `next_older_cursor` identifies the next position a caller should request after exhausting
+/// `entries`.
+#[derive(Clone, Debug, Default, PartialEq)]
+pub struct HistoryBatch {
+    /// Covered records in newest-to-oldest order.
+    pub entries: Vec<HistoryBatchEntry>,
+    /// Next position to request after exhausting `entries`.
+    pub next_older_cursor: Option<HistoryBatchCursor>,
+}
+
+struct RawHistoryBatchEntry {
+    offset: usize,
+    byte_position: u64,
+    byte_len: usize,
+    /// Oversized rows remain unbuffered until the scan determines they belong in the result.
+    bytes: Option<Vec<u8>>,
+}
+
+/// Look up a bounded batch of history records ending at `cursor`.
+///
+/// The file is opened, identity-checked, and shared-locked once. Records are counted from the
+/// oldest offset on the initial lookup. Continuation lookups scan backward from the byte position
+/// returned with the previous batch. The result retains at most 128 rows and 64 KiB of raw JSONL,
+/// except that one oversized newest row is returned alone so callers always make progress.
+///
+/// # Errors
+///
+/// Returns an I/O error when the history file cannot be opened, inspected, locked, or read.
+pub fn lookup_batch(
+    log_id: u64,
+    cursor: HistoryBatchCursor,
+    config: &HistoryConfig,
+) -> std::io::Result<HistoryBatch> {
+    let path = history_filepath(config);
+    let mut file = OpenOptions::new().read(true).open(path)?;
+    let current_log_id = log_identity(&file.metadata()?).unwrap_or(0);
+    if log_id != 0 && current_log_id != log_id {
+        return Ok(HistoryBatch::default());
+    }
+
+    for _ in 0..MAX_RETRIES {
+        match file.try_lock_shared() {
+            Ok(()) => return scan_batch(&mut file, cursor, config),
+            Err(std::fs::TryLockError::WouldBlock) => std::thread::sleep(RETRY_SLEEP),
+            Err(error) => return Err(error.into()),
+        }
+    }
+
+    Err(std::io::Error::new(
+        std::io::ErrorKind::WouldBlock,
+        "could not acquire shared history lock after multiple attempts",
+    ))
+}
+
+/// Selects the anchored backward scan only for an unchanged, uncapped history file.
+///
+/// Capped histories always use the forward scan because trimming rewrites them in place and a
+/// same-size rewrite may not be distinguishable from metadata on filesystems with coarse
+/// modification times. Falling back preserves absolute row semantics at the cost of rescanning
+/// that request from the beginning.
+fn scan_batch(
+    file: &mut File,
+    cursor: HistoryBatchCursor,
+    config: &HistoryConfig,
+) -> std::io::Result<HistoryBatch> {
+    let metadata = file.metadata()?;
+    let revision = HistoryFileRevision {
+        len: metadata.len(),
+        modified: metadata.modified().ok(),
+    };
+    if config.max_bytes.is_none()
+        && let Some(anchor) = cursor.byte_anchor
+        && anchor.revision == revision
+    {
+        return scan_batch_backward(file, cursor.end_offset, anchor.position, revision);
+    }
+
+    file.seek(SeekFrom::Start(0))?;
+    let mut batch = scan_batch_forward(file, cursor.end_offset, revision)?;
+    if config.max_bytes.is_some()
+        && let Some(next_older_cursor) = &mut batch.next_older_cursor
+    {
+        next_older_cursor.byte_anchor = None;
+    }
+    Ok(batch)
+}
+
+/// Streams from byte zero through `end_offset`, retaining only the bounded newest suffix.
+///
+/// This path establishes byte positions for later continuation cursors and is also the safe
+/// fallback when an existing cursor belongs to an older file revision. Oversized rows are tracked
+/// by position and length, then materialized only if they remain in the returned suffix.
+fn scan_batch_forward(
+    file: &mut File,
+    end_offset: usize,
+    revision: HistoryFileRevision,
+) -> std::io::Result<HistoryBatch> {
+    let mut suffix = VecDeque::new();
+    let mut suffix_bytes = 0usize;
+    {
+        let mut byte_position = 0u64;
+        let mut reader = BufReader::with_capacity(HISTORY_READ_BUFFER_SIZE, &mut *file);
+
+        'rows: for offset in 0..=end_offset {
+            let mut byte_len = 0usize;
+            let mut bytes = Some(Vec::new());
+            loop {
+                let buffer = reader.fill_buf()?;
+                if buffer.is_empty() {
+                    if byte_len == 0 {
+                        break 'rows;
+                    }
+                    break;
+                }
+
+                let newline = buffer.iter().position(|byte| *byte == b'\n');
+                let consumed = newline.map_or(buffer.len(), |index| index + 1);
+                byte_len = byte_len.saturating_add(consumed);
+                if let Some(buffered) = bytes.as_mut() {
+                    if byte_len <= MAX_BATCH_BYTES {
+                        buffered.extend_from_slice(&buffer[..consumed]);
+                    } else {
+                        bytes = None;
+                    }
+                }
+                reader.consume(consumed);
+                if newline.is_some() {
+                    break;
+                }
+            }
+
+            retain_row(
+                &mut suffix,
+                &mut suffix_bytes,
+                RawHistoryBatchEntry {
+                    offset,
+                    byte_position,
+                    byte_len,
+                    bytes,
+                },
+            );
+            byte_position += byte_len as u64;
+        }
+    }
+
+    finish_materialized_batch(file, suffix.into_iter().rev().collect(), revision)
+}
+
+/// Reads complete rows backward from a validated exclusive byte boundary.
+///
+/// `end_byte_position` must be the start of the row immediately newer than `end_offset`. Scanning
+/// in reverse lets each continuation touch only its own rows while preserving absolute offsets.
+fn scan_batch_backward(
+    file: &mut File,
+    end_offset: usize,
+    end_byte_position: u64,
+    revision: HistoryFileRevision,
+) -> std::io::Result<HistoryBatch> {
+    let mut entries = Vec::new();
+    let mut entries_bytes = 0usize;
+    let mut reversed_row = Some(Vec::new());
+    let mut row_byte_len = 0usize;
+    let mut read_buffer = [0u8; HISTORY_READ_BUFFER_SIZE];
+    let mut read_end = end_byte_position;
+    let mut offset = end_offset;
+
+    while read_end > 0 {
+        let read_start = read_end.saturating_sub(HISTORY_READ_BUFFER_SIZE as u64);
+        let read_len = usize::try_from(read_end - read_start).unwrap_or(HISTORY_READ_BUFFER_SIZE);
+        file.seek(SeekFrom::Start(read_start))?;
+        file.read_exact(&mut read_buffer[..read_len])?;
+
+        for index in (0..read_len).rev() {
+            let byte = read_buffer[index];
+            if byte == b'\n' && row_byte_len > 0 {
+                if let Some(bytes) = reversed_row.as_mut() {
+                    bytes.reverse();
+                }
+                let raw = RawHistoryBatchEntry {
+                    offset,
+                    byte_position: read_start + index as u64 + 1,
+                    byte_len: row_byte_len,
+                    bytes: reversed_row.take(),
+                };
+                if !retain_newest_row(&mut entries, &mut entries_bytes, raw) {
+                    return finish_materialized_batch(file, entries, revision);
+                }
+                let Some(next_offset) = offset.checked_sub(1) else {
+                    return finish_materialized_batch(file, entries, revision);
+                };
+                offset = next_offset;
+                reversed_row = Some(vec![b'\n']);
+                row_byte_len = 1;
+            } else {
+                row_byte_len = row_byte_len.saturating_add(1);
+                if let Some(bytes) = reversed_row.as_mut() {
+                    if row_byte_len <= MAX_BATCH_BYTES {
+                        bytes.push(byte);
+                    } else {
+                        reversed_row = None;
+                    }
+                }
+            }
+        }
+        read_end = read_start;
+    }
+
+    if row_byte_len > 0 {
+        if let Some(bytes) = reversed_row.as_mut() {
+            bytes.reverse();
+        }
+        retain_newest_row(
+            &mut entries,
+            &mut entries_bytes,
+            RawHistoryBatchEntry {
+                offset,
+                byte_position: 0,
+                byte_len: row_byte_len,
+                bytes: reversed_row,
+            },
+        );
+    }
+    finish_materialized_batch(file, entries, revision)
+}
+
+fn finish_materialized_batch(
+    file: &mut File,
+    mut entries: Vec<RawHistoryBatchEntry>,
+    revision: HistoryFileRevision,
+) -> std::io::Result<HistoryBatch> {
+    for entry in &mut entries {
+        if entry.bytes.is_none() {
+            file.seek(SeekFrom::Start(entry.byte_position))?;
+            let mut bytes = vec![0; entry.byte_len];
+            file.read_exact(&mut bytes)?;
+            entry.bytes = Some(bytes);
+        }
+    }
+    finish_batch(entries, revision)
+}
+
+/// Retains the newest suffix seen by a forward scan under both row and byte caps.
+///
+/// A single oversized row replaces the suffix so the newest requested record is always returned
+/// and callers can continue to an older cursor.
+fn retain_row(
+    suffix: &mut VecDeque<RawHistoryBatchEntry>,
+    suffix_bytes: &mut usize,
+    entry: RawHistoryBatchEntry,
+) {
+    let row_bytes = entry.byte_len;
+    if row_bytes > MAX_BATCH_BYTES {
+        suffix.clear();
+        *suffix_bytes = row_bytes;
+        suffix.push_back(entry);
+        return;
+    }
+
+    *suffix_bytes += row_bytes;
+    suffix.push_back(entry);
+    while suffix.len() > MAX_BATCH_ROWS || *suffix_bytes > MAX_BATCH_BYTES {
+        if let Some(removed) = suffix.pop_front() {
+            *suffix_bytes -= removed.byte_len;
+        }
+    }
+}
+
+/// Appends one newest-to-oldest row and reports whether the backward scan should continue.
+///
+/// Returning `false` means the batch is complete. An oversized first row is retained alone;
+/// otherwise the row that would exceed a cap is left for the next batch.
+fn retain_newest_row(
+    entries: &mut Vec<RawHistoryBatchEntry>,
+    entries_bytes: &mut usize,
+    entry: RawHistoryBatchEntry,
+) -> bool {
+    let row_bytes = entry.byte_len;
+    if entries.is_empty() && row_bytes > MAX_BATCH_BYTES {
+        entries.push(entry);
+        return false;
+    }
+    if entries.len() == MAX_BATCH_ROWS || entries_bytes.saturating_add(row_bytes) > MAX_BATCH_BYTES
+    {
+        return false;
+    }
+    *entries_bytes += row_bytes;
+    entries.push(entry);
+    true
+}
+
+/// Parses newest-first rows and anchors the continuation at the oldest retained row's start.
+fn finish_batch(
+    entries: Vec<RawHistoryBatchEntry>,
+    revision: HistoryFileRevision,
+) -> std::io::Result<HistoryBatch> {
+    let next_older_cursor = entries.last().and_then(|entry| {
+        entry
+            .offset
+            .checked_sub(1)
+            .map(|end_offset| HistoryBatchCursor {
+                end_offset,
+                byte_anchor: revision.modified.map(|_| HistoryByteAnchor {
+                    position: entry.byte_position,
+                    revision,
+                }),
+            })
+    });
+    let entries = entries
+        .into_iter()
+        .map(|raw| {
+            let bytes = raw.bytes.ok_or_else(|| {
+                std::io::Error::other("retained history row was not materialized")
+            })?;
+            Ok(HistoryBatchEntry {
+                offset: raw.offset,
+                entry: try_parse_entry(&bytes),
+            })
+        })
+        .collect::<std::io::Result<Vec<_>>>()?;
+    Ok(HistoryBatch {
+        entries,
+        next_older_cursor,
+    })
+}
+
+fn try_parse_entry(raw: &[u8]) -> Option<HistoryEntry> {
+    let raw = raw.strip_suffix(b"\n").unwrap_or(raw);
+    let raw = raw.strip_suffix(b"\r").unwrap_or(raw);
+    serde_json::from_slice(raw).ok()
+}
diff --git a/codex-rs/message-history/src/batch_tests.rs b/codex-rs/message-history/src/batch_tests.rs
new file mode 100644
index 000000000000..c0e4c10d1145
--- /dev/null
+++ b/codex-rs/message-history/src/batch_tests.rs
@@ -0,0 +1,359 @@
+use std::fs::File;
+use std::io::Write;
+
+use codex_config::types::History;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+
+use super::*;
+
+fn entry(offset: usize, text: impl Into<String>) -> HistoryEntry {
+    HistoryEntry {
+        session_id: "session".to_string(),
+        ts: offset as u64,
+        text: text.into(),
+    }
+}
+
+fn write_entries(home: &TempDir, entries: &[HistoryEntry]) -> HistoryConfig {
+    let mut file = File::create(home.path().join(HISTORY_FILENAME)).expect("create history");
+    for entry in entries {
+        serde_json::to_writer(&mut file, entry).expect("serialize entry");
+        writeln!(file).expect("write entry");
+    }
+    HistoryConfig::new(home.path(), &History::default())
+}
+
+async fn batch_for(entries: &[HistoryEntry], end_offset: usize) -> (TempDir, HistoryBatch) {
+    let home = TempDir::new().expect("temp dir");
+    let config = write_entries(&home, entries);
+    let (log_id, _) = history_metadata(&config).await;
+    let batch = lookup_batch(log_id, HistoryBatchCursor::new(end_offset), &config)
+        .expect("read history batch");
+    (home, batch)
+}
+
+#[tokio::test]
+async fn search_batch_returns_bounded_newest_first_absolute_offsets() {
+    let entries: Vec<_> = (0..400)
+        .map(|offset| entry(offset, format!("row {offset}")))
+        .collect();
+    let (home, batch) = batch_for(&entries, /*end_offset*/ 399).await;
+
+    assert_eq!(batch.entries.len(), 128);
+    assert_eq!(batch.entries.first().map(|entry| entry.offset), Some(399));
+    assert_eq!(batch.entries.last().map(|entry| entry.offset), Some(272));
+    let next_cursor = batch.next_older_cursor.expect("older cursor");
+    assert_eq!(next_cursor.end_offset(), 271);
+    assert_eq!(batch.entries[0].entry, Some(entries[399].clone()));
+
+    let config = HistoryConfig::new(home.path(), &History::default());
+    let (log_id, _) = history_metadata(&config).await;
+    let mut offsets: Vec<_> = batch.entries.iter().map(|entry| entry.offset).collect();
+    let mut cursor = Some(next_cursor);
+    while let Some(next_cursor) = cursor {
+        let older = lookup_batch(log_id, next_cursor, &config).expect("read older history batch");
+        offsets.extend(older.entries.iter().map(|entry| entry.offset));
+        cursor = older.next_older_cursor;
+    }
+    assert_eq!(offsets, (0..400).rev().collect::<Vec<_>>());
+}
+
+#[tokio::test]
+async fn search_batch_invalidates_byte_cursor_after_in_place_rewrite() {
+    let original: Vec<_> = (0..400)
+        .map(|offset| entry(offset, format!("old row {offset}")))
+        .collect();
+    let (home, batch) = batch_for(&original, /*end_offset*/ 399).await;
+    let cursor = batch.next_older_cursor.expect("older cursor");
+    let config = HistoryConfig::new(home.path(), &History::default());
+    let (log_id, _) = history_metadata(&config).await;
+    let original_len = std::fs::metadata(home.path().join(HISTORY_FILENAME))
+        .expect("history metadata")
+        .len();
+
+    let replacement: Vec<_> = (0..500)
+        .map(|offset| entry(offset, format!("replacement row {offset} with padding")))
+        .collect();
+    let replacement_config = write_entries(&home, &replacement);
+    let replacement_len = std::fs::metadata(home.path().join(HISTORY_FILENAME))
+        .expect("replacement history metadata")
+        .len();
+    assert!(replacement_len > original_len);
+    let (replacement_log_id, _) = history_metadata(&replacement_config).await;
+    assert_eq!(replacement_log_id, log_id);
+
+    let older =
+        lookup_batch(log_id, cursor, &replacement_config).expect("read rewritten history batch");
+    assert_eq!(older.entries.len(), 128);
+    assert_eq!(older.entries[0].offset, 271);
+    assert_eq!(older.entries[0].entry, Some(replacement[271].clone()));
+    assert_eq!(older.entries[127].offset, 144);
+    assert_eq!(older.entries[127].entry, Some(replacement[144].clone()));
+}
+
+#[tokio::test]
+async fn search_batch_rescans_capped_history_after_same_size_rewrite() {
+    let home = TempDir::new().expect("temp dir");
+    let original: Vec<_> = (0..400)
+        .map(|offset| entry(offset, "a".repeat(20)))
+        .collect();
+    write_entries(&home, &original);
+    let path = home.path().join(HISTORY_FILENAME);
+    let original_metadata = std::fs::metadata(&path).expect("history metadata");
+    let original_modified = original_metadata.modified().expect("history modified time");
+    let history = History {
+        max_bytes: Some(original_metadata.len() as usize),
+        ..History::default()
+    };
+    let config = HistoryConfig::new(home.path(), &history);
+    let (log_id, _) = history_metadata(&config).await;
+    let first = lookup_batch(log_id, HistoryBatchCursor::new(/*end_offset*/ 399), &config)
+        .expect("read history batch");
+    let cursor = first.next_older_cursor.expect("older cursor");
+
+    let mut replacement: Vec<_> = (0..400)
+        .map(|offset| entry(offset, "b".repeat(20)))
+        .collect();
+    replacement[0].text.push_str(&"b".repeat(10));
+    replacement[399].text.truncate(10);
+    write_entries(&home, &replacement);
+    let file = File::options()
+        .write(true)
+        .open(&path)
+        .expect("open replacement history");
+    file.set_times(std::fs::FileTimes::new().set_modified(original_modified))
+        .expect("restore history modified time");
+    let replacement_metadata = std::fs::metadata(&path).expect("replacement history metadata");
+    assert_eq!(replacement_metadata.len(), original_metadata.len());
+    assert_eq!(
+        replacement_metadata.modified().ok(),
+        Some(original_modified)
+    );
+
+    let older = lookup_batch(log_id, cursor, &config).expect("read rewritten history batch");
+    assert_eq!(
+        older,
+        HistoryBatch {
+            entries: (144..=271)
+                .rev()
+                .map(|offset| HistoryBatchEntry {
+                    offset,
+                    entry: Some(replacement[offset].clone()),
+                })
+                .collect(),
+            next_older_cursor: Some(HistoryBatchCursor::new(/*end_offset*/ 143)),
+        }
+    );
+}
+
+#[tokio::test]
+async fn search_batch_stitches_chunks_and_keeps_malformed_offsets() {
+    let home = TempDir::new().expect("temp dir");
+    let first = entry(/*offset*/ 0, "a".repeat(HISTORY_READ_BUFFER_SIZE + 17));
+    let third = entry(/*offset*/ 2, "third");
+    let contents = format!(
+        "{}\nnot-json\n{}\n",
+        serde_json::to_string(&first).expect("serialize first"),
+        serde_json::to_string(&third).expect("serialize third")
+    );
+    std::fs::write(home.path().join(HISTORY_FILENAME), contents).expect("write history");
+    let config = HistoryConfig::new(home.path(), &History::default());
+    let (log_id, _) = history_metadata(&config).await;
+
+    assert_eq!(
+        lookup_batch(log_id, HistoryBatchCursor::new(/*end_offset*/ 2), &config,)
+            .expect("read history batch"),
+        HistoryBatch {
+            entries: vec![
+                HistoryBatchEntry {
+                    offset: 2,
+                    entry: Some(third),
+                },
+                HistoryBatchEntry {
+                    offset: 1,
+                    entry: None,
+                },
+                HistoryBatchEntry {
+                    offset: 0,
+                    entry: Some(first),
+                },
+            ],
+            next_older_cursor: None,
+        }
+    );
+}
+
+#[tokio::test]
+async fn search_batch_preserves_identity_append_trim_and_short_file_semantics() {
+    let home = TempDir::new().expect("temp dir");
+    let initial = vec![entry(/*offset*/ 0, "zero"), entry(/*offset*/ 1, "one")];
+    let config = write_entries(&home, &initial);
+    let (log_id, _) = history_metadata(&config).await;
+    assert_eq!(
+        lookup_batch(
+            log_id.wrapping_add(1),
+            HistoryBatchCursor::new(/*end_offset*/ 1),
+            &config,
+        )
+        .expect("read history batch"),
+        HistoryBatch::default()
+    );
+
+    let mut file = std::fs::OpenOptions::new()
+        .append(true)
+        .open(home.path().join(HISTORY_FILENAME))
+        .expect("open history");
+    serde_json::to_writer(&mut file, &entry(/*offset*/ 2, "appended")).expect("serialize append");
+    writeln!(file).expect("append entry");
+    let batch = lookup_batch(log_id, HistoryBatchCursor::new(/*end_offset*/ 1), &config)
+        .expect("read history batch");
+    assert_eq!(
+        batch.entries,
+        vec![
+            HistoryBatchEntry {
+                offset: 1,
+                entry: Some(initial[1].clone()),
+            },
+            HistoryBatchEntry {
+                offset: 0,
+                entry: Some(initial[0].clone()),
+            },
+        ]
+    );
+
+    let newest = "c".repeat(200);
+    let history = History {
+        max_bytes: Some(newest.len() + 80),
+        ..History::default()
+    };
+    let trimmed_config = HistoryConfig::new(home.path(), &history);
+    append_entry(&newest, "session", &trimmed_config)
+        .await
+        .expect("append and trim");
+    let trimmed = lookup_batch(
+        log_id,
+        HistoryBatchCursor::new(/*end_offset*/ 20),
+        &trimmed_config,
+    )
+    .expect("read trimmed history batch");
+    assert_eq!(trimmed.entries.len(), 1);
+    assert_eq!(trimmed.entries[0].offset, 0);
+    assert_eq!(
+        trimmed.entries[0].entry.as_ref().map(|entry| &entry.text),
+        Some(&newest)
+    );
+    assert_eq!(trimmed.next_older_cursor, None);
+}
+
+#[tokio::test]
+async fn search_batch_enforces_byte_cap_and_oversized_row_progress() {
+    let entries: Vec<_> = (0..5)
+        .map(|offset| {
+            entry(
+                offset,
+                char::from(b'a' + offset as u8).to_string().repeat(20_000),
+            )
+        })
+        .collect();
+    let (_home, batch) = batch_for(&entries, /*end_offset*/ 4).await;
+    assert_eq!(batch.entries.len(), 3);
+    assert_eq!(batch.entries.first().map(|entry| entry.offset), Some(4));
+    assert_eq!(batch.entries.last().map(|entry| entry.offset), Some(2));
+    assert_eq!(
+        batch.next_older_cursor.expect("older cursor").end_offset(),
+        1
+    );
+
+    let entries = vec![
+        entry(/*offset*/ 0, "small"),
+        entry(/*offset*/ 1, "x".repeat(70_000)),
+    ];
+    let (home, oversized) = batch_for(&entries, /*end_offset*/ 1).await;
+    assert_eq!(oversized.entries.len(), 1);
+    assert_eq!(oversized.entries[0].entry, Some(entries[1].clone()));
+    let next_cursor = oversized.next_older_cursor.expect("older cursor");
+    assert_eq!(next_cursor.end_offset(), 0);
+    let config = HistoryConfig::new(home.path(), &History::default());
+    let (log_id, _) = history_metadata(&config).await;
+    let next = lookup_batch(log_id, next_cursor, &config).expect("read older history batch");
+    assert_eq!(next.entries[0].entry, Some(entries[0].clone()));
+    assert_eq!(next.next_older_cursor, None);
+
+    let entries = vec![
+        entry(/*offset*/ 0, "x".repeat(70_000)),
+        entry(/*offset*/ 1, "newest"),
+    ];
+    let (home, newest) = batch_for(&entries, /*end_offset*/ 1).await;
+    assert_eq!(
+        newest.entries,
+        vec![HistoryBatchEntry {
+            offset: 1,
+            entry: Some(entries[1].clone()),
+        }]
+    );
+    let next_cursor = newest.next_older_cursor.expect("older cursor");
+    let config = HistoryConfig::new(home.path(), &History::default());
+    let (log_id, _) = history_metadata(&config).await;
+    let oversized = lookup_batch(log_id, next_cursor, &config).expect("read oversized older row");
+    assert_eq!(
+        oversized,
+        HistoryBatch {
+            entries: vec![HistoryBatchEntry {
+                offset: 0,
+                entry: Some(entries[0].clone()),
+            }],
+            next_older_cursor: None,
+        }
+    );
+}
+
+#[tokio::test]
+async fn search_batch_defers_oversized_row_during_backward_scan() {
+    let mut entries = vec![
+        entry(/*offset*/ 0, "oldest"),
+        entry(/*offset*/ 1, "x".repeat(70_000)),
+    ];
+    entries.extend((2..8).map(|offset| entry(offset, "x".repeat(20_000))));
+    let (home, newest) = batch_for(&entries, /*end_offset*/ 7).await;
+    assert_eq!(
+        newest
+            .entries
+            .iter()
+            .map(|entry| entry.offset)
+            .collect::<Vec<_>>(),
+        vec![7, 6, 5]
+    );
+
+    let config = HistoryConfig::new(home.path(), &History::default());
+    let (log_id, _) = history_metadata(&config).await;
+    let middle = lookup_batch(
+        log_id,
+        newest.next_older_cursor.expect("middle cursor"),
+        &config,
+    )
+    .expect("read middle history batch");
+    assert_eq!(
+        middle
+            .entries
+            .iter()
+            .map(|entry| entry.offset)
+            .collect::<Vec<_>>(),
+        vec![4, 3, 2]
+    );
+
+    let oversized = lookup_batch(
+        log_id,
+        middle.next_older_cursor.expect("oversized cursor"),
+        &config,
+    )
+    .expect("read oversized history row");
+    assert_eq!(oversized.entries[0].entry, Some(entries[1].clone()));
+    assert_eq!(
+        oversized
+            .next_older_cursor
+            .expect("oldest cursor")
+            .end_offset(),
+        0
+    );
+}
diff --git a/codex-rs/message-history/src/lib.rs b/codex-rs/message-history/src/lib.rs
index 18a6d791eecf..4b2ce273b0b3 100644
--- a/codex-rs/message-history/src/lib.rs
+++ b/codex-rs/message-history/src/lib.rs
@@ -37,6 +37,12 @@ use tokio::io::AsyncReadExt;
 use codex_config::types::History;
 use codex_config::types::HistoryPersistence;
 
+mod batch;
+pub use batch::HistoryBatch;
+pub use batch::HistoryBatchCursor;
+pub use batch::HistoryBatchEntry;
+pub use batch::lookup_batch;
+
 #[cfg(unix)]
 use std::os::unix::fs::OpenOptionsExt;
 #[cfg(unix)]
@@ -433,5 +439,8 @@ fn log_identity(_metadata: &std::fs::Metadata) -> Option<u64> {
     None
 }
 
+#[cfg(test)]
+#[path = "batch_tests.rs"]
+mod batch_tests;
 #[cfg(test)]
 mod tests;