mirror of
https://github.com/openai/codex.git
synced 2026-11-02 08:19:52 +00:00
app-server: Update thread/name/set to support not-loaded threads (#13282)
Currently `thread/name/set` does only work for loaded threads. Expand the scope to also support persisted but not-yet-loaded ones for a more predictable API surface. This will make it possible to rename threads discovered via `thread/list` and similar operations.
This commit is contained in:
@@ -129,7 +129,7 @@ Example with notification opt-out:
|
|||||||
- `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`).
|
- `thread/status/changed` — notification emitted when a loaded thread’s status changes (`threadId` + new `status`).
|
||||||
- `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success and emits `thread/archived`.
|
- `thread/archive` — move a thread’s rollout file into the archived directory; returns `{}` on success and emits `thread/archived`.
|
||||||
- `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, then emits `thread/closed`.
|
- `thread/unsubscribe` — unsubscribe this connection from thread turn/item events. If this was the last subscriber, the server shuts down and unloads the thread, then emits `thread/closed`.
|
||||||
- `thread/name/set` — set or update a thread’s user-facing name; returns `{}` on success. Thread names are not required to be unique; name lookups resolve to the most recently updated thread.
|
- `thread/name/set` — set or update a thread’s user-facing name for either a loaded thread or a persisted rollout; returns `{}` on success. Thread names are not required to be unique; name lookups resolve to the most recently updated thread.
|
||||||
- `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`.
|
- `thread/unarchive` — move an archived rollout file back into the sessions directory; returns the restored `thread` on success and emits `thread/unarchived`.
|
||||||
- `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
|
- `thread/compact/start` — trigger conversation history compaction for a thread; returns `{}` immediately while progress streams through standard turn/item notifications.
|
||||||
- `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
|
- `thread/backgroundTerminals/clean` — terminate all running background terminals for a thread (experimental; requires `capabilities.experimentalApi`); returns `{}` when the cleanup request is accepted.
|
||||||
|
|||||||
@@ -141,6 +141,7 @@ use codex_app_server_protocol::ThreadListParams;
|
|||||||
use codex_app_server_protocol::ThreadListResponse;
|
use codex_app_server_protocol::ThreadListResponse;
|
||||||
use codex_app_server_protocol::ThreadLoadedListParams;
|
use codex_app_server_protocol::ThreadLoadedListParams;
|
||||||
use codex_app_server_protocol::ThreadLoadedListResponse;
|
use codex_app_server_protocol::ThreadLoadedListResponse;
|
||||||
|
use codex_app_server_protocol::ThreadNameUpdatedNotification;
|
||||||
use codex_app_server_protocol::ThreadReadParams;
|
use codex_app_server_protocol::ThreadReadParams;
|
||||||
use codex_app_server_protocol::ThreadReadResponse;
|
use codex_app_server_protocol::ThreadReadResponse;
|
||||||
use codex_app_server_protocol::ThreadRealtimeAppendAudioParams;
|
use codex_app_server_protocol::ThreadRealtimeAppendAudioParams;
|
||||||
@@ -2341,6 +2342,14 @@ impl CodexMessageProcessor {
|
|||||||
|
|
||||||
async fn thread_set_name(&self, request_id: ConnectionRequestId, params: ThreadSetNameParams) {
|
async fn thread_set_name(&self, request_id: ConnectionRequestId, params: ThreadSetNameParams) {
|
||||||
let ThreadSetNameParams { thread_id, name } = params;
|
let ThreadSetNameParams { thread_id, name } = params;
|
||||||
|
let thread_id = match ThreadId::from_string(&thread_id) {
|
||||||
|
Ok(id) => id,
|
||||||
|
Err(err) => {
|
||||||
|
self.send_invalid_request_error(request_id, format!("invalid thread id: {err}"))
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
let Some(name) = codex_core::util::normalize_thread_name(&name) else {
|
let Some(name) = codex_core::util::normalize_thread_name(&name) else {
|
||||||
self.send_invalid_request_error(
|
self.send_invalid_request_error(
|
||||||
request_id,
|
request_id,
|
||||||
@@ -2350,15 +2359,43 @@ impl CodexMessageProcessor {
|
|||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
|
|
||||||
let (_, thread) = match self.load_thread(&thread_id).await {
|
if let Ok(thread) = self.thread_manager.get_thread(thread_id).await {
|
||||||
Ok(v) => v,
|
if let Err(err) = thread.submit(Op::SetThreadName { name }).await {
|
||||||
Err(error) => {
|
self.send_internal_error(request_id, format!("failed to set thread name: {err}"))
|
||||||
self.outgoing.send_error(request_id, error).await;
|
.await;
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
|
||||||
|
|
||||||
if let Err(err) = thread.submit(Op::SetThreadName { name }).await {
|
self.outgoing
|
||||||
|
.send_response(request_id, ThreadSetNameResponse {})
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
let thread_exists =
|
||||||
|
match find_thread_path_by_id_str(&self.config.codex_home, &thread_id.to_string()).await
|
||||||
|
{
|
||||||
|
Ok(Some(_)) => true,
|
||||||
|
Ok(None) => false,
|
||||||
|
Err(err) => {
|
||||||
|
self.send_invalid_request_error(
|
||||||
|
request_id,
|
||||||
|
format!("failed to locate thread id {thread_id}: {err}"),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
if !thread_exists {
|
||||||
|
self.send_invalid_request_error(request_id, format!("thread not found: {thread_id}"))
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Err(err) =
|
||||||
|
codex_core::append_thread_name(&self.config.codex_home, thread_id, &name).await
|
||||||
|
{
|
||||||
self.send_internal_error(request_id, format!("failed to set thread name: {err}"))
|
self.send_internal_error(request_id, format!("failed to set thread name: {err}"))
|
||||||
.await;
|
.await;
|
||||||
return;
|
return;
|
||||||
@@ -2367,6 +2404,13 @@ impl CodexMessageProcessor {
|
|||||||
self.outgoing
|
self.outgoing
|
||||||
.send_response(request_id, ThreadSetNameResponse {})
|
.send_response(request_id, ThreadSetNameResponse {})
|
||||||
.await;
|
.await;
|
||||||
|
let notification = ThreadNameUpdatedNotification {
|
||||||
|
thread_id: thread_id.to_string(),
|
||||||
|
thread_name: Some(name),
|
||||||
|
};
|
||||||
|
self.outgoing
|
||||||
|
.send_server_notification(ServerNotification::ThreadNameUpdated(notification))
|
||||||
|
.await;
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn thread_unarchive(
|
async fn thread_unarchive(
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ use codex_app_server_protocol::SessionSource;
|
|||||||
use codex_app_server_protocol::ThreadItem;
|
use codex_app_server_protocol::ThreadItem;
|
||||||
use codex_app_server_protocol::ThreadListParams;
|
use codex_app_server_protocol::ThreadListParams;
|
||||||
use codex_app_server_protocol::ThreadListResponse;
|
use codex_app_server_protocol::ThreadListResponse;
|
||||||
|
use codex_app_server_protocol::ThreadNameUpdatedNotification;
|
||||||
use codex_app_server_protocol::ThreadReadParams;
|
use codex_app_server_protocol::ThreadReadParams;
|
||||||
use codex_app_server_protocol::ThreadReadResponse;
|
use codex_app_server_protocol::ThreadReadResponse;
|
||||||
use codex_app_server_protocol::ThreadResumeParams;
|
use codex_app_server_protocol::ThreadResumeParams;
|
||||||
@@ -220,25 +221,6 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
|
|||||||
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
let mut mcp = McpProcess::new(codex_home.path()).await?;
|
||||||
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
|
||||||
|
|
||||||
// `thread/name/set` operates on loaded threads (via ThreadManager). A rollout existing on disk
|
|
||||||
// is not enough; we must `thread/resume` first to load it into the running server.
|
|
||||||
let pre_resume_id = mcp
|
|
||||||
.send_thread_resume_request(ThreadResumeParams {
|
|
||||||
thread_id: conversation_id.clone(),
|
|
||||||
..Default::default()
|
|
||||||
})
|
|
||||||
.await?;
|
|
||||||
let pre_resume_resp: JSONRPCResponse = timeout(
|
|
||||||
DEFAULT_READ_TIMEOUT,
|
|
||||||
mcp.read_stream_until_response_message(RequestId::Integer(pre_resume_id)),
|
|
||||||
)
|
|
||||||
.await??;
|
|
||||||
let ThreadResumeResponse {
|
|
||||||
thread: pre_resumed,
|
|
||||||
..
|
|
||||||
} = to_response::<ThreadResumeResponse>(pre_resume_resp)?;
|
|
||||||
assert_eq!(pre_resumed.id, conversation_id);
|
|
||||||
|
|
||||||
// Set a user-facing thread title.
|
// Set a user-facing thread title.
|
||||||
let new_name = "My renamed thread";
|
let new_name = "My renamed thread";
|
||||||
let set_id = mcp
|
let set_id = mcp
|
||||||
@@ -253,6 +235,15 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
|
|||||||
)
|
)
|
||||||
.await??;
|
.await??;
|
||||||
let _: ThreadSetNameResponse = to_response::<ThreadSetNameResponse>(set_resp)?;
|
let _: ThreadSetNameResponse = to_response::<ThreadSetNameResponse>(set_resp)?;
|
||||||
|
let notification = timeout(
|
||||||
|
DEFAULT_READ_TIMEOUT,
|
||||||
|
mcp.read_stream_until_notification_message("thread/name/updated"),
|
||||||
|
)
|
||||||
|
.await??;
|
||||||
|
let notification: ThreadNameUpdatedNotification =
|
||||||
|
serde_json::from_value(notification.params.expect("thread/name/updated params"))?;
|
||||||
|
assert_eq!(notification.thread_id, conversation_id);
|
||||||
|
assert_eq!(notification.thread_name.as_deref(), Some(new_name));
|
||||||
|
|
||||||
// Read should now surface `thread.name`, and the wire payload must include `name`.
|
// Read should now surface `thread.name`, and the wire payload must include `name`.
|
||||||
let read_id = mcp
|
let read_id = mcp
|
||||||
|
|||||||
@@ -116,6 +116,7 @@ pub use rollout::RolloutRecorder;
|
|||||||
pub use rollout::RolloutRecorderParams;
|
pub use rollout::RolloutRecorderParams;
|
||||||
pub use rollout::SESSIONS_SUBDIR;
|
pub use rollout::SESSIONS_SUBDIR;
|
||||||
pub use rollout::SessionMeta;
|
pub use rollout::SessionMeta;
|
||||||
|
pub use rollout::append_thread_name;
|
||||||
pub use rollout::find_archived_thread_path_by_id_str;
|
pub use rollout::find_archived_thread_path_by_id_str;
|
||||||
#[deprecated(note = "use find_thread_path_by_id_str")]
|
#[deprecated(note = "use find_thread_path_by_id_str")]
|
||||||
pub use rollout::find_conversation_path_by_id_str;
|
pub use rollout::find_conversation_path_by_id_str;
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ pub use list::find_thread_path_by_id_str as find_conversation_path_by_id_str;
|
|||||||
pub use list::rollout_date_parts;
|
pub use list::rollout_date_parts;
|
||||||
pub use recorder::RolloutRecorder;
|
pub use recorder::RolloutRecorder;
|
||||||
pub use recorder::RolloutRecorderParams;
|
pub use recorder::RolloutRecorderParams;
|
||||||
|
pub use session_index::append_thread_name;
|
||||||
pub use session_index::find_thread_name_by_id;
|
pub use session_index::find_thread_name_by_id;
|
||||||
pub use session_index::find_thread_path_by_name_str;
|
pub use session_index::find_thread_path_by_name_str;
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user