Fix thread catalog delivery races (#26009)

Add revision-based bootstrap ordering, refresh spawned threads after listener attachment, preserve catalog notifications under in-process backpressure, and correct the generated recency timestamp type.

codex-rs/app-server-protocol/schema/json/ServerNotification.json

@@ -3746,11 +3746,18 @@
     },
     "ThreadCatalogChangedNotification": {
       "properties": {
+        "revision": {
+          "description": "Monotonically increasing revision for this app-server process.",
+          "format": "uint64",
+          "minimum": 0.0,
+          "type": "integer"
+        },
         "thread": {
           "$ref": "#/definitions/ThreadSummary"
         }
       },
       "required": [
+        "revision",
         "thread"
       ],
       "type": "object"

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json

@@ -17079,11 +17079,18 @@
       "ThreadCatalogChangedNotification": {
         "$schema": "http://json-schema.org/draft-07/schema#",
         "properties": {
+          "revision": {
+            "description": "Monotonically increasing revision for this app-server process.",
+            "format": "uint64",
+            "minimum": 0.0,
+            "type": "integer"
+          },
           "thread": {
             "$ref": "#/definitions/v2/ThreadSummary"
           }
         },
         "required": [
+          "revision",
           "thread"
         ],
         "title": "ThreadCatalogChangedNotification",
@@ -17091,6 +17098,17 @@
       },
       "ThreadCatalogSubscribeResponse": {
         "$schema": "http://json-schema.org/draft-07/schema#",
+        "properties": {
+          "revision": {
+            "description": "Highest catalog notification revision published before this subscription was acknowledged.",
+            "format": "uint64",
+            "minimum": 0.0,
+            "type": "integer"
+          }
+        },
+        "required": [
+          "revision"
+        ],
         "title": "ThreadCatalogSubscribeResponse",
         "type": "object"
       },

codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json

@@ -14858,11 +14858,18 @@
     "ThreadCatalogChangedNotification": {
       "$schema": "http://json-schema.org/draft-07/schema#",
       "properties": {
+        "revision": {
+          "description": "Monotonically increasing revision for this app-server process.",
+          "format": "uint64",
+          "minimum": 0.0,
+          "type": "integer"
+        },
         "thread": {
           "$ref": "#/definitions/ThreadSummary"
         }
       },
       "required": [
+        "revision",
         "thread"
       ],
       "title": "ThreadCatalogChangedNotification",
@@ -14870,6 +14877,17 @@
     },
     "ThreadCatalogSubscribeResponse": {
       "$schema": "http://json-schema.org/draft-07/schema#",
+      "properties": {
+        "revision": {
+          "description": "Highest catalog notification revision published before this subscription was acknowledged.",
+          "format": "uint64",
+          "minimum": 0.0,
+          "type": "integer"
+        }
+      },
+      "required": [
+        "revision"
+      ],
       "title": "ThreadCatalogSubscribeResponse",
       "type": "object"
     },

codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json

@@ -291,11 +291,18 @@
     }
   },
   "properties": {
+    "revision": {
+      "description": "Monotonically increasing revision for this app-server process.",
+      "format": "uint64",
+      "minimum": 0.0,
+      "type": "integer"
+    },
     "thread": {
       "$ref": "#/definitions/ThreadSummary"
     }
   },
   "required": [
+    "revision",
     "thread"
   ],
   "title": "ThreadCatalogChangedNotification",

codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogSubscribeResponse.json

@@ -1,5 +1,16 @@
 {
   "$schema": "http://json-schema.org/draft-07/schema#",
+  "properties": {
+    "revision": {
+      "description": "Highest catalog notification revision published before this subscription was acknowledged.",
+      "format": "uint64",
+      "minimum": 0.0,
+      "type": "integer"
+    }
+  },
+  "required": [
+    "revision"
+  ],
   "title": "ThreadCatalogSubscribeResponse",
   "type": "object"
 }

codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogChangedNotification.ts

@@ -3,4 +3,8 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 import type { ThreadSummary } from "./ThreadSummary";
 
-export type ThreadCatalogChangedNotification = { thread: ThreadSummary, };
+export type ThreadCatalogChangedNotification = {
+/**
+ * Monotonically increasing revision for this app-server process.
+ */
+revision: number, thread: ThreadSummary, };

codex-rs/app-server-protocol/schema/typescript/v2/ThreadCatalogSubscribeResponse.ts

@@ -2,4 +2,8 @@
 
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 
-export type ThreadCatalogSubscribeResponse = Record<string, never>;
+export type ThreadCatalogSubscribeResponse = {
+/**
+ * Highest catalog notification revision published before this subscription was acknowledged.
+ */
+revision: number, };

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, 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, };
+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: number | 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, };

codex-rs/app-server-protocol/src/protocol/v2/thread.rs

@@ -1098,7 +1098,11 @@ pub struct ThreadListParams {
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
-pub struct ThreadCatalogSubscribeResponse {}
+pub struct ThreadCatalogSubscribeResponse {
+    /// Highest catalog notification revision published before this subscription was acknowledged.
+    #[ts(type = "number")]
+    pub revision: u64,
+}
 
 #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
 #[serde(rename_all = "camelCase")]
@@ -1434,6 +1438,9 @@ pub struct ThreadStatusChangedNotification {
 #[serde(rename_all = "camelCase")]
 #[ts(export_to = "v2/")]
 pub struct ThreadCatalogChangedNotification {
+    /// Monotonically increasing revision for this app-server process.
+    #[ts(type = "number")]
+    pub revision: u64,
     pub thread: ThreadSummary,
 }
 

codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs

@@ -212,6 +212,7 @@ pub struct ThreadSummary {
     pub updated_at: i64,
     #[ts(type = "number")]
     pub updated_at_ms: i64,
+    #[ts(type = "number | null")]
     pub recency_at: Option<i64>,
     #[ts(type = "number | null")]
     pub recency_at_ms: Option<i64>,

codex-rs/app-server/README.md

@@ -421,12 +421,13 @@ Enable `capabilities.experimentalApi` during initialization, then use `thread/li
 } }
 ```
 
-`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.
+`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. Treat the response `revision` as the baseline, call `thread/list`, then apply buffered and future notifications with a higher revision in revision order. Ignore notifications at or below the last applied revision and filter summaries 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": {} }
+{ "id": 21, "result": { "revision": 12 } }
 { "method": "threadCatalog/changed", "params": {
+    "revision": 13,
     "thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "updatedAtMs": 1730832222000, "recencyAt": 1730832222, "recencyAtMs": 1730832222000, "archivedAt": null }
 } }
 ```

codex-rs/app-server/src/in_process.rs

@@ -106,6 +106,7 @@ fn server_notification_requires_delivery(notification: &ServerNotification) -> b
         notification,
         ServerNotification::TurnCompleted(_)
             | ServerNotification::ThreadSettingsUpdated(_)
+            | ServerNotification::ThreadCatalogChanged(_)
             | ServerNotification::ExternalAgentConfigImportCompleted(_)
     )
 }
@@ -733,6 +734,8 @@ mod tests {
     use codex_app_server_protocol::ConfigRequirementsReadResponse;
     use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification;
     use codex_app_server_protocol::SessionSource as ApiSessionSource;
+    use codex_app_server_protocol::ThreadCatalogChangedNotification;
+    use codex_app_server_protocol::ThreadCatalogSubscribeResponse;
     use codex_app_server_protocol::ThreadStartParams;
     use codex_app_server_protocol::ThreadStartResponse;
     use codex_app_server_protocol::Turn;
@@ -879,6 +882,76 @@ mod tests {
             .expect("in-process runtime should shutdown cleanly");
     }
 
+    #[tokio::test]
+    async fn in_process_catalog_notifications_survive_queue_pressure() {
+        let mut client =
+            start_test_client_with_capacity(SessionSource::Cli, /*channel_capacity*/ 1).await;
+        let subscribed = request_when_ready(&client, || ClientRequest::ThreadCatalogSubscribe {
+            request_id: RequestId::Integer(10),
+            params: None,
+        })
+        .await
+        .expect("subscribe should succeed");
+        let subscribed: ThreadCatalogSubscribeResponse =
+            serde_json::from_value(subscribed).expect("subscribe response should parse");
+
+        let mut started_ids = Vec::new();
+        for request_id in [11, 12] {
+            let started = request_when_ready(&client, || ClientRequest::ThreadStart {
+                request_id: RequestId::Integer(request_id),
+                params: ThreadStartParams::default(),
+            })
+            .await
+            .expect("thread/start should succeed");
+            let started: ThreadStartResponse =
+                serde_json::from_value(started).expect("thread/start response should parse");
+            started_ids.push(started.thread.id);
+        }
+
+        let first = next_catalog_change(&mut client).await;
+        let second = next_catalog_change(&mut client).await;
+        assert_eq!(vec![first.thread.id, second.thread.id], started_ids,);
+        assert!(first.revision > subscribed.revision);
+        assert!(second.revision > first.revision);
+
+        client
+            .shutdown()
+            .await
+            .expect("in-process runtime should shutdown cleanly");
+    }
+
+    async fn next_catalog_change(
+        client: &mut InProcessClientHandle,
+    ) -> ThreadCatalogChangedNotification {
+        timeout(SHUTDOWN_TIMEOUT, async {
+            loop {
+                if let Some(InProcessServerEvent::ServerNotification(
+                    ServerNotification::ThreadCatalogChanged(notification),
+                )) = client.next_event().await
+                {
+                    return notification;
+                }
+            }
+        })
+        .await
+        .expect("catalog notification should arrive")
+    }
+
+    async fn request_when_ready(
+        client: &InProcessClientHandle,
+        request: impl Fn() -> ClientRequest,
+    ) -> PendingClientRequestResponse {
+        loop {
+            match client.request(request()).await {
+                Ok(response) => return response,
+                Err(err) if err.kind() == ErrorKind::WouldBlock => {
+                    tokio::task::yield_now().await;
+                }
+                Err(err) => panic!("request transport should work: {err}"),
+            }
+        }
+    }
+
     #[test]
     fn guaranteed_delivery_helpers_cover_terminal_server_notifications() {
         assert!(server_notification_requires_delivery(

codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs

@@ -4,6 +4,7 @@ use super::*;
 pub(crate) struct ThreadCatalogSubscriptions {
     outgoing: Arc<OutgoingMessageSender>,
     connection_ids: Arc<Mutex<HashSet<ConnectionId>>>,
+    revision: Arc<Mutex<u64>>,
 }
 
 impl ThreadCatalogSubscriptions {
@@ -11,6 +12,7 @@ impl ThreadCatalogSubscriptions {
         Self {
             outgoing,
             connection_ids: Arc::new(Mutex::new(HashSet::new())),
+            revision: Arc::new(Mutex::new(0)),
         }
     }
 
@@ -18,8 +20,10 @@ impl ThreadCatalogSubscriptions {
         &self,
         connection_id: ConnectionId,
     ) -> ThreadCatalogSubscribeResponse {
-        self.connection_ids.lock().await.insert(connection_id);
-        ThreadCatalogSubscribeResponse {}
+        let mut connection_ids = self.connection_ids.lock().await;
+        connection_ids.insert(connection_id);
+        let revision = *self.revision.lock().await;
+        ThreadCatalogSubscribeResponse { revision }
     }
 
     pub(super) async fn unsubscribe(
@@ -45,10 +49,16 @@ impl ThreadCatalogSubscriptions {
         if connection_ids.is_empty() {
             return;
         }
+        let revision = {
+            let mut revision = self.revision.lock().await;
+            *revision = revision.saturating_add(1);
+            *revision
+        };
         self.outgoing
             .send_server_notification_to_connections(
                 &connection_ids,
                 ServerNotification::ThreadCatalogChanged(ThreadCatalogChangedNotification {
+                    revision,
                     thread,
                 }),
             )

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -2658,6 +2658,7 @@ impl ThreadRequestProcessor {
                 "thread",
             );
         }
+        self.publish_thread_catalog_change(thread_id).await;
     }
 
     async fn thread_resume_inner(

codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs

@@ -7,6 +7,7 @@ use codex_app_server_protocol::JSONRPCNotification;
 use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ThreadCatalogChangedNotification;
+use codex_app_server_protocol::ThreadCatalogSubscribeResponse;
 use codex_app_server_protocol::ThreadInjectItemsParams;
 use codex_app_server_protocol::ThreadInjectItemsResponse;
 use codex_app_server_protocol::ThreadListParams;
@@ -28,10 +29,11 @@ const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
 async fn catalog_subscription_reports_new_and_injected_thread() -> Result<()> {
     let codex_home = TempDir::new()?;
     let mut app = start_app(&codex_home).await?;
-    subscribe(&mut app).await?;
+    let subscribed = subscribe(&mut app).await?;
 
     let started = start_thread(&mut app, ThreadStartParams::default()).await?;
     let changed = read_catalog_change(&mut app).await?;
+    assert!(changed.revision > subscribed.revision);
     assert_eq!(changed.thread.id, started.id);
     assert_eq!(changed.thread.preview, "");
     assert!(
@@ -53,8 +55,11 @@ async fn catalog_subscription_reports_new_and_injected_thread() -> Result<()> {
         .await?;
     let _: ThreadInjectItemsResponse = read_response(&mut app, inject_id).await?;
     let injected = read_catalog_change(&mut app).await?;
+    assert!(injected.revision > changed.revision);
     assert_eq!(injected.thread.id, started.id);
     assert!(injected.thread.updated_at_ms >= changed.thread.updated_at_ms);
+    let resubscribed = subscribe(&mut app).await?;
+    assert!(resubscribed.revision >= injected.revision);
 
     Ok(())
 }
@@ -152,13 +157,11 @@ async fn start_app(codex_home: &TempDir) -> Result<TestAppServer> {
     Ok(app)
 }
 
-async fn subscribe(app: &mut TestAppServer) -> Result<()> {
+async fn subscribe(app: &mut TestAppServer) -> Result<ThreadCatalogSubscribeResponse> {
     let subscribe_id = app
         .send_raw_request("threadCatalog/subscribe", /*params*/ None)
         .await?;
-    let _: codex_app_server_protocol::ThreadCatalogSubscribeResponse =
-        read_response(app, subscribe_id).await?;
-    Ok(())
+    read_response(app, subscribe_id).await
 }
 
 async fn start_thread(

codex-rs/app-server/tests/suite/v2/turn_start.rs

@@ -43,6 +43,8 @@ use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::ServerRequestResolvedNotification;
 use codex_app_server_protocol::SubAgentActivityKind;
 use codex_app_server_protocol::TextElement;
+use codex_app_server_protocol::ThreadCatalogChangedNotification;
+use codex_app_server_protocol::ThreadCatalogSubscribeResponse;
 use codex_app_server_protocol::ThreadDeleteParams;
 use codex_app_server_protocol::ThreadDeleteResponse;
 use codex_app_server_protocol::ThreadDeletedNotification;
@@ -3391,7 +3393,7 @@ async fn turn_start_streams_apply_patch_change_updates_v2() -> Result<()> {
 }
 
 #[tokio::test]
-async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<()> {
+async fn turn_start_emits_spawn_agent_item_and_catalog_metadata_v2() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
     const CHILD_PROMPT: &str = "child: do work";
@@ -3454,6 +3456,15 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<()
 
     let mut mcp = TestAppServer::new_with_auto_env(codex_home.path()).await?;
     timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+    let subscribe_id = mcp
+        .send_raw_request("threadCatalog/subscribe", /*params*/ None)
+        .await?;
+    let subscribe_response = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(subscribe_id)),
+    )
+    .await??;
+    let subscribed: ThreadCatalogSubscribeResponse = to_response(subscribe_response)?;
 
     let thread_req = mcp
         .send_thread_start_request_with_auto_env(ThreadStartParams {
@@ -3570,6 +3581,32 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<()
     );
     assert_eq!(agent_state.message, None);
 
+    let catalog_notification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_matching_notification(
+            "spawned thread catalog summary",
+            |notification| {
+                notification.method == "threadCatalog/changed"
+                    && notification
+                        .params
+                        .as_ref()
+                        .and_then(|params| params["thread"]["id"].as_str())
+                        == Some(receiver_thread_id.as_str())
+            },
+        ),
+    )
+    .await??;
+    let catalog_changed: ThreadCatalogChangedNotification = serde_json::from_value(
+        catalog_notification
+            .params
+            .expect("threadCatalog/changed params"),
+    )?;
+    assert!(catalog_changed.revision > subscribed.revision);
+    assert_eq!(
+        catalog_changed.thread.parent_thread_id.as_deref(),
+        Some(thread.id.as_str())
+    );
+
     let turn_completed = timeout(DEFAULT_READ_TIMEOUT, async {
         loop {
             let turn_completed_notif = mcp