mirror of
https://github.com/openai/codex.git
synced 2026-10-28 18:58:46 +00:00
Address thread catalog review feedback
This commit is contained in:
@@ -5094,6 +5094,20 @@
|
||||
"preview": {
|
||||
"type": "string"
|
||||
},
|
||||
"recencyAt": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"recencyAtMs": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"sessionId": {
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+14
@@ -19664,6 +19664,20 @@
|
||||
"preview": {
|
||||
"type": "string"
|
||||
},
|
||||
"recencyAt": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"recencyAtMs": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"sessionId": {
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+14
@@ -17443,6 +17443,20 @@
|
||||
"preview": {
|
||||
"type": "string"
|
||||
},
|
||||
"recencyAt": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"recencyAtMs": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"sessionId": {
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
+14
@@ -234,6 +234,20 @@
|
||||
"preview": {
|
||||
"type": "string"
|
||||
},
|
||||
"recencyAt": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"recencyAtMs": {
|
||||
"format": "int64",
|
||||
"type": [
|
||||
"integer",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"sessionId": {
|
||||
"type": "string"
|
||||
},
|
||||
|
||||
@@ -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, };
|
||||
|
||||
@@ -212,6 +212,9 @@ pub struct ThreadSummary {
|
||||
pub updated_at: i64,
|
||||
#[ts(type = "number")]
|
||||
pub updated_at_ms: i64,
|
||||
pub recency_at: Option<i64>,
|
||||
#[ts(type = "number | null")]
|
||||
pub recency_at_ms: Option<i64>,
|
||||
#[ts(type = "number | null")]
|
||||
pub archived_at: Option<i64>,
|
||||
pub path: Option<PathBuf>,
|
||||
|
||||
@@ -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 }
|
||||
} }
|
||||
```
|
||||
|
||||
|
||||
@@ -4,7 +4,6 @@ use super::*;
|
||||
pub(crate) struct ThreadCatalogSubscriptions {
|
||||
outgoing: Arc<OutgoingMessageSender>,
|
||||
connection_ids: Arc<Mutex<HashSet<ConnectionId>>>,
|
||||
ephemeral_threads: Arc<Mutex<HashMap<String, ThreadSummary>>>,
|
||||
}
|
||||
|
||||
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<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())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<TestAppServer> {
|
||||
write_mock_responses_config_toml(
|
||||
codex_home.path(),
|
||||
|
||||
Reference in New Issue
Block a user