mirror of
https://github.com/openai/codex.git
synced 2026-10-30 19:18:49 +00:00
Broadcast thread section changes to connected clients (#58380)
## Why Clients need to refresh section membership and ordering when another client moves a thread. ## What changed Broadcast `thread/section/changed` after a successful `thread/section/move`, including moves within the same section. Send the response before the notification, and include `threadId` and the destination `sectionId`, using `null` for removal. Failed moves do not emit the notification. Add the notification to the protocol schemas, TypeScript exports, and Python SDK models and registry, and document its behavior. ## Testing Add a WebSocket integration test with separate writer and observer clients covering section insertion, reordering, removal, and a failed move. GitOrigin-RevId: 128ba7eddc9aae8e3bac766f47e4a2c55c64a6a5
This commit is contained in:
@@ -6201,6 +6201,26 @@
|
||||
},
|
||||
"type": "object"
|
||||
},
|
||||
"ThreadSectionChangedNotification": {
|
||||
"description": "Broadcast after a successful move into, within, or out of a section.",
|
||||
"properties": {
|
||||
"sectionId": {
|
||||
"description": "Destination section, or `null` when the thread was removed from its section.",
|
||||
"type": [
|
||||
"string",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"threadId": {
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"sectionId",
|
||||
"threadId"
|
||||
],
|
||||
"type": "object"
|
||||
},
|
||||
"ThreadSettings": {
|
||||
"properties": {
|
||||
"activePermissionProfile": {
|
||||
@@ -7408,6 +7428,26 @@
|
||||
"title": "Thread/name/updatedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
{
|
||||
"properties": {
|
||||
"method": {
|
||||
"enum": [
|
||||
"thread/section/changed"
|
||||
],
|
||||
"title": "Thread/section/changedNotificationMethod",
|
||||
"type": "string"
|
||||
},
|
||||
"params": {
|
||||
"$ref": "#/definitions/ThreadSectionChangedNotification"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"method",
|
||||
"params"
|
||||
],
|
||||
"title": "Thread/section/changedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
{
|
||||
"properties": {
|
||||
"method": {
|
||||
|
||||
+42
@@ -4882,6 +4882,26 @@
|
||||
"title": "Thread/name/updatedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
{
|
||||
"properties": {
|
||||
"method": {
|
||||
"enum": [
|
||||
"thread/section/changed"
|
||||
],
|
||||
"title": "Thread/section/changedNotificationMethod",
|
||||
"type": "string"
|
||||
},
|
||||
"params": {
|
||||
"$ref": "#/definitions/v2/ThreadSectionChangedNotification"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"method",
|
||||
"params"
|
||||
],
|
||||
"title": "Thread/section/changedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
{
|
||||
"properties": {
|
||||
"method": {
|
||||
@@ -25245,6 +25265,28 @@
|
||||
},
|
||||
"type": "object"
|
||||
},
|
||||
"ThreadSectionChangedNotification": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"description": "Broadcast after a successful move into, within, or out of a section.",
|
||||
"properties": {
|
||||
"sectionId": {
|
||||
"description": "Destination section, or `null` when the thread was removed from its section.",
|
||||
"type": [
|
||||
"string",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"threadId": {
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"sectionId",
|
||||
"threadId"
|
||||
],
|
||||
"title": "ThreadSectionChangedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
"ThreadSectionCreateParams": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"description": "Parameters for creating an independently persisted thread section.",
|
||||
|
||||
+42
@@ -17143,6 +17143,26 @@
|
||||
"title": "Thread/name/updatedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
{
|
||||
"properties": {
|
||||
"method": {
|
||||
"enum": [
|
||||
"thread/section/changed"
|
||||
],
|
||||
"title": "Thread/section/changedNotificationMethod",
|
||||
"type": "string"
|
||||
},
|
||||
"params": {
|
||||
"$ref": "#/definitions/ThreadSectionChangedNotification"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"method",
|
||||
"params"
|
||||
],
|
||||
"title": "Thread/section/changedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
{
|
||||
"properties": {
|
||||
"method": {
|
||||
@@ -22948,6 +22968,28 @@
|
||||
},
|
||||
"type": "object"
|
||||
},
|
||||
"ThreadSectionChangedNotification": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"description": "Broadcast after a successful move into, within, or out of a section.",
|
||||
"properties": {
|
||||
"sectionId": {
|
||||
"description": "Destination section, or `null` when the thread was removed from its section.",
|
||||
"type": [
|
||||
"string",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"threadId": {
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"sectionId",
|
||||
"threadId"
|
||||
],
|
||||
"title": "ThreadSectionChangedNotification",
|
||||
"type": "object"
|
||||
},
|
||||
"ThreadSectionCreateParams": {
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"description": "Parameters for creating an independently persisted thread section.",
|
||||
|
||||
+22
@@ -0,0 +1,22 @@
|
||||
{
|
||||
"$schema": "http://json-schema.org/draft-07/schema#",
|
||||
"description": "Broadcast after a successful move into, within, or out of a section.",
|
||||
"properties": {
|
||||
"sectionId": {
|
||||
"description": "Destination section, or `null` when the thread was removed from its section.",
|
||||
"type": [
|
||||
"string",
|
||||
"null"
|
||||
]
|
||||
},
|
||||
"threadId": {
|
||||
"type": "string"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"sectionId",
|
||||
"threadId"
|
||||
],
|
||||
"title": "ThreadSectionChangedNotification",
|
||||
"type": "object"
|
||||
}
|
||||
BIN
Binary file not shown.
BIN
Binary file not shown.
File diff suppressed because one or more lines are too long
+2
-1
File diff suppressed because one or more lines are too long
+12
@@ -0,0 +1,12 @@
|
||||
// GENERATED CODE! DO NOT MODIFY BY HAND!
|
||||
|
||||
// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
|
||||
|
||||
/**
|
||||
* Broadcast after a successful move into, within, or out of a section.
|
||||
*/
|
||||
export type ThreadSectionChangedNotification = { threadId: string,
|
||||
/**
|
||||
* Destination section, or `null` when the thread was removed from its section.
|
||||
*/
|
||||
sectionId: string | null, };
|
||||
@@ -582,6 +582,7 @@ export type { ThreadSearchResult } from "./ThreadSearchResult";
|
||||
export type { ThreadSearchSortKey } from "./ThreadSearchSortKey";
|
||||
export type { ThreadSection } from "./ThreadSection";
|
||||
export type { ThreadSectionAppearance } from "./ThreadSectionAppearance";
|
||||
export type { ThreadSectionChangedNotification } from "./ThreadSectionChangedNotification";
|
||||
export type { ThreadSectionCreateParams } from "./ThreadSectionCreateParams";
|
||||
export type { ThreadSectionCreateResponse } from "./ThreadSectionCreateResponse";
|
||||
export type { ThreadSectionDeleteParams } from "./ThreadSectionDeleteParams";
|
||||
|
||||
@@ -2017,6 +2017,7 @@ server_notification_definitions! {
|
||||
ThreadReverted => "thread/reverted" (v2::ThreadRevertedNotification),
|
||||
SkillsChanged => "skills/changed" (v2::SkillsChangedNotification),
|
||||
ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification),
|
||||
ThreadSectionChanged => "thread/section/changed" (v2::ThreadSectionChangedNotification),
|
||||
ThreadAttachmentUpdated => "thread/attachment/updated" (v2::ThreadAttachmentUpdatedNotification),
|
||||
ThreadGoalUpdated => "thread/goal/updated" (v2::ThreadGoalUpdatedNotification),
|
||||
#[experimental("thread/prediction/updated")]
|
||||
|
||||
@@ -1130,6 +1130,20 @@ pub struct ThreadSectionMoveParams {
|
||||
#[ts(export_to = "v2/")]
|
||||
pub struct ThreadSectionMoveResponse {}
|
||||
|
||||
/// Broadcast after a successful move into, within, or out of a section.
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export_to = "v2/")]
|
||||
pub struct ThreadSectionChangedNotification {
|
||||
pub thread_id: String,
|
||||
/// Destination section, or `null` when the thread was removed from its section.
|
||||
#[schemars(
|
||||
required,
|
||||
schema_with = "crate::protocol::serde_helpers::nullable_string_schema"
|
||||
)]
|
||||
pub section_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, JsonSchema, TS)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
#[ts(rename_all = "lowercase")]
|
||||
|
||||
@@ -44,6 +44,13 @@ These errors are not evidence that history is exhausted; callers can retry.
|
||||
A successful response with `nextCursor: null` still indicates exhaustion.
|
||||
Default scan-and-repair requests retain their filesystem fallback.
|
||||
|
||||
# Thread section moves
|
||||
|
||||
After a successful `thread/section/move`, the app-server broadcasts
|
||||
`thread/section/changed` to its connected clients. The notification carries
|
||||
`threadId` and the destination `sectionId`, which is null for a removal.
|
||||
Moves within a section also emit this event so clients can refresh its order.
|
||||
|
||||
# Shutdown admission errors
|
||||
|
||||
When graceful shutdown rejects new work, app-server returns JSON-RPC error
|
||||
|
||||
@@ -1470,7 +1470,9 @@ impl MessageProcessor {
|
||||
.await
|
||||
}
|
||||
ClientRequest::ThreadSectionMove { params, .. } => {
|
||||
self.thread_processor.thread_section_move(params).await
|
||||
self.thread_processor
|
||||
.thread_section_move(request_id.clone(), params)
|
||||
.await
|
||||
}
|
||||
ClientRequest::ThreadSectionList { params, .. } => {
|
||||
self.thread_processor.thread_section_list(params).await
|
||||
|
||||
@@ -66,6 +66,7 @@ pub(crate) fn without_notification_media(notification: ServerNotification) -> Se
|
||||
| ServerNotification::ThreadReverted(_)
|
||||
| ServerNotification::SkillsChanged(_)
|
||||
| ServerNotification::ThreadNameUpdated(_)
|
||||
| ServerNotification::ThreadSectionChanged(_)
|
||||
| ServerNotification::ThreadAttachmentUpdated(_)
|
||||
| ServerNotification::ThreadGoalUpdated(_)
|
||||
| ServerNotification::ThreadPredictionUpdated(_)
|
||||
|
||||
@@ -32,6 +32,7 @@ use codex_app_server_protocol::ThreadSampleParams;
|
||||
use codex_app_server_protocol::ThreadSampleResponse;
|
||||
use codex_app_server_protocol::ThreadSection;
|
||||
use codex_app_server_protocol::ThreadSectionAppearance;
|
||||
use codex_app_server_protocol::ThreadSectionChangedNotification;
|
||||
use codex_app_server_protocol::ThreadSectionMoveParams;
|
||||
use codex_app_server_protocol::ThreadSectionMoveResponse;
|
||||
use codex_config::types::WindowsSandboxModeToml;
|
||||
@@ -785,6 +786,7 @@ impl ThreadRequestProcessor {
|
||||
|
||||
pub(crate) async fn thread_section_move(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
params: ThreadSectionMoveParams,
|
||||
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
|
||||
let ThreadSectionMoveParams {
|
||||
@@ -812,17 +814,23 @@ impl ThreadRequestProcessor {
|
||||
})
|
||||
.transpose()?;
|
||||
|
||||
{
|
||||
let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
|
||||
self.thread_manager
|
||||
.move_thread_to_section(thread_uuid, section_id.as_deref(), before_thread_uuid)
|
||||
.await
|
||||
.map_err(|err| core_thread_write_error("move thread in section", err))?;
|
||||
}
|
||||
|
||||
Ok(Some(ClientResponsePayload::ThreadSectionMove(
|
||||
ThreadSectionMoveResponse {},
|
||||
)))
|
||||
let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
|
||||
self.thread_manager
|
||||
.move_thread_to_section(thread_uuid, section_id.as_deref(), before_thread_uuid)
|
||||
.await
|
||||
.map_err(|err| core_thread_write_error("move thread in section", err))?;
|
||||
self.outgoing
|
||||
.send_response(request_id, ThreadSectionMoveResponse {})
|
||||
.await;
|
||||
self.outgoing
|
||||
.send_server_notification(ServerNotification::ThreadSectionChanged(
|
||||
ThreadSectionChangedNotification {
|
||||
thread_id: thread_uuid.to_string(),
|
||||
section_id,
|
||||
},
|
||||
))
|
||||
.await;
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
pub(crate) async fn thread_memory_mode_set(
|
||||
|
||||
@@ -791,7 +791,7 @@ pub(super) async fn send_initialize_request(
|
||||
.await
|
||||
}
|
||||
|
||||
async fn start_thread(stream: &mut WsClient, id: i64) -> Result<String> {
|
||||
pub(super) async fn start_thread(stream: &mut WsClient, id: i64) -> Result<String> {
|
||||
send_request(
|
||||
stream,
|
||||
"thread/start",
|
||||
|
||||
@@ -1,3 +1,11 @@
|
||||
use super::connection_handling_websocket::connect_websocket;
|
||||
use super::connection_handling_websocket::read_error_for_id;
|
||||
use super::connection_handling_websocket::read_notification_for_method;
|
||||
use super::connection_handling_websocket::read_response_for_id;
|
||||
use super::connection_handling_websocket::send_initialize_request;
|
||||
use super::connection_handling_websocket::send_request;
|
||||
use super::connection_handling_websocket::spawn_websocket_server;
|
||||
use super::connection_handling_websocket::start_thread;
|
||||
use anyhow::Result;
|
||||
use app_test_support::MockResponsesConfig;
|
||||
use app_test_support::TestAppServer;
|
||||
@@ -38,6 +46,58 @@ use uuid::Uuid;
|
||||
|
||||
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
|
||||
|
||||
#[tokio::test]
|
||||
async fn thread_section_moves_notify_other_clients() -> Result<()> {
|
||||
let responses = create_mock_responses_server_repeating_assistant("Done").await;
|
||||
let codex_home = TempDir::new()?;
|
||||
MockResponsesConfig::new(&responses.uri())
|
||||
.enable_feature(Feature::Sqlite)
|
||||
.write(codex_home.path())?;
|
||||
let (mut server, address) = spawn_websocket_server(codex_home.path()).await?;
|
||||
let mut writer = connect_websocket(address).await?;
|
||||
let mut observer = connect_websocket(address).await?;
|
||||
for client in [&mut writer, &mut observer] {
|
||||
send_initialize_request(client, /*id*/ 1, "section-test").await?;
|
||||
read_response_for_id(client, /*id*/ 1).await?;
|
||||
}
|
||||
let first = start_thread(&mut writer, /*id*/ 2).await?;
|
||||
let second = start_thread(&mut writer, /*id*/ 3).await?;
|
||||
|
||||
// A failed write must not invalidate the observer's last known section.
|
||||
send_request(
|
||||
&mut writer,
|
||||
"thread/section/move",
|
||||
/*id*/ 4,
|
||||
Some(json!({"threadId": first, "sectionId": Uuid::new_v4().to_string()})),
|
||||
)
|
||||
.await?;
|
||||
read_error_for_id(&mut writer, /*id*/ 4).await?;
|
||||
|
||||
let pinned = Some(PINNED_THREAD_SECTION_ID);
|
||||
for (thread, destination, before) in [
|
||||
(&first, pinned, None),
|
||||
(&second, pinned, None),
|
||||
(&second, pinned, Some(&first)),
|
||||
(&second, None, None),
|
||||
] {
|
||||
send_request(
|
||||
&mut writer,
|
||||
"thread/section/move",
|
||||
/*id*/ 5,
|
||||
Some(json!({"threadId": thread, "sectionId": destination, "beforeThreadId": before})),
|
||||
)
|
||||
.await?;
|
||||
read_response_for_id(&mut writer, /*id*/ 5).await?;
|
||||
let changed = read_notification_for_method(&mut observer, "thread/section/changed").await?;
|
||||
assert_eq!(
|
||||
changed.params,
|
||||
Some(json!({"threadId": thread, "sectionId": destination}))
|
||||
);
|
||||
}
|
||||
server.kill().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn section_request_error(
|
||||
server: &mut TestAppServer,
|
||||
method: &str,
|
||||
|
||||
@@ -205,6 +205,7 @@ pub(super) fn server_notification_thread_target(
|
||||
ServerNotification::ProjectChanged(_)
|
||||
| ServerNotification::SkillsChanged(_)
|
||||
| ServerNotification::AccountUpdated(_)
|
||||
| ServerNotification::ThreadSectionChanged(_)
|
||||
| ServerNotification::GatewayOAuthChanged(_)
|
||||
| ServerNotification::AccountRateLimitsUpdated(_)
|
||||
| ServerNotification::AppListUpdated(_)
|
||||
|
||||
@@ -379,6 +379,7 @@ impl ChatWidget {
|
||||
| ServerNotification::ThreadStarted(_)
|
||||
| ServerNotification::ThreadStatusChanged(_)
|
||||
| ServerNotification::ThreadReadStateChanged(_)
|
||||
| ServerNotification::ThreadSectionChanged(_)
|
||||
| ServerNotification::ThreadReverted(_)
|
||||
| ServerNotification::ThreadQueueChanged(_)
|
||||
| ServerNotification::ThreadPredictionUpdated(_)
|
||||
|
||||
@@ -77,6 +77,7 @@ from .v2_all import ThreadRealtimeStartedNotification
|
||||
from .v2_all import ThreadRealtimeTranscriptDeltaNotification
|
||||
from .v2_all import ThreadRealtimeTranscriptDoneNotification
|
||||
from .v2_all import ThreadRevertedNotification
|
||||
from .v2_all import ThreadSectionChangedNotification
|
||||
from .v2_all import ThreadSettingsUpdatedNotification
|
||||
from .v2_all import ThreadStartedNotification
|
||||
from .v2_all import ThreadStatusChangedNotification
|
||||
@@ -162,6 +163,7 @@ KnownNotificationPayload: TypeAlias = (
|
||||
| ThreadRealtimeTranscriptDeltaNotification
|
||||
| ThreadRealtimeTranscriptDoneNotification
|
||||
| ThreadRevertedNotification
|
||||
| ThreadSectionChangedNotification
|
||||
| ThreadSettingsUpdatedNotification
|
||||
| ThreadStartedNotification
|
||||
| ThreadStatusChangedNotification
|
||||
@@ -250,6 +252,7 @@ NOTIFICATION_MODELS: dict[str, type[KnownNotificationPayload]] = {
|
||||
"thread/realtime/transcript/delta": ThreadRealtimeTranscriptDeltaNotification,
|
||||
"thread/realtime/transcript/done": ThreadRealtimeTranscriptDoneNotification,
|
||||
"thread/reverted": ThreadRevertedNotification,
|
||||
"thread/section/changed": ThreadSectionChangedNotification,
|
||||
"thread/settings/updated": ThreadSettingsUpdatedNotification,
|
||||
"thread/started": ThreadStartedNotification,
|
||||
"thread/status/changed": ThreadStatusChangedNotification,
|
||||
|
||||
@@ -6308,6 +6308,20 @@ class ThreadSectionAppearance(BaseModel):
|
||||
icon: str | None = None
|
||||
|
||||
|
||||
class ThreadSectionChangedNotification(BaseModel):
|
||||
model_config = ConfigDict(
|
||||
populate_by_name=True,
|
||||
)
|
||||
section_id: Annotated[
|
||||
str | None,
|
||||
Field(
|
||||
alias="sectionId",
|
||||
description="Destination section, or `null` when the thread was removed from its section.",
|
||||
),
|
||||
]
|
||||
thread_id: Annotated[str, Field(alias="threadId")]
|
||||
|
||||
|
||||
class ThreadSectionCreateParams(BaseModel):
|
||||
model_config = ConfigDict(
|
||||
populate_by_name=True,
|
||||
@@ -9697,6 +9711,23 @@ class ThreadNameUpdatedServerNotification(BaseModel):
|
||||
params: ThreadNameUpdatedNotification
|
||||
|
||||
|
||||
class ThreadSectionChangedServerNotification(BaseModel):
|
||||
model_config = ConfigDict(
|
||||
populate_by_name=True,
|
||||
)
|
||||
emitted_at_ms: Annotated[
|
||||
int | None,
|
||||
Field(
|
||||
alias="emittedAtMs",
|
||||
description="Unix timestamp (in milliseconds) when app-server emitted this notification.",
|
||||
),
|
||||
] = None
|
||||
method: Annotated[
|
||||
Literal["thread/section/changed"], Field(title="Thread/section/changedNotificationMethod")
|
||||
]
|
||||
params: ThreadSectionChangedNotification
|
||||
|
||||
|
||||
class ThreadAttachmentUpdatedServerNotification(BaseModel):
|
||||
model_config = ConfigDict(
|
||||
populate_by_name=True,
|
||||
@@ -13634,6 +13665,7 @@ class ServerNotification(
|
||||
| ThreadRevertedServerNotification
|
||||
| SkillsChangedServerNotification
|
||||
| ThreadNameUpdatedServerNotification
|
||||
| ThreadSectionChangedServerNotification
|
||||
| ThreadAttachmentUpdatedServerNotification
|
||||
| ThreadGoalUpdatedServerNotification
|
||||
| ThreadPredictionUpdatedServerNotification
|
||||
@@ -13722,6 +13754,7 @@ class ServerNotification(
|
||||
| ThreadRevertedServerNotification
|
||||
| SkillsChangedServerNotification
|
||||
| ThreadNameUpdatedServerNotification
|
||||
| ThreadSectionChangedServerNotification
|
||||
| ThreadAttachmentUpdatedServerNotification
|
||||
| ThreadGoalUpdatedServerNotification
|
||||
| ThreadPredictionUpdatedServerNotification
|
||||
|
||||
Reference in New Issue
Block a user