agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add sorting/backwardsCursor to thread/list and new thread/turns/list api (#17305)

Browse Source 
To improve performance of UI loads from the app, add two main
improvements:
1. The `thread/list` api now gets a `sortDirection` request field and a
`backwardsCursor` to the response, which lets you paginate forwards and
backwards from a window. This lets you fetch the first few items to
display immediately while you paginate to fill in history, then can
paginate "backwards" on future loads to catch up with any changes since
the last UI load without a full reload of the entire data set.
2. Added a new `thread/turns/list` api which also has sortDirection and
backwardsCursor for the same behavior as `thread/list`, allowing you the
same small-fetch for immediate display followed by background fill-in
and resync catchup.
This commit is contained in:
David de Regt
2026-04-17 11:49:02 -07:00
committed by GitHub
Unverified
parent 29bc2ad2f4
commit eaf78e43f2
54 changed files with 3510 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+420 -9
View File
@@ -19,6 +19,7 @@ use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
use crate::thread_status::ThreadWatchManager;
use crate::thread_status::resolve_thread_status;
use chrono::DateTime;
use chrono::Duration as ChronoDuration;
use chrono::SecondsFormat;
use chrono::Utc;
use codex_analytics::AnalyticsEventsClient;
@@ -123,6 +124,7 @@ use codex_app_server_protocol::SkillsConfigWriteParams;
use codex_app_server_protocol::SkillsConfigWriteResponse;
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::ThreadArchiveParams;
use codex_app_server_protocol::ThreadArchiveResponse;
@@ -177,6 +179,8 @@ use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStartedNotification;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadTurnsListResponse;
use codex_app_server_protocol::ThreadUnarchiveParams;
use codex_app_server_protocol::ThreadUnarchiveResponse;
use codex_app_server_protocol::ThreadUnarchivedNotification;
@@ -325,6 +329,7 @@ use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams;
use codex_thread_store::ListThreadsParams as StoreListThreadsParams;
use codex_thread_store::LocalThreadStore;
use codex_thread_store::ReadThreadParams as StoreReadThreadParams;
use codex_thread_store::SortDirection as StoreSortDirection;
use codex_thread_store::StoredThread;
use codex_thread_store::ThreadSortKey as StoreThreadSortKey;
use codex_thread_store::ThreadStore;
@@ -377,6 +382,8 @@ use token_usage_replay::send_thread_token_usage_update_to_connection;
const THREAD_LIST_DEFAULT_LIMIT: usize = 25;
const THREAD_LIST_MAX_LIMIT: usize = 100;
const THREAD_TURNS_DEFAULT_LIMIT: usize = 25;
const THREAD_TURNS_MAX_LIMIT: usize = 100;
struct ThreadListFilters {
model_providers: Option<Vec<String>>,
@@ -951,6 +958,10 @@ impl CodexMessageProcessor {
self.thread_read(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadTurnsList { request_id, params } => {
self.thread_turns_list(to_connection_request_id(request_id), params)
.await;
}
ClientRequest::ThreadShellCommand { request_id, params } => {
self.thread_shell_command(to_connection_request_id(request_id), params)
.await;
@@ -3641,6 +3652,7 @@ impl CodexMessageProcessor {
cursor,
limit,
sort_key,
sort_direction,
model_providers,
source_kinds,
archived,
@@ -3663,11 +3675,13 @@ impl CodexMessageProcessor {
ThreadSortKey::CreatedAt => StoreThreadSortKey::CreatedAt,
ThreadSortKey::UpdatedAt => StoreThreadSortKey::UpdatedAt,
};
let (summaries, next_cursor) = match self
let sort_direction = sort_direction.unwrap_or(SortDirection::Desc);
let list_result = self
.list_threads_common(
requested_page_size,
cursor,
store_sort_key,
sort_direction,
ThreadListFilters {
model_providers,
source_kinds,
@@ -3676,14 +3690,17 @@ impl CodexMessageProcessor {
search_term,
},
)
.await
{
.await;
let (summaries, next_cursor) = match list_result {
Ok(r) => r,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let backwards_cursor = summaries.first().and_then(|summary| {
thread_backwards_cursor_for_sort_key(summary, store_sort_key, sort_direction)
});
let mut threads = Vec::with_capacity(summaries.len());
let mut thread_ids = HashSet::with_capacity(summaries.len());
let mut status_ids = Vec::with_capacity(summaries.len());
@@ -3704,7 +3721,7 @@ impl CodexMessageProcessor {
.loaded_statuses_for_threads(status_ids)
.await;
let data = threads
let data: Vec<_> = threads
.into_iter()
.map(|(conversation_id, mut thread)| {
if let Some(title) = names.get(&conversation_id).cloned() {
@@ -3716,7 +3733,11 @@ impl CodexMessageProcessor {
thread
})
.collect();
let response = ThreadListResponse { data, next_cursor };
let response = ThreadListResponse {
data,
next_cursor,
backwards_cursor,
};
self.outgoing.send_response(request_id, response).await;
}
@@ -3832,7 +3853,7 @@ impl CodexMessageProcessor {
)));
};
let has_live_in_progress_turn = if let Some(loaded_thread) = loaded_thread {
let has_live_in_progress_turn = if let Some(loaded_thread) = loaded_thread.as_ref() {
matches!(loaded_thread.agent_status().await, AgentStatus::Running)
} else {
false
@@ -3959,6 +3980,157 @@ impl CodexMessageProcessor {
Ok(())
}
async fn thread_turns_list(
&self,
request_id: ConnectionRequestId,
params: ThreadTurnsListParams,
) {
let ThreadTurnsListParams {
thread_id,
cursor,
limit,
sort_direction,
} = params;
let thread_uuid = match ThreadId::from_string(&thread_id) {
Ok(id) => id,
Err(err) => {
self.send_invalid_request_error(request_id, format!("invalid thread id: {err}"))
.await;
return;
}
};
let state_db_ctx = get_state_db(&self.config).await;
let mut rollout_path = self
.resolve_rollout_path(thread_uuid, state_db_ctx.as_ref())
.await;
if rollout_path.is_none() {
rollout_path =
match find_thread_path_by_id_str(&self.config.codex_home, &thread_uuid.to_string())
.await
{
Ok(Some(path)) => Some(path),
Ok(None) => match find_archived_thread_path_by_id_str(
&self.config.codex_home,
&thread_uuid.to_string(),
)
.await
{
Ok(path) => path,
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate archived thread id {thread_uuid}: {err}"),
)
.await;
return;
}
},
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {thread_uuid}: {err}"),
)
.await;
return;
}
};
}
if rollout_path.is_none() {
match self.thread_manager.get_thread(thread_uuid).await {
Ok(thread) => {
rollout_path = thread.rollout_path();
if rollout_path.is_none() {
self.send_invalid_request_error(
request_id,
"ephemeral threads do not support thread/turns/list".to_string(),
)
.await;
return;
}
}
Err(_) => {
self.send_invalid_request_error(
request_id,
format!("thread not loaded: {thread_uuid}"),
)
.await;
return;
}
}
}
let Some(rollout_path) = rollout_path.as_ref() else {
self.send_internal_error(
request_id,
format!("failed to locate rollout for thread {thread_uuid}"),
)
.await;
return;
};
match read_rollout_items_from_rollout(rollout_path).await {
Ok(items) => {
// This API optimizes network transfer by letting clients page through a
// thread's turns incrementally, but it still replays the entire rollout on
// every request. Rollback and compaction events can change earlier turns, so
// the server has to rebuild the full turn list until turn metadata is indexed
// separately.
let has_live_in_progress_turn =
match self.thread_manager.get_thread(thread_uuid).await {
Ok(thread) => matches!(thread.agent_status().await, AgentStatus::Running),
Err(_) => false,
};
let turns = reconstruct_thread_turns_from_rollout_items(
&items,
self.thread_watch_manager
.loaded_status_for_thread(&thread_uuid.to_string())
.await,
has_live_in_progress_turn,
);
let page = match paginate_thread_turns(
turns,
cursor.as_deref(),
limit,
sort_direction.unwrap_or(SortDirection::Desc),
) {
Ok(page) => page,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return;
}
};
let response = ThreadTurnsListResponse {
data: page.turns,
next_cursor: page.next_cursor,
backwards_cursor: page.backwards_cursor,
};
self.outgoing.send_response(request_id, response).await;
}
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
self.send_invalid_request_error(
request_id,
format!(
"thread {thread_uuid} is not materialized yet; thread/turns/list is unavailable before first user message"
),
)
.await;
}
Err(err) => {
self.send_internal_error(
request_id,
format!(
"failed to load rollout `{}` for thread {thread_uuid}: {err}",
rollout_path.display()
),
)
.await;
}
}
}
pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver<ThreadId> {
self.thread_manager.subscribe_thread_created()
}
@@ -4980,6 +5152,7 @@ impl CodexMessageProcessor {
requested_page_size: usize,
cursor: Option<String>,
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
filters: ThreadListFilters,
) -> Result<(Vec<ConversationSummary>, Option<String>), JSONRPCErrorError> {
let ThreadListFilters {
@@ -5008,6 +5181,10 @@ impl CodexMessageProcessor {
let fallback_provider = self.config.model_provider_id.clone();
let (allowed_sources_vec, source_kind_filter) = compute_source_filters(source_kinds);
let allowed_sources = allowed_sources_vec.as_slice();
let store_sort_direction = match sort_direction {
SortDirection::Asc => StoreSortDirection::Asc,
SortDirection::Desc => StoreSortDirection::Desc,
};
while remaining > 0 {
let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
@@ -5017,6 +5194,7 @@ impl CodexMessageProcessor {
page_size,
cursor: cursor_obj.clone(),
sort_key,
sort_direction: store_sort_direction,
allowed_sources: allowed_sources.to_vec(),
model_providers: model_provider_filter.clone(),
archived,
@@ -5025,12 +5203,17 @@ impl CodexMessageProcessor {
.await
.map_err(thread_store_list_error)?;
let mut filtered = Vec::with_capacity(page.items.len());
let mut candidate_summaries = Vec::with_capacity(page.items.len());
for it in page.items {
let Some(summary) = summary_from_stored_thread(it, fallback_provider.as_str())
else {
continue;
};
candidate_summaries.push(summary);
}
let mut filtered = Vec::with_capacity(candidate_summaries.len());
for summary in candidate_summaries {
if source_kind_filter
.as_ref()
.is_none_or(|filter| source_kind_matches(&summary.source, filter))
@@ -9339,8 +9522,18 @@ fn summary_from_stored_thread(
conversation_id: thread.thread_id,
path,
preview: thread.first_user_message.unwrap_or(thread.preview),
timestamp: Some(thread.created_at.to_rfc3339_opts(SecondsFormat::Secs, true)),
updated_at: Some(thread.updated_at.to_rfc3339_opts(SecondsFormat::Secs, true)),
// Preserve millisecond precision from the thread store so thread/list cursors
// round-trip the same ordering key used by pagination queries.
timestamp: Some(
thread
.created_at
.to_rfc3339_opts(SecondsFormat::Millis, true),
),
updated_at: Some(
thread
.updated_at
.to_rfc3339_opts(SecondsFormat::Millis, true),
),
model_provider: if thread.model_provider.is_empty() {
fallback_provider.to_string()
} else {
@@ -9764,17 +9957,188 @@ pub(crate) fn summary_to_thread(
}
}
fn thread_backwards_cursor_for_sort_key(
summary: &ConversationSummary,
sort_key: StoreThreadSortKey,
sort_direction: SortDirection,
) -> Option<String> {
let timestamp = match sort_key {
StoreThreadSortKey::CreatedAt => summary.timestamp.as_deref(),
StoreThreadSortKey::UpdatedAt => summary
.updated_at
.as_deref()
.or(summary.timestamp.as_deref()),
};
let timestamp = parse_datetime(timestamp)?;
// The state DB stores unique millisecond timestamps. Offset the reverse cursor by one
// millisecond so the opposite-direction query includes the page anchor.
let timestamp = match sort_direction {
SortDirection::Asc => timestamp.checked_add_signed(ChronoDuration::milliseconds(1))?,
SortDirection::Desc => timestamp.checked_sub_signed(ChronoDuration::milliseconds(1))?,
};
Some(timestamp.to_rfc3339_opts(SecondsFormat::Millis, true))
}
struct ThreadTurnsPage {
turns: Vec<Turn>,
next_cursor: Option<String>,
backwards_cursor: Option<String>,
}
#[derive(serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct ThreadTurnsCursor {
turn_id: String,
include_anchor: bool,
}
fn paginate_thread_turns(
turns: Vec<Turn>,
cursor: Option<&str>,
limit: Option<u32>,
sort_direction: SortDirection,
) -> Result<ThreadTurnsPage, JSONRPCErrorError> {
if turns.is_empty() {
return Ok(ThreadTurnsPage {
turns: Vec::new(),
next_cursor: None,
backwards_cursor: None,
});
}
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 anchor_index = anchor
.as_ref()
.and_then(|anchor| turns.iter().position(|turn| turn.id == anchor.turn_id));
if anchor.is_some() && anchor_index.is_none() {
return Err(JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: "invalid cursor: anchor turn is no longer present".to_string(),
data: None,
});
}
let mut keyed_turns: Vec<_> = turns.into_iter().enumerate().collect();
match sort_direction {
SortDirection::Asc => {
if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) {
keyed_turns.retain(|(index, _)| {
if anchor.include_anchor {
*index >= anchor_index
} else {
*index > anchor_index
}
});
}
}
SortDirection::Desc => {
keyed_turns.reverse();
if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) {
keyed_turns.retain(|(index, _)| {
if anchor.include_anchor {
*index <= anchor_index
} else {
*index < anchor_index
}
});
}
}
}
let more_turns_available = keyed_turns.len() > page_size;
keyed_turns.truncate(page_size);
let backwards_cursor = keyed_turns
.first()
.map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, /*include_anchor*/ true))
.transpose()?;
let next_cursor = if more_turns_available {
keyed_turns
.last()
.map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, /*include_anchor*/ false))
.transpose()?
} else {
None
};
let turns = keyed_turns.into_iter().map(|(_, turn)| turn).collect();
Ok(ThreadTurnsPage {
turns,
next_cursor,
backwards_cursor,
})
}
fn serialize_thread_turns_cursor(
turn_id: &str,
include_anchor: bool,
) -> Result<String, JSONRPCErrorError> {
serde_json::to_string(&ThreadTurnsCursor {
turn_id: turn_id.to_string(),
include_anchor,
})
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to serialize cursor: {err}"),
data: None,
})
}
fn parse_thread_turns_cursor(cursor: &str) -> Result<ThreadTurnsCursor, JSONRPCErrorError> {
serde_json::from_str(cursor).map_err(|_| JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("invalid cursor: {cursor}"),
data: None,
})
}
fn reconstruct_thread_turns_from_rollout_items(
items: &[RolloutItem],
loaded_status: ThreadStatus,
has_live_in_progress_turn: bool,
) -> Vec<Turn> {
let mut turns = build_turns_from_rollout_items(items);
normalize_thread_turns_status(&mut turns, loaded_status, has_live_in_progress_turn);
turns
}
fn normalize_thread_turns_status(
turns: &mut [Turn],
loaded_status: ThreadStatus,
has_live_in_progress_turn: bool,
) {
let status = resolve_thread_status(loaded_status, has_live_in_progress_turn);
if matches!(status, ThreadStatus::Active { .. }) {
return;
}
for turn in turns {
if matches!(turn.status, TurnStatus::InProgress) {
turn.status = TurnStatus::Interrupted;
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::outgoing_message::OutgoingEnvelope;
use crate::outgoing_message::OutgoingMessage;
use anyhow::Result;
use chrono::DateTime;
use chrono::Utc;
use codex_app_server_protocol::ServerRequestPayload;
use codex_app_server_protocol::ToolRequestUserInputParams;
use codex_protocol::ThreadId;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_thread_store::StoredThread;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
use pretty_assertions::assert_eq;
@@ -9824,6 +10188,53 @@ mod tests {
validate_dynamic_tools(&tools).expect("valid schema");
}
#[test]
fn summary_from_stored_thread_preserves_millisecond_precision() {
let created_at =
DateTime::parse_from_rfc3339("2025-01-02T03:04:05.678Z").expect("valid timestamp");
let updated_at =
DateTime::parse_from_rfc3339("2025-01-02T03:04:06.789Z").expect("valid timestamp");
let thread_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000123").expect("valid thread");
let stored_thread = StoredThread {
thread_id,
rollout_path: Some(PathBuf::from("/tmp/thread.jsonl")),
forked_from_id: None,
preview: "preview".to_string(),
name: None,
model_provider: "openai".to_string(),
model: None,
reasoning_effort: None,
created_at: created_at.with_timezone(&Utc),
updated_at: updated_at.with_timezone(&Utc),
archived_at: None,
cwd: PathBuf::from("/tmp"),
cli_version: "0.0.0".to_string(),
source: SessionSource::Cli,
agent_nickname: None,
agent_role: None,
agent_path: None,
git_info: None,
approval_mode: AskForApproval::OnRequest,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
token_usage: None,
first_user_message: Some("first user message".to_string()),
history: None,
};
let summary =
summary_from_stored_thread(stored_thread, "fallback").expect("summary should exist");
assert_eq!(
summary.timestamp.as_deref(),
Some("2025-01-02T03:04:05.678Z")
);
assert_eq!(
summary.updated_at.as_deref(),
Some("2025-01-02T03:04:06.789Z")
);
}
#[test]
fn config_load_error_marks_cloud_requirements_failures_for_relogin() {
let err = std::io::Error::other(CloudRequirementsLoadError::new(