Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions codex-rs/app-server-protocol/src/protocol/v2.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1205,6 +1205,9 @@ pub struct ThreadListParams {
/// Optional provider filter; when set, only sessions recorded under these
/// providers are returned. When present but empty, includes all providers.
pub model_providers: Option<Vec<String>>,
/// Optional archived filter; when set to true, only archived threads are returned.
/// If false or null, only non-archived threads are returned.
pub archived: Option<bool>,
}

#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, JsonSchema, TS)]
Expand Down
3 changes: 2 additions & 1 deletion codex-rs/app-server/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -147,6 +147,7 @@ To branch from a stored session, call `thread/fork` with the `thread.id`. This c
- `limit` — server defaults to a reasonable page size if unset.
- `sortKey` — `created_at` (default) or `updated_at`.
- `modelProviders` — restrict results to specific providers; unset, null, or an empty array will include all providers.
- `archived` — when `true`, list archived threads only. When `false` or `null`, list non-archived threads (default).

Example:

Expand Down Expand Up @@ -187,7 +188,7 @@ Use `thread/archive` to move the persisted rollout (stored as a JSONL file on di
{ "id": 21, "result": {} }
```

An archived thread will not appear in future calls to `thread/list`.
An archived thread will not appear in `thread/list` unless `archived` is set to `true`.

### Example: Start a turn (send user input)

Expand Down
59 changes: 43 additions & 16 deletions codex-rs/app-server/src/codex_message_processor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1621,6 +1621,7 @@ impl CodexMessageProcessor {
limit,
sort_key,
model_providers,
archived,
} = params;

let requested_page_size = limit
Expand All @@ -1632,7 +1633,13 @@ impl CodexMessageProcessor {
ThreadSortKey::UpdatedAt => CoreThreadSortKey::UpdatedAt,
};
let (summaries, next_cursor) = match self
.list_threads_common(requested_page_size, cursor, model_providers, core_sort_key)
.list_threads_common(
requested_page_size,
cursor,
model_providers,
core_sort_key,
archived.unwrap_or(false),
)
.await
{
Ok(r) => r,
Expand Down Expand Up @@ -2198,6 +2205,7 @@ impl CodexMessageProcessor {
cursor,
model_providers,
CoreThreadSortKey::UpdatedAt,
false,
)
.await
{
Expand All @@ -2217,6 +2225,7 @@ impl CodexMessageProcessor {
cursor: Option<String>,
model_providers: Option<Vec<String>>,
sort_key: CoreThreadSortKey,
archived: bool,
) -> Result<(Vec<ConversationSummary>, Option<String>), JSONRPCErrorError> {
let mut cursor_obj: Option<RolloutCursor> = match cursor.as_ref() {
Some(cursor_str) => {
Expand Down Expand Up @@ -2247,21 +2256,39 @@ impl CodexMessageProcessor {

while remaining > 0 {
let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
let page = RolloutRecorder::list_threads(
&self.config.codex_home,
page_size,
cursor_obj.as_ref(),
sort_key,
INTERACTIVE_SESSION_SOURCES,
model_provider_filter.as_deref(),
fallback_provider.as_str(),
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to list threads: {err}"),
data: None,
})?;
let page = if archived {
RolloutRecorder::list_archived_threads(
&self.config.codex_home,
page_size,
cursor_obj.as_ref(),
sort_key,
INTERACTIVE_SESSION_SOURCES,
model_provider_filter.as_deref(),
fallback_provider.as_str(),
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to list threads: {err}"),
data: None,
})?
} else {
RolloutRecorder::list_threads(
&self.config.codex_home,
page_size,
cursor_obj.as_ref(),
sort_key,
INTERACTIVE_SESSION_SOURCES,
model_provider_filter.as_deref(),
fallback_provider.as_str(),
)
.await
.map_err(|err| JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to list threads: {err}"),
data: None,
})?
};

let mut filtered = page
.items
Expand Down
82 changes: 81 additions & 1 deletion codex-rs/app-server/tests/suite/v2/thread_list.rs
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,11 @@ use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadSortKey;
use codex_core::ARCHIVED_SESSIONS_SUBDIR;
use codex_protocol::protocol::GitInfo as CoreGitInfo;
use pretty_assertions::assert_eq;
use std::cmp::Reverse;
use std::fs;
use std::fs::FileTimes;
use std::fs::OpenOptions;
use std::path::Path;
Expand All @@ -36,8 +38,9 @@ async fn list_threads(
cursor: Option<String>,
limit: Option<u32>,
providers: Option<Vec<String>>,
archived: Option<bool>,
) -> Result<ThreadListResponse> {
list_threads_with_sort(mcp, cursor, limit, providers, None).await
list_threads_with_sort(mcp, cursor, limit, providers, None, archived).await
}

async fn list_threads_with_sort(
Expand All @@ -46,13 +49,15 @@ async fn list_threads_with_sort(
limit: Option<u32>,
providers: Option<Vec<String>>,
sort_key: Option<ThreadSortKey>,
archived: Option<bool>,
) -> Result<ThreadListResponse> {
let request_id = mcp
.send_thread_list_request(codex_app_server_protocol::ThreadListParams {
cursor,
limit,
sort_key,
model_providers: providers,
archived,
})
.await?;
let resp: JSONRPCResponse = timeout(
Expand Down Expand Up @@ -125,6 +130,7 @@ async fn thread_list_basic_empty() -> Result<()> {
None,
Some(10),
Some(vec!["mock_provider".to_string()]),
None,
)
.await?;
assert!(data.is_empty());
Expand Down Expand Up @@ -187,6 +193,7 @@ async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> {
None,
Some(2),
Some(vec!["mock_provider".to_string()]),
None,
)
.await?;
assert_eq!(data1.len(), 2);
Expand All @@ -211,6 +218,7 @@ async fn thread_list_pagination_next_cursor_none_on_last_page() -> Result<()> {
Some(cursor1),
Some(2),
Some(vec!["mock_provider".to_string()]),
None,
)
.await?;
assert!(data2.len() <= 2);
Expand Down Expand Up @@ -260,6 +268,7 @@ async fn thread_list_respects_provider_filter() -> Result<()> {
None,
Some(10),
Some(vec!["other_provider".to_string()]),
None,
)
.await?;
assert_eq!(data.len(), 1);
Expand Down Expand Up @@ -309,6 +318,7 @@ async fn thread_list_fetches_until_limit_or_exhausted() -> Result<()> {
None,
Some(8),
Some(vec!["target_provider".to_string()]),
None,
)
.await?;
assert_eq!(
Expand Down Expand Up @@ -353,6 +363,7 @@ async fn thread_list_enforces_max_limit() -> Result<()> {
None,
Some(200),
Some(vec!["mock_provider".to_string()]),
None,
)
.await?;
assert_eq!(
Expand Down Expand Up @@ -398,6 +409,7 @@ async fn thread_list_stops_when_not_enough_filtered_results_exist() -> Result<()
None,
Some(10),
Some(vec!["target_provider".to_string()]),
None,
)
.await?;
assert_eq!(
Expand Down Expand Up @@ -444,6 +456,7 @@ async fn thread_list_includes_git_info() -> Result<()> {
None,
Some(10),
Some(vec!["mock_provider".to_string()]),
None,
)
.await?;
let thread = data
Expand Down Expand Up @@ -502,6 +515,7 @@ async fn thread_list_default_sorts_by_created_at() -> Result<()> {
Some(10),
Some(vec!["mock_provider".to_string()]),
None,
None,
)
.await?;

Expand Down Expand Up @@ -562,6 +576,7 @@ async fn thread_list_sort_updated_at_orders_by_mtime() -> Result<()> {
Some(10),
Some(vec!["mock_provider".to_string()]),
Some(ThreadSortKey::UpdatedAt),
None,
)
.await?;

Expand Down Expand Up @@ -625,6 +640,7 @@ async fn thread_list_updated_at_paginates_with_cursor() -> Result<()> {
Some(2),
Some(vec!["mock_provider".to_string()]),
Some(ThreadSortKey::UpdatedAt),
None,
)
.await?;
let ids_page1: Vec<_> = page1.iter().map(|thread| thread.id.as_str()).collect();
Expand All @@ -640,6 +656,7 @@ async fn thread_list_updated_at_paginates_with_cursor() -> Result<()> {
Some(2),
Some(vec!["mock_provider".to_string()]),
Some(ThreadSortKey::UpdatedAt),
None,
)
.await?;
let ids_page2: Vec<_> = page2.iter().map(|thread| thread.id.as_str()).collect();
Expand Down Expand Up @@ -678,6 +695,7 @@ async fn thread_list_created_at_tie_breaks_by_uuid() -> Result<()> {
None,
Some(10),
Some(vec!["mock_provider".to_string()]),
None,
)
.await?;

Expand Down Expand Up @@ -730,6 +748,7 @@ async fn thread_list_updated_at_tie_breaks_by_uuid() -> Result<()> {
Some(10),
Some(vec!["mock_provider".to_string()]),
Some(ThreadSortKey::UpdatedAt),
None,
)
.await?;

Expand Down Expand Up @@ -769,6 +788,7 @@ async fn thread_list_updated_at_uses_mtime() -> Result<()> {
Some(10),
Some(vec!["mock_provider".to_string()]),
Some(ThreadSortKey::UpdatedAt),
None,
)
.await?;

Expand All @@ -786,6 +806,65 @@ async fn thread_list_updated_at_uses_mtime() -> Result<()> {
Ok(())
}

#[tokio::test]
async fn thread_list_archived_filter() -> Result<()> {
let codex_home = TempDir::new()?;
create_minimal_config(codex_home.path())?;

let active_id = create_fake_rollout(
codex_home.path(),
"2025-03-01T10-00-00",
"2025-03-01T10:00:00Z",
"Active",
Some("mock_provider"),
None,
)?;
let archived_id = create_fake_rollout(
codex_home.path(),
"2025-03-01T09-00-00",
"2025-03-01T09:00:00Z",
"Archived",
Some("mock_provider"),
None,
)?;

let archived_dir = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR);
fs::create_dir_all(&archived_dir)?;
let archived_source = rollout_path(codex_home.path(), "2025-03-01T09-00-00", &archived_id);
let archived_dest = archived_dir.join(
archived_source
.file_name()
.expect("archived rollout should have a file name"),
);
fs::rename(&archived_source, &archived_dest)?;

let mut mcp = init_mcp(codex_home.path()).await?;

let ThreadListResponse { data, .. } = list_threads(
&mut mcp,
None,
Some(10),
Some(vec!["mock_provider".to_string()]),
None,
)
.await?;
assert_eq!(data.len(), 1);
assert_eq!(data[0].id, active_id);

let ThreadListResponse { data, .. } = list_threads(
&mut mcp,
None,
Some(10),
Some(vec!["mock_provider".to_string()]),
Some(true),
)
.await?;
assert_eq!(data.len(), 1);
assert_eq!(data[0].id, archived_id);

Ok(())
}

#[tokio::test]
async fn thread_list_invalid_cursor_returns_error() -> Result<()> {
let codex_home = TempDir::new()?;
Expand All @@ -799,6 +878,7 @@ async fn thread_list_invalid_cursor_returns_error() -> Result<()> {
limit: Some(2),
sort_key: None,
model_providers: Some(vec!["mock_provider".to_string()]),
archived: None,
})
.await?;
let error: JSONRPCError = timeout(
Expand Down
Loading
Loading