diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 60b80f6b54..c811600717 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -12,6 +12,13 @@ exclusions and filters with each returned cursor. Clients that display a saved manual order can load those summaries separately, then exclude the entire saved ID set when listing the remainder. This field does not save or sort that order. +DB-only requests (`useStateDbOnly: true`) return JSON-RPC error `-32603` +when the local state DB cannot serve the query, even for an empty cwd filter. +Listing also returns `-32603` if the store repeats a cursor while filling a page. +These errors are not evidence that history is exhausted; callers can retry. +A successful response with `nextCursor: null` still indicates exhaustion. +Default scan-and-repair requests retain their filesystem fallback. + # Guardian circuit-breaker errors Set `auto_review.circuit_break_action = "strict"` to include `TooManyDenials` in diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 0b952c23d4..9bb4b7f6f5 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -5503,7 +5503,7 @@ impl ThreadRequestProcessor { excluded_thread_ids, } = filters; let mut cursor_obj = cursor; - let mut last_cursor = cursor_obj.clone(); + let mut seen_cursors: HashSet = cursor_obj.iter().cloned().collect(); let mut remaining = requested_page_size; let mut items = Vec::with_capacity(requested_page_size); let mut next_cursor: Option = None; @@ -5580,20 +5580,16 @@ impl ThreadRequestProcessor { remaining = requested_page_size.saturating_sub(items.len()); next_cursor = page.next_cursor; - if remaining == 0 { - break; - } - let Some(cursor_val) = next_cursor.clone() else { break; }; - // Break if our pagination would reuse the same cursor again; this avoids - // an infinite loop when filtering drops everything on the page. - if last_cursor.as_ref() == Some(&cursor_val) { - next_cursor = None; + // Check full pages too: returning a repeated cursor would loop on the next request. + if !seen_cursors.insert(cursor_val.clone()) { + return Err(internal_error("thread listing returned a repeated cursor")); + } + if remaining == 0 { break; } - last_cursor = Some(cursor_val.clone()); cursor_obj = Some(cursor_val); } diff --git a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs index 8284d0f137..34883c89d0 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs @@ -190,6 +190,106 @@ async fn thread_start_defaults_to_legacy_without_history_list_support() -> Resul Ok(()) } +#[tokio::test] +async fn thread_list_repeated_cursor_returns_internal_error() -> Result<()> { + let codex_home = TempDir::new()?; + let store_id = Uuid::new_v4().to_string(); + create_config_toml_with_thread_store(codex_home.path(), "http://127.0.0.1:1", &store_id)?; + let thread_store = InMemoryThreadStore::for_id(store_id.clone()); + let _in_memory_store = InMemoryThreadStoreId { store_id }; + thread_store + .repeat_list_threads_cursor_for_testing("repeated-cursor") + .await; + let client = start_in_process_server(codex_home.path()).await?; + + let error = client + .request(ClientRequest::ThreadList { + request_id: RequestId::Integer(1), + params: ThreadListParams { + excluded_thread_ids: None, + originators: None, + cursor: None, + limit: Some(1), + sort_key: None, + sort_direction: None, + model_providers: Some(Vec::new()), + source_kinds: None, + archived: None, + section_id: None, + project_id: None, + cwd: None, + use_state_db_only: false, + search_term: None, + parent_thread_id: None, + ancestor_thread_id: None, + }, + }) + .await? + .expect_err("a repeated refill cursor must fail thread/list"); + assert_eq!(error.code, -32603); + assert_eq!(error.message, "thread listing returned a repeated cursor"); + + let thread_id = ThreadId::new(); + thread_store + .create_thread(StoreCreateThreadParams { + creator_user_id: None, + creator_account_id: None, + session_id: thread_id.into(), + thread_id, + extra_config: None, + forked_from_id: None, + parent_thread_id: None, + source: SessionSource::Cli, + thread_source: None, + originator: "test_originator".to_string(), + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + selected_capability_roots: Vec::new(), + multi_agent_version: None, + history_mode: Default::default(), + history_base: None, + subagent_history_start_ordinal: None, + initial_window_id: Uuid::now_v7().to_string(), + runtime_workspace_roots: None, + metadata: ThreadPersistenceMetadata { + cwd: Some(codex_home.path().to_path_buf()), + model_provider: "mock_provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }) + .await?; + + let error = client + .request(ClientRequest::ThreadList { + request_id: RequestId::Integer(2), + params: ThreadListParams { + excluded_thread_ids: None, + originators: None, + cursor: Some("repeated-cursor".to_string()), + limit: Some(1), + sort_key: None, + sort_direction: None, + model_providers: Some(Vec::new()), + source_kinds: None, + archived: None, + section_id: None, + project_id: None, + cwd: None, + use_state_db_only: false, + search_term: None, + parent_thread_id: None, + ancestor_thread_id: None, + }, + }) + .await? + .expect_err("a repeated full-page cursor must fail thread/list"); + assert_eq!(error.code, -32603); + assert_eq!(error.message, "thread listing returned a repeated cursor"); + + client.shutdown().await?; + Ok(()) +} + #[tokio::test] async fn thread_attachment_operations_without_sqlite_return_method_not_found() -> Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index 95df4fb098..35dcb81241 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -1144,6 +1144,122 @@ async fn thread_search_filters_by_source_kind() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_list_db_only_errors_distinguish_unavailable_database_from_empty_history() +-> Result<()> { + use codex_app_server::in_process; + use codex_app_server::in_process::InProcessStartArgs; + use codex_app_server_protocol::ClientInfo; + use codex_app_server_protocol::InitializeParams; + use codex_arg0::Arg0DispatchPaths; + use codex_config::CloudConfigBundleLoader; + use codex_config::LoaderOverrides; + use codex_config::NoopThreadConfigLoader; + use codex_core::config::ConfigBuilder; + use codex_exec_server::EnvironmentManager; + use codex_feedback::CodexFeedback; + use std::sync::Arc; + + let home = TempDir::new()?; + let other_home = TempDir::new()?; + create_minimal_config(home.path())?; + let loader_overrides = LoaderOverrides::without_managed_config_for_tests(); + let config = Arc::new( + ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .fallback_cwd(Some(home.path().to_path_buf())) + .loader_overrides(loader_overrides.clone()) + .build() + .await?, + ); + let right_db = codex_state::StateRuntime::init( + config.sqlite_config().clone(), + config.model_provider_id.clone(), + ) + .await?; + let wrong_db = codex_state::StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(other_home.path().abs()), + config.model_provider_id.clone(), + ) + .await?; + // A rollout file must not conceal a failed database-only read. + create_fake_rollout( + home.path(), + "2025-01-03T12-00-00", + "2025-01-03T12:00:00Z", + "only in the rollout", + Some("mock_provider"), + /*git_info*/ None, + )?; + + for (state_db, should_succeed) in [ + (None, false), + (Some(wrong_db), false), + (Some(right_db), true), + ] { + let client = in_process::start(InProcessStartArgs { + arg0_paths: Arg0DispatchPaths::default(), + config: Arc::clone(&config), + cli_overrides: Vec::new(), + loader_overrides: loader_overrides.clone(), + strict_config: false, + cloud_config_bundle: CloudConfigBundleLoader::default(), + embedded_network_policy: Default::default(), + thread_config_loader: Arc::new(NoopThreadConfigLoader), + feedback: CodexFeedback::new(), + log_db: None, + state_db, + environment_manager: Arc::new(EnvironmentManager::default_for_tests()), + config_warnings: Vec::new(), + session_source: CoreSessionSource::Cli, + enable_codex_api_key_env: false, + initialize: InitializeParams { + client_info: ClientInfo { + name: "codex-app-server-tests".into(), + title: None, + version: "0.1.0".into(), + }, + capabilities: None, + }, + channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY, + }) + .await?; + for archived in [false, true] { + for cwd in [json!(null), json!([])] { + let result = client + .request(ClientRequest::ThreadList { + request_id: RequestId::Integer(1), + params: serde_json::from_value(json!({ + "useStateDbOnly": true, + "modelProviders": [], + "archived": archived, + "cwd": cwd, + }))?, + }) + .await?; + if should_succeed { + let page: ThreadListResponse = serde_json::from_value( + result.expect("a healthy, empty database should list successfully"), + )?; + assert_eq!( + page, + ThreadListResponse { + data: vec![], + next_cursor: None, + backwards_cursor: None + } + ); + } else { + let error = + result.expect_err("an unavailable database must not look exhausted"); + assert_eq!(error.code, -32603); + } + } + } + } + Ok(()) +} + #[tokio::test] async fn thread_list_state_db_only_returns_sqlite_without_jsonl_repair() -> Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index f99f2c5a89..765d83ccc7 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -546,10 +546,6 @@ impl RolloutRecorder { ThreadListArchiveFilter::Active => false, ThreadListArchiveFilter::Archived => true, }; - if cwd_filters.is_some_and(<[std::path::PathBuf]>::is_empty) { - return Ok(ThreadsPage::default()); - } - if matches!(repair_mode, ThreadListRepairMode::StateDbOnly) { return state_db::list_threads_db( state_db_ctx.as_deref(), @@ -571,6 +567,9 @@ impl RolloutRecorder { .map(Into::into) .ok_or_else(|| std::io::Error::other("failed to list threads from state database")); } + if cwd_filters.is_some_and(<[std::path::PathBuf]>::is_empty) { + return Ok(ThreadsPage::default()); + } let listing_has_metadata_filters = !allowed_sources.is_empty() || model_providers.is_some() diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 152c1d0021..898e6fdfde 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -526,6 +526,7 @@ pub struct InMemoryThreadStore { #[derive(Default)] struct InMemoryThreadStoreState { calls: InMemoryThreadStoreCalls, + repeated_list_threads_cursor: Option, created_threads: HashMap, histories: HashMap>, metadata_updates: HashMap, @@ -566,6 +567,11 @@ impl InMemoryThreadStore { self.state.lock().await.calls.clone() } + /// Makes every thread-list page return the same cursor for public API error-path tests. + pub async fn repeat_list_threads_cursor_for_testing(&self, cursor: impl Into) { + self.state.lock().await.repeated_list_threads_cursor = Some(cursor.into()); + } + /// Makes metadata updates apply normally while returning no materialized thread. pub fn omit_metadata_update_result_for_testing(&self) { self.omit_metadata_update_result @@ -753,7 +759,7 @@ impl InMemoryThreadStore { items.sort_by_key(|item| item.thread_id.to_string()); Ok(ThreadPage { items, - next_cursor: None, + next_cursor: state.repeated_list_threads_cursor.clone(), }) }