openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 17:38:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Address thread catalog review feedback

Browse Source
This commit is contained in:
Brent Traut
2026-06-24 11:19:36 -07:00
parent 98262bb933
commit 961355389f
23 changed files with 343 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
+6 -6
View File
@@ -5204,9 +5204,9 @@
},
"method": {
"enum": [
"thread/catalog/subscribe"
"threadCatalog/subscribe"
],
"title": "Thread/catalog/subscribeRequestMethod",
"title": "ThreadCatalog/subscribeRequestMethod",
"type": "string"
},
"params": {
@@ -5217,7 +5217,7 @@
"id",
"method"
],
"title": "Thread/catalog/subscribeRequest",
"title": "ThreadCatalog/subscribeRequest",
"type": "object"
},
{
@@ -5227,9 +5227,9 @@
},
"method": {
"enum": [
"thread/catalog/unsubscribe"
"threadCatalog/unsubscribe"
],
"title": "Thread/catalog/unsubscribeRequestMethod",
"title": "ThreadCatalog/unsubscribeRequestMethod",
"type": "string"
},
"params": {
@@ -5240,7 +5240,7 @@
"id",
"method"
],
"title": "Thread/catalog/unsubscribeRequest",
"title": "ThreadCatalog/unsubscribeRequest",
"type": "object"
},
{
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+14 -4
View File
@@ -5041,6 +5041,10 @@
"format": "int64",
"type": "integer"
},
"createdAtMs": {
"format": "int64",
"type": "integer"
},
"cwd": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -5109,11 +5113,16 @@
"updatedAt": {
"format": "int64",
"type": "integer"
},
"updatedAtMs": {
"format": "int64",
"type": "integer"
}
},
"required": [
"cliVersion",
"createdAt",
"createdAtMs",
"cwd",
"ephemeral",
"id",
@@ -5121,7 +5130,8 @@
"preview",
"sessionId",
"source",
"updatedAt"
"updatedAt",
"updatedAtMs"
],
"type": "object"
},
@@ -5839,9 +5849,9 @@
"properties": {
"method": {
"enum": [
"thread/catalog/changed"
"threadCatalog/changed"
],
"title": "Thread/catalog/changedNotificationMethod",
"title": "ThreadCatalog/changedNotificationMethod",
"type": "string"
},
"params": {
@@ -5852,7 +5862,7 @@
"method",
"params"
],
"title": "Thread/catalog/changedNotification",
"title": "ThreadCatalog/changedNotification",
"type": "object"
},
{
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+20 -10
View File
@@ -643,9 +643,9 @@
},
"method": {
"enum": [
"thread/catalog/subscribe"
"threadCatalog/subscribe"
],
"title": "Thread/catalog/subscribeRequestMethod",
"title": "ThreadCatalog/subscribeRequestMethod",
"type": "string"
},
"params": {
@@ -656,7 +656,7 @@
"id",
"method"
],
"title": "Thread/catalog/subscribeRequest",
"title": "ThreadCatalog/subscribeRequest",
"type": "object"
},
{
@@ -666,9 +666,9 @@
},
"method": {
"enum": [
"thread/catalog/unsubscribe"
"threadCatalog/unsubscribe"
],
"title": "Thread/catalog/unsubscribeRequestMethod",
"title": "ThreadCatalog/unsubscribeRequestMethod",
"type": "string"
},
"params": {
@@ -679,7 +679,7 @@
"id",
"method"
],
"title": "Thread/catalog/unsubscribeRequest",
"title": "ThreadCatalog/unsubscribeRequest",
"type": "object"
},
{
@@ -4218,9 +4218,9 @@
"properties": {
"method": {
"enum": [
"thread/catalog/changed"
"threadCatalog/changed"
],
"title": "Thread/catalog/changedNotificationMethod",
"title": "ThreadCatalog/changedNotificationMethod",
"type": "string"
},
"params": {
@@ -4231,7 +4231,7 @@
"method",
"params"
],
"title": "Thread/catalog/changedNotification",
"title": "ThreadCatalog/changedNotification",
"type": "object"
},
{
@@ -19611,6 +19611,10 @@
"format": "int64",
"type": "integer"
},
"createdAtMs": {
"format": "int64",
"type": "integer"
},
"cwd": {
"$ref": "#/definitions/v2/AbsolutePathBuf"
},
@@ -19679,11 +19683,16 @@
"updatedAt": {
"format": "int64",
"type": "integer"
},
"updatedAtMs": {
"format": "int64",
"type": "integer"
}
},
"required": [
"cliVersion",
"createdAt",
"createdAtMs",
"cwd",
"ephemeral",
"id",
@@ -19691,7 +19700,8 @@
"preview",
"sessionId",
"source",
"updatedAt"
"updatedAt",
"updatedAtMs"
],
"type": "object"
},
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+20 -10
View File
@@ -1655,9 +1655,9 @@
},
"method": {
"enum": [
"thread/catalog/subscribe"
"threadCatalog/subscribe"
],
"title": "Thread/catalog/subscribeRequestMethod",
"title": "ThreadCatalog/subscribeRequestMethod",
"type": "string"
},
"params": {
@@ -1668,7 +1668,7 @@
"id",
"method"
],
"title": "Thread/catalog/subscribeRequest",
"title": "ThreadCatalog/subscribeRequest",
"type": "object"
},
{
@@ -1678,9 +1678,9 @@
},
"method": {
"enum": [
"thread/catalog/unsubscribe"
"threadCatalog/unsubscribe"
],
"title": "Thread/catalog/unsubscribeRequestMethod",
"title": "ThreadCatalog/unsubscribeRequestMethod",
"type": "string"
},
"params": {
@@ -1691,7 +1691,7 @@
"id",
"method"
],
"title": "Thread/catalog/unsubscribeRequest",
"title": "ThreadCatalog/unsubscribeRequest",
"type": "object"
},
{
@@ -12626,9 +12626,9 @@
"properties": {
"method": {
"enum": [
"thread/catalog/changed"
"threadCatalog/changed"
],
"title": "Thread/catalog/changedNotificationMethod",
"title": "ThreadCatalog/changedNotificationMethod",
"type": "string"
},
"params": {
@@ -12639,7 +12639,7 @@
"method",
"params"
],
"title": "Thread/catalog/changedNotification",
"title": "ThreadCatalog/changedNotification",
"type": "object"
},
{
@@ -17390,6 +17390,10 @@
"format": "int64",
"type": "integer"
},
"createdAtMs": {
"format": "int64",
"type": "integer"
},
"cwd": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -17458,11 +17462,16 @@
"updatedAt": {
"format": "int64",
"type": "integer"
},
"updatedAtMs": {
"format": "int64",
"type": "integer"
}
},
"required": [
"cliVersion",
"createdAt",
"createdAtMs",
"cwd",
"ephemeral",
"id",
@@ -17470,7 +17479,8 @@
"preview",
"sessionId",
"source",
"updatedAt"
"updatedAt",
"updatedAtMs"
],
"type": "object"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadCatalogChangedNotification.json
Generated
+11 -1
View File
@@ -181,6 +181,10 @@
"format": "int64",
"type": "integer"
},
"createdAtMs": {
"format": "int64",
"type": "integer"
},
"cwd": {
"$ref": "#/definitions/AbsolutePathBuf"
},
@@ -249,11 +253,16 @@
"updatedAt": {
"format": "int64",
"type": "integer"
},
"updatedAtMs": {
"format": "int64",
"type": "integer"
}
},
"required": [
"cliVersion",
"createdAt",
"createdAtMs",
"cwd",
"ephemeral",
"id",
@@ -261,7 +270,8 @@
"preview",
"sessionId",
"source",
"updatedAt"
"updatedAt",
"updatedAtMs"
],
"type": "object"
}
codex-rs/app-server-protocol/schema/typescript/ClientRequest.ts
Generated
+1 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts
Generated
+1 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/v2/ThreadSummary.ts
Generated
+1 -1
View File
@@ -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, updatedAt: 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, 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/common.rs
+3 -3
View File
@@ -618,12 +618,12 @@ client_request_definitions! {
serialization: None,
response: v2::ThreadListResponse,
},
ThreadCatalogSubscribe => "thread/catalog/subscribe" {
ThreadCatalogSubscribe => "threadCatalog/subscribe" {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
serialization: None,
response: v2::ThreadCatalogSubscribeResponse,
},
ThreadCatalogUnsubscribe => "thread/catalog/unsubscribe" {
ThreadCatalogUnsubscribe => "threadCatalog/unsubscribe" {
params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
serialization: None,
response: v2::ThreadCatalogUnsubscribeResponse,
@@ -1612,7 +1612,7 @@ server_notification_definitions! {
Error => "error" (v2::ErrorNotification),
ThreadStarted => "thread/started" (v2::ThreadStartedNotification),
ThreadStatusChanged => "thread/status/changed" (v2::ThreadStatusChangedNotification),
ThreadCatalogChanged => "thread/catalog/changed" (v2::ThreadCatalogChangedNotification),
ThreadCatalogChanged => "threadCatalog/changed" (v2::ThreadCatalogChangedNotification),
ThreadArchived => "thread/archived" (v2::ThreadArchivedNotification),
ThreadDeleted => "thread/deleted" (v2::ThreadDeletedNotification),
ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification),
codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs
+4
View File
@@ -207,7 +207,11 @@ pub struct ThreadSummary {
#[ts(type = "number")]
pub created_at: i64,
#[ts(type = "number")]
pub created_at_ms: i64,
#[ts(type = "number")]
pub updated_at: i64,
#[ts(type = "number")]
pub updated_at_ms: i64,
#[ts(type = "number | null")]
pub archived_at: Option<i64>,
pub path: Option<PathBuf>,
codex-rs/app-server/README.md
+4 -4
View File
@@ -421,13 +421,13 @@ Enable `capabilities.experimentalApi` during initialization, then use `thread/li
} }
```
`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.
`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.
```json
{ "method": "thread/catalog/subscribe", "id": 21 }
{ "method": "threadCatalog/subscribe", "id": 21 }
{ "id": 21, "result": {} }
{ "method": "thread/catalog/changed", "params": {
"thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "archivedAt": null }
{ "method": "threadCatalog/changed", "params": {
"thread": { "id": "thr_z", "preview": "Fix tests", "updatedAt": 1730832222, "updatedAtMs": 1730832222000, "archivedAt": null }
} }
```
codex-rs/app-server/src/message_processor.rs
+2 -1
View File
@@ -477,7 +477,9 @@ impl MessageProcessor {
Arc::clone(&thread_manager),
outgoing.clone(),
Arc::clone(&config),
Arc::clone(&thread_store),
thread_state_manager.clone(),
thread_catalog_subscriptions.clone(),
state_db.clone(),
Arc::clone(&goal_service),
);
@@ -508,7 +510,6 @@ impl MessageProcessor {
Arc::clone(&config),
config_manager.clone(),
pending_thread_unloads,
Arc::clone(&thread_store),
thread_state_manager,
thread_watch_manager,
thread_catalog_subscriptions,
codex-rs/app-server/src/request_processors/thread_catalog_subscription.rs
+34
View File
@@ -4,6 +4,7 @@ use super::*;
pub(crate) struct ThreadCatalogSubscriptions {
outgoing: Arc<OutgoingMessageSender>,
connection_ids: Arc<Mutex<HashSet<ConnectionId>>>,
ephemeral_threads: Arc<Mutex<HashMap<String, ThreadSummary>>>,
}
impl ThreadCatalogSubscriptions {
@@ -11,6 +12,7 @@ impl ThreadCatalogSubscriptions {
Self {
outgoing,
connection_ids: Arc::new(Mutex::new(HashSet::new())),
ephemeral_threads: Arc::new(Mutex::new(HashMap::new())),
}
}
@@ -35,6 +37,12 @@ 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()
@@ -84,4 +92,30 @@ 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<ThreadSummary> {
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())
}
}
codex-rs/app-server/src/request_processors/thread_goal_processor.rs
+23
View File
@@ -10,17 +10,25 @@ pub(crate) struct ThreadGoalRequestProcessor {
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
config: Arc<Config>,
thread_store: Arc<dyn ThreadStore>,
thread_state_manager: ThreadStateManager,
thread_catalog_subscriptions: ThreadCatalogSubscriptions,
state_db: Option<StateDbHandle>,
goal_service: Arc<GoalService>,
}
impl ThreadGoalRequestProcessor {
#[expect(
clippy::too_many_arguments,
reason = "request processor dependencies stay explicit"
)]
pub(crate) fn new(
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
config: Arc<Config>,
thread_store: Arc<dyn ThreadStore>,
thread_state_manager: ThreadStateManager,
thread_catalog_subscriptions: ThreadCatalogSubscriptions,
state_db: Option<StateDbHandle>,
goal_service: Arc<GoalService>,
) -> Self {
@@ -28,7 +36,9 @@ impl ThreadGoalRequestProcessor {
thread_manager,
outgoing,
config,
thread_store,
thread_state_manager,
thread_catalog_subscriptions,
state_db,
goal_service,
}
@@ -158,6 +168,7 @@ impl ThreadGoalRequestProcessor {
.await;
self.emit_thread_goal_updated_ordered(thread_id, goal, listener_command_tx)
.await;
self.publish_thread_catalog_change(thread_id).await;
outcome.apply_runtime_effects(&self.goal_service).await;
Ok(())
}
@@ -212,10 +223,22 @@ impl ThreadGoalRequestProcessor {
if cleared {
self.emit_thread_goal_cleared_ordered(thread_id, listener_command_tx)
.await;
self.publish_thread_catalog_change(thread_id).await;
}
Ok(())
}
async fn publish_thread_catalog_change(&self, thread_id: ThreadId) {
self.thread_catalog_subscriptions
.publish_thread_change(
&self.thread_store,
thread_id,
self.config.model_provider_id.as_str(),
&self.config.cwd,
)
.await;
}
async fn state_db_for_materialized_thread(
&self,
thread_id: ThreadId,
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+24 -20
View File
@@ -8,7 +8,6 @@ pub(super) struct ListenerTaskContext {
pub(super) thread_state_manager: ThreadStateManager,
pub(super) outgoing: Arc<OutgoingMessageSender>,
pub(super) pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
pub(super) thread_store: Arc<dyn ThreadStore>,
pub(super) thread_watch_manager: ThreadWatchManager,
pub(super) thread_catalog_subscriptions: ThreadCatalogSubscriptions,
pub(super) thread_list_state_permit: Arc<Semaphore>,
@@ -122,17 +121,6 @@ impl UnloadingState {
}
}
fn catalog_worthy_event(event: &EventMsg) -> bool {
matches!(
event,
EventMsg::TurnStarted(_)
| EventMsg::TurnComplete(_)
| EventMsg::TurnAborted(_)
| EventMsg::ThreadSettingsApplied(_)
| EventMsg::ThreadRolledBack(_)
)
}
pub(super) enum ThreadShutdownResult {
Complete,
SubmitFailed,
@@ -280,7 +268,6 @@ pub(super) async fn ensure_listener_task_running(
thread_manager,
thread_state_manager,
pending_thread_unloads,
thread_store,
thread_watch_manager,
thread_catalog_subscriptions,
thread_list_state_permit,
@@ -290,6 +277,8 @@ 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! {
@@ -315,6 +304,22 @@ pub(super) async fn ensure_listener_task_running(
)
.await;
}
stored_thread = async {
match &mut thread_metadata_updated_rx {
Some(rx) => rx.recv().await.ok(),
None => futures::future::pending().await,
}
} => {
if let Some(stored_thread) = stored_thread {
thread_catalog_subscriptions
.publish_thread_summary(thread_summary_from_stored_thread(
stored_thread,
fallback_model_provider.as_str(),
&fallback_cwd,
))
.await;
}
}
event = conversation.next_event() => {
let event = match event {
Ok(event) => event,
@@ -366,14 +371,13 @@ pub(super) async fn ensure_listener_task_running(
fallback_model_provider.clone(),
)
.await;
if catalog_worthy_event(&event.msg) {
if is_ephemeral
&& let Some(summary) = thread_catalog_subscriptions
.update_ephemeral_thread_summary(conversation_id, &event.msg)
.await
{
thread_catalog_subscriptions
.publish_thread_change(
&thread_store,
conversation_id,
fallback_model_provider.as_str(),
&fallback_cwd,
)
.publish_thread_summary(summary)
.await;
}
}
codex-rs/app-server/src/request_processors/thread_processor.rs
+16 -3
View File
@@ -876,7 +876,6 @@ impl ThreadRequestProcessor {
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
thread_store: Arc::clone(&self.thread_store),
thread_watch_manager: self.thread_watch_manager.clone(),
thread_catalog_subscriptions: self.thread_catalog_subscriptions.clone(),
thread_list_state_permit: self.thread_list_state_permit.clone(),
@@ -979,7 +978,6 @@ impl ThreadRequestProcessor {
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
thread_store: Arc::clone(&self.thread_store),
thread_watch_manager: self.thread_watch_manager.clone(),
thread_catalog_subscriptions: self.thread_catalog_subscriptions.clone(),
thread_list_state_permit: self.thread_list_state_permit.clone(),
@@ -1239,6 +1237,12 @@ impl ThreadRequestProcessor {
otel.name = "app_server.thread_start.config_snapshot",
))
.await;
let stored_thread = thread
.read_thread(
/*include_archived*/ true, /*include_history*/ false,
)
.await
.ok();
let mut thread = build_thread_from_snapshot(
thread_id,
session_configured.session_id.to_string(),
@@ -1309,7 +1313,16 @@ 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 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 notif = thread_started_notification(thread);
listener_task_context
.outgoing
codex-rs/app-server/src/request_processors/thread_summary.rs
+9 -2
View File
@@ -280,9 +280,14 @@ pub(super) fn thread_summary_from_stored_thread(
fallback_provider: &str,
fallback_cwd: &AbsolutePathBuf,
) -> ThreadSummary {
let archived_at = thread.archived_at.map(|dt| dt.timestamp());
let created_at_ms = thread.created_at.timestamp_millis();
let updated_at_ms = thread.updated_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);
thread_summary_from_thread(thread, archived_at)
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
}
pub(super) fn thread_summary_from_thread(
@@ -298,7 +303,9 @@ pub(super) fn thread_summary_from_thread(
ephemeral: thread.ephemeral,
model_provider: thread.model_provider,
created_at: thread.created_at,
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),
archived_at,
path: thread.path,
cwd: thread.cwd,
codex-rs/app-server/src/request_processors/turn_processor.rs
-4
View File
@@ -76,7 +76,6 @@ pub(crate) struct TurnRequestProcessor {
config: Arc<Config>,
config_manager: ConfigManager,
pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
thread_store: Arc<dyn ThreadStore>,
thread_state_manager: ThreadStateManager,
thread_watch_manager: ThreadWatchManager,
thread_catalog_subscriptions: ThreadCatalogSubscriptions,
@@ -135,7 +134,6 @@ impl TurnRequestProcessor {
config: Arc<Config>,
config_manager: ConfigManager,
pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
thread_store: Arc<dyn ThreadStore>,
thread_state_manager: ThreadStateManager,
thread_watch_manager: ThreadWatchManager,
thread_catalog_subscriptions: ThreadCatalogSubscriptions,
@@ -151,7 +149,6 @@ impl TurnRequestProcessor {
config,
config_manager,
pending_thread_unloads,
thread_store,
thread_state_manager,
thread_watch_manager,
thread_catalog_subscriptions,
@@ -1425,7 +1422,6 @@ impl TurnRequestProcessor {
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
thread_store: Arc::clone(&self.thread_store),
thread_watch_manager: self.thread_watch_manager.clone(),
thread_catalog_subscriptions: self.thread_catalog_subscriptions.clone(),
thread_list_state_permit: self.thread_list_state_permit.clone(),
codex-rs/app-server/tests/suite/v2/thread_catalog_subscription.rs
+93 -31
View File
@@ -7,6 +7,8 @@ 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::ThreadInjectItemsParams;
use codex_app_server_protocol::ThreadInjectItemsResponse;
use codex_app_server_protocol::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadSetNameParams;
@@ -23,41 +25,37 @@ use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(10);
#[tokio::test]
async fn catalog_subscription_reports_new_empty_thread() -> Result<()> {
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 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::<ThreadStartResponse>(&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);
let started = start_thread(&mut app, ThreadStartParams::default()).await?;
let changed = read_catalog_change(&mut app).await?;
assert_eq!(changed.thread.id, started.id);
assert_eq!(changed.thread.preview, "");
assert!(
!started
.thread
.path
.expect("thread path should be present")
.exists()
);
let inject_id = app
.send_thread_inject_items_request(ThreadInjectItemsParams {
thread_id: started.id.clone(),
items: vec![serde_json::json!({
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "injected"}],
})],
})
.await?;
let _: ThreadInjectItemsResponse = read_response(&mut app, inject_id).await?;
let injected = read_catalog_change(&mut app).await?;
assert_eq!(injected.thread.id, started.id);
assert!(injected.thread.updated_at_ms >= changed.thread.updated_at_ms);
Ok(())
}
@@ -82,11 +80,7 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()>
)?;
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?;
subscribe(&mut app).await?;
let list_id = app
.send_thread_list_request(ThreadListParams {
@@ -98,6 +92,7 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()>
source_kinds: None,
archived: None,
cwd: None,
parent_thread_id: None,
use_state_db_only: false,
search_term: None,
})
@@ -115,12 +110,12 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()>
let notification: JSONRPCNotification = timeout(
DEFAULT_READ_TIMEOUT,
app.read_stream_until_notification_message("thread/catalog/changed"),
app.read_stream_until_notification_message("threadCatalog/changed"),
)
.await??;
let raw_params = notification
.params
.expect("thread/catalog/changed should have params");
.expect("threadCatalog/changed should have params");
assert_eq!(raw_params["thread"].get("turns"), None);
let changed: ThreadCatalogChangedNotification = serde_json::from_value(raw_params)?;
@@ -134,6 +129,41 @@ async fn catalog_subscription_reports_thread_outside_loaded_page() -> Result<()>
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<TestAppServer> {
write_mock_responses_config_toml(
codex_home.path(),
@@ -149,6 +179,38 @@ async fn start_app(codex_home: &TempDir) -> Result<TestAppServer> {
Ok(app)
}
async fn subscribe(app: &mut TestAppServer) -> Result<()> {
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(())
}
async fn start_thread(
app: &mut TestAppServer,
params: ThreadStartParams,
) -> Result<codex_app_server_protocol::Thread> {
let start_id = app.send_thread_start_request(params).await?;
let started = read_response::<ThreadStartResponse>(app, start_id).await?;
Ok(started.thread)
}
async fn read_catalog_change(app: &mut TestAppServer) -> Result<ThreadCatalogChangedNotification> {
let notification: JSONRPCNotification = timeout(
DEFAULT_READ_TIMEOUT,
app.read_stream_until_notification_message("threadCatalog/changed"),
)
.await??;
serde_json::from_value(
notification
.params
.ok_or_else(|| anyhow::anyhow!("threadCatalog/changed should have params"))?,
)
.map_err(Into::into)
}
async fn rename_thread(app: &mut TestAppServer, thread_id: String, name: &str) -> Result<()> {
let rename_id = app
.send_thread_set_name_request(ThreadSetNameParams {
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+13
View File
@@ -1080,6 +1080,14 @@ async fn thread_resume_keeps_paused_goal_paused() -> Result<()> {
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let subscribe_id = mcp
.send_raw_request("threadCatalog/subscribe", /*params*/ None)
.await?;
let _: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(subscribe_id)),
)
.await??;
let goal_id = mcp
.send_raw_request(
@@ -1102,6 +1110,11 @@ async fn thread_resume_keeps_paused_goal_paused() -> Result<()> {
mcp.read_stream_until_notification_message("thread/goal/updated"),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("threadCatalog/changed"),
)
.await??;
mcp.clear_message_buffer();
let resume_id = mcp
codex-rs/core/src/codex_thread.rs
+6
View File
@@ -416,6 +416,12 @@ impl CodexThread {
self.codex.next_event().await
}
pub fn subscribe_thread_metadata_updated(
&self,
) -> Option<tokio::sync::broadcast::Receiver<StoredThread>> {
self.codex.session.subscribe_thread_metadata_updated()
}
pub async fn agent_status(&self) -> AgentStatus {
self.codex.agent_status().await
}
codex-rs/core/src/session/mod.rs
+8
View File
@@ -153,6 +153,7 @@ use codex_thread_store::LiveThreadInitGuard;
use codex_thread_store::LocalThreadStore;
use codex_thread_store::ReadThreadParams;
use codex_thread_store::ResumeThreadParams;
use codex_thread_store::StoredThread;
use codex_thread_store::ThreadPersistenceMetadata;
use codex_thread_store::ThreadStore;
use codex_utils_path_uri::PathUri;
@@ -1154,6 +1155,13 @@ impl Session {
self.services.live_thread.as_ref()
}
pub(crate) fn subscribe_thread_metadata_updated(
&self,
) -> Option<tokio::sync::broadcast::Receiver<StoredThread>> {
self.live_thread()
.map(LiveThread::subscribe_metadata_updated)
}
/// Flush rollout writes and return the final durability-barrier result.
#[instrument(name = "session.flush_rollout", level = "trace", skip_all)]
pub(crate) async fn flush_rollout(&self) -> std::io::Result<()> {
codex-rs/thread-store/src/live_thread.rs
+30 -5
View File
@@ -8,6 +8,7 @@ use codex_rollout::RolloutPersistenceTelemetry;
use codex_rollout::measure_and_filter_rollout_items;
use codex_rollout::persisted_rollout_items;
use tokio::sync::Mutex;
use tokio::sync::broadcast;
use tracing::warn;
use crate::AppendThreadItemsParams;
@@ -35,6 +36,7 @@ pub struct LiveThread {
thread_store: Arc<dyn ThreadStore>,
metadata_sync: Arc<Mutex<ThreadMetadataSync>>,
persistence_telemetry: RolloutPersistenceTelemetry,
metadata_updated_tx: broadcast::Sender<StoredThread>,
}
/// Owns a live thread while session initialization is still fallible.
@@ -87,6 +89,8 @@ impl Drop for LiveThreadInitGuard {
}
impl LiveThread {
const METADATA_UPDATED_CHANNEL_CAPACITY: usize = 16;
pub async fn create(
thread_store: Arc<dyn ThreadStore>,
params: CreateThreadParams,
@@ -94,11 +98,13 @@ impl LiveThread {
let thread_id = params.thread_id;
let metadata_sync = ThreadMetadataSync::for_create(&params).await;
thread_store.create_thread(params).await?;
let (metadata_updated_tx, _) = broadcast::channel(Self::METADATA_UPDATED_CHANNEL_CAPACITY);
Ok(Self {
thread_id,
thread_store,
metadata_sync: Arc::new(Mutex::new(metadata_sync)),
persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id),
metadata_updated_tx,
})
}
@@ -130,14 +136,20 @@ impl LiveThread {
}
}
}
let (metadata_updated_tx, _) = broadcast::channel(Self::METADATA_UPDATED_CHANNEL_CAPACITY);
Ok(Self {
thread_id,
thread_store,
metadata_sync: Arc::new(Mutex::new(metadata_sync)),
persistence_telemetry: RolloutPersistenceTelemetry::new(thread_id),
metadata_updated_tx,
})
}
pub fn subscribe_metadata_updated(&self) -> broadcast::Receiver<StoredThread> {
self.metadata_updated_tx.subscribe()
}
#[tracing::instrument(
level = "trace",
skip_all,
@@ -172,13 +184,15 @@ impl LiveThread {
.await
.observe_appended_items(canonical_items.as_slice());
if let Some(update) = update {
self.thread_store
let stored_thread = self
.thread_store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id: self.thread_id,
patch: update.patch.clone(),
include_archived: true,
})
.await?;
self.publish_metadata_updated(stored_thread);
self.metadata_sync
.lock()
.await
@@ -240,7 +254,8 @@ impl LiveThread {
include_archived: bool,
) -> ThreadStoreResult<()> {
self.flush_pending_metadata_update().await?;
self.thread_store
let stored_thread = self
.thread_store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id: self.thread_id,
patch: ThreadMetadataPatch {
@@ -250,6 +265,7 @@ impl LiveThread {
include_archived,
})
.await?;
self.publish_metadata_updated(stored_thread);
Ok(())
}
@@ -259,13 +275,16 @@ impl LiveThread {
include_archived: bool,
) -> ThreadStoreResult<StoredThread> {
self.flush_pending_metadata_update().await?;
self.thread_store
let stored_thread = self
.thread_store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id: self.thread_id,
patch,
include_archived,
})
.await
.await?;
self.publish_metadata_updated(stored_thread.clone());
Ok(stored_thread)
}
/// Returns the live local rollout path for legacy local-only callers.
@@ -306,17 +325,23 @@ impl LiveThread {
let Some(update) = update else {
return Ok(());
};
self.thread_store
let stored_thread = self
.thread_store
.update_thread_metadata(UpdateThreadMetadataParams {
thread_id: self.thread_id,
patch: update.patch.clone(),
include_archived: true,
})
.await?;
self.publish_metadata_updated(stored_thread);
self.metadata_sync
.lock()
.await
.mark_pending_update_applied(&update);
Ok(())
}
fn publish_metadata_updated(&self, stored_thread: StoredThread) {
let _ = self.metadata_updated_tx.send(stored_thread);
}
}