diff --git a/codex-rs/app-server-protocol/schema/json/ClientRequest.json b/codex-rs/app-server-protocol/schema/json/ClientRequest.json index 06a057533e..e3623b2b03 100644 --- a/codex-rs/app-server-protocol/schema/json/ClientRequest.json +++ b/codex-rs/app-server-protocol/schema/json/ClientRequest.json @@ -4942,6 +4942,42 @@ ], "type": "object" }, + "ThreadReadStateOperation": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "read" + ], + "title": "ReadThreadReadStateOperationType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "ReadThreadReadStateOperation", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "unread" + ], + "title": "UnreadThreadReadStateOperationType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "UnreadThreadReadStateOperation", + "type": "object" + } + ] + }, "ThreadRealtimeAudioChunk": { "description": "EXPERIMENTAL - thread realtime audio chunk.", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index 3a91f47408..5801d70f38 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -23517,6 +23517,42 @@ "title": "ThreadReadStateChangedNotification", "type": "object" }, + "ThreadReadStateOperation": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "read" + ], + "title": "ReadThreadReadStateOperationType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "ReadThreadReadStateOperation", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "unread" + ], + "title": "UnreadThreadReadStateOperationType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "UnreadThreadReadStateOperation", + "type": "object" + } + ] + }, "ThreadRealtimeAudioChunk": { "description": "EXPERIMENTAL - thread realtime audio chunk.", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 8147a4ed26..edc42d3cf4 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -21231,6 +21231,42 @@ "title": "ThreadReadStateChangedNotification", "type": "object" }, + "ThreadReadStateOperation": { + "oneOf": [ + { + "properties": { + "type": { + "enum": [ + "read" + ], + "title": "ReadThreadReadStateOperationType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "ReadThreadReadStateOperation", + "type": "object" + }, + { + "properties": { + "type": { + "enum": [ + "unread" + ], + "title": "UnreadThreadReadStateOperationType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "UnreadThreadReadStateOperation", + "type": "object" + } + ] + }, "ThreadRealtimeAudioChunk": { "description": "EXPERIMENTAL - thread realtime audio chunk.", "properties": { diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index 71f272c2f6..f409b73b9d 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index 6bdbd16795..96be2b3865 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadStateOperation.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadStateOperation.ts new file mode 100644 index 0000000000..a6759a5ec7 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadStateOperation.ts @@ -0,0 +1,5 @@ +// 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. + +export type ThreadReadStateOperation = { "type": "read" } | { "type": "unread" }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts index 26ee4dcc7e..4072345343 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -541,6 +541,7 @@ export type { ThreadReadParams } from "./ThreadReadParams"; export type { ThreadReadResponse } from "./ThreadReadResponse"; export type { ThreadReadState } from "./ThreadReadState"; export type { ThreadReadStateChangedNotification } from "./ThreadReadStateChangedNotification"; +export type { ThreadReadStateOperation } from "./ThreadReadStateOperation"; export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk"; export type { ThreadRealtimeBemItemPresentation } from "./ThreadRealtimeBemItemPresentation"; export type { ThreadRealtimeClosedNotification } from "./ThreadRealtimeClosedNotification"; diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 262464aede..ec94ae843d 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -860,6 +860,12 @@ client_request_definitions! { serialization: thread_id(params.thread_id), response: v2::ThreadReadResponse, }, + #[experimental("thread/readState/update")] + ThreadReadStateUpdate => "thread/readState/update" { + params: v2::ThreadReadStateUpdateParams, + serialization: thread_id(params.thread_id), + response: v2::ThreadReadStateUpdateResponse, + }, ThreadTurnsList => "thread/turns/list" { params: v2::ThreadTurnsListParams, // Explicitly concurrent: this primarily reads append-only rollout storage. diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 4495fd0fa7..a88c8608d6 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -2042,6 +2042,35 @@ pub enum ThreadUnreadPosition { }, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde( + tag = "type", + rename_all = "camelCase", + rename_all_fields = "camelCase" +)] +#[ts(tag = "type", rename_all = "camelCase", export_to = "v2/")] +pub enum ThreadReadStateOperation { + Read, + Unread, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadReadStateUpdateParams { + pub thread_id: String, + pub operation: ThreadReadStateOperation, + /// Revision of the row or transcript actually seen. Conflicts require a fresh user action. + pub expected_revision: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadReadStateUpdateResponse { + pub read_state: ThreadReadState, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 3c8e4f6ffd..bc4bbe3b17 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -43,8 +43,17 @@ clears a removed position; explicit thread-start marks survive. Loading snapshot never acknowledges activity; empty threads start read and forked history does not publish inherited turns as new results. -Enabling `initialize.capabilities.experimentalApi` permits receiving read-state -notifications. +Enabling `initialize.capabilities.experimentalApi` permits updating read state and +receiving its notifications. + +`thread/readState/update` takes `{threadId, expectedRevision, operation}`, +where operation is `{type:"read"}` or `{type:"unread"}`. +Use the revision of the row or transcript actually seen. An unread edit +starts at `threadStart`; a read edit clears the position. Both require an +exact revision match, and both conflicts return `-32600` with +`{reason:"readStateConflict", readState:}` as error data. +Do not blindly retry: the original action may precede an unseen result or +another window's mark. Unavailable threads fail; writes never change recency. Started, resumed or forked threads send `thread/readState/changed` alongside ordinary thread events; thread/unsubscribe or disconnect stops delivery. diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 61987a1fb0..7a1c09aa48 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1518,6 +1518,9 @@ impl MessageProcessor { ClientRequest::ThreadRead { params, .. } => { self.thread_processor.thread_read(&request_id, params).await } + ClientRequest::ThreadReadStateUpdate { params, .. } => { + self.thread_processor.thread_read_state_update(params).await + } ClientRequest::ThreadTurnsList { params, .. } => { self.thread_processor.thread_turns_list(params).await } diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index bea69d7cf6..5af5b0126c 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2946,7 +2946,7 @@ impl ThreadRequestProcessor { } /// Builds the API view for `thread/read` from persisted metadata plus optional live state. - async fn read_thread_view( + pub(super) async fn read_thread_view( &self, thread_id: ThreadId, include_turns: bool, @@ -6054,14 +6054,14 @@ pub(super) fn normalize_thread_turns_status( } } -enum ThreadReadViewError { +pub(super) enum ThreadReadViewError { InvalidRequest(String), Unsupported(&'static str), Internal(String), JsonRpc(JSONRPCErrorError), } -fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError { +pub(super) fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError { match err { ThreadReadViewError::InvalidRequest(message) => invalid_request(message), ThreadReadViewError::Unsupported(operation) => { diff --git a/codex-rs/app-server/src/request_processors/thread_read_state.rs b/codex-rs/app-server/src/request_processors/thread_read_state.rs index 33bb9f090a..a64c31b55e 100644 --- a/codex-rs/app-server/src/request_processors/thread_read_state.rs +++ b/codex-rs/app-server/src/request_processors/thread_read_state.rs @@ -1,9 +1,15 @@ //! Read state for durable local threads. Receipt changes use the thread's usual subscribers. use super::*; +use crate::error_code::method_not_found; use codex_app_server_protocol::ThreadReadState; use codex_app_server_protocol::ThreadReadStateChangedNotification; +use codex_app_server_protocol::ThreadReadStateOperation; +use codex_app_server_protocol::ThreadReadStateUpdateParams; +use codex_app_server_protocol::ThreadReadStateUpdateResponse; use codex_app_server_protocol::ThreadUnreadPosition; +use codex_state::ReadStateOperation; +use codex_state::ReadStateUpdate; // These scopes have product notification policy the local server cannot resolve. // Do not advertise read state for them until their policy reaches this publisher. @@ -145,3 +151,51 @@ pub(super) async fn publish( } None } + +impl ThreadRequestProcessor { + pub(crate) async fn thread_read_state_update( + &self, + params: ThreadReadStateUpdateParams, + ) -> Result, JSONRPCErrorError> { + let id = ThreadId::from_string(¶ms.thread_id) + .map_err(|err| invalid_params(format!("invalid thread id: {err}")))?; + let db = self + .state_db + .as_ref() + .ok_or_else(|| method_not_found("durable thread read state unavailable"))?; + let thread = self + .read_thread_view(id, /*include_turns*/ false) + .await + .map_err(super::thread_processor::thread_read_view_error)?; + if !eligible(&thread) { + return Err(invalid_request("thread read state unavailable")); + } + let operation = match params.operation { + ThreadReadStateOperation::Read => ReadStateOperation::Read, + ThreadReadStateOperation::Unread => ReadStateOperation::Unread, + }; + let outcome = db + .update_thread_read_state(id, ¶ms.expected_revision, operation) + .await + .map_err(|err| internal_error(format!("read state was not saved: {err}")))?; + match outcome { + ReadStateUpdate::Applied(state) => { + notify(db, &self.thread_state_manager, &self.outgoing, id).await; + Ok(Some( + ThreadReadStateUpdateResponse { + read_state: to_api(state), + } + .into(), + )) + } + ReadStateUpdate::Conflict(state) => { + let mut error = invalid_request("read state changed"); + error.data = Some( + serde_json::json!({ "reason": "readStateConflict", "readState": to_api(state) }), + ); + Err(error) + } + ReadStateUpdate::Unavailable => Err(invalid_request("thread read state unavailable")), + } + } +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_read_state.rs b/codex-rs/app-server/tests/suite/v2/thread_read_state.rs index 32fe3923d0..da6b3aa1da 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read_state.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read_state.rs @@ -1,14 +1,19 @@ use anyhow::Context; use anyhow::Result; +use app_test_support::DEFAULT_CLIENT_NAME; use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; use app_test_support::create_mock_responses_server_repeating_assistant; +use codex_app_server_protocol::ClientInfo; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeCapabilities; +use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadReadStateChangedNotification; +use codex_app_server_protocol::ThreadReadStateUpdateResponse; use codex_app_server_protocol::ThreadRevertParams; use codex_app_server_protocol::ThreadRevertResponse; use codex_app_server_protocol::ThreadStartParams; @@ -23,6 +28,40 @@ use tokio::time::timeout; const TIMEOUT: Duration = Duration::from_secs(15); +#[tokio::test] +async fn read_state_methods_require_experimental_capability() -> Result<()> { + let mut server = TestAppServer::builder().without_auto_env().build().await?; + server + .initialize_with_capabilities( + ClientInfo { + name: DEFAULT_CLIENT_NAME.to_string(), + title: None, + version: "0.1.0".to_string(), + }, + Some(InitializeCapabilities { + experimental_api: false, + ..Default::default() + }), + ) + .await?; + let id = server + .send_raw_request( + "thread/readState/update", + Some(json!({ + "threadId": "unknown", + "expectedRevision": "none", "operation": {"type": "unread"} + })), + ) + .await?; + let error = timeout( + TIMEOUT, + server.read_stream_until_error_message(RequestId::Integer(id)), + ) + .await??; + assert!(error.error.message.contains("experimental"), "{error:?}"); + Ok(()) +} + #[tokio::test] async fn read_state_tracks_terminal_activity_and_reverted_history() -> Result<()> { let responses = create_mock_responses_server_repeating_assistant("Ready").await; @@ -104,6 +143,64 @@ async fn read_state_tracks_terminal_activity_and_reverted_history() -> Result<() .await??; assert_eq!(changed.read_state, first); + let id = server + .send_raw_request( + "thread/readState/update", + Some(json!({ + "threadId": thread_id, + "expectedRevision": first.revision, "operation": {"type": "unread"} + })), + ) + .await?; + let marked: ThreadReadStateUpdateResponse = + timeout(TIMEOUT, server.read_response(id)).await??; + assert_eq!( + marked.read_state.first_unread, + Some(codex_app_server_protocol::ThreadUnreadPosition::ThreadStart) + ); + let changed: ThreadReadStateChangedNotification = timeout( + TIMEOUT, + server.read_notification("thread/readState/changed"), + ) + .await??; + assert_eq!(changed.read_state, marked.read_state); + + // The initial snapshot predates this explicit unread mark. Both stale actions + // must conflict, and neither may erase the manual marker or change the revision. + for operation in ["read", "unread"] { + let id = server + .send_raw_request( + "thread/readState/update", + Some(json!({ + "threadId": thread_id, + "expectedRevision": first.revision, + "operation": {"type": operation} + })), + ) + .await?; + let error = timeout( + TIMEOUT, + server.read_stream_until_error_message(RequestId::Integer(id)), + ) + .await??; + assert_eq!( + error.error.data, + Some(json!({"reason": "readStateConflict", "readState": marked.read_state})) + ); + } + let id = server + .send_raw_request( + "thread/readState/update", + Some(json!({ + "threadId": thread_id, + "expectedRevision": marked.read_state.revision, + "operation": {"type": "read"} + })), + ) + .await?; + let fresh: ThreadReadStateUpdateResponse = timeout(TIMEOUT, server.read_response(id)).await??; + assert_eq!(fresh.read_state.first_unread, None); + let _: ThreadRevertResponse = server .request(|request_id| ClientRequest::ThreadRevert { request_id, diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 394f97bf99..012abd9e1e 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -5761,6 +5761,29 @@ class ThreadReadParams(BaseModel): thread_id: Annotated[str, Field(alias="threadId")] +class ReadThreadReadStateOperation(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + type: Annotated[Literal["read"], Field(title="ReadThreadReadStateOperationType")] + + +class UnreadThreadReadStateOperation(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + type: Annotated[Literal["unread"], Field(title="UnreadThreadReadStateOperationType")] + + +class ThreadReadStateOperation( + RootModel[ReadThreadReadStateOperation | UnreadThreadReadStateOperation] +): + model_config = ConfigDict( + populate_by_name=True, + ) + root: ReadThreadReadStateOperation | UnreadThreadReadStateOperation + + class ThreadRealtimeAudioChunk(BaseModel): model_config = ConfigDict( populate_by_name=True,