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 bc34bb89dd..01986e2001 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 @@ -23435,6 +23435,28 @@ "title": "ThreadReadResponse", "type": "object" }, + "ThreadReadState": { + "description": "The first unread position in a durable local thread. A null position means read. The opaque revision guards edits against activity or marks the caller has not seen.", + "properties": { + "firstUnread": { + "anyOf": [ + { + "$ref": "#/definitions/v2/ThreadUnreadPosition" + }, + { + "type": "null" + } + ] + }, + "revision": { + "type": "string" + } + }, + "required": [ + "revision" + ], + "type": "object" + }, "ThreadRealtimeAudioChunk": { "description": "EXPERIMENTAL - thread realtime audio chunk.", "properties": { @@ -25340,6 +25362,48 @@ "title": "ThreadUnarchivedNotification", "type": "object" }, + "ThreadUnreadPosition": { + "oneOf": [ + { + "description": "Explicitly marked unread, including when the thread has no turns.", + "properties": { + "type": { + "enum": [ + "threadStart" + ], + "title": "ThreadStartThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "ThreadStartThreadUnreadPosition", + "type": "object" + }, + { + "description": "The earliest completed result needing attention.", + "properties": { + "turnId": { + "type": "string" + }, + "type": { + "enum": [ + "turn" + ], + "title": "TurnThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "turnId", + "type" + ], + "title": "TurnThreadUnreadPosition", + "type": "object" + } + ] + }, "ThreadUnsubscribeParams": { "$schema": "http://json-schema.org/draft-07/schema#", "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 61aac9999f..95ffa2fe63 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 @@ -21149,6 +21149,28 @@ "title": "ThreadReadResponse", "type": "object" }, + "ThreadReadState": { + "description": "The first unread position in a durable local thread. A null position means read. The opaque revision guards edits against activity or marks the caller has not seen.", + "properties": { + "firstUnread": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadUnreadPosition" + }, + { + "type": "null" + } + ] + }, + "revision": { + "type": "string" + } + }, + "required": [ + "revision" + ], + "type": "object" + }, "ThreadRealtimeAudioChunk": { "description": "EXPERIMENTAL - thread realtime audio chunk.", "properties": { @@ -23054,6 +23076,48 @@ "title": "ThreadUnarchivedNotification", "type": "object" }, + "ThreadUnreadPosition": { + "oneOf": [ + { + "description": "Explicitly marked unread, including when the thread has no turns.", + "properties": { + "type": { + "enum": [ + "threadStart" + ], + "title": "ThreadStartThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "ThreadStartThreadUnreadPosition", + "type": "object" + }, + { + "description": "The earliest completed result needing attention.", + "properties": { + "turnId": { + "type": "string" + }, + "type": { + "enum": [ + "turn" + ], + "title": "TurnThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "turnId", + "type" + ], + "title": "TurnThreadUnreadPosition", + "type": "object" + } + ] + }, "ThreadUnsubscribeParams": { "$schema": "http://json-schema.org/draft-07/schema#", "properties": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json index ab43908fbe..bbf8ed7d34 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json @@ -2262,6 +2262,28 @@ } ] }, + "ThreadReadState": { + "description": "The first unread position in a durable local thread. A null position means read. The opaque revision guards edits against activity or marks the caller has not seen.", + "properties": { + "firstUnread": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadUnreadPosition" + }, + { + "type": "null" + } + ] + }, + "revision": { + "type": "string" + } + }, + "required": [ + "revision" + ], + "type": "object" + }, "ThreadSection": { "description": "An independently persisted, user-visible thread section.", "properties": { @@ -2388,6 +2410,48 @@ } ] }, + "ThreadUnreadPosition": { + "oneOf": [ + { + "description": "Explicitly marked unread, including when the thread has no turns.", + "properties": { + "type": { + "enum": [ + "threadStart" + ], + "title": "ThreadStartThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "ThreadStartThreadUnreadPosition", + "type": "object" + }, + { + "description": "The earliest completed result needing attention.", + "properties": { + "turnId": { + "type": "string" + }, + "type": { + "enum": [ + "turn" + ], + "title": "TurnThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "turnId", + "type" + ], + "title": "TurnThreadUnreadPosition", + "type": "object" + } + ] + }, "Turn": { "properties": { "completedAt": { diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json index d824e57036..9c4975d2ba 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json @@ -2262,6 +2262,28 @@ } ] }, + "ThreadReadState": { + "description": "The first unread position in a durable local thread. A null position means read. The opaque revision guards edits against activity or marks the caller has not seen.", + "properties": { + "firstUnread": { + "anyOf": [ + { + "$ref": "#/definitions/ThreadUnreadPosition" + }, + { + "type": "null" + } + ] + }, + "revision": { + "type": "string" + } + }, + "required": [ + "revision" + ], + "type": "object" + }, "ThreadSection": { "description": "An independently persisted, user-visible thread section.", "properties": { @@ -2388,6 +2410,48 @@ } ] }, + "ThreadUnreadPosition": { + "oneOf": [ + { + "description": "Explicitly marked unread, including when the thread has no turns.", + "properties": { + "type": { + "enum": [ + "threadStart" + ], + "title": "ThreadStartThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "ThreadStartThreadUnreadPosition", + "type": "object" + }, + { + "description": "The earliest completed result needing attention.", + "properties": { + "turnId": { + "type": "string" + }, + "type": { + "enum": [ + "turn" + ], + "title": "TurnThreadUnreadPositionType", + "type": "string" + } + }, + "required": [ + "turnId", + "type" + ], + "title": "TurnThreadUnreadPosition", + "type": "object" + } + ] + }, "Turn": { "properties": { "completedAt": { 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 c9b39369e9..2571775b78 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 6644afa894..9e9b36b47c 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/ThreadListResponse.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadListResponse.ts index 51757e24c3..71a34fdb34 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadListResponse.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadListResponse.ts @@ -3,16 +3,14 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { Thread } from "./Thread"; -export type ThreadListResponse = { data: Array, -/** +export type ThreadListResponse = {data: Array, /** * Opaque cursor to pass to the next call to continue after the last item. * if None, there are no more items to return. */ -nextCursor: string | null, -/** +nextCursor: string | null, /** * Opaque cursor to pass as `cursor` when reversing `sortDirection`. * This is only populated when the page contains at least one thread. * Use it with the opposite `sortDirection`; for timestamp sorts it anchors * at the start of the page timestamp so same-second updates are not skipped. */ -backwardsCursor: string | null, }; +backwardsCursor: string | null}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadResponse.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadResponse.ts index a6da50649c..e3ecf38f5d 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadResponse.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadResponse.ts @@ -3,4 +3,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { Thread } from "./Thread"; -export type ThreadReadResponse = { thread: Thread, }; +export type ThreadReadResponse = {thread: Thread}; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadState.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadState.ts new file mode 100644 index 0000000000..8324a78a75 --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadReadState.ts @@ -0,0 +1,10 @@ +// 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. +import type { ThreadUnreadPosition } from "./ThreadUnreadPosition"; + +/** + * The first unread position in a durable local thread. A null position means read. + * The opaque revision guards edits against activity or marks the caller has not seen. + */ +export type ThreadReadState = { firstUnread: ThreadUnreadPosition | null, revision: string, }; diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/ThreadUnreadPosition.ts b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadUnreadPosition.ts new file mode 100644 index 0000000000..da5d8b24dc --- /dev/null +++ b/codex-rs/app-server-protocol/schema/typescript/v2/ThreadUnreadPosition.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 ThreadUnreadPosition = { "type": "threadStart" } | { "type": "turn", turnId: string, }; 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 a90d0548a3..acc4e3c9e8 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/index.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/index.ts @@ -539,6 +539,7 @@ export type { ThreadProjectUpdatedNotification } from "./ThreadProjectUpdatedNot export type { ThreadQueueChangedNotification } from "./ThreadQueueChangedNotification"; export type { ThreadReadParams } from "./ThreadReadParams"; export type { ThreadReadResponse } from "./ThreadReadResponse"; +export type { ThreadReadState } from "./ThreadReadState"; export type { ThreadRealtimeAudioChunk } from "./ThreadRealtimeAudioChunk"; export type { ThreadRealtimeBemItemPresentation } from "./ThreadRealtimeBemItemPresentation"; export type { ThreadRealtimeClosedNotification } from "./ThreadRealtimeClosedNotification"; @@ -600,6 +601,7 @@ export type { ThreadTurnsListResponse } from "./ThreadTurnsListResponse"; export type { ThreadUnarchiveParams } from "./ThreadUnarchiveParams"; export type { ThreadUnarchiveResponse } from "./ThreadUnarchiveResponse"; export type { ThreadUnarchivedNotification } from "./ThreadUnarchivedNotification"; +export type { ThreadUnreadPosition } from "./ThreadUnreadPosition"; export type { ThreadUnsubscribeParams } from "./ThreadUnsubscribeParams"; export type { ThreadUnsubscribeResponse } from "./ThreadUnsubscribeResponse"; export type { ThreadUnsubscribeStatus } from "./ThreadUnsubscribeStatus"; 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 120bf66ff9..609e266a07 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1559,11 +1559,15 @@ pub enum SortDirection { Desc, } -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadListResponse { pub data: Vec, + /// Local receipts for eligible threads on this page, keyed by thread ID. + /// Null means durable read state is unavailable; an absent key means that row is ineligible. + #[experimental("thread/list.readStates")] + pub read_states: Option>, /// Opaque cursor to pass to the next call to continue after the last item. /// if None, there are no more items to return. pub next_cursor: Option, @@ -1708,11 +1712,14 @@ pub struct ThreadReadParams { pub include_turns: bool, } -#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS, ExperimentalApi)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] pub struct ThreadReadResponse { pub thread: Thread, + /// Local receipt. Null means this thread's durable read state is unavailable. + #[experimental("thread/read.readState")] + pub read_state: Option, } #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] @@ -2011,6 +2018,30 @@ pub struct ThreadStatusChangedNotification { pub status: ThreadStatus, } +/// The first unread position in a durable local thread. A null position means read. +/// The opaque revision guards edits against activity or marks the caller has not seen. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadReadState { + pub first_unread: Option, + pub revision: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(tag = "type", rename_all = "camelCase")] +#[ts(tag = "type", rename_all = "camelCase", export_to = "v2/")] +pub enum ThreadUnreadPosition { + /// Explicitly marked unread, including when the thread has no turns. + ThreadStart, + /// The earliest completed result needing attention. + Turn { + #[serde(rename = "turnId")] + #[ts(rename = "turnId")] + turn_id: String, + }, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, 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 1852530983..b88d0c87ff 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -19,6 +19,29 @@ 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 read state (experimental) + +Local durable ordinary user threads expose a `readState` on the +`thread/read` response. `thread/list` returns `readStates`, a map from eligible +thread IDs on the returned page to their receipts. Null (or absent on older servers) +means durable read state is unavailable; a missing map entry means that row is +ineligible. Ephemeral, not-yet-persisted, service and product-owned threads are +ineligible. Shared `Thread` objects, including search and lifecycle responses, +carry no receipts. After start, resume or fork, refresh with `thread/read`. +Each local thread has one receipt, with the same storage ownership as the thread. + +A read state is `{firstUnread, revision}`. `firstUnread` is null when read, +`{type:"threadStart"}` for an explicit unread mark (even for an empty thread), +or `{type:"turn", turnId}` for the earliest eligible completed result. +`revision` is an opaque string, scoped to this thread. + +Only newly delivered, durable, visible terminal results advance read state, +before `turn/completed`. Every new result changes the revision, even when an +older turn remains the first unread position. Reverting a paginated thread +clears a removed position; explicit thread-start marks survive. Loading snapshots +never acknowledges activity; empty threads start read and forked history does +not publish inherited turns as new results. + # Guardian circuit-breaker errors Set `auto_review.circuit_break_action = "strict"` to include `TooManyDenials` in diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 22249329ed..0edce83c3f 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -570,6 +570,7 @@ mod thread_fork_goal; mod thread_input; mod thread_processor; mod thread_queue_processor; +mod thread_read_state; mod thread_sections; mod token_usage_replay; mod turn_processor; diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index 4418577ee6..2b455908cc 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -341,10 +341,12 @@ pub(super) async fn ensure_listener_task_running( // Track the event before emitting any typed translations // so thread-local state such as raw event opt-in stays // synchronized with the conversation. - let raw_events_enabled = { + let (raw_events_enabled, terminal_summary) = { let mut thread_state = thread_state.lock().await; thread_state.track_current_turn_event(&event.id, &event.msg); - thread_state.experimental_raw_events + let terminal = matches!(event.msg, EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_)) + .then(|| thread_state.turn_summary.clone()); + (thread_state.experimental_raw_events, terminal) }; if matches!( &event.msg, @@ -362,6 +364,12 @@ pub(super) async fn ensure_listener_task_running( conversation_id, ); + // Make reads after turn/completed reflect the committed frontier. + if let Some(summary) = terminal_summary { + super::thread_read_state::publish( + &event, conversation_id, &conversation, &summary, + ).await; + } apply_bespoke_event_handling( event.clone(), conversation_id, 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 abc2daf877..d9ba44f647 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -2218,6 +2218,53 @@ impl ThreadRequestProcessor { thread_state.lock().await.take_shutdown_drain_waiter(); return Err(err); } + // The listener is drained: even an interrupted turn is durable now. Inspect only the + // suffix being reverted and remember the revision, not a collection of turn IDs. + let receipt_to_retire = async { + let Some(db) = self.state_db.as_ref() else { + return anyhow::Ok(None); + }; + let Some(state) = db + .thread_read_states(&[thread_id]) + .await? + .remove(&thread_id) + else { + return Ok(None); + }; + let Some(turn_id) = state.first_unread_turn.as_ref().filter(|id| !id.is_empty()) else { + return Ok(None); + }; + let mut cursor = None; + loop { + let page = self + .thread_store + .list_turns(codex_thread_store::ListTurnsParams { + thread_id, + include_archived: false, + cursor: cursor.clone(), + page_size: THREAD_TURNS_MAX_LIMIT, + sort_direction: codex_thread_store::SortDirection::Desc, + items_view: codex_thread_store::StoredTurnItemsView::NotLoaded, + }) + .await?; + for turn in page.turns { + if turn.turn_id == *turn_id { + return Ok(Some(state.revision)); + } + if turn.turn_id == before_turn_id { + return Ok(None); + } + } + let Some(next) = page.next_cursor else { + return Ok(None); + }; + if cursor.as_ref() == Some(&next) { + anyhow::bail!("thread/revert turn cursor did not advance"); + } + cursor = Some(next); + } + } + .await; if self .thread_manager .remove_thread(&thread_id) @@ -2243,6 +2290,27 @@ impl ThreadRequestProcessor { }) .await .map_err(|err| thread_store_mutation_error("revert", err)); + // Receipt lookup or update errors cannot prevent reload or misreport a committed revert. + // A concurrent mark wins over this cleanup, even while the thread was being truncated. + if revert_result.is_ok() + && let Some(db) = self.state_db.as_ref() + { + let retirement = match receipt_to_retire { + Ok(Some(revision)) => db + .update_thread_read_state( + thread_id, + &revision, + codex_state::ReadStateOperation::Read, + ) + .await + .map(|_| ()), + Ok(None) => Ok(()), + Err(err) => Err(err), + }; + if let Err(err) = retirement { + tracing::warn!("reverted history but could not update unread position: {err}") + } + } let response = self .reload_paginated_thread( request_id, @@ -2662,8 +2730,10 @@ impl ThreadRequestProcessor { |thread| thread, ) .await; + let read_states = super::thread_read_state::snapshots(self.state_db.as_ref(), &data).await; Ok(ThreadListResponse { data, + read_states, next_cursor, backwards_cursor, }) @@ -2854,7 +2924,13 @@ impl ThreadRequestProcessor { .read_thread_view(thread_uuid, include_turns) .await .map_err(thread_read_view_error)?; - Ok(ThreadReadResponse { thread }) + let read_state = super::thread_read_state::snapshots( + self.state_db.as_ref(), + std::slice::from_ref(&thread), + ) + .await + .and_then(|mut states| states.remove(&thread.id)); + Ok(ThreadReadResponse { thread, read_state }) } /// Builds the API view for `thread/read` from persisted metadata plus optional live state. 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 new file mode 100644 index 0000000000..5f3d8aeadf --- /dev/null +++ b/codex-rs/app-server/src/request_processors/thread_read_state.rs @@ -0,0 +1,122 @@ +//! Read state for durable local threads returned by the thread store. + +use super::*; +use codex_app_server_protocol::ThreadReadState; +use codex_app_server_protocol::ThreadUnreadPosition; + +// These scopes have product notification policy the local server cannot resolve. +// Do not advertise read state for them until their policy reaches this publisher. +fn eligible(thread: &Thread) -> bool { + !thread.ephemeral + && !matches!( + thread.source, + codex_app_server_protocol::SessionSource::SubAgent(_) + | codex_app_server_protocol::SessionSource::Unknown + ) + && matches!( + thread.thread_source, + None | Some(codex_app_server_protocol::ThreadSource::User) + ) +} + +fn to_api(state: codex_state::ThreadReadState) -> ThreadReadState { + ThreadReadState { + first_unread: state.first_unread_turn.map(|turn_id| { + if turn_id.is_empty() { + ThreadUnreadPosition::ThreadStart + } else { + ThreadUnreadPosition::Turn { turn_id } + } + }), + revision: state.revision, + } +} + +/// Read receipts only for the local store's list/read results, in one bounded batch. +pub(super) async fn snapshots( + db: Option<&StateDbHandle>, + threads: &[Thread], +) -> Option> { + let db = db?; + let ids = threads + .iter() + .filter(|t| eligible(t)) + .filter_map(|t| ThreadId::from_string(&t.id).ok()) + .collect::>(); + match db.thread_read_states(&ids).await { + Ok(states) => Some( + states + .into_iter() + .map(|(id, state)| (id.to_string(), to_api(state))) + .collect(), + ), + Err(err) => { + tracing::warn!("thread read state unavailable: {err}"); + None + } + } +} + +/// Runs only on newly delivered Core events, never on history decoding or replay. +pub(super) async fn publish( + event: &codex_protocol::protocol::Event, + id: ThreadId, + conversation: &Arc, + summary: &crate::thread_state::TurnSummary, +) -> Option { + let db = conversation.state_db()?; + let config = conversation.config_snapshot().await; + if config.ephemeral + || matches!( + config.session_source, + codex_protocol::protocol::SessionSource::SubAgent(_) + | codex_protocol::protocol::SessionSource::Internal(_) + | codex_protocol::protocol::SessionSource::Unknown + ) + || !matches!( + config.thread_source, + None | Some(codex_protocol::protocol::ThreadSource::User) + ) + { + return None; + } + let has_error = summary.last_error.is_some(); + let final_text = match &summary.last_agent_message { + Some(ThreadItem::AgentMessage { text, .. }) => Some(text.as_str()), + _ => None, + }; + let visible = match &event.msg { + EventMsg::TurnComplete(_) if has_error => true, + EventMsg::TurnComplete(_) => { + let text = final_text?; + if text.contains("DONT_NOTIFY") { + return None; + } + // An unfinished goal will resume automatically; it is not a terminal result. + match db.thread_goals().get_thread_goal(id).await { + Ok(Some(goal)) if goal.status == codex_state::ThreadGoalStatus::Active => false, + Ok(_) => !text.trim().is_empty(), + Err(err) => { + tracing::warn!("could not resolve terminal attention policy: {err}"); + return None; + } + } + } + EventMsg::TurnAborted(aborted) => { + aborted.error.is_some() || has_error || final_text.is_some() + } + _ => false, + }; + if visible { + if let Err(err) = conversation.flush_rollout().await { + tracing::warn!("cannot publish attention before the terminal result is durable: {err}"); + return None; + } + match db.publish_thread_attention(id, &event.id).await { + Ok(true) => return Some(db), + Ok(false) => {} + Err(err) => tracing::warn!("failed to publish terminal attention: {err}"), + } + } + None +} diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 9166598de5..68167089eb 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/codex-rs/app-server/tests/suite/v2/mod.rs @@ -149,6 +149,7 @@ mod thread_name_persistence; mod thread_name_websocket; mod thread_queue; mod thread_read; +mod thread_read_state; mod thread_resume; mod thread_revert; mod thread_sections; diff --git a/codex-rs/app-server/tests/suite/v2/rollout_migration.rs b/codex-rs/app-server/tests/suite/v2/rollout_migration.rs index da20bdbea8..a9fb5aa9c3 100644 --- a/codex-rs/app-server/tests/suite/v2/rollout_migration.rs +++ b/codex-rs/app-server/tests/suite/v2/rollout_migration.rs @@ -91,7 +91,7 @@ async fn runtime_enabled_legacy_migration_preserves_cold_resume_model_context() include_turns: false, }) .await?; - let ThreadReadResponse { thread: read } = secondary.read_response(read_id).await?; + let ThreadReadResponse { thread: read, .. } = secondary.read_response(read_id).await?; if read.history_mode == ThreadHistoryMode::Paginated { return Ok::<(), anyhow::Error>(()); } diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index b11ff57b5b..e17c4a9a0a 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -353,7 +353,9 @@ async fn thread_archive_without_turns(history_mode: ThreadHistoryMode) -> Result .await??; assert_eq!(archived_notification.thread_id, thread.id); - let ThreadReadResponse { thread: archived } = mcp + let ThreadReadResponse { + thread: archived, .. + } = mcp .request(|request_id| ClientRequest::ThreadRead { request_id, params: ThreadReadParams { diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index 03835a08bd..d0c9e1a80c 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -1292,6 +1292,7 @@ async fn thread_fork_can_cut_before_unfinished_stored_turn() -> Result<()> { .await?; let ThreadReadResponse { thread: source_thread, + .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; assert_eq!(source_thread.turns.len(), 2); assert_eq!(source_thread.turns[1].id, unfinished_turn_id); diff --git a/codex-rs/app-server/tests/suite/v2/thread_list.rs b/codex-rs/app-server/tests/suite/v2/thread_list.rs index 35dcb81241..f697f9e141 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_list.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_list.rs @@ -1005,7 +1005,7 @@ async fn thread_search_returns_content_matches() -> Result<()> { include_turns: false, }) .await?; - let ThreadReadResponse { thread } = + let ThreadReadResponse { thread, .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(request_id)).await??; pinned_threads.push(thread); } @@ -1245,6 +1245,7 @@ async fn thread_list_db_only_errors_distinguish_unavailable_database_from_empty_ page, ThreadListResponse { data: vec![], + read_states: Some(HashMap::new()), next_cursor: None, backwards_cursor: None } diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index a79255bfdb..00e65fee24 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -291,6 +291,7 @@ async fn thread_read_preserves_file_id_and_root_from_completed_turn( .await?; let ThreadReadResponse { thread: loaded_thread, + .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; let ThreadItem::UserMessage { content, .. } = &loaded_thread.turns[0].items[0] else { panic!("expected completed turn to start with a user message"); @@ -316,6 +317,7 @@ async fn thread_read_preserves_file_id_and_root_from_completed_turn( .await?; let ThreadReadResponse { thread: restarted_thread, + .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; let ThreadItem::UserMessage { content, .. } = &restarted_thread.turns[0].items[0] else { panic!("expected persisted turn to start with a user message"); @@ -370,7 +372,7 @@ async fn paginated_stored_thread_routes_projected_turns() -> Result<()> { include_turns: false, }) .await?; - let ThreadReadResponse { thread } = + let ThreadReadResponse { thread, .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated); assert!(thread.turns.is_empty()); @@ -1202,7 +1204,7 @@ async fn thread_read_can_return_archived_threads_by_id() -> Result<()> { include_turns: false, }) .await?; - let ThreadReadResponse { thread } = + let ThreadReadResponse { thread, .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; assert_eq!(thread.id, conversation_id); @@ -1874,6 +1876,7 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( .await?; let ThreadReadResponse { thread: unloaded_thread, + .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; assert_eq!(unloaded_thread.turns, expected_full_turns); @@ -2229,6 +2232,7 @@ async fn paginated_history_lists_and_legacy_reads_use_projected_turns_and_items( .await?; let ThreadReadResponse { thread: loaded_thread, + .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; assert_eq!(&loaded_thread.turns[..2], expected_full_turns); assert_eq!( 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 new file mode 100644 index 0000000000..ccc01fef0b --- /dev/null +++ b/codex-rs/app-server/tests/suite/v2/thread_read_state.rs @@ -0,0 +1,121 @@ +use anyhow::Context; +use anyhow::Result; +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::ClientRequest; +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::ThreadRevertParams; +use codex_app_server_protocol::ThreadRevertResponse; +use codex_app_server_protocol::ThreadStartParams; +use codex_app_server_protocol::TurnStartParams; +use codex_app_server_protocol::UserInput; +use codex_features::Feature; +use pretty_assertions::assert_eq; +use serde_json::json; +use tempfile::TempDir; +use tokio::time::Duration; +use tokio::time::timeout; + +const TIMEOUT: Duration = Duration::from_secs(15); + +#[tokio::test] +async fn read_state_tracks_terminal_activity_and_reverted_history() -> Result<()> { + let responses = create_mock_responses_server_repeating_assistant("Ready").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&responses.uri()) + .enable_feature(Feature::Sqlite) + .write(codex_home.path())?; + let mut server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let started = server + .start_thread(ThreadStartParams { + model: Some("mock-model".into()), + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let thread_id = started.thread.id; + let completed = timeout( + TIMEOUT, + server.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread_id.clone(), + input: vec![UserInput::Text { + text: "Finish this task.".into(), + text_elements: vec![], + }], + ..Default::default() + }), + ) + .await??; + let loaded: ThreadReadResponse = server + .request(|request_id| ClientRequest::ThreadRead { + request_id, + params: ThreadReadParams { + thread_id: thread_id.clone(), + include_turns: false, + }, + }) + .await?; + let first = loaded + .read_state + .context("durable completed user thread should have read state")?; + assert_eq!( + first.first_unread, + Some(codex_app_server_protocol::ThreadUnreadPosition::Turn { + turn_id: completed.turn.id.clone() + }) + ); + + let reloaded: ThreadReadResponse = server + .request(|request_id| ClientRequest::ThreadRead { + request_id, + params: ThreadReadParams { + thread_id: thread_id.clone(), + include_turns: false, + }, + }) + .await?; + assert_eq!( + reloaded.read_state, + Some(first.clone()), + "loading a snapshot must never acknowledge the activity" + ); + + let id = server + .send_raw_request("thread/list", Some(json!({"limit": 1}))) + .await?; + let listed: ThreadListResponse = timeout(TIMEOUT, server.read_response(id)).await??; + let receipts = listed.read_states.context("durable list receipts")?; + assert_eq!(listed.data.len(), 1); + assert_eq!(listed.data[0].id, thread_id); + assert_eq!(receipts, [(thread_id.clone(), first.clone())].into()); + + let _: ThreadRevertResponse = server + .request(|request_id| ClientRequest::ThreadRevert { + request_id, + params: ThreadRevertParams { + thread_id: thread_id.clone(), + before_turn_id: completed.turn.id, + }, + }) + .await?; + let reverted: ThreadReadResponse = server + .request(|request_id| ClientRequest::ThreadRead { + request_id, + params: ThreadReadParams { + thread_id, + include_turns: false, + }, + }) + .await?; + let after = reverted.read_state.context("reverted snapshot")?; + assert_eq!(after.first_unread, None); + assert_ne!(after.revision, first.revision); + Ok(()) +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 1413da3b7e..0c005f7dab 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -1421,7 +1421,7 @@ async fn thread_resume_preserves_acknowledged_model_effort_and_approvals_reviewe include_turns: false, }) .await?; - let ThreadReadResponse { thread: read } = + let ThreadReadResponse { thread: read, .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; assert_eq!(read.cwd.as_path(), persisted_cwd); diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index fcc328d28b..50041ea975 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -124,6 +124,7 @@ async fn thread_unarchive_moves_rollout_back_into_sessions_directory() -> Result .await?; let ThreadReadResponse { thread: pinned_thread, + .. } = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??; assert_eq!(pinned_thread.section, Some(pinned_section.clone())); let pinned_entered_at = pinned_thread diff --git a/codex-rs/state/migrations/0060_thread_read_state.sql b/codex-rs/state/migrations/0061_thread_read_state.sql similarity index 100% rename from codex-rs/state/migrations/0060_thread_read_state.sql rename to codex-rs/state/migrations/0061_thread_read_state.sql diff --git a/codex-rs/tui/src/app_server_session/thread_list_tests.rs b/codex-rs/tui/src/app_server_session/thread_list_tests.rs index 7ce2c5509a..fe1ae9ee02 100644 --- a/codex-rs/tui/src/app_server_session/thread_list_tests.rs +++ b/codex-rs/tui/src/app_server_session/thread_list_tests.rs @@ -65,7 +65,7 @@ async fn thread_list_preserves_filters_and_retries_only_legacy_cwd_errors() -> R if code == 0 || retry { assert_eq!( serde_json::to_value(result?)?, - json!({"data": [], "nextCursor": "next", "backwardsCursor": null}) + json!({"data": [], "readStates": null, "nextCursor": "next", "backwardsCursor": null}) ); } else { assert!(format!("{:#}", result.unwrap_err()).contains(message)); diff --git a/sdk/python/src/openai_codex/generated/v2_all.py b/sdk/python/src/openai_codex/generated/v2_all.py index 0a8d8fa442..beba74ebd6 100644 --- a/sdk/python/src/openai_codex/generated/v2_all.py +++ b/sdk/python/src/openai_codex/generated/v2_all.py @@ -6297,6 +6297,28 @@ class ThreadUnarchivedNotification(BaseModel): thread_id: Annotated[str, Field(alias="threadId")] +class ThreadStartThreadUnreadPosition(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + type: Annotated[Literal["threadStart"], Field(title="ThreadStartThreadUnreadPositionType")] + + +class TurnThreadUnreadPosition(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + turn_id: Annotated[str, Field(alias="turnId")] + type: Annotated[Literal["turn"], Field(title="TurnThreadUnreadPositionType")] + + +class ThreadUnreadPosition(RootModel[ThreadStartThreadUnreadPosition | TurnThreadUnreadPosition]): + model_config = ConfigDict( + populate_by_name=True, + ) + root: ThreadStartThreadUnreadPosition | TurnThreadUnreadPosition + + class ThreadUnsubscribeParams(BaseModel): model_config = ConfigDict( populate_by_name=True, @@ -10202,6 +10224,14 @@ class ThreadListParams(BaseModel): ] = None +class ThreadReadState(BaseModel): + model_config = ConfigDict( + populate_by_name=True, + ) + first_unread: Annotated[ThreadUnreadPosition | None, Field(alias="firstUnread")] = None + revision: str + + class TranscriptSegmentThreadRealtimeItem(BaseModel): model_config = ConfigDict( populate_by_name=True,