From 98262bb9336a484fa09998b4d185ced120622f3e Mon Sep 17 00:00:00 2001 From: Brent Traut Date: Thu, 4 Jun 2026 09:46:19 -0700 Subject: [PATCH] codex: address PR review feedback (#26009) --- .../v2/ThreadCatalogChangedNotification.json | 5 -- codex-rs/app-server/README.md | 2 +- .../thread_catalog_subscription.rs | 3 + .../request_processors/thread_processor.rs | 6 ++ .../src/request_processors/thread_summary.rs | 7 ++ .../suite/v2/thread_catalog_subscription.rs | 69 +++++++++++++++---- 6 files changed, 74 insertions(+), 18 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json index c9ba34a7b2..7a42120171 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json @@ -151,11 +151,6 @@ "type": "string" }, "ThreadSource": { - "enum": [ - "user", - "subagent", - "memory_consolidation" - ], "type": "string" }, "ThreadSummary": { diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 561659d5d7..0a1db57621 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -421,7 +421,7 @@ Enable `capabilities.experimentalApi` during initialization, then use `thread/li } } ``` -`thread/catalog/subscribe` watches all future thread metadata changes without draining existing history. Install the subscription before calling `thread/list` when you need a race-free sidebar bootstrap; after the subscribe response arrives, buffer catalog notifications until the list response arrives, then apply the complete summaries idempotently and filter them client-side. The summary contains sidebar metadata such as id, preview/name, cwd, timestamps, archive state, git info, and source; it does not contain turns, items, messages, deltas, tool state, status, or runtime history. +`thread/catalog/subscribe` watches all future thread metadata changes without draining existing history. For a race-free sidebar bootstrap, start buffering `thread/catalog/changed` notifications before sending the subscribe request. After the subscribe response arrives, call `thread/list`, then apply the buffered complete summaries idempotently and filter them client-side. The summary contains sidebar metadata such as id, preview/name, cwd, timestamps, archive state, git info, and source; it does not contain turns, items, messages, deltas, tool state, status, or runtime history. ```json { "method": "thread/catalog/subscribe", "id": 21 } diff --git a/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs b/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs index a59d0865aa..1192e9c6c6 100644 --- a/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs +++ b/codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs @@ -62,6 +62,9 @@ impl ThreadCatalogSubscriptions { fallback_provider: &str, fallback_cwd: &AbsolutePathBuf, ) { + if self.connection_ids.lock().await.is_empty() { + return; + } let stored_thread = match thread_store .read_thread(StoreReadThreadParams { thread_id, 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 0f2bd6a206..9025156cd2 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1309,6 +1309,7 @@ impl ThreadRequestProcessor { reasoning_effort: config_snapshot.reasoning_effort, multi_agent_mode: config_snapshot.multi_agent_mode, }; + let catalog_summary = thread_summary_from_thread(thread.clone(), /*archived_at*/ None); let notif = thread_started_notification(thread); listener_task_context .outgoing @@ -1319,6 +1320,11 @@ impl ThreadRequestProcessor { )) .await; + listener_task_context + .thread_catalog_subscriptions + .publish_thread_summary(catalog_summary) + .await; + listener_task_context .outgoing .send_server_notification(ServerNotification::ThreadStarted(notif)) diff --git a/codex-rs/app-server/src/request_processors/thread_summary.rs b/codex-rs/app-server/src/request_processors/thread_summary.rs index bf04c2098d..d67da42d8f 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -282,6 +282,13 @@ pub(super) fn thread_summary_from_stored_thread( ) -> ThreadSummary { let archived_at = thread.archived_at.map(|dt| dt.timestamp()); let (thread, _) = thread_from_stored_thread(thread, fallback_provider, fallback_cwd); + thread_summary_from_thread(thread, archived_at) +} + +pub(super) fn thread_summary_from_thread( + thread: Thread, + archived_at: Option, +) -> ThreadSummary { ThreadSummary { id: thread.id, session_id: thread.session_id, diff --git a/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs b/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs index 232e4a360f..ff5a006866 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs @@ -12,6 +12,8 @@ use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadSetNameParams; use codex_app_server_protocol::ThreadSetNameResponse; use codex_app_server_protocol::ThreadSortKey; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::ThreadStartResponse; use pretty_assertions::assert_eq; use serde::de::DeserializeOwned; use tempfile::TempDir; @@ -20,18 +22,48 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10); +#[tokio::test] +async fn catalog_subscription_reports_new_empty_thread() -> Result<()> { + let codex_home = TempDir::new()?; + let mut app = start_app(&codex_home).await?; + + let subscribe_id = app + .send_raw_request("thread/catalog/subscribe", /*params*/ None) + .await?; + let _: codex_app_server_protocol::ThreadCatalogSubscribeResponse = + read_response(&mut app, subscribe_id).await?; + + let start_id = app + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let started = read_response::(&mut app, start_id).await?; + let notification: JSONRPCNotification = timeout( + DEFAULT_READ_TIMEOUT, + app.read_stream_until_notification_message("thread/catalog/changed"), + ) + .await??; + let changed: ThreadCatalogChangedNotification = serde_json::from_value( + notification + .params + .expect("thread/catalog/changed should have params"), + )?; + + assert_eq!(changed.thread.id, started.thread.id); + assert_eq!(changed.thread.preview, ""); + assert!( + !started + .thread + .path + .expect("thread path should be present") + .exists() + ); + + Ok(()) +} + #[tokio::test] async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> { let codex_home = TempDir::new()?; - write_mock_responses_config_toml( - codex_home.path(), - "http://localhost:1", - &Default::default(), - i64::MAX, - /*requires_openai_auth*/ None, - "mock_provider", - "", - )?; let older_thread_id = create_fake_rollout( codex_home.path(), "2025-01-05T12-00-00", @@ -48,9 +80,7 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> Some("mock_provider"), /*git_info*/ None, )?; - - let mut app = TestAppServer::new(codex_home.path()).await?; - timeout(DEFAULT_READ_TIMEOUT, app.initialize()).await??; + let mut app = start_app(&codex_home).await?; let subscribe_id = app .send_raw_request("thread/catalog/subscribe", /*params*/ None) @@ -104,6 +134,21 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> Ok(()) } +async fn start_app(codex_home: &TempDir) -> Result { + write_mock_responses_config_toml( + codex_home.path(), + "http://localhost:1", + &Default::default(), + i64::MAX, + /*requires_openai_auth*/ None, + "mock_provider", + "", + )?; + let mut app = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, app.initialize()).await??; + Ok(app) +} + async fn rename_thread(app: &mut TestAppServer, thread_id: String, name: &str) -> Result<()> { let rename_id = app .send_thread_set_name_request(ThreadSetNameParams {