openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add app-server support for reverting paginated threads (#38440)

Browse Source 
## What changed

- Add the experimental `thread/revert` request, which replaces a loaded paginated thread's durable history with the prefix before `beforeTurnId` while preserving the thread ID.
- Interrupt any active turn, reload the replacement history without unloading the thread, preserve mutable thread settings, and emit `thread/reverted`.
- Return backward pagination cursors for the retained turns and items, reject stale rollout paths after a revert, and document that local file changes are unaffected.

## Testing

- Cover history replacement, pagination, stale paths, missing turns, subsequent turns, active-turn interruption, and preserved thread settings.

GitOrigin-RevId: 435a8163ad4f310cc47219b436f6bdde21f2b83a
This commit is contained in:
Owen Lin
2026-08-13 21:45:02 +00:00
committed by copyberry
parent 1992f8c018
commit 4343b2bdc4
24 changed files with 939 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+31
View File
@@ -5066,6 +5066,17 @@
],
"type": "object"
},
"ThreadRevertedNotification": {
"properties": {
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"type": "object"
},
"ThreadSection": {
"description": "An independently persisted, user-visible thread section.",
"properties": {
@@ -6149,6 +6160,26 @@
"title": "Thread/closedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"thread/reverted"
],
"title": "Thread/revertedNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadRevertedNotification"
}
},
"required": [
"method",
"params"
],
"title": "Thread/revertedNotification",
"type": "object"
},
{
"properties": {
"method": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+33
View File
@@ -4477,6 +4477,26 @@
"title": "Thread/closedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"thread/reverted"
],
"title": "Thread/revertedNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/v2/ThreadRevertedNotification"
}
},
"required": [
"method",
"params"
],
"title": "Thread/revertedNotification",
"type": "object"
},
{
"properties": {
"method": {
@@ -20968,6 +20988,19 @@
"title": "ThreadResumeResponse",
"type": "object"
},
"ThreadRevertedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"title": "ThreadRevertedNotification",
"type": "object"
},
"ThreadRollbackParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "DEPRECATED: `thread/rollback` will be removed soon.",
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+33
View File
@@ -14447,6 +14447,26 @@
"title": "Thread/closedNotification",
"type": "object"
},
{
"properties": {
"method": {
"enum": [
"thread/reverted"
],
"title": "Thread/revertedNotificationMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadRevertedNotification"
}
},
"required": [
"method",
"params"
],
"title": "Thread/revertedNotification",
"type": "object"
},
{
"properties": {
"method": {
@@ -18720,6 +18740,19 @@
"title": "ThreadResumeResponse",
"type": "object"
},
"ThreadRevertedNotification": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"title": "ThreadRevertedNotification",
"type": "object"
},
"ThreadRollbackParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "DEPRECATED: `thread/rollback` will be removed soon.",
codex-rs/app-server-protocol/schema/json/v2/ThreadRevertedNotification.json
Generated
+13
View File
@@ -0,0 +1,13 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"threadId": {
"type": "string"
}
},
"required": [
"threadId"
],
"title": "ThreadRevertedNotification",
"type": "object"
}
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/typescript/ServerNotification.ts
Generated
+2 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/ServerNotificationEnvelope.ts
Generated
+2 -1
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/v2/ThreadRevertedNotification.ts
Generated
+5
View File
@@ -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 ThreadRevertedNotification = { threadId: string, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
Generated
+1
View File
@@ -478,6 +478,7 @@ export type { ThreadRealtimeTranscriptDoneNotification } from "./ThreadRealtimeT
export type { ThreadResumeInitialTurnsPageParams } from "./ThreadResumeInitialTurnsPageParams";
export type { ThreadResumeParams } from "./ThreadResumeParams";
export type { ThreadResumeResponse } from "./ThreadResumeResponse";
export type { ThreadRevertedNotification } from "./ThreadRevertedNotification";
export type { ThreadRollbackParams } from "./ThreadRollbackParams";
export type { ThreadRollbackResponse } from "./ThreadRollbackResponse";
export type { ThreadSearchResult } from "./ThreadSearchResult";
codex-rs/app-server-protocol/src/protocol/common.rs
+8
View File
@@ -646,6 +646,12 @@ client_request_definitions! {
serialization: thread_id(params.thread_id),
response: v2::ThreadRollbackResponse,
},
#[experimental("thread/revert")]
ThreadRevert => "thread/revert" {
params: v2::ThreadRevertParams,
serialization: thread_id(params.thread_id),
response: v2::ThreadRevertResponse,
},
ThreadList => "thread/list" {
params: v2::ThreadListParams,
inspect_params: true,
@@ -1711,6 +1717,8 @@ server_notification_definitions! {
ThreadDeleted => "thread/deleted" (v2::ThreadDeletedNotification),
ThreadUnarchived => "thread/unarchived" (v2::ThreadUnarchivedNotification),
ThreadClosed => "thread/closed" (v2::ThreadClosedNotification),
#[experimental("thread/reverted")]
ThreadReverted => "thread/reverted" (v2::ThreadRevertedNotification),
SkillsChanged => "skills/changed" (v2::SkillsChangedNotification),
ThreadNameUpdated => "thread/name/updated" (v2::ThreadNameUpdatedNotification),
ThreadGoalUpdated => "thread/goal/updated" (v2::ThreadGoalUpdatedNotification),
codex-rs/app-server-protocol/src/protocol/v2/thread.rs
+42 -4
View File
@@ -437,17 +437,17 @@ pub struct ThreadResumeResponse {
#[experimental("thread/resume.initialTurnsPage")]
#[serde(default)]
pub initial_turns_page: Option<TurnsPage>,
/// Opaque head cursor for hydrating paginated turns backwards.
/// Opaque cursor for hydrating paginated turns backwards.
///
/// Pass this as `cursor` to `thread/turns/list` with
/// `sortDirection: "desc"`. The first page includes the cursor's head turn.
/// `sortDirection: "desc"`. The first page includes the turn identified by the cursor.
#[experimental("thread/resume.turnsBackwardsCursor")]
#[serde(default)]
pub turns_backwards_cursor: Option<String>,
/// Opaque head cursor for hydrating paginated items backwards.
/// Opaque cursor for hydrating paginated items backwards.
///
/// Pass this as `cursor` to `thread/items/list` with
/// `sortDirection: "desc"`. The first page includes the cursor's head item.
/// `sortDirection: "desc"`. The first page includes the item identified by the cursor.
#[experimental("thread/resume.itemsBackwardsCursor")]
#[serde(default)]
pub items_backwards_cursor: Option<String>,
@@ -1117,6 +1117,37 @@ pub struct ThreadRollbackResponse {
pub thread: Thread,
}
/// Replace a paginated thread's durable history with the prefix before one turn.
///
/// This only changes persisted conversation history. It does not revert local file changes.
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRevertParams {
pub thread_id: String,
/// Turn excluded from the replacement history, together with every later turn.
pub before_turn_id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRevertResponse {
/// Updated loaded thread metadata. `turns` is always empty; hydrate retained history through
/// `thread/turns/list`.
pub thread: Thread,
/// Opaque cursor for hydrating paginated turns backwards.
///
/// Pass this as `cursor` to `thread/turns/list` with
/// `sortDirection: "desc"`. The first page includes the turn identified by the cursor.
pub turns_backwards_cursor: Option<String>,
/// Opaque cursor for hydrating paginated items backwards.
///
/// Pass this as `cursor` to `thread/items/list` with
/// `sortDirection: "desc"`. The first page includes the item identified by the cursor.
pub items_backwards_cursor: Option<String>,
}
/// Parameters for listing independently persisted thread sections.
#[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
@@ -1704,6 +1735,13 @@ pub struct ThreadUnarchivedNotification {
pub struct ThreadClosedNotification {
pub thread_id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRevertedNotification {
pub thread_id: String,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
codex-rs/app-server/README.md
+4 -3
View File
@@ -198,6 +198,7 @@ Example with notification opt-out:
- `thread/backgroundTerminals/list` — list running background terminals for a loaded thread (experimental; requires `capabilities.experimentalApi`); returns `data` with the running terminal ids.
- `thread/backgroundTerminals/terminate` — terminate one running background terminal by app-server `processId` (experimental; requires `capabilities.experimentalApi`); returns whether a process was terminated.
- `thread/rollback` — deprecated and will be removed soon. Drop the last N turns from the agent’s in-memory context and persist a rollback marker in the rollout so future resumes see the pruned history; returns the updated `thread` (with `turns` populated) on success. Paginated threads do not support rollback.
- `thread/revert` — experimental. Replace a loaded paginated thread's durable history with the prefix strictly before `beforeTurnId` while preserving its thread id. The operation interrupts an active turn if needed, leaves older rollout files immutable, reloads the thread, returns updated thread metadata with empty `turns` plus pagination cursors, and emits `thread/reverted`. It does not revert local file changes.
- `turn/start` — add user input to a thread and begin Codex generation; responds with the initial `turn` object and streams `turn/started`, `item/*`, and `turn/completed` notifications. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Experimental `runtimeWorkspaceRoots` supplies the default roots for newly resolved environment selections. Explicit `environments[].runtimeWorkspaceRoots` override that fallback with environment-native absolute paths. Prefer experimental `permissions` profile selection by id for permission overrides; the legacy `sandboxPolicy` field is still accepted but cannot be combined with `permissions`. For `collaborationMode`, `settings.developer_instructions: null` means "use built-in instructions for the selected mode". Deprecated experimental `multiAgentMode` is ignored; Ultra reasoning effort selects proactive behavior.
- `thread/inject_items` — append raw Responses API items to a loaded thread’s model-visible history without starting a user turn; returns `{}` on success.
- `turn/steer` — add user input to an already in-flight regular turn without starting a new turn; returns the active `turnId` that accepted the input. `clientUserMessageId` is optional; when supplied, the corresponding `userMessage` item echoes it as `clientId`. Review and manual compaction turns reject `turn/steer`.
@@ -352,7 +353,7 @@ To continue a stored session, call `thread/resume` with the `thread.id` you prev
By default, `thread/resume` includes the reconstructed turn history in `thread.turns`. Experimental clients can pass `excludeTurns: true` to return only thread metadata and live resume state, then call `thread/turns/list` separately if they want to page the turn history over the network. A cold paginated resume can still replay persisted `thread/tokenUsage/updated` when it can identify the corresponding stored turn; resuming an already-loaded thread waits for the next live update.
Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the cursor's head row, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet.
Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the row identified by the cursor, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet.
Only one app-server process can hold a paginated thread open for writing at a time. If another process already owns the thread, `thread/resume`, `thread/archive`, and `thread/delete` fail with JSON-RPC error `-32600`. Archive and deletion also fail if another process owns any spawned descendant. Read-only requests remain available without resuming the thread.
@@ -375,8 +376,8 @@ Example:
} }
{ "id": 12, "result": {
"thread": { "id": "thr_123", "turns": [], … },
"turnsBackwardsCursor": "turn-head-cursor-or-null",
"itemsBackwardsCursor": "item-head-cursor-or-null"
"turnsBackwardsCursor": "turn-backwards-cursor-or-null",
"itemsBackwardsCursor": "item-backwards-cursor-or-null"
} }
{ "method": "thread/resume", "id": 13, "params": {
codex-rs/app-server/src/message_processor.rs
+10
View File
@@ -1181,6 +1181,16 @@ impl MessageProcessor {
.thread_rollback(&request_id, params, app_server_client_name.as_deref())
.await
}
ClientRequest::ThreadRevert { params, .. } => {
self.thread_processor
.thread_revert(
request_id.clone(),
params,
app_server_client_name.clone(),
client_version.clone(),
)
.await
}
ClientRequest::ThreadList { params, .. } => {
self.thread_processor.thread_list(params).await
}
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+16 -5
View File
@@ -346,6 +346,14 @@ pub(super) async fn ensure_listener_task_running(
fallback_model_provider.clone(),
)
.await;
if matches!(event.msg, EventMsg::ShutdownComplete)
&& let Some(completion_tx) = thread_state
.lock()
.await
.take_shutdown_drain_waiter()
{
let _ = completion_tx.send(());
}
}
unloading_watchers_open = unloading_state.wait_for_unloading_trigger() => {
if !unloading_watchers_open {
@@ -421,11 +429,14 @@ pub(super) async fn unload_thread_without_subscribers(
tokio::spawn(async move {
match wait_for_thread_shutdown(&thread).await {
ThreadShutdownResult::Complete => {
if thread_manager.remove_thread(&thread_id).await.is_none() {
info!("thread {thread_id} was already removed before teardown finalized");
thread_watch_manager
.remove_thread(&thread_id.to_string())
.await;
// A delayed unload can finish after thread/revert replaces this runtime under
// the same thread ID. Only the runtime that scheduled this unload may remove it.
if thread_manager
.remove_thread_if_matches(&thread_id, &thread)
.await
.is_none()
{
info!("thread {thread_id} was replaced or removed before teardown finalized");
pending_thread_unloads.lock().await.remove(&thread_id);
return;
}
codex-rs/app-server/src/request_processors/thread_processor.rs
+285 -5
View File
@@ -4,6 +4,9 @@ use super::turn_processor::can_accept_direct_input;
use super::*;
use crate::error_code::method_not_found;
use codex_app_server_protocol::SelectedCapabilityRoot;
use codex_app_server_protocol::ThreadRevertParams;
use codex_app_server_protocol::ThreadRevertResponse;
use codex_app_server_protocol::ThreadRevertedNotification;
use codex_app_server_protocol::ThreadSection;
use codex_app_server_protocol::ThreadSectionAppearance;
use codex_app_server_protocol::ThreadSectionMoveParams;
@@ -33,6 +36,12 @@ struct ThreadListFilters {
relation_filter: Option<StoreThreadRelationFilter>,
}
struct ThreadRevertRuntimeSnapshot {
config: Config,
settings: ThreadConfigSnapshot,
client_mcp_extensions: ClientMcpExtensions,
}
fn collect_resume_override_mismatches(
request: &ThreadResumeParams,
config_snapshot: &ThreadConfigSnapshot,
@@ -554,6 +563,30 @@ impl ThreadRequestProcessor {
.map(|()| None)
}
pub(crate) async fn thread_revert(
&self,
request_id: ConnectionRequestId,
params: ThreadRevertParams,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
let (response, thread_id) = self
.thread_revert_response(
&request_id,
params,
app_server_client_name,
app_server_client_version,
)
.await?;
self.outgoing.send_response(request_id, response).await;
self.outgoing
.send_server_notification(ServerNotification::ThreadReverted(
ThreadRevertedNotification { thread_id },
))
.await;
Ok(None)
}
pub(crate) async fn thread_archive(
&self,
request_id: ConnectionRequestId,
@@ -1532,7 +1565,7 @@ impl ThreadRequestProcessor {
archive_thread_ids.push(thread_id);
}
}
Err(err) => return Err(thread_store_archive_error("archive", err)),
Err(err) => return Err(thread_store_mutation_error("archive", err)),
}
for descendant_thread_id in subtree_thread_ids.iter().copied().skip(1) {
match self
@@ -1573,7 +1606,7 @@ impl ThreadRequestProcessor {
writer_lock_thread_ids: subtree_thread_ids,
})
.await
.map_err(|err| thread_store_archive_error("archive", err))?
.map_err(|err| thread_store_mutation_error("archive", err))?
.into_iter()
.map(|thread_id| thread_id.to_string())
.collect();
@@ -1828,7 +1861,7 @@ impl ThreadRequestProcessor {
.thread_store
.unarchive_thread(StoreArchiveThreadParams { thread_id })
.await
.map_err(|err| thread_store_archive_error("unarchive", err))?;
.map_err(|err| thread_store_mutation_error("unarchive", err))?;
let (mut thread, _) =
thread_from_stored_thread(stored_thread, fallback_provider.as_str(), &self.config.cwd);
@@ -1851,6 +1884,227 @@ impl ThreadRequestProcessor {
self.thread_rollback_start(request_id, params).await
}
async fn thread_revert_response(
&self,
request_id: &ConnectionRequestId,
params: ThreadRevertParams,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
) -> Result<(ThreadRevertResponse, String), JSONRPCErrorError> {
let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?;
let ThreadRevertParams {
thread_id,
before_turn_id,
} = params;
let (thread_id, thread) = self.load_thread(&thread_id).await?;
let config_snapshot = thread.config_snapshot().await;
if !matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) {
return Err(invalid_request(
"thread/revert only supports paginated threads",
));
}
let runtime_snapshot = ThreadRevertRuntimeSnapshot {
config: thread.config().await.as_ref().clone(),
settings: config_snapshot,
client_mcp_extensions: thread.client_mcp_extensions(),
};
// Subscribe before shutdown so a pending idle unload either rejects this request or can
// no longer race the replacement runtime. The same listener then drains Core's shutdown
// events before we replace it.
if matches!(
self.ensure_conversation_listener(
thread_id,
request_id.connection_id,
/*raw_events_enabled*/ false,
)
.await?,
EnsureConversationListenerResult::ConnectionClosed
) {
return Err(internal_error(format!(
"connection closed before thread {thread_id} could be reverted"
)));
}
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
let shutdown_drain_rx = thread_state.lock().await.register_shutdown_drain_waiter();
match wait_for_thread_shutdown(&thread).await {
ThreadShutdownResult::Complete => {}
ThreadShutdownResult::SubmitFailed => {
thread_state.lock().await.take_shutdown_drain_waiter();
return Err(internal_error(format!(
"failed to shut down thread {thread_id} before revert"
)));
}
ThreadShutdownResult::TimedOut => {
thread_state.lock().await.take_shutdown_drain_waiter();
return Err(internal_error(format!(
"timed out shutting down thread {thread_id} before revert"
)));
}
}
let drain_result = tokio::time::timeout(Duration::from_secs(10), shutdown_drain_rx)
.await
.map_err(|_| {
internal_error(format!(
"timed out waiting for thread {thread_id} listener to drain shutdown events"
))
})
.and_then(|result| {
result.map_err(|_| {
internal_error(format!(
"thread {thread_id} listener stopped before draining shutdown events"
))
})
});
if let Err(err) = drain_result {
thread_state.lock().await.take_shutdown_drain_waiter();
return Err(err);
}
if self
.thread_manager
.remove_thread(&thread_id)
.await
.is_none()
{
return Err(internal_error(format!(
"thread {thread_id} disappeared before revert"
)));
}
// Keep thread state and subscriptions across the internal reload. Full teardown would
// force clients to call thread/resume after a successful revert.
self.outgoing
.cancel_requests_for_thread(thread_id, /*error*/ None)
.await;
let revert_result = self
.thread_store
.revert_thread(codex_thread_store::RevertThreadParams {
thread_id,
before_turn_id,
})
.await
.map_err(|err| thread_store_mutation_error("revert", err));
let response = self
.reload_paginated_thread(
request_id,
thread_id,
runtime_snapshot,
app_server_client_name,
app_server_client_version,
)
.await?;
revert_result?;
Ok((response, thread_id.to_string()))
}
async fn reload_paginated_thread(
&self,
request_id: &ConnectionRequestId,
thread_id: ThreadId,
runtime_snapshot: ThreadRevertRuntimeSnapshot,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
) -> Result<ThreadRevertResponse, JSONRPCErrorError> {
let ThreadRevertRuntimeSnapshot {
config,
settings,
client_mcp_extensions,
} = runtime_snapshot;
let thread_id_string = thread_id.to_string();
let stored_thread = self
.read_stored_thread_for_resume(
thread_id_string.as_str(),
/*path*/ None,
/*include_history*/ false,
)
.await?;
let (thread_history, resume_source_thread) = self
.load_resume_initial_history_from_stored_thread(stored_thread)
.await?;
let response_history = thread_history.clone();
let NewThread {
thread_id: resumed_thread_id,
thread: codex_thread,
session_configured,
..
} = self
.thread_manager
.resume_thread_with_history(
config,
thread_history,
self.auth_manager.clone(),
self.request_trace_context(request_id).await,
client_mcp_extensions,
)
.await
.map_err(|err| internal_error(format!("error reloading thread after revert: {err}")))?;
if resumed_thread_id != thread_id {
return Err(internal_error(format!(
"thread {thread_id} reloaded as {resumed_thread_id} after revert"
)));
}
codex_thread
.restore_thread_settings(settings)
.await
.map_err(|err| {
internal_error(format!(
"failed to restore thread settings after revert: {err}"
))
})?;
Self::set_app_server_client_info(
codex_thread.as_ref(),
app_server_client_name,
app_server_client_version,
)
.await?;
let SessionConfiguredEvent { rollout_path, .. } = session_configured;
let rollout_path = rollout_path.ok_or_else(|| {
internal_error(format!(
"rollout path missing after reloading thread {thread_id}"
))
})?;
log_listener_attach_result(
self.ensure_conversation_listener(
thread_id,
request_id.connection_id,
/*raw_events_enabled*/ false,
)
.await,
thread_id,
request_id.connection_id,
"thread",
);
let mut thread = self
.load_thread_from_resume_source_or_send_internal(
thread_id,
codex_thread.as_ref(),
&response_history,
rollout_path.as_path(),
Some(resume_source_thread),
/*include_turns*/ false,
)
.await
.map_err(internal_error)?;
self.thread_watch_manager.upsert_thread(&thread.id).await;
let thread_status = self
.thread_watch_manager
.loaded_status_for_thread(&thread.id)
.await;
set_thread_status_and_interrupt_stale_turns(
&mut thread,
thread_status,
/*has_live_in_progress_turn*/ false,
);
let (turns_backwards_cursor, items_backwards_cursor) =
Self::paginated_resume_backwards_cursors(self.thread_store.as_ref(), thread_id).await?;
Ok(ThreadRevertResponse {
thread,
turns_backwards_cursor,
items_backwards_cursor,
})
}
async fn thread_rollback_start(
&self,
request_id: &ConnectionRequestId,
@@ -2857,7 +3111,7 @@ impl ThreadRequestProcessor {
// durable next cursor still starts after the last returned stored turn.
let page_size = thread_turns_page_size(params.limit);
if page_size == 1 {
// ThreadStore does not accept an empty page. Use its head cursor as
// ThreadStore does not accept an empty page. Use its backwards cursor as
// the next cursor so the omitted durable row is returned next.
let mut page = self
.paginated_resume_initial_turns_page(thread_id, params)
@@ -3883,6 +4137,32 @@ impl ThreadRequestProcessor {
};
let stored_thread = result.map_err(thread_store_resume_read_error)?;
if let Some(requested_path) = path
&& matches!(stored_thread.history_mode, ThreadHistoryMode::Paginated)
{
let current_thread = self
.thread_store
.read_thread(StoreReadThreadParams {
thread_id: stored_thread.thread_id,
include_archived: true,
include_history: false,
})
.await
.map_err(thread_store_resume_read_error)?;
if let Some(current_path) = current_thread.rollout_path.as_ref()
&& !path_utils::paths_match_after_normalization(
codex_rollout::plain_rollout_path(requested_path).as_path(),
codex_rollout::plain_rollout_path(current_path).as_path(),
)
{
return Err(invalid_request(format!(
"cannot resume paginated thread {} with stale path: requested {}, current {}; omit path and resume by thread id",
stored_thread.thread_id,
requested_path.display(),
current_path.display()
)));
}
}
if stored_thread.archived_at.is_some() {
let thread_id = stored_thread.thread_id;
return Err(invalid_request(format!(
@@ -5188,7 +5468,7 @@ pub(super) fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPC
}
}
fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError {
fn thread_store_mutation_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError {
match err {
ThreadStoreError::InvalidRequest { message } | ThreadStoreError::Conflict { message } => {
invalid_request(message)
codex-rs/app-server/src/thread_state.rs
+14
View File
@@ -94,6 +94,9 @@ pub(crate) struct ThreadState {
pub(crate) pending_rollbacks: Option<ConnectionRequestId>,
pub(crate) turn_summary: TurnSummary,
pub(crate) last_terminal_turn_id: Option<String>,
/// Lets an internal runtime replacement wait until the old listener has processed Core's
/// `ShutdownComplete` event before that listener is superseded.
shutdown_drain_waiter: Option<oneshot::Sender<()>>,
pub(crate) cancel_tx: Option<oneshot::Sender<()>>,
pub(crate) experimental_raw_events: bool,
pub(crate) listener_generation: u64,
@@ -135,6 +138,7 @@ impl ThreadState {
if let Some(cancel_tx) = self.cancel_tx.take() {
let _ = cancel_tx.send(());
}
self.shutdown_drain_waiter = None;
self.listener_command_tx = None;
self.current_turn_history.reset();
self.listener_thread = None;
@@ -155,6 +159,16 @@ impl ThreadState {
self.current_turn_history.active_turn_snapshot()
}
pub(crate) fn register_shutdown_drain_waiter(&mut self) -> oneshot::Receiver<()> {
let (completion_tx, completion_rx) = oneshot::channel();
self.shutdown_drain_waiter = Some(completion_tx);
completion_rx
}
pub(crate) fn take_shutdown_drain_waiter(&mut self) -> Option<oneshot::Sender<()>> {
self.shutdown_drain_waiter.take()
}
pub(crate) fn track_current_turn_event(&mut self, event_turn_id: &str, event: &EventMsg) {
if let EventMsg::TurnStarted(payload) = event {
self.turn_summary.started_at = payload.started_at;
codex-rs/app-server/tests/suite/v2/mod.rs
+1
View File
@@ -92,6 +92,7 @@ mod thread_metadata_update;
mod thread_name_websocket;
mod thread_read;
mod thread_resume;
mod thread_revert;
mod thread_rollback;
mod thread_sections;
mod thread_settings_update;
codex-rs/app-server/tests/suite/v2/thread_revert.rs
+379
View File
@@ -0,0 +1,379 @@
use anyhow::Result;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
use app_test_support::create_final_assistant_message_sse_response;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::create_mock_responses_server_sequence;
use app_test_support::create_request_user_input_sse_response;
use codex_app_server_protocol::AskForApproval;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::InitializeCapabilities;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadItemsListParams;
use codex_app_server_protocol::ThreadItemsListResponse;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRevertParams;
use codex_app_server_protocol::ThreadRevertResponse;
use codex_app_server_protocol::ThreadRevertedNotification;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadTurnsListResponse;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::openai_models::ReasoningEffort;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
async fn thread_revert_replaces_paginated_history_before_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build()
.await?;
initialize_experimental(&mut mcp).await?;
let ThreadStartResponse { thread, .. } = mcp
.start_thread(ThreadStartParams {
history_mode: Some(ThreadHistoryMode::Paginated),
..Default::default()
})
.await?;
let stale_rollout_path = thread.path.clone().expect("thread rollout path");
let mut turn_ids = Vec::new();
for text in ["first", "second"] {
let completed = mcp
.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
turn_ids.push(completed.turn.id);
}
let ThreadRevertResponse {
thread: reverted_thread,
turns_backwards_cursor,
items_backwards_cursor,
} = mcp
.request(|request_id| ClientRequest::ThreadRevert {
request_id,
params: ThreadRevertParams {
thread_id: thread.id.clone(),
before_turn_id: turn_ids[1].clone(),
},
})
.await?;
let reverted: ThreadRevertedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("thread/reverted"),
)
.await??;
assert_eq!(reverted.thread_id, thread.id);
assert_eq!(reverted_thread.id, thread.id);
assert!(reverted_thread.turns.is_empty());
assert!(items_backwards_cursor.is_some());
assert_eq!(
turn_ids_from_cursor(
&mut mcp,
&thread.id,
turns_backwards_cursor,
/*sort_direction*/ None,
)
.await?,
turn_ids[..1]
);
let ThreadItemsListResponse {
data: reverted_items,
..
} = mcp
.request(|request_id| ClientRequest::ThreadItemsList {
request_id,
params: ThreadItemsListParams {
thread_id: thread.id.clone(),
turn_id: None,
cursor: items_backwards_cursor,
limit: None,
sort_direction: None,
},
})
.await?;
assert!(!reverted_items.is_empty());
assert!(
reverted_items
.iter()
.all(|item| item.turn_id == turn_ids[0])
);
mcp.shutdown_gracefully().await?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build()
.await?;
initialize_experimental(&mut mcp).await?;
let stale_resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
path: Some(stale_rollout_path),
..Default::default()
})
.await?;
let stale_resume_error: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(stale_resume_id)),
)
.await??;
assert!(
stale_resume_error.error.message.contains("stale path")
&& stale_resume_error
.error
.message
.contains("omit path and resume by thread id"),
"unexpected resume error: {}",
stale_resume_error.error.message,
);
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
..Default::default()
})
.await?;
let _: ThreadResumeResponse =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(resume_id)).await??;
let invalid_revert_id = mcp
.send_raw_request(
"thread/revert",
Some(serde_json::to_value(ThreadRevertParams {
thread_id: thread.id.clone(),
before_turn_id: "missing-turn".to_string(),
})?),
)
.await?;
let invalid_revert_error: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(invalid_revert_id)),
)
.await??;
assert_eq!(
invalid_revert_error.error.message,
"turn not found: missing-turn"
);
let third_turn = mcp
.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "third".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let requests = server.received_requests().await.expect("response requests");
let model_input = requests
.iter()
.rev()
.find(|request| request.url.path().ends_with("/responses"))
.expect("third turn response request")
.body_json::<serde_json::Value>()?["input"]
.clone();
let model_input = serde_json::to_string(&model_input)?;
assert!(model_input.contains("first"));
assert!(!model_input.contains("second"));
assert!(model_input.contains("third"));
assert_eq!(
turn_ids_from_cursor(
&mut mcp,
&thread.id,
/*cursor*/ None,
Some(SortDirection::Asc),
)
.await?,
vec![turn_ids[0].clone(), third_turn.turn.id]
);
Ok(())
}
#[tokio::test]
async fn thread_revert_interrupts_active_turn_and_keeps_thread_loaded() -> Result<()> {
let home = TempDir::new()?;
let server = create_mock_responses_server_sequence(vec![
create_final_assistant_message_sse_response("first")?,
create_request_user_input_sse_response("call_blocked")?,
create_final_assistant_message_sse_response("third")?,
])
.await;
MockResponsesConfig::new(&server.uri()).write(home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(home.path())
.build()
.await?;
initialize_experimental(&mut mcp).await?;
let ThreadStartResponse { thread, .. } = mcp
.start_thread(ThreadStartParams {
history_mode: Some(ThreadHistoryMode::Paginated),
..Default::default()
})
.await?;
let first_turn = mcp
.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "first".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let TurnStartResponse { turn: active_turn } = mcp
.request(|request_id| ClientRequest::TurnStart {
request_id,
params: TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "sleep".to_string(),
text_elements: Vec::new(),
}],
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Plan,
settings: Settings {
model: "mock-model".to_string(),
reasoning_effort: Some(ReasoningEffort::Medium),
developer_instructions: None,
},
}),
approval_policy: Some(AskForApproval::Never),
..Default::default()
},
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_request_message(),
)
.await??;
let ThreadRevertResponse {
thread: reverted_thread,
turns_backwards_cursor,
items_backwards_cursor,
} = mcp
.request(|request_id| ClientRequest::ThreadRevert {
request_id,
params: ThreadRevertParams {
thread_id: thread.id.clone(),
before_turn_id: active_turn.id.clone(),
},
})
.await?;
let completed: TurnCompletedNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("turn/completed"),
)
.await??;
assert_eq!(completed.thread_id, thread.id);
assert_eq!(completed.turn.status, TurnStatus::Interrupted);
assert!(reverted_thread.turns.is_empty());
assert!(items_backwards_cursor.is_some());
assert_eq!(
turn_ids_from_cursor(
&mut mcp,
&thread.id,
turns_backwards_cursor,
/*sort_direction*/ None,
)
.await?,
vec![first_turn.turn.id]
);
let resumed: ThreadResumeResponse = mcp
.request(|request_id| ClientRequest::ThreadResume {
request_id,
params: ThreadResumeParams {
thread_id: thread.id.clone(),
..Default::default()
},
})
.await?;
assert_eq!(resumed.approval_policy, AskForApproval::Never);
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: thread.id,
input: vec![UserInput::Text {
text: "third".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
Ok(())
}
async fn turn_ids_from_cursor(
mcp: &mut TestAppServer,
thread_id: &str,
cursor: Option<String>,
sort_direction: Option<SortDirection>,
) -> Result<Vec<String>> {
let ThreadTurnsListResponse { data, .. } = mcp
.request(|request_id| ClientRequest::ThreadTurnsList {
request_id,
params: ThreadTurnsListParams {
thread_id: thread_id.to_string(),
cursor,
limit: None,
sort_direction,
items_view: None,
},
})
.await?;
Ok(data.into_iter().map(|turn| turn.id).collect())
}
async fn initialize_experimental(mcp: &mut TestAppServer) -> Result<()> {
let initialized = timeout(
DEFAULT_READ_TIMEOUT,
mcp.initialize_with_capabilities(
ClientInfo {
name: "test-client".to_string(),
title: None,
version: "0.1.0".to_string(),
},
Some(InitializeCapabilities {
experimental_api: true,
request_attestation: false,
opt_out_notification_methods: None,
mcp_server_openai_form_elicitation: false,
extensions: None,
}),
),
)
.await??;
assert!(matches!(initialized, JSONRPCMessage::Response(_)));
Ok(())
}
codex-rs/core/src/codex_thread.rs
+36
View File
@@ -20,6 +20,7 @@ use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::mcp::CallToolResult;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::models::ActivePermissionProfile;
use codex_protocol::models::ContentItem;
use codex_protocol::models::PermissionProfile;
@@ -125,6 +126,22 @@ impl ThreadConfigSnapshot {
collaboration_mode: self.collaboration_mode,
}
}
fn into_thread_settings_overrides(self) -> CodexThreadSettingsOverrides {
CodexThreadSettingsOverrides {
environments: Some(self.environments),
profile_workspace_roots: Some(self.profile_workspace_roots),
approval_policy: Some(self.approval_policy),
approvals_reviewer: Some(self.approvals_reviewer),
permission_profile: Some(self.permission_profile),
active_permission_profile: self.active_permission_profile,
summary: self.reasoning_summary,
service_tier: Some(self.service_tier),
collaboration_mode: Some(self.collaboration_mode),
personality: self.personality,
..Default::default()
}
}
}
/// Thread settings overrides that app-server validates before starting a turn.
@@ -406,6 +423,20 @@ impl CodexThread {
self.session.preview_settings(&updates).await
}
/// Restores effective mutable settings captured from another loaded runtime.
///
/// Runtime replacement uses this after resume so clients keep their current thread settings
/// rather than reverting to the original layer-backed config.
pub async fn restore_thread_settings(
&self,
snapshot: ThreadConfigSnapshot,
) -> ConstraintResult<()> {
let updates = self
.thread_settings_update(snapshot.into_thread_settings_overrides())
.await;
self.session.update_settings(updates).await
}
async fn thread_settings_update(
&self,
overrides: CodexThreadSettingsOverrides,
@@ -617,6 +648,11 @@ impl CodexThread {
self.session.thread_config_snapshot().await
}
/// Returns the MCP extensions declared by the client that created this runtime.
pub fn client_mcp_extensions(&self) -> ClientMcpExtensions {
self.session.services.client_mcp_extensions.clone()
}
/// Returns the files that supplied the thread's loaded model instructions.
pub async fn instruction_sources(&self) -> Vec<PathUri> {
self.session.instruction_sources().await
codex-rs/core/src/thread_manager.rs
+20
View File
@@ -1040,6 +1040,26 @@ impl ThreadManager {
self.state.threads.write().await.remove(thread_id)
}
/// Removes a thread only if `thread_id` still maps to `expected`.
///
/// Delayed cleanup uses this to avoid removing a replacement runtime registered under the
/// same thread ID.
pub async fn remove_thread_if_matches(
&self,
thread_id: &ThreadId,
expected: &Arc<CodexThread>,
) -> Option<Arc<CodexThread>> {
let mut threads = self.state.threads.write().await;
if threads
.get(thread_id)
.is_some_and(|thread| Arc::ptr_eq(thread, expected))
{
threads.remove(thread_id)
} else {
None
}
}
/// Tries to shut down all tracked threads concurrently within the provided timeout.
/// Threads that complete shutdown are removed from the manager; incomplete shutdowns
/// remain tracked so callers can retry or inspect them later.
codex-rs/tui/src/app/app_server_event_targets.rs
+1
View File
@@ -51,6 +51,7 @@ pub(super) fn server_notification_thread_target(
ServerNotification::ThreadStatusChanged(notification) => {
Some(notification.thread_id.as_str())
}
ServerNotification::ThreadReverted(notification) => Some(notification.thread_id.as_str()),
ServerNotification::ThreadArchived(notification) => Some(notification.thread_id.as_str()),
ServerNotification::ThreadDeleted(notification) => Some(notification.thread_id.as_str()),
ServerNotification::ThreadUnarchived(notification) => Some(notification.thread_id.as_str()),
codex-rs/tui/src/app/tests/safety_buffering.rs
+2 -2
View File
@@ -747,10 +747,10 @@ goals = true
let expected_source_tokens = if committed_steer.is_some() { 150 } else { 50 };
assert_eq!(source_goal.objective, RETRY_GOAL);
assert_eq!(source_goal.tokens_used, expected_source_tokens);
assert_eq!(source_goal.time_used_seconds, 12);
assert!(source_goal.time_used_seconds >= 12);
assert_eq!(retry_goal.objective, RETRY_GOAL);
assert!(retry_goal.tokens_used >= expected_source_tokens);
assert!(retry_goal.time_used_seconds >= 12);
assert!(retry_goal.time_used_seconds >= source_goal.time_used_seconds);
let request_bodies = server
.requests()
codex-rs/tui/src/chatwidget/protocol.rs
+1
View File
@@ -198,6 +198,7 @@ impl ChatWidget {
| ServerNotification::AccountRateLimitsUpdated(_)
| ServerNotification::ThreadStarted(_)
| ServerNotification::ThreadStatusChanged(_)
| ServerNotification::ThreadReverted(_)
| ServerNotification::ThreadArchived(_)
| ServerNotification::ThreadDeleted(_)
| ServerNotification::ThreadUnarchived(_)