Bound legacy resume preview scans (#39033)

## Why

Building a resume-picker preview for a legacy session should not require loading its full thread history.

## What changed

- Read up to six preview lines by scanning at most 1 MiB from the end of a local legacy rollout.
- Fall back to the existing history hydration path when the local file is unavailable, the bounded scan is incomplete, or the tail contains a rollback.
- Keep the initial `thread/read` request turn-free while loading the preview.

## Testing

Added coverage for compaction, rollbacks, oversized records, exhausted scan budgets, and app-server request parameters.

GitOrigin-RevId: a8950f03d8b56323f268fd3355c6a38a16cb9833
This commit is contained in:
Charlie Marsh
2026-08-17 17:14:46 +00:00
committed by copyberry
parent dc473903ab
commit a8525b139c
3 changed files with 369 additions and 15 deletions
@@ -1123,6 +1123,21 @@ async fn paginated_workflows_never_request_full_thread_history() -> Result<()> {
assert!(!recorded_params(&requests, "thread/items/list").is_empty());
let previous_read_count = paginated_reads.len();
let preview = crate::resume_picker::load_transcript_preview(
&mut app_server,
legacy_thread_id,
Some(&app.config),
)
.await?;
assert!(!preview.is_empty());
let preview_reads = recorded_params(&requests, "thread/read");
let preview_include_turns = preview_reads[previous_read_count..]
.iter()
.map(|params| params["includeTurns"].as_bool().unwrap_or(false))
.collect::<Vec<_>>();
assert_eq!(preview_include_turns, vec![false]);
let previous_read_count = preview_reads.len();
crate::thread_transcript::load_session_transcript(
&mut app_server,
legacy_thread_id,
@@ -1,5 +1,9 @@
use std::borrow::Cow;
use std::collections::HashSet;
use std::fs::File;
use std::io::Read;
use std::io::Seek;
use std::io::SeekFrom;
use std::path::Path;
use super::TranscriptPreviewLine;
@@ -8,17 +12,24 @@ use crate::app_server_session::AppServerSession;
use crate::app_server_session::HISTORY_ITEM_PAGE_LIMIT;
use crate::app_server_session::HISTORY_ITEM_SCAN_LIMIT;
use crate::app_server_session::HistoryHydrationScope;
use crate::app_server_session::ThreadParamsMode;
use crate::git_action_directives::parse_assistant_markdown;
use crate::inline_visualization::InlineVisualizationContext;
use crate::legacy_core::config::Config;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadItem;
use codex_protocol::ThreadId;
use codex_protocol::protocol::EventMsg;
use codex_rollout::ReverseJsonlScanner;
use codex_rollout::RolloutItem;
use codex_rollout::RolloutLine;
use codex_rollout::ScanOutcome;
const MAX_TRANSCRIPT_PREVIEW_LINES: usize = 6;
const TRANSCRIPT_PREVIEW_ITEMS_PAGE_SIZE: u32 = 6;
const MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES: usize = 1024 * 1024;
/// Loads the newest nonblank transcript lines within a bounded paginated-item scan.
/// Loads the newest transcript lines from bounded paginated items or a local legacy rollout tail.
pub(crate) async fn load_transcript_preview(
app_server: &mut AppServerSession,
thread_id: ThreadId,
@@ -28,19 +39,6 @@ pub(crate) async fn load_transcript_preview(
.thread_read(thread_id, /*include_turns*/ false)
.await
.map_err(std::io::Error::other)?;
if thread.history_mode == ThreadHistoryMode::Legacy {
app_server
.hydrate_initial_thread_history(
&mut thread,
/*turn_cursor*/ None,
/*item_cursor*/ None,
/*config*/ None,
HistoryHydrationScope::Initial,
)
.await
.map_err(std::io::Error::other)?;
}
let cwd = thread.cwd.as_path();
let inline_visualization_context = config.and_then(|config| {
ThreadId::from_string(&thread.id)
.ok()
@@ -49,6 +47,40 @@ pub(crate) async fn load_transcript_preview(
let mut lines = Vec::with_capacity(MAX_TRANSCRIPT_PREVIEW_LINES);
match thread.history_mode {
ThreadHistoryMode::Legacy => {
if matches!(app_server.thread_params_mode(), ThreadParamsMode::Embedded)
&& let Some(path) = thread.path.as_ref()
&& path
.file_name()
.and_then(|name| name.to_str())
.is_some_and(|name| name.ends_with(".jsonl"))
{
let path = path.clone();
let cwd = thread.cwd.clone();
let inline_visualization_context = inline_visualization_context.clone();
let scanned = tokio::task::spawn_blocking(move || {
scan_legacy_transcript_preview(
path.as_path(),
cwd.as_path(),
inline_visualization_context.as_ref(),
)
})
.await;
if let Ok(Ok(Some(mut scanned_lines))) = scanned {
scanned_lines.reverse();
return Ok(scanned_lines);
}
}
app_server
.hydrate_initial_thread_history(
&mut thread,
/*turn_cursor*/ None,
/*item_cursor*/ None,
/*config*/ None,
HistoryHydrationScope::Initial,
)
.await
.map_err(std::io::Error::other)?;
append_transcript_preview_lines(
&mut lines,
thread
@@ -56,11 +88,12 @@ pub(crate) async fn load_transcript_preview(
.iter()
.rev()
.flat_map(|turn| turn.items.iter().rev()),
cwd,
thread.cwd.as_path(),
inline_visualization_context.as_ref(),
);
}
ThreadHistoryMode::Paginated => {
let cwd = thread.cwd.as_path();
let mut cursor = None;
let mut seen_cursors = HashSet::new();
let mut scanned_items = 0_usize;
@@ -109,6 +142,89 @@ pub(crate) async fn load_transcript_preview(
Ok(lines)
}
/// Scans a bounded legacy rollout tail, falling back when its history cannot be proven complete.
fn scan_legacy_transcript_preview(
path: &Path,
cwd: &Path,
inline_visualization_context: Option<&InlineVisualizationContext>,
) -> std::io::Result<Option<Vec<TranscriptPreviewLine>>> {
let mut lines = Vec::with_capacity(MAX_TRANSCRIPT_PREVIEW_LINES);
let file = File::open(path)?;
let end_offset = file.metadata()?.len();
let minimum_offset = end_offset.saturating_sub(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES as u64);
let mut scanner = ReverseJsonlScanner::new_at(
BoundedRolloutTailReader {
file,
minimum_offset,
},
end_offset,
)?
.with_max_record_bytes(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES);
loop {
let outcome = match scanner.scan_next::<RolloutLine>() {
Ok(Some(outcome)) => outcome,
Ok(None) => break,
Err(error) if error.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(None),
Err(error) => return Err(error),
};
let ScanOutcome::Parsed(line) = outcome else {
continue;
};
match line.item {
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => return Ok(None),
RolloutItem::EventMsg(EventMsg::UserMessage(message)) => {
append_transcript_preview_text(
&mut lines,
TranscriptPreviewSpeaker::User,
&message.message,
cwd,
inline_visualization_context,
);
}
RolloutItem::EventMsg(EventMsg::AgentMessage(message)) => {
append_transcript_preview_text(
&mut lines,
TranscriptPreviewSpeaker::Assistant,
&message.message,
cwd,
inline_visualization_context,
);
}
_ => {}
}
if lines.len() == MAX_TRANSCRIPT_PREVIEW_LINES {
break;
}
}
Ok(Some(lines))
}
/// Prevents reverse scans from reading past the bounded suffix of a rollout.
struct BoundedRolloutTailReader {
file: File,
minimum_offset: u64,
}
impl Read for BoundedRolloutTailReader {
fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
self.file.read(buffer)
}
}
impl Seek for BoundedRolloutTailReader {
fn seek(&mut self, position: SeekFrom) -> std::io::Result<u64> {
let offset = self.file.seek(position)?;
if offset < self.minimum_offset {
return Err(std::io::Error::new(
std::io::ErrorKind::UnexpectedEof,
"legacy transcript preview exceeded its scan budget",
));
}
Ok(offset)
}
}
/// Appends the newest preview lines from items already ordered newest-first.
fn append_transcript_preview_lines<'a>(
lines: &mut Vec<TranscriptPreviewLine>,
@@ -14,6 +14,10 @@ use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::UserInput;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::UserMessageEvent;
use codex_rollout::CompactedItem;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use tempfile::tempdir;
@@ -25,6 +29,225 @@ fn preview_line(speaker: TranscriptPreviewSpeaker, text: &str) -> TranscriptPrev
}
}
fn rollout_user_message(text: &str) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: text.to_string(),
..Default::default()
}))
}
fn rollout_agent_message(text: &str) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
message: text.to_string(),
phase: None,
memory_citation: None,
}))
}
fn write_preview_rollout(path: &Path, items: Vec<RolloutItem>) {
let contents = items
.into_iter()
.map(|item| {
serde_json::to_string(&RolloutLine {
timestamp: String::from("2026-07-19T12:00:00Z"),
ordinal: None,
item,
})
.expect("serialize rollout line")
})
.collect::<Vec<_>>()
.join("\n");
std::fs::write(path, format!("{contents}\n")).expect("write rollout");
}
#[test]
fn legacy_transcript_preview_scans_tail_across_compaction() {
let temp_dir = tempdir().expect("tempdir");
let path = temp_dir.path().join("rollout.jsonl");
write_preview_rollout(
&path,
vec![
rollout_user_message("older user"),
rollout_agent_message("older assistant"),
RolloutItem::Compacted(CompactedItem {
message: String::from("summary that is not transcript text"),
replacement_history: None,
window_number: None,
first_window_id: None,
previous_window_id: None,
window_id: None,
}),
rollout_user_message("recent user"),
rollout_agent_message("commentary one\ncommentary two"),
rollout_agent_message("final one\n\n final two "),
],
);
let mut lines = scan_legacy_transcript_preview(
&path,
temp_dir.path(),
/*inline_visualization_context*/ None,
)
.expect("scan rollout")
.expect("bounded tail");
lines.reverse();
assert_eq!(
lines,
vec![
preview_line(TranscriptPreviewSpeaker::Assistant, "older assistant"),
preview_line(TranscriptPreviewSpeaker::User, "recent user"),
preview_line(TranscriptPreviewSpeaker::Assistant, "commentary one"),
preview_line(TranscriptPreviewSpeaker::Assistant, "commentary two"),
preview_line(TranscriptPreviewSpeaker::Assistant, "final one"),
preview_line(TranscriptPreviewSpeaker::Assistant, "final two"),
]
);
}
#[test]
fn legacy_transcript_preview_falls_back_when_rollback_is_in_the_tail() {
let temp_dir = tempdir().expect("tempdir");
let path = temp_dir.path().join("rollout.jsonl");
write_preview_rollout(
&path,
vec![
rollout_user_message("rolled back user"),
rollout_agent_message("rolled back assistant"),
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
num_turns: 1,
})),
rollout_user_message("recent user"),
rollout_agent_message("recent assistant"),
],
);
let lines = scan_legacy_transcript_preview(
&path,
temp_dir.path(),
/*inline_visualization_context*/ None,
)
.expect("scan rollout");
assert_eq!(lines, None);
}
#[test]
fn legacy_transcript_preview_stops_before_rollback_outside_the_tail() {
let temp_dir = tempdir().expect("tempdir");
let path = temp_dir.path().join("rollout.jsonl");
write_preview_rollout(
&path,
vec![
rollout_user_message("rolled back user"),
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
num_turns: 1,
})),
rollout_user_message("recent user"),
rollout_agent_message("one\ntwo\nthree\nfour\nfive\nsix"),
],
);
let mut lines = scan_legacy_transcript_preview(
&path,
temp_dir.path(),
/*inline_visualization_context*/ None,
)
.expect("scan rollout")
.expect("bounded tail");
lines.reverse();
assert_eq!(
lines,
vec![
preview_line(TranscriptPreviewSpeaker::Assistant, "one"),
preview_line(TranscriptPreviewSpeaker::Assistant, "two"),
preview_line(TranscriptPreviewSpeaker::Assistant, "three"),
preview_line(TranscriptPreviewSpeaker::Assistant, "four"),
preview_line(TranscriptPreviewSpeaker::Assistant, "five"),
preview_line(TranscriptPreviewSpeaker::Assistant, "six"),
]
);
}
#[test]
fn legacy_transcript_preview_falls_back_for_oversized_visible_record() {
let temp_dir = tempdir().expect("tempdir");
let path = temp_dir.path().join("rollout.jsonl");
write_preview_rollout(
&path,
vec![rollout_agent_message(
&"x".repeat(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES),
)],
);
let lines = scan_legacy_transcript_preview(
&path,
temp_dir.path(),
/*inline_visualization_context*/ None,
)
.expect("scan rollout");
assert_eq!(lines, None);
}
#[test]
fn legacy_transcript_preview_falls_back_for_oversized_hidden_record() {
let temp_dir = tempdir().expect("tempdir");
let path = temp_dir.path().join("rollout.jsonl");
write_preview_rollout(
&path,
vec![
rollout_user_message("older user"),
RolloutItem::Compacted(CompactedItem {
message: "x".repeat(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES),
replacement_history: None,
window_number: None,
first_window_id: None,
previous_window_id: None,
window_id: None,
}),
rollout_agent_message("recent assistant"),
],
);
let lines = scan_legacy_transcript_preview(
&path,
temp_dir.path(),
/*inline_visualization_context*/ None,
)
.expect("scan rollout");
assert_eq!(lines, None);
}
#[test]
fn legacy_transcript_preview_falls_back_when_scan_budget_is_exhausted() {
let temp_dir = tempdir().expect("tempdir");
let path = temp_dir.path().join("rollout.jsonl");
let compacted = RolloutItem::Compacted(CompactedItem {
message: "x".repeat(MAX_LEGACY_TRANSCRIPT_PREVIEW_SCAN_BYTES / 8),
replacement_history: None,
window_number: None,
first_window_id: None,
previous_window_id: None,
window_id: None,
});
let mut items = vec![rollout_user_message("older user")];
items.extend(std::iter::repeat_n(compacted, 9));
items.push(rollout_agent_message("recent assistant"));
write_preview_rollout(&path, items);
let lines = scan_legacy_transcript_preview(
&path,
temp_dir.path(),
/*inline_visualization_context*/ None,
)
.expect("scan rollout");
assert_eq!(lines, None);
}
#[test]
fn transcript_preview_reverse_scan_stops_before_older_items() {
let newest = ThreadItem::AgentMessage {