Add experimental app-server thread read-state updates (#52395)

## Why

Clients need to mark threads read or unread without overwriting unseen results or another window's explicit mark.

## What changed

Add `thread/readState/update` for eligible durable local threads, gated by `initialize.capabilities.experimentalApi`. Requests include `threadId`, `expectedRevision`, and an operation of `{type:"read"}` or `{type:"unread"}`.

Marking read clears `firstUnread`; marking unread sets it to `threadStart`. Successful updates return the read state and notify subscribers through `thread/readState/changed`. Both operations require an exact revision match. Stale requests return `-32600` with `readStateConflict` and the current read state, requiring a fresh user action rather than a blind retry. Updates do not change thread recency.

## Testing

Add app-server coverage for experimental capability gating, explicit unread marks and notifications, conflicts for both stale operations, and marking read with a fresh revision.

GitOrigin-RevId: 775f6e9584dc91bd3c2ee9a309144b5baf0edffa
This commit is contained in:
Brent Traut
2026-10-09 03:42:19 +00:00
committed by copyberry
parent 82883da25e
commit e45069d770
15 changed files with 340 additions and 5 deletions
@@ -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": {
@@ -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": {
@@ -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": {
@@ -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" };
@@ -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";
@@ -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.
@@ -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/")]
+11 -2
View File
@@ -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:<current>}` 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.
@@ -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
}
@@ -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) => {
@@ -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<Option<ClientResponsePayload>, JSONRPCErrorError> {
let id = ThreadId::from_string(&params.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, &params.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")),
}
}
}
@@ -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,
@@ -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,