From aea5bfcabfe1be3a94b3aa711b0ff6a4ed8716a2 Mon Sep 17 00:00:00 2001 From: Brent Traut Date: Mon, 22 Jun 2026 11:45:56 -0700 Subject: [PATCH] Address thread catalog review feedback --- .../schema/json/ServerNotification.json | 14 ++++++ .../codex_app_server_protocol.schemas.json | 14 ++++++ .../codex_app_server_protocol.v2.schemas.json | 14 ++++++ .../v2/ThreadCatalogChangedNotification.json | 14 ++++++ .../schema/typescript/v2/ThreadSummary.ts | 2 +- .../src/protocol/v2/thread_data.rs | 3 ++ codex-rs/app-server/README.md | 4 +- .../thread_catalog_subscription.rs | 34 --------------- .../request_processors/thread_lifecycle.rs | 12 +----- .../request_processors/thread_processor.rs | 32 ++++++++------ .../src/request_processors/thread_summary.rs | 6 +++ .../suite/v2/thread_catalog_subscription.rs | 43 ++++--------------- 12 files changed, 95 insertions(+), 97 deletions(-) diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index eacd4ced53..886f72b7ad 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -5094,6 +5094,20 @@ "preview": { "type": "string" }, + "recencyAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "recencyAtMs": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "sessionId": { "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 3e6dd79cb4..ce67d57b46 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -19664,6 +19664,20 @@ "preview": { "type": "string" }, + "recencyAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "recencyAtMs": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "sessionId": { "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 2b13aa04f5..73c32766b2 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -17443,6 +17443,20 @@ "preview": { "type": "string" }, + "recencyAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "recencyAtMs": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "sessionId": { "type": "string" }, 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 56d0e12f49..e3bc3e1835 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json @@ -234,6 +234,20 @@ "preview": { "type": "string" }, + "recencyAt": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, + "recencyAtMs": { + "format": "int64", + "type": [ + "integer", + "null" + ] + }, "sessionId": { "type": "string" }, diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts index ee4ed5bf17..8758ab9f36 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts @@ -6,4 +6,4 @@ import type { GitInfo } from "./GitInfo"; import type { SessionSource } from "./SessionSource"; import type { ThreadSource } from "./ThreadSource"; -export type ThreadSummary = { id: string, sessionId: string, forkedFromId: string | null, parentThreadId: string | null, preview: string, ephemeral: boolean, modelProvider: string, createdAt: number, createdAtMs: number, updatedAt: number, updatedAtMs: number, archivedAt: number | null, path: string | null, cwd: AbsolutePathBuf, cliVersion: string, source: SessionSource, threadSource: ThreadSource | null, agentNickname: string | null, agentRole: string | null, gitInfo: GitInfo | null, name: string | null, }; +export type ThreadSummary = { id: string, sessionId: string, forkedFromId: string | null, parentThreadId: string | null, preview: string, ephemeral: boolean, modelProvider: string, createdAt: number, createdAtMs: number, updatedAt: number, updatedAtMs: number, recencyAt: bigint | null, recencyAtMs: number | null, archivedAt: number | null, path: string | null, cwd: AbsolutePathBuf, cliVersion: string, source: SessionSource, threadSource: ThreadSource | null, agentNickname: string | null, agentRole: string | null, gitInfo: GitInfo | null, name: string | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs index 5446821056..e3e85057f9 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs @@ -212,6 +212,9 @@ pub struct ThreadSummary { pub updated_at: i64, #[ts(type = "number")] pub updated_at_ms: i64, + pub recency_at: Option, + #[ts(type = "number | null")] + pub recency_at_ms: Option, #[ts(type = "number | null")] pub archived_at: Option, pub path: Option, diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 31cb4b13fb..405ac6c476 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -421,13 +421,13 @@ Enable `capabilities.experimentalApi` during initialization, then use `thread/li } } ``` -`threadCatalog/subscribe` watches all future thread metadata changes without draining existing history. For a race-free sidebar bootstrap, start buffering `threadCatalog/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, second and millisecond timestamps, archive state, git info, and source; it does not contain turns, items, messages, deltas, tool state, status, or runtime history. +`threadCatalog/subscribe` watches future persisted thread metadata changes without draining existing history. For a race-free sidebar bootstrap, start buffering `threadCatalog/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, second and millisecond timestamps, archive state, git info, and source; it does not contain turns, items, messages, deltas, tool state, status, or runtime history. ```json { "method": "threadCatalog/subscribe", "id": 21 } { "id": 21, "result": {} } { "method": "threadCatalog/changed", "params": { - "thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "updatedAtMs": 1730832222000, "archivedAt": null } + "thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "updatedAtMs": 1730832222000, "recencyAt": 1730832222, "recencyAtMs": 1730832222000, "archivedAt": null } } } ``` 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 99037eceef..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 @@ -4,7 +4,6 @@ use super::*; pub(crate) struct ThreadCatalogSubscriptions { outgoing: Arc, connection_ids: Arc>>, - ephemeral_threads: Arc>>, } impl ThreadCatalogSubscriptions { @@ -12,7 +11,6 @@ impl ThreadCatalogSubscriptions { Self { outgoing, connection_ids: Arc::new(Mutex::new(HashSet::new())), - ephemeral_threads: Arc::new(Mutex::new(HashMap::new())), } } @@ -37,12 +35,6 @@ impl ThreadCatalogSubscriptions { } pub(super) async fn publish_thread_summary(&self, thread: ThreadSummary) { - if thread.ephemeral { - self.ephemeral_threads - .lock() - .await - .insert(thread.id.clone(), thread.clone()); - } let connection_ids = self .connection_ids .lock() @@ -92,30 +84,4 @@ impl ThreadCatalogSubscriptions { thread_summary_from_stored_thread(stored_thread, fallback_provider, fallback_cwd); self.publish_thread_summary(summary).await; } - - pub(super) async fn update_ephemeral_thread_summary( - &self, - thread_id: ThreadId, - event: &EventMsg, - ) -> Option { - let mut threads = self.ephemeral_threads.lock().await; - let thread = threads.get_mut(&thread_id.to_string())?; - match event { - EventMsg::UserMessage(user) if thread.preview.is_empty() => { - thread.preview = user.message.clone(); - } - EventMsg::UserMessage(_) - | EventMsg::ThreadSettingsApplied(_) - | EventMsg::ThreadRolledBack(_) - | EventMsg::TurnComplete(_) - | EventMsg::TurnAborted(_) => {} - _ => return None, - } - let updated_at_ms = - i64::try_from(time::OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000) - .ok()?; - thread.updated_at = updated_at_ms / 1_000; - thread.updated_at_ms = updated_at_ms; - Some(thread.clone()) - } } diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 3233a00092..0530e0db2f 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -241,6 +241,7 @@ pub(super) async fn ensure_listener_task_running( .await; let thread_settings_baseline = thread_settings_from_config_snapshot(&conversation.config_snapshot().await); + let mut thread_metadata_updated_rx = conversation.subscribe_thread_metadata_updated(); let (mut listener_command_rx, listener_generation) = { let mut thread_state = thread_state.lock().await; if thread_state.listener_matches(&conversation) { @@ -277,8 +278,6 @@ pub(super) async fn ensure_listener_task_running( .. } = listener_task_context; let outgoing_for_task = Arc::clone(&outgoing); - let mut thread_metadata_updated_rx = conversation.subscribe_thread_metadata_updated(); - let is_ephemeral = conversation.config_snapshot().await.ephemeral; tokio::spawn(async move { loop { tokio::select! { @@ -371,15 +370,6 @@ pub(super) async fn ensure_listener_task_running( fallback_model_provider.clone(), ) .await; - if is_ephemeral - && let Some(summary) = thread_catalog_subscriptions - .update_ephemeral_thread_summary(conversation_id, &event.msg) - .await - { - thread_catalog_subscriptions - .publish_thread_summary(summary) - .await; - } } unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => { if !unloading_watchers_open { 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 1ac02251cd..adf5217ebb 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1313,16 +1313,18 @@ impl ThreadRequestProcessor { reasoning_effort: config_snapshot.reasoning_effort, multi_agent_mode: config_snapshot.multi_agent_mode, }; - let catalog_summary = stored_thread.map_or_else( - || thread_summary_from_thread(thread.clone(), /*archived_at*/ None), - |stored_thread| { - thread_summary_from_stored_thread( - stored_thread, - thread.model_provider.as_str(), - &thread.cwd, - ) - }, - ); + let catalog_summary = (!thread.ephemeral).then(|| { + stored_thread.map_or_else( + || thread_summary_from_thread(thread.clone(), /*archived_at*/ None), + |stored_thread| { + thread_summary_from_stored_thread( + stored_thread, + thread.model_provider.as_str(), + &thread.cwd, + ) + }, + ) + }); let notif = thread_started_notification(thread); listener_task_context .outgoing @@ -1333,10 +1335,12 @@ impl ThreadRequestProcessor { )) .await; - listener_task_context - .thread_catalog_subscriptions - .publish_thread_summary(catalog_summary) - .await; + if let Some(catalog_summary) = catalog_summary { + listener_task_context + .thread_catalog_subscriptions + .publish_thread_summary(catalog_summary) + .await; + } listener_task_context .outgoing 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 574ba7140b..94134c2b96 100644 --- a/codex-rs/app-server/src/request_processors/thread_summary.rs +++ b/codex-rs/app-server/src/request_processors/thread_summary.rs @@ -282,11 +282,13 @@ pub(super) fn thread_summary_from_stored_thread( ) -> ThreadSummary { let created_at_ms = thread.created_at.timestamp_millis(); let updated_at_ms = thread.updated_at.timestamp_millis(); + let recency_at_ms = thread.recency_at.timestamp_millis(); let archived_at = thread.archived_at.as_ref().map(chrono::DateTime::timestamp); let (thread, _) = thread_from_stored_thread(thread, fallback_provider, fallback_cwd); let mut summary = thread_summary_from_thread(thread, archived_at); summary.created_at_ms = created_at_ms; summary.updated_at_ms = updated_at_ms; + summary.recency_at_ms = Some(recency_at_ms); summary } @@ -306,6 +308,10 @@ pub(super) fn thread_summary_from_thread( created_at_ms: thread.created_at.saturating_mul(1_000), updated_at: thread.updated_at, updated_at_ms: thread.updated_at.saturating_mul(1_000), + recency_at: thread.recency_at, + recency_at_ms: thread + .recency_at + .map(|recency_at| recency_at.saturating_mul(1_000)), archived_at, path: thread.path, cwd: thread.cwd, 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 a8ea2e5a92..7c8f4bac62 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 @@ -124,46 +124,19 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()> changed.thread.name.as_deref(), Some("Renamed outside first page") ); + assert!(changed.thread.recency_at.is_some()); + assert_eq!( + changed + .thread + .recency_at_ms + .map(|recency_at_ms| recency_at_ms / 1_000), + changed.thread.recency_at + ); assert_eq!(changed.thread.archived_at, None); Ok(()) } -#[tokio::test] -async fn catalog_subscription_reports_ephemeral_thread_updates() -> Result<()> { - let codex_home = TempDir::new()?; - let mut app = start_app(&codex_home).await?; - subscribe(&mut app).await?; - - let thread = start_thread( - &mut app, - ThreadStartParams { - ephemeral: Some(true), - ..Default::default() - }, - ) - .await?; - let created = read_catalog_change(&mut app).await?; - - let turn_id = app - .send_turn_start_request(codex_app_server_protocol::TurnStartParams { - thread_id: thread.id.clone(), - input: vec![codex_app_server_protocol::UserInput::Text { - text: "Ephemeral catalog preview".to_string(), - text_elements: Vec::new(), - }], - ..Default::default() - }) - .await?; - let _: codex_app_server_protocol::TurnStartResponse = read_response(&mut app, turn_id).await?; - let changed = read_catalog_change(&mut app).await?; - assert_eq!(changed.thread.id, thread.id); - assert_eq!(changed.thread.preview, "Ephemeral catalog preview"); - assert!(changed.thread.updated_at_ms >= created.thread.updated_at_ms); - - Ok(()) -} - async fn start_app(codex_home: &TempDir) -> Result { write_mock_responses_config_toml( codex_home.path(),