openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-06 20:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bound resumed TUI history

Browse Source 
Resuming a 707 MiB rollout left Codex with a 16.1 GiB physical
footprint. The app server rebuilt every turn, the TUI retained another
rendered representation, and several resume handoffs cloned the full
rollout.

Have the TUI request the newest 100 full turns in the resume response.
Reconstruct descending pages with a bounded history builder, falling
back to full reconstruction when rollback or unmatched lifecycle events
make discarded history relevant. Show a notice when older turns were
omitted.

Move stored rollout history across resume ownership boundaries, omit
the redundant SessionConfigured replay clone for app-server resumes,
and retain no initial-message transcript in live thread metadata.
Preserve restored token usage by attributing it to a turn from the
initial page.

This bounds retained replay state and avoids full-history clones;
loading the rollout file itself remains eager.
This commit is contained in:
Tamir Duberstein
2026-06-15 10:54:22 -04:00
parent 77e452a15e
commit 8af393b4f9
23 changed files with 609 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/thread_history.rs
+127 -4
View File
@@ -83,12 +83,35 @@ pub fn build_turns_from_rollout_items(items: &[RolloutItem]) -> Vec<Turn> {
builder.finish()
}
/// Convert persisted [`RolloutItem`] entries into at most the newest `limit`
/// turns.
///
/// Returns `None` when a rollback occurs after older turns have been discarded,
/// because those turns may become visible again and require a full replay.
pub fn build_last_turns_from_rollout_items(
items: &[RolloutItem],
limit: usize,
) -> Option<Vec<Turn>> {
if limit == 0 {
return Some(Vec::new());
}
let mut builder = ThreadHistoryBuilder::with_turn_limit(limit);
for item in items {
builder.handle_rollout_item(item);
}
builder.finish_bounded()
}
pub struct ThreadHistoryBuilder {
turns: Vec<Turn>,
current_turn: Option<PendingTurn>,
next_item_index: i64,
current_rollout_index: usize,
next_rollout_index: usize,
turn_limit: Option<usize>,
discarded_turns: bool,
bounded_history_invalidated: bool,
}
impl Default for ThreadHistoryBuilder {
@@ -105,6 +128,16 @@ impl ThreadHistoryBuilder {
next_item_index: 1,
current_rollout_index: 0,
next_rollout_index: 0,
turn_limit: None,
discarded_turns: false,
bounded_history_invalidated: false,
}
}
fn with_turn_limit(turn_limit: usize) -> Self {
Self {
turn_limit: Some(turn_limit),
..Self::new()
}
}
@@ -117,6 +150,11 @@ impl ThreadHistoryBuilder {
self.turns
}
fn finish_bounded(mut self) -> Option<Vec<Turn>> {
self.finish_current_turn();
(!self.bounded_history_invalidated).then_some(self.turns)
}
pub fn active_turn_snapshot(&self) -> Option<Turn> {
self.current_turn
.as_ref()
@@ -937,6 +975,11 @@ impl ThreadHistoryBuilder {
turn.duration_ms = payload.duration_ms;
return;
}
if self.discarded_turns {
self.bounded_history_invalidated = true;
return;
}
}
// If the event has no ID (or refers to an unknown turn), fall back to the active turn.
@@ -988,6 +1031,11 @@ impl ThreadHistoryBuilder {
return;
}
if self.discarded_turns {
self.bounded_history_invalidated = true;
return;
}
// If the completion event cannot be matched, apply it to the active turn.
if let Some(current_turn) = self.current_turn.as_mut() {
mark_completed(current_turn);
@@ -1007,6 +1055,10 @@ impl ThreadHistoryBuilder {
fn handle_thread_rollback(&mut self, payload: &ThreadRolledBackEvent) {
self.finish_current_turn();
if self.discarded_turns {
self.bounded_history_invalidated = true;
}
let n = usize::try_from(payload.num_turns).unwrap_or(usize::MAX);
if n >= self.turns.len() {
self.turns.clear();
@@ -1024,6 +1076,13 @@ impl ThreadHistoryBuilder {
return;
}
self.turns.push(Turn::from(turn));
if let Some(turn_limit) = self.turn_limit
&& self.turns.len() > turn_limit
{
let discard_count = self.turns.len() - turn_limit;
self.turns.drain(..discard_count);
self.discarded_turns = true;
}
}
}
@@ -1075,10 +1134,12 @@ impl ThreadHistoryBuilder {
return;
}
warn!(
item_id = item.id(),
"dropping turn-scoped item for unknown turn id `{turn_id}`"
);
if !self.discarded_turns {
warn!(
item_id = item.id(),
"dropping turn-scoped item for unknown turn id `{turn_id}`"
);
}
}
fn upsert_item_in_current_turn(&mut self, item: ThreadItem) {
@@ -1874,6 +1935,68 @@ mod tests {
);
}
#[test]
fn bounded_history_matches_the_full_history_tail() {
let events = ["First", "Second", "Third", "Fourth"]
.into_iter()
.flat_map(|message| {
[
EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: message.to_string(),
images: None,
text_elements: Vec::new(),
local_images: Vec::new(),
..Default::default()
}),
EventMsg::AgentMessage(AgentMessageEvent {
message: format!("Answer to {message}"),
phase: None,
memory_citation: None,
}),
]
})
.map(RolloutItem::EventMsg)
.collect::<Vec<_>>();
let full_history = build_turns_from_rollout_items(&events);
assert_eq!(
build_last_turns_from_rollout_items(&events, 2),
Some(full_history[2..].to_vec())
);
}
#[test]
fn bounded_history_falls_back_when_rollback_exposes_discarded_turns() {
let mut events = ["First", "Second", "Third"]
.into_iter()
.flat_map(|message| {
[
EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: message.to_string(),
images: None,
text_elements: Vec::new(),
local_images: Vec::new(),
..Default::default()
}),
EventMsg::AgentMessage(AgentMessageEvent {
message: format!("Answer to {message}"),
phase: None,
memory_citation: None,
}),
]
})
.map(RolloutItem::EventMsg)
.collect::<Vec<_>>();
events.push(RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
ThreadRolledBackEvent { num_turns: 1 },
)));
assert_eq!(build_last_turns_from_rollout_items(&events, 2), None);
}
#[test]
fn thread_rollback_clears_all_turns_when_num_turns_exceeds_history() {
let events = vec![
codex-rs/app-server/README.md
+1 -1
View File
@@ -299,7 +299,7 @@ Valid `personality` values are `"friendly"`, `"pragmatic"`, and `"none"`. When `
To continue a stored session, call `thread/resume` with the `thread.id` you previously recorded. The response shape matches `thread/start`. When the stored session includes persisted token usage, the server emits `thread/tokenUsage/updated` immediately after the response so clients can render restored usage before the next turn starts. You can also pass the same configuration overrides supported by `thread/start`, including `approvalsReviewer`.
By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. In that mode the server also skips replaying restored `thread/tokenUsage/updated`, which avoids rebuilding turns just to attribute historical usage.
By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. In that mode the server skips replaying restored `thread/tokenUsage/updated` unless `initialTurnsPage` supplies a turn to own the update.
Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted controls use its defaults. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination.
codex-rs/app-server/src/request_processors.rs
+3
View File
@@ -274,6 +274,7 @@ use codex_app_server_protocol::WindowsSandboxSetupCompletedNotification;
use codex_app_server_protocol::WindowsSandboxSetupMode;
use codex_app_server_protocol::WindowsSandboxSetupStartParams;
use codex_app_server_protocol::WindowsSandboxSetupStartResponse;
use codex_app_server_protocol::build_last_turns_from_rollout_items;
use codex_arg0::Arg0DispatchPaths;
use codex_backend_client::AddCreditsNudgeCreditType as BackendAddCreditsNudgeCreditType;
use codex_backend_client::Client as BackendClient;
@@ -521,6 +522,8 @@ use crate::thread_state::ConnectionCapabilities;
use crate::thread_state::ThreadListenerCommand;
use crate::thread_state::ThreadState;
use crate::thread_state::ThreadStateManager;
use token_usage_replay::latest_token_usage_turn_id;
use token_usage_replay::latest_token_usage_turn_id_from_page;
use token_usage_replay::latest_token_usage_turn_id_from_rollout_items;
use token_usage_replay::send_thread_token_usage_update_to_connection;
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+21 -9
View File
@@ -575,7 +575,6 @@ pub(super) async fn handle_pending_thread_resume_request(
thread_status,
has_live_in_progress_turn,
);
let token_usage_thread = pending.include_turns.then(|| thread.clone());
let mut initial_turns_page = if let Some(params) = pending.initial_turns_page.as_ref() {
match super::thread_processor::build_thread_resume_initial_turns_page(
&pending.history_items,
@@ -593,6 +592,26 @@ pub(super) async fn handle_pending_thread_resume_request(
} else {
None
};
let token_usage_turn_id = if pending.include_turns {
Some(
latest_token_usage_turn_id_from_rollout_items(
&pending.history_items,
thread.turns.as_slice(),
)
.unwrap_or_else(|| latest_token_usage_turn_id(thread.turns.as_slice())),
)
} else {
pending
.initial_turns_page
.as_ref()
.zip(initial_turns_page.as_ref())
.map(|(params, page)| {
latest_token_usage_turn_id_from_page(
page.data.as_slice(),
params.sort_direction.unwrap_or(SortDirection::Desc),
)
})
};
if pending.redact_resume_payloads {
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
@@ -664,20 +683,13 @@ pub(super) async fn handle_pending_thread_resume_request(
initial_turns_page,
};
outgoing.send_response(request_id, response).await;
// Match cold resume: metadata-only resume should attach the listener without
// paying the cost of turn reconstruction for historical usage replay.
if let Some(token_usage_thread) = token_usage_thread {
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
&pending.history_items,
token_usage_thread.turns.as_slice(),
);
if let Some(token_usage_turn_id) = token_usage_turn_id {
// Rejoining a loaded thread has the same UI contract as a cold resume, but
// uses the live conversation state instead of reconstructing a new session.
send_thread_token_usage_update_to_connection(
outgoing,
connection_id,
conversation_id,
&token_usage_thread,
conversation.as_ref(),
token_usage_turn_id,
)
codex-rs/app-server/src/request_processors/thread_processor.rs
+174 -105
View File
@@ -2576,9 +2576,9 @@ impl ThreadRequestProcessor {
.await
.map(|thread_history| (thread_history, None))
} else if let Some(stored_thread) = stored_thread_from_running_probe {
self.stored_thread_to_initial_history(&stored_thread)
.await
.map(|thread_history| (thread_history, Some(*stored_thread)))
let mut stored_thread = *stored_thread;
self.take_stored_thread_initial_history(&mut stored_thread)
.map(|thread_history| (thread_history, Some(stored_thread)))
} else {
self.resume_thread_from_rollout(&thread_id, path.as_ref())
.await
@@ -2629,11 +2629,38 @@ impl ThreadRequestProcessor {
}
};
let response_history = thread_history.clone();
let forked_preview = match &thread_history {
InitialHistory::Forked(items) => Some(preview_from_rollout_items(items)),
InitialHistory::New | InitialHistory::Resumed(_) | InitialHistory::Cleared => None,
};
// A cold resume has no live turn to merge. Build the bounded page before core takes
// ownership of the full rollout so metadata-only clients do not require a history clone.
let prepared_initial_turns_page = if !include_turns {
if let Some(params) = initial_turns_page.as_ref() {
match build_thread_resume_initial_turns_page(
thread_history.rollout_items(),
ThreadStatus::Idle,
/*has_live_running_thread*/ false,
/*active_turn*/ None,
params,
) {
Ok(page) => Some(page),
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
}
}
} else {
None
}
} else {
None
};
let response_history = include_turns.then(|| thread_history.clone());
match self
.thread_manager
.resume_thread_with_history(
.resume_thread_with_history_omitting_initial_messages(
config,
thread_history,
self.auth_manager.clone(),
@@ -2682,7 +2709,8 @@ impl ThreadRequestProcessor {
.load_thread_from_resume_source_or_send_internal(
thread_id,
codex_thread.as_ref(),
&response_history,
response_history.as_ref(),
forked_preview.as_deref(),
rollout_path.as_path(),
resume_source_thread,
include_turns,
@@ -2725,10 +2753,17 @@ impl ThreadRequestProcessor {
let active_permission_profile = thread_response_active_permission_profile(
config_snapshot.active_permission_profile,
);
let token_usage_thread = include_turns.then(|| thread.clone());
let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() {
let initial_turns_sort_direction = initial_turns_page
.as_ref()
.and_then(|params| params.sort_direction)
.unwrap_or(SortDirection::Desc);
let mut initial_turns_page = if let Some(page) = prepared_initial_turns_page {
Some(page)
} else if let (Some(params), Some(response_history)) =
(initial_turns_page.as_ref(), response_history.as_ref())
{
match build_thread_resume_initial_turns_page(
&response_history.get_rollout_items(),
response_history.rollout_items(),
thread.status.clone(),
/*has_live_running_thread*/ false,
/*active_turn*/ None,
@@ -2743,6 +2778,23 @@ impl ThreadRequestProcessor {
} else {
None
};
let token_usage_turn_id = if let Some(response_history) = response_history.as_ref()
{
Some(
latest_token_usage_turn_id_from_rollout_items(
response_history.rollout_items(),
thread.turns.as_slice(),
)
.unwrap_or_else(|| latest_token_usage_turn_id(thread.turns.as_slice())),
)
} else {
initial_turns_page.as_ref().map(|page| {
latest_token_usage_turn_id_from_page(
page.data.as_slice(),
initial_turns_sort_direction,
)
})
};
if redact_resume_payloads {
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
@@ -2768,13 +2820,7 @@ impl ThreadRequestProcessor {
let connection_id = request_id.connection_id;
self.outgoing.send_response(request_id, response).await;
// `excludeTurns` is explicitly the cheap resume path, so avoid
// rebuilding history only to attribute a replayed usage update.
if let Some(token_usage_thread) = token_usage_thread {
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
&response_history.get_rollout_items(),
token_usage_thread.turns.as_slice(),
);
if let Some(token_usage_turn_id) = token_usage_turn_id {
// The client needs restored usage before it starts another turn.
// Sending after the response preserves JSON-RPC request ordering while
// still filling the status line before the next turn lifecycle begins.
@@ -2782,7 +2828,6 @@ impl ThreadRequestProcessor {
&self.outgoing,
connection_id,
thread_id,
&token_usage_thread,
codex_thread.as_ref(),
token_usage_turn_id,
)
@@ -2930,10 +2975,11 @@ impl ThreadRequestProcessor {
}
let redact_resume_payloads =
should_redact_thread_resume_payloads(app_server_client_name.as_deref());
let mut source_thread = source_thread;
let history_items = source_thread
.history
.as_ref()
.map(|history| history.items.clone())
.take()
.map(|history| history.items)
.ok_or_else(|| {
internal_error(format!(
"thread {existing_thread_id} did not include persisted history"
@@ -2957,10 +3003,8 @@ impl ThreadRequestProcessor {
)
.await?;
let mut summary_source_thread = source_thread;
summary_source_thread.history = None;
let mut thread_summary = self.stored_thread_to_api_thread(
summary_source_thread,
source_thread,
config_snapshot.model_provider_id.as_str(),
/*include_turns*/ false,
);
@@ -3029,12 +3073,10 @@ impl ThreadRequestProcessor {
thread_id: &str,
path: Option<&PathBuf>,
) -> Result<(InitialHistory, StoredThread), JSONRPCErrorError> {
let stored_thread = self
let mut stored_thread = self
.read_stored_thread_for_resume(thread_id, path, /*include_history*/ true)
.await?;
let history = self
.stored_thread_to_initial_history(&stored_thread)
.await?;
let history = self.take_stored_thread_initial_history(&mut stored_thread)?;
Ok((history, stored_thread))
}
@@ -3079,15 +3121,15 @@ impl ThreadRequestProcessor {
}
#[tracing::instrument(level = "trace", skip_all)]
async fn stored_thread_to_initial_history(
fn take_stored_thread_initial_history(
&self,
stored_thread: &StoredThread,
stored_thread: &mut StoredThread,
) -> Result<InitialHistory, JSONRPCErrorError> {
let thread_id = stored_thread.thread_id;
let history = stored_thread
.history
.as_ref()
.map(|history| history.items.clone())
.take()
.map(|history| history.items)
.ok_or_else(|| {
internal_error(format!(
"thread {thread_id} did not include persisted history"
@@ -3137,94 +3179,85 @@ impl ThreadRequestProcessor {
&self,
thread_id: ThreadId,
thread: &CodexThread,
thread_history: &InitialHistory,
thread_history: Option<&InitialHistory>,
forked_preview: Option<&str>,
rollout_path: &Path,
resume_source_thread: Option<StoredThread>,
include_turns: bool,
) -> std::result::Result<Thread, String> {
let config_snapshot = thread.config_snapshot().await;
let session_id = thread.session_configured().session_id.to_string();
let thread = match thread_history {
InitialHistory::Resumed(resumed) => {
let fallback_provider = config_snapshot.model_provider_id.as_str();
if let Some(stored_thread) = resume_source_thread {
let stored_thread =
if let Some(rollout_path) = stored_thread.rollout_path.clone() {
self.thread_store
.read_thread_by_rollout_path(StoreReadThreadByRolloutPathParams {
rollout_path,
include_archived: true,
include_history: false,
})
.await
.unwrap_or(StoredThread {
history: None,
..stored_thread
})
} else {
self.thread_store
.read_thread(StoreReadThreadParams {
thread_id: stored_thread.thread_id,
include_archived: true,
include_history: false,
})
.await
.unwrap_or(StoredThread {
history: None,
..stored_thread
})
};
Ok(thread_from_stored_thread(
stored_thread,
fallback_provider,
&self.config.cwd,
)
.0)
} else {
match self
.thread_store
.read_thread(StoreReadThreadParams {
thread_id: resumed.conversation_id,
let thread = if let Some(preview) = forked_preview {
let mut thread = build_thread_from_snapshot(
thread_id,
session_id.clone(),
&config_snapshot,
Some(rollout_path.into()),
);
thread.preview = preview.to_string();
Ok(thread)
} else {
let fallback_provider = config_snapshot.model_provider_id.as_str();
if let Some(stored_thread) = resume_source_thread {
let stored_thread = if let Some(rollout_path) = stored_thread.rollout_path.clone() {
self.thread_store
.read_thread_by_rollout_path(StoreReadThreadByRolloutPathParams {
rollout_path,
include_archived: true,
include_history: false,
})
.await
{
Ok(stored_thread) => Ok(thread_from_stored_thread(
stored_thread,
fallback_provider,
&self.config.cwd,
)
.0),
Err(read_err) => {
Err(format!("failed to read thread from store: {read_err}"))
}
}
.unwrap_or(StoredThread {
history: None,
..stored_thread
})
} else {
self.thread_store
.read_thread(StoreReadThreadParams {
thread_id: stored_thread.thread_id,
include_archived: true,
include_history: false,
})
.await
.unwrap_or(StoredThread {
history: None,
..stored_thread
})
};
Ok(thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd).0)
} else {
match self
.thread_store
.read_thread(StoreReadThreadParams {
thread_id,
include_archived: true,
include_history: false,
})
.await
{
Ok(stored_thread) => Ok(thread_from_stored_thread(
stored_thread,
fallback_provider,
&self.config.cwd,
)
.0),
Err(read_err) => Err(format!("failed to read thread from store: {read_err}")),
}
}
InitialHistory::Forked(items) => {
let mut thread = build_thread_from_snapshot(
thread_id,
session_id.clone(),
&config_snapshot,
Some(rollout_path.into()),
);
thread.preview = preview_from_rollout_items(items);
Ok(thread)
}
InitialHistory::New | InitialHistory::Cleared => Err(format!(
"failed to build resume response for thread {thread_id}: initial history missing"
)),
};
let mut thread = thread?;
thread.id = thread_id.to_string();
thread.session_id = session_id;
thread.path = Some(rollout_path.to_path_buf());
if include_turns {
let history_items = thread_history.get_rollout_items();
let Some(thread_history) = thread_history else {
return Err(format!(
"failed to build resume response for thread {thread_id}: initial history missing"
));
};
populate_thread_turns_from_history(
&mut thread,
&history_items,
thread_history.rollout_items(),
/*active_turn*/ None,
);
}
@@ -3494,14 +3527,14 @@ impl ThreadRequestProcessor {
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
&history_items,
token_usage_thread.turns.as_slice(),
);
)
.unwrap_or_else(|| latest_token_usage_turn_id(token_usage_thread.turns.as_slice()));
// Mirror the resume contract for forks: the new thread is usable as soon
// as the response arrives, so restored usage must follow immediately.
send_thread_token_usage_update_to_connection(
&self.outgoing,
connection_id,
thread_id,
&token_usage_thread,
forked_thread.as_ref(),
token_usage_turn_id,
)
@@ -3729,10 +3762,7 @@ fn paginate_thread_turns(
}
let anchor = cursor.map(parse_thread_turns_cursor).transpose()?;
let page_size = limit
.map(|value| value as usize)
.unwrap_or(THREAD_TURNS_DEFAULT_LIMIT)
.clamp(1, THREAD_TURNS_MAX_LIMIT);
let page_size = thread_turns_page_size(limit);
let anchor_index = anchor
.as_ref()
@@ -3793,6 +3823,13 @@ fn paginate_thread_turns(
})
}
fn thread_turns_page_size(limit: Option<u32>) -> usize {
limit
.map(|value| value as usize)
.unwrap_or(THREAD_TURNS_DEFAULT_LIMIT)
.clamp(1, THREAD_TURNS_MAX_LIMIT)
}
fn serialize_thread_turns_cursor(
turn_id: &str,
include_anchor: bool,
@@ -3844,6 +3881,38 @@ pub(super) fn build_thread_resume_initial_turns_page(
active_turn: Option<Turn>,
params: &ThreadResumeInitialTurnsPageParams,
) -> Result<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> {
let sort_direction = params.sort_direction.unwrap_or(SortDirection::Desc);
if matches!(sort_direction, SortDirection::Desc) {
let page_size = thread_turns_page_size(params.limit);
if let Some(mut turns) =
build_last_turns_from_rollout_items(items, page_size.saturating_add(1))
{
let has_live_in_progress_turn = has_live_running_thread
|| active_turn
.as_ref()
.is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress));
normalize_thread_turns_status(
&mut turns,
loaded_status.clone(),
has_live_in_progress_turn,
);
if let Some(active_turn) = active_turn.clone() {
merge_turn_history_with_active_turn(&mut turns, active_turn);
}
apply_thread_turns_items_view(
&mut turns,
params.items_view.unwrap_or(TurnItemsView::Summary),
);
let page =
paginate_thread_turns(turns, /*cursor*/ None, params.limit, sort_direction)?;
return Ok(codex_app_server_protocol::TurnsPage {
data: page.turns,
next_cursor: page.next_cursor,
backwards_cursor: page.backwards_cursor,
});
}
}
build_thread_turns_page_response(
items,
loaded_status,
@@ -3852,7 +3921,7 @@ pub(super) fn build_thread_resume_initial_turns_page(
ThreadTurnsPageOptions {
cursor: None,
limit: params.limit,
sort_direction: params.sort_direction.unwrap_or(SortDirection::Desc),
sort_direction,
items_view: params.items_view.unwrap_or(TurnItemsView::Summary),
},
)
codex-rs/app-server/src/request_processors/token_usage_replay.rs
+21 -8
View File
@@ -12,7 +12,7 @@
use std::sync::Arc;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadHistoryBuilder;
use codex_app_server_protocol::ThreadTokenUsage;
use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification;
@@ -37,16 +37,15 @@ pub(super) async fn send_thread_token_usage_update_to_connection(
outgoing: &Arc<OutgoingMessageSender>,
connection_id: ConnectionId,
thread_id: ThreadId,
thread: &Thread,
conversation: &CodexThread,
token_usage_turn_id: Option<String>,
token_usage_turn_id: String,
) {
let Some(info) = conversation.token_usage_info().await else {
return;
};
let notification = ThreadTokenUsageUpdatedNotification {
thread_id: thread_id.to_string(),
turn_id: token_usage_turn_id.unwrap_or_else(|| latest_token_usage_turn_id(thread)),
turn_id: token_usage_turn_id,
token_usage: ThreadTokenUsage::from(info),
};
outgoing
@@ -102,17 +101,31 @@ pub(super) fn latest_token_usage_turn_id_from_rollout_items(
/// Normal replay derives the owner from the rollout position of the latest
/// `TokenCount` event. This fallback only preserves a stable wire shape for
/// unusual histories where that rollout information cannot be read.
fn latest_token_usage_turn_id(thread: &Thread) -> String {
thread
.turns
pub(super) fn latest_token_usage_turn_id(turns: &[Turn]) -> String {
turns
.iter()
.rev()
.find(|turn| matches!(turn.status, TurnStatus::Completed | TurnStatus::Failed))
.or_else(|| thread.turns.last())
.or_else(|| turns.last())
.map(|turn| turn.id.clone())
.unwrap_or_default()
}
pub(super) fn latest_token_usage_turn_id_from_page(
turns: &[Turn],
sort_direction: SortDirection,
) -> String {
match sort_direction {
SortDirection::Asc => latest_token_usage_turn_id(turns),
SortDirection::Desc => turns
.iter()
.find(|turn| matches!(turn.status, TurnStatus::Completed | TurnStatus::Failed))
.or_else(|| turns.first())
.map(|turn| turn.id.clone())
.unwrap_or_default(),
}
}
#[cfg(test)]
mod tests {
use super::*;
codex-rs/app-server/tests/suite/v2/thread_read.rs
+2 -2
View File
@@ -660,7 +660,7 @@ async fn thread_resume_initial_turns_page_matches_requested_turns_list_page() ->
thread_id: conversation_id.clone(),
cursor: None,
limit: Some(2),
sort_direction: Some(SortDirection::Asc),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::NotLoaded),
})
.await?;
@@ -677,7 +677,7 @@ async fn thread_resume_initial_turns_page_matches_requested_turns_list_page() ->
exclude_turns: true,
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
limit: Some(2),
sort_direction: Some(SortDirection::Asc),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::NotLoaded),
}),
..Default::default()
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+19 -4
View File
@@ -29,6 +29,7 @@ use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadGoalClearResponse;
use codex_app_server_protocol::ThreadGoalSetResponse;
use codex_app_server_protocol::ThreadGoalStatus;
@@ -1564,7 +1565,7 @@ async fn thread_resume_emits_restored_token_usage_before_next_turn() -> Result<(
}
#[tokio::test]
async fn thread_resume_skips_restored_token_usage_when_turns_are_excluded() -> Result<()> {
async fn thread_resume_replays_usage_for_initial_page_but_not_metadata_only() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
@@ -1583,6 +1584,12 @@ async fn thread_resume_skips_restored_token_usage_when_turns_are_excluded() -> R
let first_resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id.clone(),
exclude_turns: true,
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
limit: Some(1),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::NotLoaded),
}),
..Default::default()
})
.await?;
@@ -1591,9 +1598,17 @@ async fn thread_resume_skips_restored_token_usage_when_turns_are_excluded() -> R
mcp.read_stream_until_response_message(RequestId::Integer(first_resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } =
to_response::<ThreadResumeResponse>(first_resume_resp)?;
let expected_turn_id = thread.turns[0].id.clone();
let ThreadResumeResponse {
thread,
initial_turns_page,
..
} = to_response::<ThreadResumeResponse>(first_resume_resp)?;
assert!(thread.turns.is_empty());
let expected_turn_id = initial_turns_page
.expect("resume should include the requested initial turns page")
.data[0]
.id
.clone();
let first_note = timeout(
DEFAULT_READ_TIMEOUT,
codex-rs/core/src/codex_delegate.rs
+2
View File
@@ -48,6 +48,7 @@ use crate::session::Codex;
use crate::session::CodexSpawnArgs;
use crate::session::CodexSpawnOk;
use crate::session::SUBMISSION_CHANNEL_CAPACITY;
use crate::session::SessionConfiguredInitialMessages;
use crate::session::emit_subagent_session_started;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
@@ -96,6 +97,7 @@ pub(crate) async fn run_codex_thread_interactive(
mcp_manager: Arc::clone(&parent_session.services.mcp_manager),
extensions: Arc::clone(&parent_session.services.extensions),
conversation_history,
session_configured_initial_messages: SessionConfiguredInitialMessages::Include,
session_source: SessionSource::SubAgent(subagent_source.clone()),
forked_from_thread_id,
parent_thread_id: Some(parent_session.thread_id),
codex-rs/core/src/session/mod.rs
+9
View File
@@ -401,6 +401,12 @@ pub struct CodexSpawnOk {
pub thread_id: ThreadId,
}
#[derive(Clone, Copy)]
pub(crate) enum SessionConfiguredInitialMessages {
Include,
Omit,
}
pub(crate) struct CodexSpawnArgs {
pub(crate) config: Config,
pub(crate) user_instructions: LoadedUserInstructions,
@@ -413,6 +419,7 @@ pub(crate) struct CodexSpawnArgs {
pub(crate) mcp_manager: Arc<McpManager>,
pub(crate) extensions: Arc<codex_extension_api::ExtensionRegistry<crate::config::Config>>,
pub(crate) conversation_history: InitialHistory,
pub(crate) session_configured_initial_messages: SessionConfiguredInitialMessages,
pub(crate) session_source: SessionSource,
pub(crate) forked_from_thread_id: Option<ThreadId>,
pub(crate) parent_thread_id: Option<ThreadId>,
@@ -499,6 +506,7 @@ impl Codex {
mcp_manager,
extensions,
conversation_history,
session_configured_initial_messages,
session_source,
forked_from_thread_id,
parent_thread_id,
@@ -653,6 +661,7 @@ impl Codex {
tx_event.clone(),
agent_status_tx.clone(),
conversation_history,
session_configured_initial_messages,
session_source_clone,
skills_manager,
plugins_manager,
codex-rs/core/src/session/session.rs
+8 -3
View File
@@ -480,6 +480,7 @@ impl Session {
tx_event: Sender<Event>,
agent_status: watch::Sender<AgentStatus>,
initial_history: InitialHistory,
session_configured_initial_messages: SessionConfiguredInitialMessages,
session_source: SessionSource,
skills_manager: Arc<SkillsManager>,
plugins_manager: Arc<PluginsManager>,
@@ -1065,9 +1066,13 @@ impl Session {
let mut guard = network_policy_decider_session.write().await;
*guard = Arc::downgrade(&sess);
}
// Dispatch the SessionConfiguredEvent first and then report any errors.
// If resuming, include converted initial messages in the payload so UIs can render them immediately.
let initial_messages = initial_history.get_event_msgs();
// Dispatch the SessionConfiguredEvent first and then report any errors. Direct core
// clients use initial messages to replay a resume. App-server clients reconstruct
// turns separately and can omit this otherwise redundant full-history clone.
let initial_messages = match session_configured_initial_messages {
SessionConfiguredInitialMessages::Include => initial_history.get_event_msgs(),
SessionConfiguredInitialMessages::Omit => None,
};
let events = std::iter::once(Event {
id: INITIAL_SUBMIT_ID.to_owned(),
msg: EventMsg::SessionConfigured(SessionConfiguredEvent {
codex-rs/core/src/session/tests/guardian_tests.rs
+1
View File
@@ -715,6 +715,7 @@ async fn guardian_subagent_does_not_inherit_parent_exec_policy_rules() {
mcp_manager,
extensions: codex_extension_api::empty_extension_registry(),
conversation_history: InitialHistory::New,
session_configured_initial_messages: SessionConfiguredInitialMessages::Include,
session_source: SessionSource::SubAgent(SubAgentSource::Other(
GUARDIAN_REVIEWER_NAME.to_string(),
)),
codex-rs/core/src/thread_manager.rs
+53 -1
View File
@@ -12,6 +12,7 @@ use crate::session::Codex;
use crate::session::CodexSpawnArgs;
use crate::session::CodexSpawnOk;
use crate::session::INITIAL_SUBMIT_ID;
use crate::session::SessionConfiguredInitialMessages;
use crate::session::resolve_multi_agent_version;
use crate::shell_snapshot::ShellSnapshot;
use crate::tasks::InterruptedTurnHistoryMarker;
@@ -631,6 +632,7 @@ impl ThreadManager {
Box::pin(self.state.spawn_thread_with_source(
options.config,
options.initial_history,
SessionConfiguredInitialMessages::Include,
Arc::clone(&self.state.auth_manager),
self.agent_control(),
session_source,
@@ -710,6 +712,41 @@ impl ThreadManager {
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread> {
self.resume_thread_with_history_initial_messages(
config,
initial_history,
auth_manager,
parent_trace,
SessionConfiguredInitialMessages::Include,
)
.await
}
pub async fn resume_thread_with_history_omitting_initial_messages(
&self,
config: Config,
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread> {
self.resume_thread_with_history_initial_messages(
config,
initial_history,
auth_manager,
parent_trace,
SessionConfiguredInitialMessages::Omit,
)
.await
}
async fn resume_thread_with_history_initial_messages(
&self,
config: Config,
initial_history: InitialHistory,
auth_manager: Arc<AuthManager>,
parent_trace: Option<W3cTraceContext>,
initial_messages: SessionConfiguredInitialMessages,
) -> CodexResult<NewThread> {
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
@@ -721,6 +758,7 @@ impl ThreadManager {
Box::pin(self.state.spawn_thread_with_source(
config,
initial_history,
initial_messages,
auth_manager,
self.agent_control(),
session_source,
@@ -784,6 +822,7 @@ impl ThreadManager {
Box::pin(self.state.spawn_thread_with_source(
config,
initial_history,
SessionConfiguredInitialMessages::Include,
auth_manager,
self.agent_control(),
session_source,
@@ -1209,6 +1248,7 @@ impl ThreadManagerState {
Box::pin(self.spawn_thread_with_source(
config,
InitialHistory::New,
SessionConfiguredInitialMessages::Include,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
@@ -1246,6 +1286,7 @@ impl ThreadManagerState {
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
SessionConfiguredInitialMessages::Include,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
@@ -1284,6 +1325,7 @@ impl ThreadManagerState {
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
SessionConfiguredInitialMessages::Include,
Arc::clone(&self.auth_manager),
agent_control,
session_source,
@@ -1323,6 +1365,7 @@ impl ThreadManagerState {
Box::pin(self.spawn_thread_with_source(
config,
initial_history,
SessionConfiguredInitialMessages::Include,
auth_manager,
agent_control,
self.session_source.clone(),
@@ -1346,6 +1389,7 @@ impl ThreadManagerState {
&self,
config: Config,
initial_history: InitialHistory,
session_configured_initial_messages: SessionConfiguredInitialMessages,
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
session_source: SessionSource,
@@ -1415,6 +1459,7 @@ impl ThreadManagerState {
mcp_manager: Arc::clone(&self.mcp_manager),
extensions: Arc::clone(&self.extensions),
conversation_history: initial_history,
session_configured_initial_messages,
session_source,
forked_from_thread_id,
parent_thread_id,
@@ -1464,9 +1509,16 @@ impl ThreadManagerState {
{
let mut threads = self.threads.write().await;
if let std::collections::hash_map::Entry::Vacant(e) = threads.entry(thread_id) {
// The caller receives the initial messages in `NewThread`. Retaining them in the
// thread metadata would keep a second copy of the resumed transcript alive even
// though later `session_configured()` callers only need session metadata.
let retained_session_configured = SessionConfiguredEvent {
initial_messages: None,
..session_configured.clone()
};
let thread = Arc::new(CodexThread::new(
codex,
session_configured.clone(),
retained_session_configured,
session_configured.rollout_path.clone(),
session_source,
));
codex-rs/core/src/thread_manager_tests.rs
+10 -1
View File
@@ -986,7 +986,7 @@ async fn rollout_path_resume_and_fork_read_history_through_thread_store() {
.join("rollouts/source.jsonl")
.to_path_buf();
let resumed = manager
.resume_thread_with_history(
.resume_thread_with_history_omitting_initial_messages(
config.clone(),
InitialHistory::Resumed(ResumedHistory {
conversation_id: source.thread_id,
@@ -998,6 +998,15 @@ async fn rollout_path_resume_and_fork_read_history_through_thread_store() {
)
.await
.expect("seed rollout path in store");
assert_eq!(
resumed.session_configured.initial_messages, None,
"app-server resumes should not materialize replay-only initial messages"
);
assert_eq!(
resumed.thread.session_configured().initial_messages,
None,
"the live thread should not retain replay-only initial messages"
);
resumed
.thread
.shutdown_and_wait()
codex-rs/protocol/src/protocol.rs
+8 -3
View File
@@ -2439,10 +2439,15 @@ impl InitialHistory {
}
pub fn get_rollout_items(&self) -> Vec<RolloutItem> {
self.rollout_items().to_vec()
}
/// Returns the persisted rollout items without cloning them.
pub fn rollout_items(&self) -> &[RolloutItem] {
match self {
InitialHistory::New | InitialHistory::Cleared => Vec::new(),
InitialHistory::Resumed(resumed) => resumed.history.clone(),
InitialHistory::Forked(items) => items.clone(),
InitialHistory::New | InitialHistory::Cleared => &[],
InitialHistory::Resumed(resumed) => resumed.history.as_slice(),
InitialHistory::Forked(items) => items.as_slice(),
}
}
codex-rs/thread-store/src/live_thread.rs
+18 -8
View File
@@ -100,21 +100,25 @@ impl LiveThread {
pub async fn resume(
thread_store: Arc<dyn ThreadStore>,
mut params: ResumeThreadParams,
params: ResumeThreadParams,
) -> ThreadStoreResult<Self> {
let thread_id = params.thread_id;
let should_load_history = params.history.is_none();
let include_archived = params.include_archived;
thread_store.resume_thread(params.clone()).await?;
if should_load_history {
match thread_store
let metadata = params.metadata.clone();
let metadata_sync = (!should_load_history).then(|| ThreadMetadataSync::for_resume(&params));
thread_store.resume_thread(params).await?;
let metadata_sync = if let Some(metadata_sync) = metadata_sync {
metadata_sync
} else {
let history = match thread_store
.load_history(LoadThreadHistoryParams {
thread_id,
include_archived,
})
.await
{
Ok(history) => params.history = Some(history.items),
Ok(history) => history,
Err(err) => {
if let Err(discard_err) = thread_store.discard_thread(thread_id).await {
warn!(
@@ -123,9 +127,15 @@ impl LiveThread {
}
return Err(err);
}
}
}
let metadata_sync = ThreadMetadataSync::for_resume(&params);
};
ThreadMetadataSync::for_resume(&ResumeThreadParams {
thread_id,
rollout_path: None,
history: Some(history.items),
include_archived,
metadata,
})
};
Ok(Self {
thread_id,
thread_store,
codex-rs/tui/src/app.rs
+6 -2
View File
@@ -1058,8 +1058,12 @@ See the Codex keymap documentation for supported actions and examples."
let initial_session_started_at = Instant::now();
if let Some(started) = initial_started_thread {
let thread_id = started.session.thread_id;
app.enqueue_primary_thread_session(started.session, started.turns)
.await?;
app.enqueue_primary_thread_session(
started.session,
started.turns,
started.history_truncated,
)
.await?;
if should_prompt_for_paused_goal_after_startup_resume {
app.maybe_prompt_resume_paused_goal_after_resume(&mut app_server, thread_id)
.await;
codex-rs/tui/src/app/session_lifecycle.rs
+12 -4
View File
@@ -513,8 +513,12 @@ impl App {
.set_queue_submissions_until_session_configured(/*queue*/ false);
match result {
Ok(started) => {
self.enqueue_primary_thread_session(started.session, started.turns)
.await?;
self.enqueue_primary_thread_session(
started.session,
started.turns,
started.history_truncated,
)
.await?;
self.chat_widget.maybe_send_next_queued_input();
}
Err(err) => {
@@ -611,8 +615,12 @@ impl App {
initial_user_message,
);
self.replace_chat_widget(ChatWidget::new_with_app_event(init));
self.enqueue_primary_thread_session(started.session, started.turns)
.await?;
self.enqueue_primary_thread_session(
started.session,
started.turns,
started.history_truncated,
)
.await?;
self.backfill_loaded_subagent_threads(app_server).await;
Ok(())
}
codex-rs/tui/src/app/snapshots/codex_tui__app__tests__truncated_resume_history_notice.snap
+6
View File
@@ -0,0 +1,6 @@
---
source: tui/src/app/tests.rs
assertion_line: 392
expression: "truncated_history_notice.expect(\"expected truncated history notice\")"
---
• Earlier turns are omitted from this transcript to limit memory use.
codex-rs/tui/src/app/tests.rs
+28 -7
View File
@@ -209,6 +209,7 @@ async fn enqueue_primary_thread_session_replays_buffered_approval_after_attach()
app.enqueue_primary_thread_session(
test_thread_session(thread_id, test_path_buf("/tmp/project")),
Vec::new(),
/*history_truncated*/ false,
)
.await?;
@@ -257,6 +258,7 @@ async fn resolved_buffered_approval_does_not_become_actionable_after_drain() ->
app.enqueue_primary_thread_session(
test_thread_session(thread_id, test_path_buf("/tmp/project")),
Vec::new(),
/*history_truncated*/ false,
)
.await?;
while app_event_rx.try_recv().is_ok() {}
@@ -354,16 +356,21 @@ async fn enqueue_primary_thread_session_replays_turns_before_initial_prompt_subm
}],
}],
)],
/*history_truncated*/ true,
)
.await?;
let mut saw_replayed_answer = false;
let mut truncated_history_notice = None;
let mut submitted_items = None;
while let Ok(event) = app_event_rx.try_recv() {
match event {
AppEvent::InsertHistoryCell(cell) => {
let transcript = lines_to_single_string(&cell.transcript_lines(/*width*/ 80));
saw_replayed_answer |= transcript.contains("earlier prompt");
if transcript.contains(super::thread_routing::TRUNCATED_HISTORY_NOTICE) {
truncated_history_notice = Some(transcript);
}
}
AppEvent::SubmitThreadOp {
thread_id: op_thread_id,
@@ -382,6 +389,10 @@ async fn enqueue_primary_thread_session_replays_turns_before_initial_prompt_subm
saw_replayed_answer,
"expected replayed history before initial prompt submit"
);
insta::assert_snapshot!(
"truncated_resume_history_notice",
truncated_history_notice.expect("expected truncated history notice")
);
assert_eq!(
submitted_items,
Some(vec![UserInput::Text {
@@ -3512,6 +3523,7 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() {
AppServerStartedThread {
session: test_thread_session(child_thread_id, test_path_buf("/tmp/child")),
turns: Vec::new(),
history_truncated: false,
},
&mut child_snapshot,
)
@@ -4165,7 +4177,7 @@ async fn set_thread_goal_draft_materializes_long_objective_and_confirms_before_p
.start_thread(app.chat_widget.config_ref())
.await?;
let thread_id = started.session.thread_id;
app.enqueue_primary_thread_session(started.session, started.turns)
app.enqueue_primary_thread_session(started.session, started.turns, started.history_truncated)
.await?;
let objective = "x".repeat(MAX_THREAD_GOAL_OBJECTIVE_CHARS + 1);
@@ -5512,6 +5524,7 @@ async fn refreshed_snapshot_session_persists_resumed_turns() {
AppServerStartedThread {
session: resumed_session.clone(),
turns: resumed_turns.clone(),
history_truncated: false,
},
&mut snapshot,
)
@@ -5772,9 +5785,13 @@ async fn interrupt_without_active_turn_is_treated_as_handled() {
.await
.expect("thread/start should succeed");
let thread_id = started.session.thread_id;
app.enqueue_primary_thread_session(started.session, started.turns)
.await
.expect("primary thread should be registered");
app.enqueue_primary_thread_session(
started.session,
started.turns,
started.history_truncated,
)
.await
.expect("primary thread should be registered");
let op = AppCommand::interrupt();
let handled = Box::pin(app.try_submit_active_thread_op_via_app_server(
@@ -5805,9 +5822,13 @@ async fn override_turn_context_sends_thread_settings_update() {
let thread_id = started.session.thread_id;
let initial_model = started.session.model.clone();
let initial_effort = started.session.reasoning_effort.clone();
app.enqueue_primary_thread_session(started.session, started.turns)
.await
.expect("primary thread should be registered");
app.enqueue_primary_thread_session(
started.session,
started.turns,
started.history_truncated,
)
.await
.expect("primary thread should be registered");
let service_tier = ServiceTier::Fast.request_value().to_string();
let collaboration_mode = CollaborationMode {
mode: ModeKind::Plan,
codex-rs/tui/src/app/tests/startup.rs
+2
View File
@@ -162,6 +162,7 @@ async fn startup_thread_started_submits_queued_startup_input() {
Ok(AppServerStartedThread {
session: test_thread_session(thread_id, test_path_buf("/tmp/project")),
turns: Vec::new(),
history_truncated: false,
}),
)
.await
@@ -239,6 +240,7 @@ fn stale_startup_thread_started_removes_local_routing_state() -> Result<()> {
Ok(AppServerStartedThread {
session: test_thread_session(stale_thread_id, test_path_buf("/tmp/project")),
turns: Vec::new(),
history_truncated: false,
}),
)
.await?;
codex-rs/tui/src/app/thread_routing.rs
+13 -1
View File
@@ -7,6 +7,9 @@
use super::*;
use crate::session_resume::read_session_model;
pub(super) const TRUNCATED_HISTORY_NOTICE: &str =
"Earlier turns are omitted from this transcript to limit memory use.";
impl App {
pub(super) async fn shutdown_current_thread(&mut self, app_server: &mut AppServerSession) {
if let Some(thread_id) = self.chat_widget.thread_id() {
@@ -1081,6 +1084,7 @@ impl App {
&mut self,
session: ThreadSessionState,
turns: Vec<Turn>,
history_truncated: bool,
) -> Result<()> {
let thread_id = session.thread_id;
self.primary_thread_id = Some(thread_id);
@@ -1104,6 +1108,10 @@ impl App {
self.app_event_tx
.send(AppEvent::BeginInitialHistoryReplayBuffer);
}
if history_truncated {
self.chat_widget
.add_info_message(TRUNCATED_HISTORY_NOTICE.to_string(), /*hint*/ None);
}
self.chat_widget
.replay_thread_turns(turns, ReplayKind::ResumeInitialMessages);
if should_buffer_initial_replay {
@@ -1209,7 +1217,11 @@ impl App {
started: AppServerStartedThread,
snapshot: &mut ThreadEventSnapshot,
) {
let AppServerStartedThread { session, turns } = started;
let AppServerStartedThread {
session,
turns,
history_truncated: _,
} = started;
if let Some(channel) = self.thread_event_channels.get(&thread_id) {
let mut store = channel.store.lock().await;
store.set_session(session.clone(), turns.clone());
codex-rs/tui/src/app_server_session.rs
+65 -1
View File
@@ -47,6 +47,7 @@ use codex_app_server_protocol::ReviewStartResponse;
use codex_app_server_protocol::ReviewTarget;
use codex_app_server_protocol::SkillsListParams;
use codex_app_server_protocol::SkillsListResponse;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadApproveGuardianDeniedActionParams;
use codex_app_server_protocol::ThreadApproveGuardianDeniedActionResponse;
@@ -81,6 +82,7 @@ use codex_app_server_protocol::ThreadMetadataUpdateParams;
use codex_app_server_protocol::ThreadMetadataUpdateResponse;
use codex_app_server_protocol::ThreadReadParams;
use codex_app_server_protocol::ThreadReadResponse;
use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
@@ -102,6 +104,7 @@ use codex_app_server_protocol::ThreadUnsubscribeResponse;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnInterruptResponse;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnSteerParams;
@@ -133,6 +136,8 @@ use uuid::Uuid;
const JSONRPC_INVALID_REQUEST: i64 = -32600;
const JSONRPC_METHOD_NOT_FOUND: i64 = -32601;
// Replayed turns are retained both as structured thread state and rendered transcript cells.
const TUI_RESUME_TURN_LIMIT: u32 = 100;
pub(crate) const EXTERNAL_AGENT_CONFIG_IMPORT_IN_PROGRESS_MESSAGE: &str =
"A previous Claude Code import is still running. Wait for it to finish before importing again.";
const THREAD_SETTINGS_UPDATE_METHOD: &str = "thread/settings/update";
@@ -199,6 +204,7 @@ impl ThreadParamsMode {
pub(crate) struct AppServerStartedThread {
pub(crate) session: ThreadSessionState,
pub(crate) turns: Vec<Turn>,
pub(crate) history_truncated: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
@@ -1432,6 +1438,12 @@ fn thread_resume_params_from_config(
developer_instructions: with_terminal_visualization_instructions(
&config, /*control_instructions*/ None,
),
exclude_turns: true,
initial_turns_page: Some(ThreadResumeInitialTurnsPageParams {
limit: Some(TUI_RESUME_TURN_LIMIT),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::Full),
}),
..ThreadResumeParams::default()
}
}
@@ -1500,6 +1512,7 @@ async fn started_thread_from_start_response(
Ok(AppServerStartedThread {
session,
turns: response.thread.turns,
history_truncated: false,
})
}
@@ -1512,9 +1525,17 @@ async fn started_thread_from_resume_response(
thread_session_state_from_thread_resume_response(&response, config, thread_params_mode)
.await
.map_err(color_eyre::eyre::Report::msg)?;
let (turns, history_truncated) = if let Some(mut page) = response.initial_turns_page {
let history_truncated = page.next_cursor.is_some();
page.data.reverse();
(page.data, history_truncated)
} else {
(response.thread.turns, false)
};
Ok(AppServerStartedThread {
session,
turns: response.thread.turns,
turns,
history_truncated,
})
}
@@ -1530,6 +1551,7 @@ async fn started_thread_from_fork_response(
Ok(AppServerStartedThread {
session,
turns: response.thread.turns,
history_truncated: false,
})
}
@@ -2026,6 +2048,29 @@ mod tests {
assert_eq!(fork.thread_source, Some(ThreadSource::User));
}
#[tokio::test]
async fn thread_resume_requests_recent_full_turns() {
let temp_dir = tempfile::tempdir().expect("tempdir");
let config = build_config(&temp_dir).await;
let params = thread_resume_params_from_config(
config,
ThreadId::new(),
ThreadParamsMode::Embedded,
/*remote_cwd_override*/ None,
);
assert!(params.exclude_turns);
assert_eq!(
params.initial_turns_page,
Some(ThreadResumeInitialTurnsPageParams {
limit: Some(TUI_RESUME_TURN_LIMIT),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::Full),
})
);
}
#[test]
fn sandbox_mode_does_not_project_non_cwd_write_roots_for_remote_sessions() {
let cwd = test_path_buf("/workspace/project").abs();
@@ -2393,6 +2438,25 @@ mod tests {
assert_eq!(started.session.permission_profile, read_only_profile);
assert_eq!(started.turns.len(), 1);
assert_eq!(started.turns[0], response.thread.turns[0]);
assert!(!started.history_truncated);
let first_turn = response.thread.turns[0].clone();
let mut second_turn = first_turn.clone();
second_turn.id = "turn-2".to_string();
let mut paged_response = response.clone();
paged_response.thread.turns.clear();
paged_response.initial_turns_page = Some(codex_app_server_protocol::TurnsPage {
data: vec![second_turn.clone(), first_turn.clone()],
next_cursor: Some("older-turns".to_string()),
backwards_cursor: Some("newer-turns".to_string()),
});
let started =
started_thread_from_resume_response(paged_response, &config, ThreadParamsMode::Remote)
.await
.expect("paged resume response should map");
assert_eq!(started.turns, vec![first_turn, second_turn]);
assert!(started.history_truncated);
let embedded_config = ConfigBuilder::default()
.codex_home(temp_dir.path().join("embedded-codex-home"))