github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Remove the deprecated thread/rollback API (#44915)

Browse Source 
## What changed

Remove `thread/rollback`, its request and response types, generated bindings, and the core `Op::ThreadRollback` operation. Requests now follow the generic unknown-method rejection path. Document `thread/revert` as the alternative for paginated threads.

Keep historical `ThreadRolledBack` markers and legacy error deserialization so existing rollouts remain compatible with replay and migration.

## Testing

Adapt retained-context and Guardian history tests to append legacy rollback markers and resume threads, preserving coverage of surviving instructions, answers, and review history.

GitOrigin-RevId: b3da1becdf86b1869275aacb0ffc2817cee5af2e
This commit is contained in:
Owen Lin
2026-09-11 21:32:04 +00:00
committed by copyberry
parent c62d191c4c
commit 3052bbcf8c
65 changed files with 124 additions and 5472 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/ClientRequest.json
Generated
-43
View File
@@ -4941,25 +4941,6 @@
],
"type": "object"
},
"ThreadRollbackParams": {
"description": "DEPRECATED: `thread/rollback` will be removed soon.",
"properties": {
"numTurns": {
"description": "The number of turns to drop from the end of the thread. Must be >= 1.\n\nThis only modifies the thread's history and does not revert local file changes that have been made by the agent. Clients are responsible for reverting these changes.",
"format": "uint32",
"minimum": 0.0,
"type": "integer"
},
"threadId": {
"type": "string"
}
},
"required": [
"numTurns",
"threadId"
],
"type": "object"
},
"ThreadSearchSortKey": {
"enum": [
"created_at",
@@ -6325,30 +6306,6 @@
"title": "Thread/approveGuardianDeniedActionRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/rollback"
],
"title": "Thread/rollbackRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadRollbackParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/rollbackRequest",
"type": "object"
},
{
"properties": {
"id": {
codex-rs/app-server-protocol/schema/json/ServerNotification.json
Generated
+1 -1
View File
@@ -4276,7 +4276,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+1 -64
View File
@@ -684,30 +684,6 @@
"title": "Thread/approveGuardianDeniedActionRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/v2/RequestId"
},
"method": {
"enum": [
"thread/rollback"
],
"title": "Thread/rollbackRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/v2/ThreadRollbackParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/rollbackRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -20602,7 +20578,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/v2/Turn"
},
@@ -23503,45 +23479,6 @@
"title": "ThreadRevertedNotification",
"type": "object"
},
"ThreadRollbackParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "DEPRECATED: `thread/rollback` will be removed soon.",
"properties": {
"numTurns": {
"description": "The number of turns to drop from the end of the thread. Must be >= 1.\n\nThis only modifies the thread's history and does not revert local file changes that have been made by the agent. Clients are responsible for reverting these changes.",
"format": "uint32",
"minimum": 0.0,
"type": "integer"
},
"threadId": {
"type": "string"
}
},
"required": [
"numTurns",
"threadId"
],
"title": "ThreadRollbackParams",
"type": "object"
},
"ThreadRollbackResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"thread": {
"allOf": [
{
"$ref": "#/definitions/v2/Thread"
}
],
"description": "The updated thread after applying the rollback, with `turns` populated.\n\nThe ThreadItems stored in each Turn are lossy since we explicitly do not persist all agent interactions, such as command executions. This is the same behavior as `thread/resume`."
}
},
"required": [
"thread"
],
"title": "ThreadRollbackResponse",
"type": "object"
},
"ThreadSearchResult": {
"properties": {
"snippet": {
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+1 -64
View File
@@ -2214,30 +2214,6 @@
"title": "Thread/approveGuardianDeniedActionRequest",
"type": "object"
},
{
"properties": {
"id": {
"$ref": "#/definitions/RequestId"
},
"method": {
"enum": [
"thread/rollback"
],
"title": "Thread/rollbackRequestMethod",
"type": "string"
},
"params": {
"$ref": "#/definitions/ThreadRollbackParams"
}
},
"required": [
"id",
"method",
"params"
],
"title": "Thread/rollbackRequest",
"type": "object"
},
{
"properties": {
"id": {
@@ -18316,7 +18292,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
@@ -21217,45 +21193,6 @@
"title": "ThreadRevertedNotification",
"type": "object"
},
"ThreadRollbackParams": {
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "DEPRECATED: `thread/rollback` will be removed soon.",
"properties": {
"numTurns": {
"description": "The number of turns to drop from the end of the thread. Must be >= 1.\n\nThis only modifies the thread's history and does not revert local file changes that have been made by the agent. Clients are responsible for reverting these changes.",
"format": "uint32",
"minimum": 0.0,
"type": "integer"
},
"threadId": {
"type": "string"
}
},
"required": [
"numTurns",
"threadId"
],
"title": "ThreadRollbackParams",
"type": "object"
},
"ThreadRollbackResponse": {
"$schema": "http://json-schema.org/draft-07/schema#",
"properties": {
"thread": {
"allOf": [
{
"$ref": "#/definitions/Thread"
}
],
"description": "The updated thread after applying the rollback, with `turns` populated.\n\nThe ThreadItems stored in each Turn are lossy since we explicitly do not persist all agent interactions, such as command executions. This is the same behavior as `thread/resume`."
}
},
"required": [
"thread"
],
"title": "ThreadRollbackResponse",
"type": "object"
},
"ThreadSearchResult": {
"properties": {
"snippet": {
codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json
Generated
+1 -1
View File
@@ -1456,7 +1456,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json
Generated
+1 -1
View File
@@ -1247,7 +1247,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json
Generated
+1 -1
View File
@@ -1247,7 +1247,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json
Generated
+1 -1
View File
@@ -1247,7 +1247,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json
Generated
+1 -1
View File
@@ -1456,7 +1456,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadRevertResponse.json
Generated
+1 -1
View File
@@ -1247,7 +1247,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackParams.json
Generated
-21
View File
@@ -1,21 +0,0 @@
{
"$schema": "http://json-schema.org/draft-07/schema#",
"description": "DEPRECATED: `thread/rollback` will be removed soon.",
"properties": {
"numTurns": {
"description": "The number of turns to drop from the end of the thread. Must be >= 1.\n\nThis only modifies the thread's history and does not revert local file changes that have been made by the agent. Clients are responsible for reverting these changes.",
"format": "uint32",
"minimum": 0.0,
"type": "integer"
},
"threadId": {
"type": "string"
}
},
"required": [
"numTurns",
"threadId"
],
"title": "ThreadRollbackParams",
"type": "object"
}
codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json
Generated
-2736
View File
File diff suppressed because it is too large Load Diff
codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json
Generated
+1 -1
View File
@@ -1456,7 +1456,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json
Generated
+1 -1
View File
@@ -1247,7 +1247,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json
Generated
+1 -1
View File
@@ -1247,7 +1247,7 @@
"description": "Optional analytics source classification for this thread."
},
"turns": {
"description": "Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"description": "Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list.",
"items": {
"$ref": "#/definitions/Turn"
},
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/ClientRequest.ts
Generated
+1 -2
View File
File diff suppressed because one or more lines are too long
codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts
Generated
+1 -1
View File
@@ -96,7 +96,7 @@ gitInfo: GitInfo | null, /**
* Optional user-facing thread title.
*/
name: string | null, /**
* Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read`
* Only populated on `thread/resume`, `thread/fork`, and `thread/read`
* (when `includeTurns` is true) responses.
* For all other responses and notifications returning a Thread,
* the turns field will be an empty list.
codex-rs/app-server-protocol/schema/typescript/v2/ThreadRollbackParams.ts
Generated
-15
View File
@@ -1,15 +0,0 @@
// 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.
/**
* DEPRECATED: `thread/rollback` will be removed soon.
*/
export type ThreadRollbackParams = { threadId: string,
/**
* The number of turns to drop from the end of the thread. Must be >= 1.
*
* This only modifies the thread's history and does not revert local file changes
* that have been made by the agent. Clients are responsible for reverting these changes.
*/
numTurns: number, };
codex-rs/app-server-protocol/schema/typescript/v2/ThreadRollbackResponse.ts
Generated
-14
View File
@@ -1,14 +0,0 @@
// 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 { Thread } from "./Thread";
export type ThreadRollbackResponse = {
/**
* The updated thread after applying the rollback, with `turns` populated.
*
* The ThreadItems stored in each Turn are lossy since we explicitly do not
* persist all agent interactions, such as command executions. This is the same
* behavior as `thread/resume`.
*/
thread: Thread, };
codex-rs/app-server-protocol/schema/typescript/v2/index.ts
Generated
-2
View File
@@ -543,8 +543,6 @@ export type { ThreadResumeResponse } from "./ThreadResumeResponse";
export type { ThreadRevertParams } from "./ThreadRevertParams";
export type { ThreadRevertResponse } from "./ThreadRevertResponse";
export type { ThreadRevertedNotification } from "./ThreadRevertedNotification";
export type { ThreadRollbackParams } from "./ThreadRollbackParams";
export type { ThreadRollbackResponse } from "./ThreadRollbackResponse";
export type { ThreadSearchResult } from "./ThreadSearchResult";
export type { ThreadSearchSortKey } from "./ThreadSearchSortKey";
export type { ThreadSection } from "./ThreadSection";
codex-rs/app-server-protocol/src/protocol/common.rs
-5
View File
@@ -753,11 +753,6 @@ client_request_definitions! {
serialization: thread_id(params.thread_id),
response: v2::ThreadBackgroundTerminalsTerminateResponse,
},
ThreadRollback => "thread/rollback" {
params: v2::ThreadRollbackParams,
serialization: thread_id(params.thread_id),
response: v2::ThreadRollbackResponse,
},
ThreadRevert => "thread/revert" {
params: v2::ThreadRevertParams,
serialization: thread_id(params.thread_id),
codex-rs/app-server-protocol/src/protocol/v2/thread.rs
-25
View File
@@ -1241,31 +1241,6 @@ pub struct ThreadBackgroundTerminalsTerminateResponse {
pub terminated: bool,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
/// DEPRECATED: `thread/rollback` will be removed soon.
pub struct ThreadRollbackParams {
pub thread_id: String,
/// The number of turns to drop from the end of the thread. Must be >= 1.
///
/// This only modifies the thread's history and does not revert local file changes
/// that have been made by the agent. Clients are responsible for reverting these changes.
pub num_turns: u32,
}
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export_to = "v2/")]
pub struct ThreadRollbackResponse {
/// The updated thread after applying the rollback, with `turns` populated.
///
/// The ThreadItems stored in each Turn are lossy since we explicitly do not
/// persist all agent interactions, such as command executions. This is the same
/// behavior as `thread/resume`.
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.
codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs
+1 -1
View File
@@ -286,7 +286,7 @@ pub struct Thread {
/// Saved Daybreak choice, independent of turn execution. Null if unset.
#[experimental("thread.daybreakEnabled")]
pub daybreak_enabled: Option<bool>,
/// Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read`
/// Only populated on `thread/resume`, `thread/fork`, and `thread/read`
/// (when `includeTurns` is true) responses.
/// For all other responses and notifications returning a Thread,
/// the turns field will be an empty list.
codex-rs/app-server/README.md
+11
View File
@@ -204,3 +204,14 @@ in both `full` and `toolsAndAuthOnly` detail modes, including thread-scoped read
This is the server's advertised MCP capabilities object, including its `extensions`
map. It is null when the connection has not initialized successfully; capabilities
are never inferred from tools or copied from a shared catalog cache.
# Thread rollback
`thread/rollback` has been removed from the API, including its request and response
types. Requests use the generic unknown-method rejection path. Use `thread/revert`
for paginated threads instead.
Existing rollouts may contain historical `ThreadRolledBack` events. Their replay
and migration remain supported so resuming, reading, and forking those threads
preserves the surviving history. This disk compatibility does not require restoring
support for new `thread/rollback` requests.
codex-rs/app-server/src/bespoke_event_handling.rs
+2 -232
View File
@@ -1,11 +1,6 @@
use crate::error_code::internal_error;
use crate::error_code::invalid_request;
use crate::notification_media::without_notification_media;
use crate::outgoing_message::ClientRequestResult;
use crate::outgoing_message::ThreadScopedOutgoingMessageSender;
use crate::request_processors::apply_live_thread_settings;
use crate::request_processors::populate_thread_turns_from_history;
use crate::request_processors::thread_from_stored_thread;
use crate::request_processors::thread_settings_from_config_snapshot;
use crate::server_request_error::is_turn_transition_server_request_error;
use crate::thread_state::ThreadState;
@@ -70,9 +65,7 @@ use codex_app_server_protocol::ThreadRealtimeSdpNotification;
use codex_app_server_protocol::ThreadRealtimeStartedNotification;
use codex_app_server_protocol::ThreadRealtimeTranscriptDeltaNotification;
use codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification;
use codex_app_server_protocol::ThreadRollbackResponse;
use codex_app_server_protocol::ThreadSettingsUpdatedNotification;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadTokenUsage;
use codex_app_server_protocol::ThreadTokenUsageUpdatedNotification;
use codex_app_server_protocol::ToolRequestUserInputOption;
@@ -102,7 +95,6 @@ use codex_protocol::items::CollabAgentTool as CoreCollabAgentTool;
use codex_protocol::items::TurnItem as CoreTurnItem;
use codex_protocol::models::AdditionalPermissionProfile as CoreAdditionalPermissionProfile;
use codex_protocol::plan_tool::UpdatePlanArgs;
use codex_protocol::protocol::CodexErrorInfo as CoreCodexErrorInfo;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecApprovalRequestEvent;
@@ -120,7 +112,6 @@ use codex_protocol::request_permissions::RequestPermissionsResponse as CoreReque
use codex_protocol::request_user_input::RequestUserInputAnswer as CoreRequestUserInputAnswer;
use codex_protocol::request_user_input::RequestUserInputResponse as CoreRequestUserInputResponse;
use codex_shell_command::parse_command::shlex_join;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path_uri::LegacyAppPathString;
use codex_utils_path_uri::PathUri;
use std::collections::HashMap;
@@ -145,7 +136,6 @@ struct CommandExecutionCompletionItem {
command_actions: Vec<V2ParsedCommand>,
}
#[allow(clippy::too_many_arguments)]
pub(crate) async fn apply_bespoke_event_handling(
event: Event,
conversation_id: ThreadId,
@@ -154,8 +144,6 @@ pub(crate) async fn apply_bespoke_event_handling(
outgoing: ThreadScopedOutgoingMessageSender,
thread_state: Arc<tokio::sync::Mutex<ThreadState>>,
thread_watch_manager: ThreadWatchManager,
thread_list_state_permit: Arc<tokio::sync::Semaphore>,
fallback_model_provider: String,
) {
let Event {
id: event_turn_id,
@@ -1036,25 +1024,6 @@ pub(crate) async fn apply_bespoke_event_handling(
.note_system_error(&conversation_id.to_string())
.await;
let message = ev.message.clone();
let codex_error_info = ev.codex_error_info.clone();
// If this error belongs to an in-flight `thread/rollback` request, fail that request
// (and clear pending state) so subsequent rollbacks are unblocked.
//
// Don't send a notification for this error.
if matches!(
codex_error_info,
Some(CoreCodexErrorInfo::ThreadRollbackFailed)
) {
return handle_thread_rollback_failed(
conversation_id,
message,
&thread_state,
&outgoing,
)
.await;
};
if !ev.affects_turn_status() {
return;
}
@@ -1239,70 +1208,6 @@ pub(crate) async fn apply_bespoke_event_handling(
)
.await;
}
EventMsg::ThreadRolledBack(_rollback_event) => {
let pending = {
let mut state = thread_state.lock().await;
state.pending_rollbacks.take()
};
if let Some((request_id, _completion_tx)) = pending {
let _thread_list_state_permit = match thread_list_state_permit.acquire().await {
Ok(permit) => permit,
Err(err) => {
outgoing
.send_error(
request_id,
internal_error(format!(
"failed to acquire thread list state permit: {err}"
)),
)
.await;
return;
}
};
let config_snapshot = conversation.config_snapshot().await;
let stored_thread = match conversation
.read_thread(
/*include_archived*/ true, /*include_history*/ true,
)
.await
{
Ok(stored_thread) => stored_thread,
Err(err) => {
outgoing
.send_error(
request_id.clone(),
internal_error(format!(
"failed to read thread {conversation_id} after rollback: {err}"
)),
)
.await;
return;
}
};
let loaded_status = thread_watch_manager
.loaded_status_for_thread(&conversation_id.to_string())
.await;
let mut response = match thread_rollback_response_from_stored_thread(
stored_thread,
conversation.session_configured().session_id.to_string(),
fallback_model_provider.as_str(),
config_snapshot.cwd(),
loaded_status,
) {
Ok(response) => response,
Err(err) => {
outgoing
.send_error(request_id.clone(), internal_error(err))
.await;
return;
}
};
apply_live_thread_settings(&mut response.thread, &config_snapshot);
outgoing.send_response(request_id, response).await;
}
}
EventMsg::ThreadGoalUpdated(thread_goal_event) => {
let notification = ThreadGoalUpdatedNotification {
thread_id: thread_goal_event.thread_id.to_string(),
@@ -1315,7 +1220,7 @@ pub(crate) async fn apply_bespoke_event_handling(
))
.await;
}
EventMsg::ThreadQueueChanged(_) => {}
EventMsg::ThreadRolledBack(_) | EventMsg::ThreadQueueChanged(_) => {}
EventMsg::ThreadSettingsApplied(_) => {
let thread_settings =
thread_settings_from_config_snapshot(&conversation.config_snapshot().await);
@@ -1635,42 +1540,6 @@ async fn handle_turn_interrupted(
.await;
}
async fn handle_thread_rollback_failed(
_conversation_id: ThreadId,
message: String,
thread_state: &Arc<Mutex<ThreadState>>,
outgoing: &ThreadScopedOutgoingMessageSender,
) {
let pending_rollback = thread_state.lock().await.pending_rollbacks.take();
if let Some((request_id, _completion_tx)) = pending_rollback {
outgoing
.send_error(request_id, invalid_request(message))
.await;
}
}
fn thread_rollback_response_from_stored_thread(
stored_thread: codex_thread_store::StoredThread,
session_id: String,
fallback_model_provider: &str,
fallback_cwd: &AbsolutePathBuf,
loaded_status: ThreadStatus,
) -> std::result::Result<ThreadRollbackResponse, String> {
let thread_id = stored_thread.thread_id;
let (mut thread, history) =
thread_from_stored_thread(stored_thread, fallback_model_provider, fallback_cwd);
thread.session_id = session_id;
let Some(history) = history else {
return Err(format!(
"thread {thread_id} did not include persisted history after rollback"
));
};
populate_thread_turns_from_history(&mut thread, &history.items, /*active_turn*/ None);
thread.status = loaded_status;
Ok(ThreadRollbackResponse { thread })
}
async fn respond_to_pending_interrupts(
thread_state: &Arc<Mutex<ThreadState>>,
outgoing: &ThreadScopedOutgoingMessageSender,
@@ -2219,11 +2088,11 @@ mod tests {
use anyhow::Result;
use anyhow::anyhow;
use anyhow::bail;
use chrono::Utc;
use codex_app_server_protocol::AutoReviewDecisionSource;
use codex_app_server_protocol::GuardianApprovalReviewStatus;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::ServerRequest;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::TurnPlanStepStatus;
use codex_login::CodexAuth;
use codex_protocol::AgentPath;
@@ -2235,11 +2104,8 @@ mod tests {
use codex_protocol::items::TurnItem as CoreTurnItem;
use codex_protocol::models::FileSystemPermissions as CoreFileSystemPermissions;
use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions;
use codex_protocol::models::PermissionProfile;
use codex_protocol::plan_tool::PlanItemArg;
use codex_protocol::plan_tool::StepStatus;
use codex_protocol::protocol::AgentMessageEvent;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::AuthRecoveryEvent;
use codex_protocol::protocol::CreditsSnapshot;
use codex_protocol::protocol::EventMsg;
@@ -2249,13 +2115,8 @@ mod tests {
use codex_protocol::protocol::ItemStartedEvent;
use codex_protocol::protocol::RateLimitSnapshot;
use codex_protocol::protocol::RateLimitWindow;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use codex_protocol::protocol::UserMessageEvent;
use codex_rollout::RolloutItem;
use codex_thread_store::StoredThread;
use codex_thread_store::StoredThreadHistory;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_absolute_path::test_support::PathBufExt;
use codex_utils_absolute_path::test_support::test_path_buf;
@@ -2296,87 +2157,6 @@ mod tests {
Ok(envelope.notification)
}
#[test]
fn rollback_response_rebuilds_pathless_thread_from_stored_history() -> Result<()> {
let thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000789")?;
let created_at = Utc::now();
let history_items = vec![
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "before rollback".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
})),
RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
message: "after rollback".to_string(),
phase: None,
memory_citation: None,
delivery: None,
questions: None,
})),
];
let stored_thread = StoredThread {
originator: None,
thread_id,
extra_config: None,
rollout_path: None,
forked_from_id: None,
parent_thread_id: None,
preview: "fallback preview".to_string(),
name: Some("Rollback thread".to_string()),
model_provider: "openai".to_string(),
model: None,
reasoning_effort: None,
created_at,
updated_at: created_at,
recency_at: created_at,
archived_at: None,
section: None,
section_position: None,
section_entered_at: None,
project_id: None,
daybreak_enabled: None,
cwd: test_path_buf("/tmp").abs().into(),
cli_version: "0.0.0".to_string(),
source: SessionSource::Cli,
history_mode: Default::default(),
thread_source: None,
agent_nickname: None,
agent_role: None,
agent_path: None,
git_info: None,
approval_mode: AskForApproval::OnRequest,
permission_profile: PermissionProfile::read_only(),
token_usage: None,
first_user_message: Some("before rollback".to_string()),
history: Some(StoredThreadHistory {
thread_id,
items: history_items,
}),
};
let fallback_cwd = test_path_buf("/tmp").abs();
let response = thread_rollback_response_from_stored_thread(
stored_thread,
thread_id.to_string(),
"fallback-provider",
&fallback_cwd,
ThreadStatus::NotLoaded,
)
.expect("rollback response should rebuild from stored history");
assert_eq!(response.thread.id, thread_id.to_string());
assert_eq!(response.thread.path, None);
assert_eq!(response.thread.preview, "fallback preview");
assert_eq!(response.thread.name.as_deref(), Some("Rollback thread"));
assert_eq!(response.thread.status, ThreadStatus::NotLoaded);
assert_eq!(response.thread.turns.len(), 1);
assert_eq!(response.thread.turns[0].items.len(), 2);
Ok(())
}
fn turn_complete_event(turn_id: &str) -> TurnCompleteEvent {
TurnCompleteEvent {
turn_id: turn_id.to_string(),
@@ -2485,8 +2265,6 @@ mod tests {
self.outgoing.clone(),
self.thread_state.clone(),
self.thread_watch_manager.clone(),
Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)),
"test-provider".to_string(),
)
.await;
}
@@ -3330,8 +3108,6 @@ mod tests {
outgoing.clone(),
Arc::clone(&thread_state),
thread_watch_manager.clone(),
Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)),
"test-provider".to_string(),
)
.await;
@@ -3372,8 +3148,6 @@ mod tests {
outgoing.clone(),
Arc::clone(&thread_state),
thread_watch_manager.clone(),
Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)),
"test-provider".to_string(),
)
.await;
@@ -3453,8 +3227,6 @@ mod tests {
outgoing,
new_thread_state(),
thread_watch_manager.clone(),
Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)),
"test-provider".to_string(),
)
.await;
@@ -3542,8 +3314,6 @@ mod tests {
outgoing,
new_thread_state(),
ThreadWatchManager::new(),
Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)),
"test-provider".to_string(),
)
.await;
codex-rs/app-server/src/message_processor.rs
-7
View File
@@ -528,7 +528,6 @@ impl MessageProcessor {
pending_thread_unloads,
thread_state_manager,
thread_watch_manager,
thread_list_state_permit,
Arc::clone(&skills_watcher),
turn_cost_worker.as_ref().map(TurnCostWorker::handle),
);
@@ -983,7 +982,6 @@ impl MessageProcessor {
ClientRequest::ThreadStart { .. }
| ClientRequest::ThreadFork { .. }
| ClientRequest::ThreadResume { .. }
| ClientRequest::ThreadRollback { .. }
| ClientRequest::ThreadRevert { .. }
| ClientRequest::ThreadSettingsUpdate { .. }
| ClientRequest::TurnSettingsUpdate { .. }
@@ -1449,11 +1447,6 @@ impl MessageProcessor {
.thread_background_terminals_terminate(params)
.await
}
ClientRequest::ThreadRollback { params, .. } => {
self.thread_processor
.thread_rollback(&request_id, params, app_server_client_name.as_deref())
.await
}
ClientRequest::ThreadRevert { params, .. } => {
self.thread_processor
.thread_revert(
codex-rs/app-server/src/outgoing_message.rs
-8
View File
@@ -231,14 +231,6 @@ impl ThreadScopedOutgoingMessageSender {
{
self.outgoing.send_response(request_id, response).await;
}
pub(crate) async fn send_error(
&self,
request_id: ConnectionRequestId,
error: impl Into<JSONRPCErrorError>,
) {
self.outgoing.send_error(request_id, error).await;
}
}
impl OutgoingMessageSender {
codex-rs/app-server/src/request_processors.rs
-1
View File
@@ -262,7 +262,6 @@ use codex_app_server_protocol::ThreadRealtimeStopResponse;
use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadSearchOccurrence;
use codex_app_server_protocol::ThreadSearchOccurrencesParams;
use codex_app_server_protocol::ThreadSearchOccurrencesResponse;
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
-6
View File
@@ -11,8 +11,6 @@ pub(super) struct ListenerTaskContext {
pub(super) outgoing: Arc<OutgoingMessageSender>,
pub(super) pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
pub(super) thread_watch_manager: ThreadWatchManager,
pub(super) thread_list_state_permit: Arc<Semaphore>,
pub(super) fallback_model_provider: String,
pub(super) codex_home: PathBuf,
pub(super) thread_unload_delay: Duration,
pub(super) skills_watcher: Arc<SkillsWatcher>,
@@ -271,8 +269,6 @@ pub(super) async fn ensure_listener_task_running(
thread_state_manager,
pending_thread_unloads,
thread_watch_manager,
thread_list_state_permit,
fallback_model_provider,
codex_home,
turn_cost_worker,
..
@@ -353,8 +349,6 @@ pub(super) async fn ensure_listener_task_running(
thread_outgoing,
thread_state.clone(),
thread_watch_manager.clone(),
thread_list_state_permit.clone(),
fallback_model_provider.clone(),
)
.await;
if matches!(event.msg, EventMsg::ShutdownComplete)
codex-rs/app-server/src/request_processors/thread_processor.rs
-98
View File
@@ -30,9 +30,6 @@ use std::ops::ControlFlow;
pub(super) const THREAD_LIST_DEFAULT_LIMIT: usize = 25;
pub(super) const THREAD_LIST_MAX_LIMIT: usize = 100;
const CODEX_TUI_CLIENT_NAME: &str = "codex-tui";
const THREAD_ROLLBACK_DEPRECATION_SUMMARY: &str =
"thread/rollback is deprecated and will be removed soon";
const PAGINATED_FULL_HISTORY_DEPRECATION_SUMMARY: &str = "Full-history hydration is deprecated for paginated threads; use `excludeTurns: true`, then page with `thread/turns/list` and `thread/items/list`.";
const PAGINATED_THREAD_READ_DEPRECATION_SUMMARY: &str = "Full-history hydration is deprecated for paginated threads; omit `includeTurns` or set it to `false`, then page with `thread/turns/list` and `thread/items/list`.";
@@ -808,24 +805,6 @@ impl ThreadRequestProcessor {
.map(|response| Some(response.into()))
}
pub(crate) async fn thread_rollback(
&self,
request_id: &ConnectionRequestId,
params: ThreadRollbackParams,
app_server_client_name: Option<&str>,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
if app_server_client_name != Some(CODEX_TUI_CLIENT_NAME) {
self.send_deprecation_notice(
request_id.connection_id,
THREAD_ROLLBACK_DEPRECATION_SUMMARY,
)
.await;
}
self.thread_rollback_inner(request_id, params)
.await
.map(|()| None)
}
async fn send_deprecation_notice(&self, connection_id: ConnectionId, summary: &str) {
self.outgoing
.send_server_notification_to_connections(
@@ -1102,8 +1081,6 @@ impl ThreadRequestProcessor {
outgoing: Arc::clone(&self.outgoing),
pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
thread_watch_manager: self.thread_watch_manager.clone(),
thread_list_state_permit: self.thread_list_state_permit.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.to_path_buf(),
thread_unload_delay: self.config.thread_unload_delay,
skills_watcher: Arc::clone(&self.skills_watcher),
@@ -1234,8 +1211,6 @@ impl ThreadRequestProcessor {
outgoing: Arc::clone(&self.outgoing),
pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
thread_watch_manager: self.thread_watch_manager.clone(),
thread_list_state_permit: self.thread_list_state_permit.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.to_path_buf(),
thread_unload_delay: self.config.thread_unload_delay,
skills_watcher: Arc::clone(&self.skills_watcher),
@@ -2118,14 +2093,6 @@ impl ThreadRequestProcessor {
Ok((ThreadUnarchiveResponse { thread }, thread_id))
}
async fn thread_rollback_inner(
&self,
request_id: &ConnectionRequestId,
params: ThreadRollbackParams,
) -> Result<(), JSONRPCErrorError> {
self.thread_rollback_start(request_id, params).await
}
async fn thread_revert_response(
&self,
request_id: &ConnectionRequestId,
@@ -2353,71 +2320,6 @@ impl ThreadRequestProcessor {
})
}
async fn thread_rollback_start(
&self,
request_id: &ConnectionRequestId,
params: ThreadRollbackParams,
) -> Result<(), JSONRPCErrorError> {
let ThreadRollbackParams {
thread_id,
num_turns,
} = params;
if num_turns == 0 {
return Err(invalid_request("numTurns must be >= 1"));
}
let (thread_id, thread) = self.load_thread(&thread_id).await?;
ensure_direct_input_allowed(thread.as_ref()).await?;
if matches!(
thread.config_snapshot().await.history_mode,
ThreadHistoryMode::Paginated
) {
return Err(invalid_request(
"paginated threads do not support thread/rollback",
));
}
let request = request_id.clone();
let (completion_tx, completion_rx) = tokio::sync::oneshot::channel();
let rollback_already_in_progress = {
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
let mut thread_state = thread_state.lock().await;
if thread_state.pending_rollbacks.is_some() {
true
} else {
thread_state.pending_rollbacks = Some((request, completion_tx));
false
}
};
if rollback_already_in_progress {
return Err(invalid_request(
"rollback already in progress for this thread",
));
}
if let Err(err) = self
.submit_core_op(
request_id,
thread.as_ref(),
Op::ThreadRollback { num_turns },
)
.await
{
// No ThreadRollback event will arrive if an error occurs.
// Clean up and reply immediately.
let thread_state = self.thread_state_manager.thread_state(thread_id).await;
thread_state.lock().await.pending_rollbacks = None;
return Err(internal_error(format!("failed to start rollback: {err}")));
}
// The listener drops the sender after queuing the response, including errors.
// Keep the RPC's drain admission alive until then, without holding thread state.
let _ = completion_rx.await;
Ok(())
}
async fn thread_compact_start_inner(
&self,
request_id: &ConnectionRequestId,
codex-rs/app-server/src/request_processors/turn_processor.rs
-5
View File
@@ -85,7 +85,6 @@ pub(crate) struct TurnRequestProcessor {
pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
thread_state_manager: ThreadStateManager,
thread_watch_manager: ThreadWatchManager,
thread_list_state_permit: Arc<Semaphore>,
skills_watcher: Arc<SkillsWatcher>,
turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
}
@@ -149,7 +148,6 @@ impl TurnRequestProcessor {
pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>,
thread_state_manager: ThreadStateManager,
thread_watch_manager: ThreadWatchManager,
thread_list_state_permit: Arc<Semaphore>,
skills_watcher: Arc<SkillsWatcher>,
turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
) -> Self {
@@ -166,7 +164,6 @@ impl TurnRequestProcessor {
pending_thread_unloads,
thread_state_manager,
thread_watch_manager,
thread_list_state_permit,
skills_watcher,
turn_cost_worker,
}
@@ -1657,8 +1654,6 @@ impl TurnRequestProcessor {
outgoing: Arc::clone(&self.outgoing),
pending_thread_unloads: Arc::clone(&self.pending_thread_unloads),
thread_watch_manager: self.thread_watch_manager.clone(),
thread_list_state_permit: self.thread_list_state_permit.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.to_path_buf(),
thread_unload_delay: self.config.thread_unload_delay,
skills_watcher: Arc::clone(&self.skills_watcher),
codex-rs/app-server/src/thread_state.rs
-24
View File
@@ -98,7 +98,6 @@ pub(crate) struct TurnSummary {
#[derive(Default)]
pub(crate) struct ThreadState {
pub(crate) pending_interrupts: PendingInterruptQueue,
pub(crate) pending_rollbacks: Option<(ConnectionRequestId, oneshot::Sender<()>)>,
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
@@ -146,7 +145,6 @@ impl ThreadState {
let _ = cancel_tx.send(());
}
self.shutdown_drain_waiter = None;
self.pending_rollbacks = None;
self.listener_command_tx = None;
self.current_turn_history.reset();
self.listener_thread = None;
@@ -251,28 +249,6 @@ mod tests {
use codex_utils_absolute_path::AbsolutePathBuf;
use pretty_assertions::assert_eq;
#[test]
fn clear_listener_releases_pending_rollback() {
let (completion_tx, mut completion_rx) = oneshot::channel();
let mut state = ThreadState {
pending_rollbacks: Some((
ConnectionRequestId {
connection_id: ConnectionId(1),
request_id: RequestId::Integer(1),
},
completion_tx,
)),
..Default::default()
};
state.clear_listener();
assert_eq!(
completion_rx.try_recv(),
Err(oneshot::error::TryRecvError::Closed)
);
}
#[test]
fn note_thread_settings_reports_only_effective_changes() {
let mut state = ThreadState::default();
codex-rs/app-server/tests/common/test_app_server.rs
-10
View File
@@ -104,7 +104,6 @@ use codex_app_server_protocol::ThreadRealtimeListVoicesParams;
use codex_app_server_protocol::ThreadRealtimeStartParams;
use codex_app_server_protocol::ThreadRealtimeStopParams;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadSearchOccurrencesParams;
use codex_app_server_protocol::ThreadSearchParams;
use codex_app_server_protocol::ThreadSectionMoveParams;
@@ -659,15 +658,6 @@ impl TestAppServer {
self.send_request("thread/shellCommand", params).await
}
/// Send a `thread/rollback` JSON-RPC request.
pub async fn send_thread_rollback_request(
&mut self,
params: ThreadRollbackParams,
) -> anyhow::Result<i64> {
let params = Some(serde_json::to_value(params)?);
self.send_request("thread/rollback", params).await
}
/// Send a `thread/list` JSON-RPC request.
pub async fn send_thread_list_request(
&mut self,
codex-rs/app-server/tests/suite/v2/connection_handling_websocket_unix.rs
-4
View File
@@ -97,10 +97,6 @@ async fn websocket_transport_ctrl_c_waits_for_running_turn_before_exit() -> Resu
"turn/settings/update",
json!({"threadId": thread_id, "turnId": "unused", "model": "other"}),
),
(
"thread/rollback",
json!({"threadId": thread_id, "numTurns": 1}),
),
(
"thread/revert",
json!({"threadId": thread_id, "beforeTurnId": "unused"}),
codex-rs/app-server/tests/suite/v2/git_attribution.rs
+3 -13
View File
@@ -17,8 +17,6 @@ use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadRollbackResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnStartParams;
@@ -79,7 +77,7 @@ async fn git_attribution_follows_authenticated_workspace_policy() -> Result<()>
"Recovered",
"Cached",
"After switch",
"After rollback",
"After switching back",
]
.into_iter()
.map(create_final_assistant_message_sse_response)
@@ -166,14 +164,6 @@ async fn git_attribution_follows_authenticated_workspace_policy() -> Result<()>
let _: LoginAccountResponse = read_response(&mut app_server, request_id).await?;
run_turn(&mut app_server, &thread.id, "Turn after workspace switch").await?;
let request_id = app_server
.send_thread_rollback_request(ThreadRollbackParams {
thread_id: thread.id.clone(),
num_turns: 1,
})
.await?;
let _: ThreadRollbackResponse = read_response(&mut app_server, request_id).await?;
let request_id = app_server
.send_chatgpt_auth_tokens_login_request(
"e30.e30.c2ln".to_string(),
@@ -182,13 +172,13 @@ async fn git_attribution_follows_authenticated_workspace_policy() -> Result<()>
)
.await?;
let _: LoginAccountResponse = read_response(&mut app_server, request_id).await?;
run_turn(&mut app_server, &thread.id, "Turn after rollback").await?;
run_turn(&mut app_server, &thread.id, "Turn after switching back").await?;
let requests = response_mock.requests();
assert_eq!(requests.len(), 5);
for (request, expected) in requests
.into_iter()
.zip([(0, 0), (1, 0), (1, 0), (1, 1), (1, 0)])
.zip([(0, 0), (1, 0), (1, 0), (1, 1), (2, 1)])
{
let developer_text = request.message_input_texts("developer").join("\n");
assert_eq!(
codex-rs/app-server/tests/suite/v2/guardian_v2.rs
+1 -22
View File
@@ -36,12 +36,9 @@ use codex_app_server_protocol::ThreadCompactStartParams;
use codex_app_server_protocol::ThreadCompactStartResponse;
use codex_app_server_protocol::ThreadForkParams;
use codex_app_server_protocol::ThreadForkResponse;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadRollbackResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnSettingsUpdateParams;
@@ -247,7 +244,6 @@ enum ThreadLifecycle {
UserInputHookBlocked,
Resume,
Fork,
RootRollback,
RootRestriction,
RootRestrictionDuringClassification,
RootTrustedSkill,
@@ -261,8 +257,7 @@ impl ThreadLifecycle {
fn uses_root_worker(self) -> bool {
matches!(
self,
Self::RootRollback
| Self::RootRestriction
Self::RootRestriction
| Self::RootRestrictionDuringClassification
| Self::RootTrustedSkill
| Self::RootUserInputRestriction
@@ -914,7 +909,6 @@ async fn guardian_v2_routes_scoped_tool_approvals(
| ThreadLifecycle::UserInputEmpty
| ThreadLifecycle::UserInputHookFeedback
| ThreadLifecycle::UserInputHookBlocked
| ThreadLifecycle::RootRollback
| ThreadLifecycle::RootRestriction
| ThreadLifecycle::RootRestrictionDuringClassification
| ThreadLifecycle::RootTrustedSkill
@@ -963,7 +957,6 @@ async fn guardian_v2_routes_scoped_tool_approvals(
| ThreadLifecycle::UserInputEmpty
| ThreadLifecycle::UserInputHookFeedback
| ThreadLifecycle::UserInputHookBlocked
| ThreadLifecycle::RootRollback
| ThreadLifecycle::RootRestriction
| ThreadLifecycle::RootRestrictionDuringClassification
| ThreadLifecycle::RootTrustedSkill
@@ -975,8 +968,6 @@ async fn guardian_v2_routes_scoped_tool_approvals(
.start_thread(ThreadStartParams {
approval_policy: Some(AskForApproval::OnRequest),
approvals_reviewer: Some(requested_reviewer),
history_mode: matches!(lifecycle, ThreadLifecycle::RootRollback)
.then_some(ThreadHistoryMode::Legacy),
..Default::default()
})
.await?;
@@ -1560,17 +1551,6 @@ async fn guardian_v2_routes_scoped_tool_approvals(
&& (lifecycle.uses_root_worker()
|| matches!(lifecycle, ThreadLifecycle::RootUserRestriction))
{
if matches!(lifecycle, ThreadLifecycle::RootRollback) {
let rollback_id = app_server
.send_thread_rollback_request(ThreadRollbackParams {
thread_id: thread_id.clone(),
num_turns: 1,
})
.await?;
let _: ThreadRollbackResponse =
timeout(TIMEOUT, app_server.read_response(rollback_id)).await??;
}
if lifecycle.has_root_user_input() {
submit_user_input_response(
&mut app_server,
@@ -2548,7 +2528,6 @@ async fn forked_thread_ignores_persisted_guardian_score() -> Result<()> {
.await
}
#[test_case(ThreadLifecycle::RootRollback; "worker_root_rollback")]
#[test_case(ThreadLifecycle::RootRestriction; "worker_root_restriction")]
#[test_case(ThreadLifecycle::RootUserRestriction; "root_user_restriction")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
codex-rs/app-server/tests/suite/v2/guardian_v2_history_tests.rs
+8 -48
View File
@@ -1,4 +1,4 @@
//! Exercises both reviewers' evidence delivery through real compaction and rollback.
//! Exercises both reviewers' evidence delivery through real compaction and resume.
use std::sync::Arc;
use std::sync::Mutex;
@@ -24,8 +24,6 @@ use codex_app_server_protocol::ThreadCompactStartResponse;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadRollbackResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
@@ -108,7 +106,7 @@ enum ReviewCheckpoint {
#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 140, EvidenceSize::Normal, ReviewCheckpoint::Valid; "legacy source call evicted")]
#[test_case(ContextPath::Legacy, CheckpointReuse::Enabled, Some("matching"), Some("matching"), 0, EvidenceSize::OversizedAnswer, ReviewCheckpoint::Valid; "legacy answer truncation")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollback(
async fn guardians_retain_evidence_after_compaction_and_resume(
context_path: ContextPath,
checkpoint_reuse: CheckpointReuse,
parent_hash: Option<&str>,
@@ -334,8 +332,6 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
restriction.as_str(),
"Recheck the repository.",
"Inspect after resume.",
"Inspect after partial rollback.",
"Inspect a different repository.",
]
.into_iter()
.enumerate()
@@ -368,15 +364,6 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
timeout(TIMEOUT, app_server.read_notification("turn/completed")).await??;
assert_eq!(completed.turn.status, TurnStatus::Completed);
assert_eq!(compact_requests.lock().expect("request log lock").len(), 1);
} else if index >= 3 {
let id = app_server
.send_thread_rollback_request(ThreadRollbackParams {
thread_id: thread_id.clone(),
num_turns: if index == 3 { 1 } else { 3 },
})
.await?;
let _: ThreadRollbackResponse =
timeout(TIMEOUT, app_server.read_response(id)).await??;
}
let id = app_server
@@ -492,7 +479,7 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
}
continue;
}
if requires_sync && (1..=3).contains(&index) {
if requires_sync && index > 0 {
let reviews = review_requests.lock().expect("request log lock");
assert_eq!(
reviews.len(),
@@ -513,12 +500,6 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
"current user input missing from sync review: {text}"
);
}
if index == 3 {
assert!(
!text.contains("Inspect after resume."),
"rolled-back user input remains in sync review: {text}"
);
}
assert!(text.contains(USER_INPUT_RESTRICTION));
assert!(text.contains("TRUSTED USER ANSWERS START"));
assert!(text.contains("TRANSCRIPT START"));
@@ -575,11 +556,8 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
let review = &reviews[index];
let sync_input = review["input"].as_array().expect("request input array");
let async_input = request["input"].as_array().expect("request input array");
assert_eq!(sync_input.contains(&checkpoint), (1..=3).contains(&index));
assert_eq!(
async_input.contains(&checkpoint),
(1..=3).contains(&index) && compatible
);
assert_eq!(sync_input.contains(&checkpoint), index > 0);
assert_eq!(async_input.contains(&checkpoint), index > 0 && compatible);
let sync_text = sync_input
.iter()
.filter(|item| item["role"] == "user")
@@ -591,14 +569,6 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
sync_text.contains(prompt),
"current user input missing from sync review: {sync_text}"
);
if index == 3 {
for (consumer, text) in [("sync", sync_text.as_str()), ("async", transcript)] {
assert!(
!text.contains("Inspect after resume."),
"rolled-back user input remains in {consumer} review: {text}"
);
}
}
if index == 0 {
let input = parent[2]["input"].as_array().expect("request input array");
let output = input
@@ -615,12 +585,12 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
.expect("request input array");
let parent_input = serde_json::to_string(parent_items)?;
// V2 keeps bounded user history while replacing old tool output with a checkpoint.
assert_eq!(parent_input.contains(RESTRICTION), index <= 3);
assert!(parent_input.contains(RESTRICTION));
assert!(!parent_input.contains(EVIDENCE));
assert!(!parent_items.iter().any(|item| {
item["type"] == "function_call_output" && item["call_id"] == "inspect-1"
}));
if index <= 3 {
{
assert!(sync_text.contains(RESTRICTION));
if matches!(context_path, ContextPath::ThreadOwned) {
assert!(
@@ -678,17 +648,10 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
assert!(!transcript.contains("tool request_user_input call:"));
}
}
} else {
assert!(!sync_text.contains(RESTRICTION));
assert!(!sync_text.contains(EVIDENCE));
assert!(!transcript.contains(RESTRICTION));
assert!(!transcript.contains(EVIDENCE));
assert!(!transcript.contains("Recheck the repository."));
assert!(!transcript.contains("\"echoed\":\"current inspection\""));
}
}
for (consumer, text) in [("async", &content), ("sync", &sync_text)] {
if index < 4 && (matches!(context_path, ContextPath::ThreadOwned) || index == 0) {
if matches!(context_path, ContextPath::ThreadOwned) || index == 0 {
let answers = text
.split_once(">>> TRUSTED USER ANSWERS START")
.unwrap_or_else(|| {
@@ -715,9 +678,6 @@ async fn guardians_retain_evidence_after_compaction_and_discard_it_after_rollbac
},
}
} else {
if index == 4 {
assert!(!text.contains(USER_INPUT_RESTRICTION));
}
assert!(!text.contains(">>> TRUSTED USER ANSWERS START"));
}
}
codex-rs/app-server/tests/suite/v2/mod.rs
-1
View File
@@ -126,7 +126,6 @@ mod thread_queue;
mod thread_read;
mod thread_resume;
mod thread_revert;
mod thread_rollback;
mod thread_sections;
mod thread_settings_update;
mod thread_shell_command;
codex-rs/app-server/tests/suite/v2/thread_rollback.rs
-343
View File
@@ -1,343 +0,0 @@
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_unchecked;
use app_test_support::to_response;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::DeprecationNoticeNotification;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadRollbackParams;
use codex_app_server_protocol::ThreadRollbackResponse;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_protocol::openai_models::ReasoningEffort;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
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_rollback_rejects_paginated_thread() -> 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())
.without_auto_env()
.build()
.await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let start_id = mcp
.send_thread_start_request(ThreadStartParams {
history_mode: Some(ThreadHistoryMode::Paginated),
..Default::default()
})
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response(start_resp)?;
let rollback_id = mcp
.send_thread_rollback_request(ThreadRollbackParams {
thread_id: thread.id,
num_turns: 1,
})
.await?;
let rollback_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(rollback_id)),
)
.await??;
assert_eq!(rollback_err.error.code, -32600);
assert_eq!(
rollback_err.error.message,
"paginated threads do not support thread/rollback"
);
Ok(())
}
#[tokio::test]
async fn thread_rollback_does_not_emit_deprecation_notice_to_codex_tui() -> Result<()> {
let codex_home = TempDir::new()?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build()
.await?;
let initialized = timeout(
DEFAULT_READ_TIMEOUT,
mcp.initialize_with_client_info(ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "0.1.0".to_string(),
}),
)
.await??;
let JSONRPCMessage::Response(_) = initialized else {
panic!("expected initialize response, got {initialized:?}");
};
mcp.clear_message_buffer();
let rollback_id = mcp
.send_thread_rollback_request(ThreadRollbackParams {
thread_id: "00000000-0000-0000-0000-000000000001".to_string(),
num_turns: 1,
})
.await?;
loop {
let message = timeout(DEFAULT_READ_TIMEOUT, mcp.read_next_message()).await??;
match message {
JSONRPCMessage::Notification(notification) => {
assert_ne!(notification.method, "deprecationNotice");
}
JSONRPCMessage::Error(error) if error.id == RequestId::Integer(rollback_id) => {
break;
}
message => {
panic!("expected rollback error response, got {message:?}");
}
}
}
Ok(())
}
#[tokio::test]
async fn thread_rollback_drops_last_turns_and_persists_to_rollout() -> Result<()> {
// Three Codex turns hit the mock model (session start + two turn/start calls).
let responses = vec![
create_final_assistant_message_sse_response("Done")?,
create_final_assistant_message_sse_response("Done")?,
create_final_assistant_message_sse_response("Done")?,
];
let server = create_mock_responses_server_sequence_unchecked(responses).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?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
// Start a thread.
let start_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams {
model: Some("mock-model".to_string()),
history_mode: Some(ThreadHistoryMode::Legacy),
..Default::default()
})
.await?;
let start_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(start_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(start_resp)?;
// Two turns.
let first_text = "First";
let turn1_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
client_user_message_id: None,
input: vec![V2UserInput::Text {
text: first_text.to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _turn1_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn1_id)),
)
.await??;
let _completed1 = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let turn2_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
client_user_message_id: None,
input: vec![V2UserInput::Text {
text: "Second".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _turn2_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn2_id)),
)
.await??;
let _completed2 = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
drop(mcp);
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
model: Some("gpt-5.2".to_string()),
config: Some([("model_reasoning_effort".to_string(), json!("high"))].into()),
..Default::default()
})
.await?;
let ThreadResumeResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(resume_id)).await??;
// Drain the resume snapshots before checking rollback notification ordering.
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/goal/cleared"),
)
.await??;
mcp.clear_message_buffer();
// Roll back the last turn.
let rollback_id = mcp
.send_thread_rollback_request(ThreadRollbackParams {
thread_id: thread.id.clone(),
num_turns: 1,
})
.await?;
// Pipeline resume while the listener may still be handling rollback. Waiting for
// its response must not retain the thread-list permit needed by that listener.
let pipelined_resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
exclude_turns: true,
..Default::default()
})
.await?;
let deprecation_notice = timeout(DEFAULT_READ_TIMEOUT, mcp.read_next_message()).await??;
let JSONRPCMessage::Notification(deprecation_notice) = deprecation_notice else {
panic!("thread/rollback should emit deprecationNotice before its response");
};
assert_eq!(deprecation_notice.method, "deprecationNotice");
let deprecation_notice: DeprecationNoticeNotification = serde_json::from_value(
deprecation_notice
.params
.expect("deprecationNotice params should be present"),
)?;
assert_eq!(
deprecation_notice,
DeprecationNoticeNotification {
summary: "thread/rollback is deprecated and will be removed soon".to_string(),
details: None,
}
);
let rollback_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(rollback_id)),
)
.await??;
let rollback_result = rollback_resp.result.clone();
let ThreadRollbackResponse {
thread: rolled_back_thread,
} = to_response::<ThreadRollbackResponse>(rollback_resp)?;
// Wire contract: thread title field is `name`, serialized as null when unset.
let thread_json = rollback_result
.get("thread")
.and_then(Value::as_object)
.expect("thread/rollback result.thread must be an object");
assert_eq!(rolled_back_thread.name, None);
assert_eq!(rolled_back_thread.session_id, thread.session_id);
assert_eq!(
thread_json.get("name"),
Some(&Value::Null),
"thread/rollback must serialize `name: null` when unset"
);
assert_eq!(
thread_json.get("sessionId").and_then(Value::as_str),
Some(thread.session_id.as_str())
);
assert_eq!(rolled_back_thread.turns.len(), 1);
assert_eq!(rolled_back_thread.status, ThreadStatus::Idle);
assert_eq!(
(
rolled_back_thread.model.as_deref(),
rolled_back_thread.reasoning_effort
),
(Some("gpt-5.2"), Some(ReasoningEffort::High))
);
assert_eq!(rolled_back_thread.turns[0].items.len(), 2);
match &rolled_back_thread.turns[0].items[0] {
ThreadItem::UserMessage { content, .. } => {
assert_eq!(
content,
&vec![V2UserInput::Text {
text: first_text.to_string(),
text_elements: Vec::new(),
}]
);
}
other => panic!("expected user message item, got {other:?}"),
}
let _: ThreadResumeResponse =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(pipelined_resume_id)).await??;
// Resume after rollback completes to verify the pruned history.
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id,
..Default::default()
})
.await?;
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(thread.turns.len(), 1);
assert_eq!(thread.status, ThreadStatus::Idle);
assert_eq!(thread.turns[0].items.len(), 2);
match &thread.turns[0].items[0] {
ThreadItem::UserMessage { content, .. } => {
assert_eq!(
content,
&vec![V2UserInput::Text {
text: first_text.to_string(),
text_elements: Vec::new(),
}]
);
}
other => panic!("expected user message item, got {other:?}"),
}
Ok(())
}
codex-rs/app-server/tests/suite/v2/turn_start.rs
-1
View File
@@ -4594,7 +4594,6 @@ async fn direct_input_to_multi_agent_v2_subagent_is_rejected(
"turn/settings/update",
json!({"turnId": "any-child-turn", "model": "gpt-5.4"}),
),
("thread/rollback", json!({"numTurns": 1})),
("thread/revert", json!({"beforeTurnId": "any-child-turn"})),
(
"review/start",
codex-rs/core/src/context_manager/history.rs
+1 -1
View File
@@ -522,7 +522,7 @@ impl ContextManager {
/// Instruction turns are history messages that should behave like a new prompt boundary:
/// ordinary user messages and structured assistant inter-agent instructions.
///
/// This mirrors thread-rollback semantics:
/// Used only to replay historical rollback markers when reconstructing a saved rollout:
/// - `num_turns == 0` is a no-op
/// - if there are no user turns, this is a no-op
/// - if `num_turns` exceeds the number of user turns, all user turns are dropped while
codex-rs/core/src/rollout_budget.rs
-8
View File
@@ -109,14 +109,6 @@ impl RolloutBudget {
);
}
/// Forces the next sampling request for `thread_id` to restate the current remainder.
pub(crate) fn rearm_reminder(&self, thread_id: ThreadId) {
let Some(mut state) = self.lock() else {
return;
};
state.deliveries.remove(&thread_id);
}
fn lock(&self) -> Option<MutexGuard<'_, RolloutBudgetState>> {
self.state.get().map(|state| {
state
codex-rs/core/src/session/handlers.rs
-131
View File
@@ -17,15 +17,12 @@ use crate::session::turn_input;
use crate::config::Config;
use crate::context::ContextualUserFragment;
use crate::context::GuardianApprovedAction;
use crate::context::NodeReplReviewEvidence;
use crate::review_prompts::resolve_review_request;
use crate::session::spawn_review_thread;
use crate::state::ReasoningEffortPin;
use crate::tasks::CompactTask;
use crate::tasks::UserShellCommandMode;
use crate::tasks::UserShellCommandTask;
use crate::tasks::execute_user_shell_command;
use codex_history::RolloutItem;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::Event;
@@ -39,7 +36,6 @@ use codex_protocol::protocol::RealtimeVoicesList;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::WarningEvent;
use codex_protocol::request_permissions::RequestPermissionsResponse;
@@ -252,129 +248,6 @@ pub async fn compact(sess: &Arc<Session>, sub_id: String) {
sess.spawn_task(turn_context, Vec::new(), CompactTask).await;
}
pub async fn thread_rollback(sess: &Arc<Session>, sub_id: String, num_turns: u32) {
if num_turns == 0 {
sess.send_event_raw(Event {
id: sub_id,
msg: EventMsg::Error(ErrorEvent {
misalignment: None,
message: "num_turns must be >= 1".to_string(),
codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed),
}),
})
.await;
return;
}
let has_active_turn = { sess.active_turn.lock().await.is_some() };
if has_active_turn {
sess.send_event_raw(Event {
id: sub_id,
msg: EventMsg::Error(ErrorEvent {
misalignment: None,
message: "Cannot rollback while a turn is in progress.".to_string(),
codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed),
}),
})
.await;
return;
}
let turn_context = sess
.new_turn_with_default_settings(sub_id, Default::default())
.await;
let live_thread = match sess.live_thread_for_persistence("rollback thread") {
Ok(live_thread) => live_thread,
Err(_) => {
sess.send_event_raw(Event {
id: turn_context.sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
misalignment: None,
message: "thread rollback requires persisted thread history".to_string(),
codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed),
}),
})
.await;
return;
}
};
if let Err(err) = live_thread.flush().await {
sess.send_event_raw(Event {
id: turn_context.sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
misalignment: None,
message: format!("failed to flush thread persistence for rollback replay: {err}"),
codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed),
}),
})
.await;
return;
}
let stored_history = match live_thread.load_history(/*include_archived*/ false).await {
Ok(history) => history,
Err(err) => {
sess.send_event_raw(Event {
id: turn_context.sub_id.clone(),
msg: EventMsg::Error(ErrorEvent {
misalignment: None,
message: format!("failed to load thread history for rollback replay: {err}"),
codex_error_info: Some(CodexErrorInfo::ThreadRollbackFailed),
}),
})
.await;
return;
}
};
let rollback_event = ThreadRolledBackEvent { num_turns };
let rollback_msg = EventMsg::ThreadRolledBack(rollback_event.clone());
let replay_items = stored_history
.items
.into_iter()
.chain(std::iter::once(RolloutItem::EventMsg(rollback_msg.clone())))
.collect::<Vec<_>>();
sess.apply_rollout_reconstruction(turn_context.as_ref(), replay_items.as_slice())
.await;
{
let mut state = sess.state.lock().await;
// Keep the baseline while startup prewarm is retained for the first turn,
// including when its task has not established the pin yet.
if state.startup_prewarm.is_none() {
state.reasoning_effort_pin = ReasoningEffortPin::Unset;
}
}
sess.services
.thread_extension_data
.remove::<NodeReplReviewEvidence>();
sess.guardian_review_session().invalidate().await;
sess.services
.agent_control
.rollout_budget()
.rearm_reminder(sess.thread_id());
sess.recompute_token_usage(turn_context.as_ref()).await;
sess.persist_rollout_items(&[RolloutItem::EventMsg(rollback_msg.clone())])
.await;
if let Err(err) = sess.flush_rollout().await {
sess.send_event(
turn_context.as_ref(),
EventMsg::Warning(WarningEvent {
message: format!(
"Rolled the thread back, but failed to save the rollback marker. Codex will continue retrying. Error: {err}"
),
}),
)
.await;
}
sess.deliver_event_raw(Event {
id: turn_context.sub_id.clone(),
msg: rollback_msg,
})
.await;
}
pub(super) async fn persist_thread_memory_mode_update(
sess: &Arc<Session>,
mode: ThreadMemoryMode,
@@ -689,10 +562,6 @@ pub(super) async fn submission_loop(
compact(&sess, sub.id.clone()).await;
false
}
Op::ThreadRollback { num_turns } => {
thread_rollback(&sess, sub.id.clone(), num_turns).await;
false
}
Op::SetThreadMemoryMode { mode } => {
set_thread_memory_mode(&sess, sub.id.clone(), mode).await;
false
codex-rs/core/src/session/reasoning_effort.rs
+2 -2
View File
@@ -1,7 +1,7 @@
//! Cache-preserving effort updates and the request-effort baseline for a context window.
//!
//! Only trusted harness items establish overrides. Replay and rollback preserve startup
//! prewarm's baseline while it is retained; later rollback invalidates the runtime pin.
//! Only trusted harness items establish overrides. Replay preserves startup prewarm's
//! baseline while it is retained.
//! Successful compaction retires the overrides and allows a fresh request baseline.
use super::session::Session;
codex-rs/core/src/session/tests.rs
-640
View File
@@ -86,7 +86,6 @@ use codex_protocol::permissions::FileSystemSandboxPolicy;
use codex_protocol::permissions::FileSystemSandboxPolicyContext;
use codex_protocol::permissions::FileSystemSpecialPath;
use codex_protocol::protocol::EnvironmentConfigState;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::TurnEnvironmentSelections;
use codex_protocol::request_permissions::PermissionGrantScope;
@@ -165,7 +164,6 @@ use codex_protocol::protocol::RealtimeVoicesList;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::Submission;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::protocol::TokenCountEvent;
use codex_protocol::protocol::TokenUsage;
@@ -3906,608 +3904,6 @@ async fn record_initial_history_forked_hydrates_previous_turn_settings() {
);
}
#[tokio::test]
async fn thread_rollback_drops_last_turn_from_history() {
let (mut sess, tc, rx) = make_session_and_context_with_rx().await;
let rollout_path = attach_thread_persistence(
Arc::get_mut(&mut sess).expect("session should not have additional references"),
)
.await;
let initial_context = build_initial_context(&sess, &tc).await;
let turn_1 = vec![
user_message("turn 1 user"),
assistant_message("turn 1 assistant"),
];
let turn_2 = vec![
user_message("turn 2 user"),
assistant_message("turn 2 assistant"),
];
let mut full_history = Vec::new();
full_history.extend(initial_context.clone());
full_history.extend(turn_1.clone());
full_history.extend(turn_2);
sess.replace_history(full_history.clone(), Some(tc.to_turn_context_item()))
.await;
let rollout_items: Vec<RolloutItem> = full_history
.into_iter()
.map(ResponseItemEnvelope::new)
.map(RolloutItem::ResponseItem)
.collect();
sess.persist_rollout_items(&rollout_items).await;
sess.set_previous_turn_settings(Some(PreviousTurnSettings {
model: "stale-model".to_string(),
comp_hash: None,
realtime_active: Some(tc.realtime_active),
}))
.await;
{
let mut state = sess.state.lock().await;
state.set_reference_context_item(Some(tc.to_turn_context_item()));
}
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
let rollback_event = wait_for_thread_rolled_back(&rx).await;
assert_eq!(rollback_event.num_turns, 1);
let mut expected = Vec::new();
expected.extend(initial_context);
expected.extend(turn_1);
let history = sess.clone_history().await;
assert_eq!(expected, raw_history_items(&history));
assert_eq!(sess.previous_turn_settings().await, None);
assert!(sess.reference_context_item().await.is_none());
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
.await
.expect("read rollout history")
else {
panic!("expected resumed rollout history");
};
assert!(resumed.history.iter().any(|item| {
matches!(
item,
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback))
if rollback.num_turns == 1
)
}));
}
#[tokio::test]
async fn thread_rollback_clears_history_when_num_turns_exceeds_existing_turns() {
let (mut sess, tc, rx) = make_session_and_context_with_rx().await;
attach_thread_persistence(
Arc::get_mut(&mut sess).expect("session should not have additional references"),
)
.await;
let initial_context = build_initial_context(&sess, &tc).await;
let turn_1 = vec![user_message("turn 1 user")];
let mut full_history = Vec::new();
full_history.extend(initial_context.clone());
full_history.extend(turn_1);
sess.replace_history(full_history.clone(), Some(tc.to_turn_context_item()))
.await;
let rollout_items: Vec<RolloutItem> = full_history
.into_iter()
.map(ResponseItemEnvelope::new)
.map(RolloutItem::ResponseItem)
.collect();
sess.persist_rollout_items(&rollout_items).await;
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 99).await;
let rollback_event = wait_for_thread_rolled_back(&rx).await;
assert_eq!(rollback_event.num_turns, 99);
let history = sess.clone_history().await;
assert_eq!(initial_context, raw_history_items(&history));
}
#[tokio::test]
async fn thread_rollback_fails_without_persisted_thread_history() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let initial_context = build_initial_context(&sess, &tc).await;
sess.record_conversation_items(tc.as_ref(), tc.model_info(), &initial_context)
.await;
let history_before_rollback = sess.clone_history().await;
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
let error_event = wait_for_thread_rollback_failed(&rx).await;
assert_eq!(
error_event.message,
"thread rollback requires persisted thread history"
);
assert_eq!(
error_event.codex_error_info,
Some(CodexErrorInfo::ThreadRollbackFailed)
);
assert_eq!(
raw_history_items(&sess.clone_history().await),
raw_history_items(&history_before_rollback)
);
}
#[tokio::test]
async fn thread_rollback_recomputes_previous_turn_settings_and_reference_context_from_replay() {
let (mut sess, tc, rx) = make_session_and_context_with_rx().await;
attach_thread_persistence(
Arc::get_mut(&mut sess).expect("session should not have additional references"),
)
.await;
let first_context_item = tc.to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
.expect("thread settings should have turn_id");
let mut rolled_back_context_item = first_context_item.clone();
rolled_back_context_item.turn_id = Some("rolled-back-turn".to_string());
rolled_back_context_item.model = "rolled-back-model".to_string();
let rolled_back_turn_id = rolled_back_context_item
.turn_id
.clone()
.expect("thread settings should have turn_id");
let turn_one_user = user_message("turn 1 user");
let turn_one_assistant = assistant_message("turn 1 assistant");
let turn_two_user = user_message("turn 2 user");
let turn_two_assistant = assistant_message("turn 2 assistant");
sess.persist_rollout_items(&[
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: first_turn_id.clone(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::EventMsg(EventMsg::UserMessage(
codex_protocol::protocol::UserMessageEvent {
client_id: None,
message: "turn 1 user".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
},
)),
RolloutItem::TurnContext(first_context_item.clone()),
RolloutItem::ResponseItem(turn_one_user.clone().into()),
RolloutItem::ResponseItem(turn_one_assistant.clone().into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: first_turn_id,
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: rolled_back_turn_id.clone(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::EventMsg(EventMsg::UserMessage(
codex_protocol::protocol::UserMessageEvent {
client_id: None,
message: "turn 2 user".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
},
)),
RolloutItem::TurnContext(rolled_back_context_item),
RolloutItem::ResponseItem(turn_two_user.into()),
RolloutItem::ResponseItem(turn_two_assistant.into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: rolled_back_turn_id,
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
])
.await;
sess.replace_history(
vec![assistant_message("stale history")],
Some(first_context_item.clone()),
)
.await;
sess.set_previous_turn_settings(Some(PreviousTurnSettings {
model: "stale-model".to_string(),
comp_hash: None,
realtime_active: None,
}))
.await;
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
let rollback_event = wait_for_thread_rolled_back(&rx).await;
assert_eq!(rollback_event.num_turns, 1);
assert_eq!(
raw_history_items(&sess.clone_history().await),
vec![turn_one_user, turn_one_assistant]
);
assert_eq!(
sess.previous_turn_settings().await,
Some(PreviousTurnSettings {
model: tc.model_info().slug.clone(),
comp_hash: None,
realtime_active: Some(tc.realtime_active),
})
);
assert_eq!(
serde_json::to_value(sess.reference_context_item().await)
.expect("serialize replay reference context item"),
serde_json::to_value(Some(first_context_item))
.expect("serialize expected reference context item")
);
}
#[tokio::test]
async fn thread_rollback_restores_cleared_reference_context_item_after_compaction() {
let (mut sess, tc, rx) = make_session_and_context_with_rx().await;
attach_thread_persistence(
Arc::get_mut(&mut sess).expect("session should not have additional references"),
)
.await;
let first_context_item = tc.to_turn_context_item();
let first_turn_id = first_context_item
.turn_id
.clone()
.expect("thread settings should have turn_id");
let compact_turn_id = "compact-turn".to_string();
let rolled_back_turn_id = "rolled-back-turn".to_string();
let compacted_history = vec![
user_message("turn 1 user"),
user_message("summary after compaction"),
];
let first_window_id = Uuid::now_v7();
let previous_window_id = Uuid::now_v7();
let compacted_window_id = Uuid::now_v7();
sess.persist_rollout_items(&[
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: first_turn_id.clone(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "turn 1 user".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
})),
RolloutItem::TurnContext(first_context_item.clone()),
RolloutItem::ResponseItem(user_message("turn 1 user").into()),
RolloutItem::ResponseItem(assistant_message("turn 1 assistant").into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: first_turn_id,
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: compact_turn_id.clone(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::Compacted(CompactedItem {
message: "summary after compaction".to_string(),
replacement_history: Some(
compacted_history
.iter()
.cloned()
.map(ResponseItemEnvelope::new)
.collect(),
),
retained_context: None,
guardian_history: None,
mcp_resource_origins: None,
window_number: Some(7),
first_window_id: Some(first_window_id.to_string()),
previous_window_id: Some(previous_window_id.to_string()),
window_id: Some(compacted_window_id.to_string()),
compaction_response_id: None,
latest_token_usage_record: None,
}),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: compact_turn_id,
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: rolled_back_turn_id.clone(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "turn 2 user".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
})),
RolloutItem::TurnContext(TurnContextItem {
turn_id: Some(rolled_back_turn_id.clone()),
model: "rolled-back-model".to_string(),
comp_hash: None,
..first_context_item.clone()
}),
RolloutItem::ResponseItem(user_message("turn 2 user").into()),
RolloutItem::ResponseItem(assistant_message("turn 2 assistant").into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: rolled_back_turn_id,
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
])
.await;
sess.replace_history(
vec![assistant_message("stale history")],
Some(first_context_item),
)
.await;
{
let mut state = sess.state.lock().await;
state.restore_auto_compact_window(
/*window_number*/ 99,
AutoCompactWindowIds {
first_window_id: Uuid::now_v7(),
previous_window_id: Some(Uuid::now_v7()),
window_id: Uuid::now_v7(),
},
);
}
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
let rollback_event = wait_for_thread_rolled_back(&rx).await;
assert_eq!(rollback_event.num_turns, 1);
assert_eq!(
raw_history_items(&sess.clone_history().await),
compacted_history
);
assert!(sess.reference_context_item().await.is_none());
assert_eq!(
sess.state.lock().await.auto_compact_window_ids(),
AutoCompactWindowIds {
first_window_id,
previous_window_id: Some(previous_window_id),
window_id: compacted_window_id,
}
);
assert!(sess.current_window_id().await.ends_with(":7"));
}
#[tokio::test]
async fn thread_rollback_persists_marker_and_replays_cumulatively() {
let (mut sess, tc, rx) = make_session_and_context_with_rx().await;
let rollout_path = attach_thread_persistence(
Arc::get_mut(&mut sess).expect("session should not have additional references"),
)
.await;
let turn_context_item = tc.to_turn_context_item();
sess.persist_rollout_items(&[
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: "turn-1".to_string(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "turn 1 user".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
})),
RolloutItem::TurnContext(turn_context_item.clone()),
RolloutItem::ResponseItem(user_message("turn 1 user").into()),
RolloutItem::ResponseItem(assistant_message("turn 1 assistant").into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: "turn-1".to_string(),
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: "turn-2".to_string(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "turn 2 user".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
})),
RolloutItem::TurnContext(turn_context_item.clone()),
RolloutItem::ResponseItem(user_message("turn 2 user").into()),
RolloutItem::ResponseItem(assistant_message("turn 2 assistant").into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: "turn-2".to_string(),
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
RolloutItem::EventMsg(EventMsg::TurnStarted(
codex_protocol::protocol::TurnStartedEvent {
turn_id: "turn-3".to_string(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: Some(128_000),
collaboration_mode_kind: ModeKind::Default,
},
)),
RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "turn 3 user".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
..Default::default()
})),
RolloutItem::TurnContext(turn_context_item),
RolloutItem::ResponseItem(user_message("turn 3 user").into()),
RolloutItem::ResponseItem(assistant_message("turn 3 assistant").into()),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
turn_id: "turn-3".to_string(),
started_at: None,
last_agent_message: None,
error: None,
completed_at: None,
duration_ms: None,
time_to_first_token_ms: None,
})),
])
.await;
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
let first_rollback = wait_for_thread_rolled_back(&rx).await;
assert_eq!(first_rollback.num_turns, 1);
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
let second_rollback = wait_for_thread_rolled_back(&rx).await;
assert_eq!(second_rollback.num_turns, 1);
assert_eq!(
raw_history_items(&sess.clone_history().await),
vec![
user_message("turn 1 user"),
assistant_message("turn 1 assistant")
]
);
let InitialHistory::Resumed(resumed) = RolloutRecorder::get_rollout_history(&rollout_path)
.await
.expect("read rollout history")
else {
panic!("expected resumed rollout history");
};
let rollback_markers = resumed
.history
.iter()
.filter(|item| matches!(item, RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_))))
.count();
assert_eq!(rollback_markers, 2);
}
#[tokio::test]
async fn thread_rollback_fails_when_turn_in_progress() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let initial_context = build_initial_context(&sess, &tc).await;
sess.record_conversation_items(tc.as_ref(), tc.model_info(), &initial_context)
.await;
let history_before_rollback = sess.clone_history().await;
*sess.active_turn.lock().await = Some(crate::state::ActiveTurn::default());
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await;
let error_event = wait_for_thread_rollback_failed(&rx).await;
assert_eq!(
error_event.codex_error_info,
Some(CodexErrorInfo::ThreadRollbackFailed)
);
let history = sess.clone_history().await;
assert_eq!(
raw_history_items(&history_before_rollback),
raw_history_items(&history)
);
}
#[tokio::test]
async fn thread_rollback_fails_when_num_turns_is_zero() {
let (sess, tc, rx) = make_session_and_context_with_rx().await;
let initial_context = build_initial_context(&sess, &tc).await;
sess.record_conversation_items(tc.as_ref(), tc.model_info(), &initial_context)
.await;
let history_before_rollback = sess.clone_history().await;
handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 0).await;
let error_event = wait_for_thread_rollback_failed(&rx).await;
assert_eq!(error_event.message, "num_turns must be >= 1");
assert_eq!(
error_event.codex_error_info,
Some(CodexErrorInfo::ThreadRollbackFailed)
);
let history = sess.clone_history().await;
assert_eq!(
raw_history_items(&history_before_rollback),
raw_history_items(&history)
);
}
#[tokio::test]
async fn set_rate_limits_retains_previous_credits() {
let codex_home = tempfile::tempdir().expect("create temp dir");
@@ -4965,42 +4361,6 @@ fn success_flag_true_with_no_error_and_content_used() {
assert_eq!(expected, got);
}
async fn wait_for_thread_rolled_back(rx: &async_channel::Receiver<Event>) -> ThreadRolledBackEvent {
let deadline = StdDuration::from_secs(2);
let start = std::time::Instant::now();
loop {
let remaining = deadline.saturating_sub(start.elapsed());
let evt = tokio::time::timeout(remaining, rx.recv())
.await
.expect("timeout waiting for event")
.expect("event");
match evt.msg {
EventMsg::ThreadRolledBack(payload) => return payload,
_ => continue,
}
}
}
async fn wait_for_thread_rollback_failed(rx: &async_channel::Receiver<Event>) -> ErrorEvent {
let deadline = StdDuration::from_secs(2);
let start = std::time::Instant::now();
loop {
let remaining = deadline.saturating_sub(start.elapsed());
let evt = tokio::time::timeout(remaining, rx.recv())
.await
.expect("timeout waiting for event")
.expect("event");
match evt.msg {
EventMsg::Error(payload)
if payload.codex_error_info == Some(CodexErrorInfo::ThreadRollbackFailed) =>
{
return payload;
}
_ => continue,
}
}
}
async fn open_thread_persistence(session: &mut Session) -> PathBuf {
let config = session.get_config().await;
let live_thread = LiveThread::create(
codex-rs/core/src/session/turn.rs
+1 -1
View File
@@ -503,7 +503,7 @@ pub(crate) async fn run_turn(
.record_step_world_state_if_changed(&world_state, step_context.as_ref())
.await?;
// Keep the override after accepted input so ordinary turn rollback removes it too.
// Keep the override after accepted input so history truncation removes them together.
sess.record_reasoning_effort_override(step_context.as_ref())
.await;
codex-rs/core/src/state/session.rs
+1 -2
View File
@@ -31,8 +31,7 @@ use codex_utils_output_truncation::TruncationPolicy;
use tokio_util::task::AbortOnDropHandle;
/// Runtime request effort, initially unset and established by prewarm or sampling.
/// Rollback clears it after startup prewarm is consumed; successful compaction allows
/// a fresh baseline without an override.
/// Successful compaction allows a fresh baseline without an override.
pub(crate) enum ReasoningEffortPin {
Unset,
Compacted,
codex-rs/core/tests/suite/compact_resume_fork.rs
-237
View File
@@ -19,30 +19,21 @@ use codex_core::spawn::CODEX_SANDBOX_NETWORK_DISABLED_ENV_VAR;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_history::CodexHarnessMetadata;
use codex_history::RolloutItem;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::protocol::WarningEvent;
use codex_protocol::user_input::UserInput;
use core_test_support::ThreadIdle;
use core_test_support::context_snapshot;
use core_test_support::context_snapshot::ContextSnapshotOptions;
use core_test_support::context_snapshot::ContextSnapshotRenderMode;
use core_test_support::responses::ResponseMock;
use core_test_support::responses::ResponsesRequest;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once_match;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::sse;
use core_test_support::test_codex::local_selections;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
@@ -54,7 +45,6 @@ use tempfile::TempDir;
use wiremock::MockServer;
const AFTER_SECOND_RESUME: &str = "AFTER_SECOND_RESUME";
const AFTER_ROLLBACK: &str = "AFTER_ROLLBACK";
const CHECKPOINT_METADATA_KEY: &str = "replacement_history_metadata";
fn network_disabled() -> bool {
@@ -506,233 +496,6 @@ async fn compact_resume_after_second_compaction_preserves_history() -> Result<()
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
/// Scenario: rolling back behind a pre-turn compaction should replay
/// append-only history from the rollout file and keep earlier compacted
/// history visible.
async fn snapshot_rollback_past_compaction_replays_append_only_history() -> Result<()> {
if network_disabled() {
println!("Skipping test because network is disabled in this sandbox");
return Ok(());
}
const EDITED_AFTER_COMPACT: &str = "EDITED_AFTER_COMPACT";
const SECOND_REPLY: &str = "SECOND_REPLY";
let server = MockServer::start().await;
let sse1 = sse(vec![
ev_assistant_message("m1", FIRST_REPLY),
ev_completed("r1"),
]);
let sse2 = sse(vec![
ev_assistant_message("m2", SUMMARY_TEXT),
ev_completed("r2"),
]);
let sse3 = sse(vec![
ev_assistant_message("m3", SECOND_REPLY),
ev_completed("r3"),
]);
let sse4 = sse(vec![ev_completed("r4")]);
let request_log = mount_sse_sequence(&server, vec![sse1, sse2, sse3, sse4]).await;
let (_home, _config, _manager, base) = start_test_conversation(&server, /*model*/ None).await;
user_turn(&base, "hello world").await;
compact_conversation(&base).await;
user_turn(&base, EDITED_AFTER_COMPACT).await;
base.submit(Op::ThreadRollback { num_turns: 1 })
.await
.expect("submit thread rollback");
let rollback_event = wait_for_event(&base, |ev| {
if let EventMsg::Error(error) = ev {
panic!("rollback failed: {error:?}");
}
matches!(ev, EventMsg::ThreadRolledBack(_))
})
.await;
let EventMsg::ThreadRolledBack(rollback_event) = rollback_event else {
panic!("expected thread rolled back event");
};
assert_eq!(rollback_event.num_turns, 1);
user_turn(&base, AFTER_ROLLBACK).await;
let requests = request_log.requests();
assert_eq!(requests.len(), 4);
assert!(requests[1].body_contains_text(SUMMARIZATION_PROMPT));
assert!(requests[2].body_contains_text("hello world"));
assert!(requests[2].body_contains_text(SUMMARY_TEXT));
assert!(requests[2].body_contains_text(EDITED_AFTER_COMPACT));
let after_rollback_user_texts = requests[3].message_input_texts("user");
let after_rollback_last = after_rollback_user_texts
.last()
.expect("post-rollback request missing user messages");
assert_eq!(after_rollback_last, AFTER_ROLLBACK);
assert!(
requests[3].body_contains_text("hello world"),
"the first turn should remain visible after rollback behind compaction",
);
assert!(
!requests[3].body_contains_text(EDITED_AFTER_COMPACT),
"the edited post-compaction turn should be removed by rollback",
);
assert!(
requests[3].body_contains_text(SUMMARY_TEXT),
"compaction summary should remain for the preserved first turn",
);
insta::assert_snapshot!(
"rollback_past_compaction_shapes",
context_snapshot::format_labeled_requests_snapshot(
"rollback past compaction replay after rollback",
&[
("compaction request", &requests[1]),
("before rollback", &requests[2]),
("after rollback", &requests[3]),
],
&ContextSnapshotOptions::default()
.strip_capability_instructions()
.render_mode(ContextSnapshotRenderMode::KindWithTextPrefix { max_chars: 64 }),
)
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
/// Scenario: rolling back a turn that introduced persistent pre-thread settings
/// diffs should trim those context updates so the next request includes them
/// only once.
async fn snapshot_rollback_followup_turn_trims_context_updates() -> Result<()> {
if network_disabled() {
println!("Skipping test because network is disabled in this sandbox");
return Ok(());
}
const MODEL: &str = "gpt-5.4";
const TURN_ONE_USER: &str = "turn 1 user";
const TURN_TWO_USER: &str = "turn 2 user";
const FOLLOWUP_USER: &str = "follow-up user";
const ROLLED_BACK_DEV_INSTRUCTIONS: &str = "ROLLED_BACK_DEV_INSTRUCTIONS";
const PRETURN_CONTEXT_DIFF_CWD: &str = "PRETURN_CONTEXT_DIFF_CWD";
let server = MockServer::start().await;
let request_log = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_assistant_message("m1", "turn 1 assistant"),
ev_completed("r1"),
]),
sse(vec![
ev_assistant_message("m2", "turn 2 assistant"),
ev_completed("r2"),
]),
sse(vec![ev_response_created("r3"), ev_completed("r3")]),
],
)
.await;
let (_home, config, _manager, conversation) =
start_test_conversation(&server, Some(MODEL)).await;
user_turn(&conversation, TURN_ONE_USER).await;
let override_cwd = config.cwd.join(PRETURN_CONTEXT_DIFF_CWD);
std::fs::create_dir_all(&override_cwd)?;
core_test_support::submit_thread_settings(
&conversation,
ThreadSettingsOverrides {
environments: Some(local_selections(override_cwd.clone())),
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Default,
settings: Settings {
model: MODEL.to_string(),
reasoning_effort: None,
developer_instructions: Some(ROLLED_BACK_DEV_INSTRUCTIONS.to_string()),
},
}),
..Default::default()
},
)
.await?;
user_turn(&conversation, TURN_TWO_USER).await;
conversation
.submit(Op::ThreadRollback { num_turns: 1 })
.await?;
let rollback_event = wait_for_event(&conversation, |ev| {
if let EventMsg::Error(error) = ev {
panic!("rollback failed: {error:?}");
}
matches!(ev, EventMsg::ThreadRolledBack(_))
})
.await;
let EventMsg::ThreadRolledBack(rollback_event) = rollback_event else {
panic!("expected thread rolled back event");
};
assert_eq!(rollback_event.num_turns, 1);
user_turn(&conversation, FOLLOWUP_USER).await;
let requests = request_log.requests();
assert_eq!(requests.len(), 3);
let before_rollback_developer_count = requests[1]
.message_input_texts("developer")
.iter()
.filter(|text| text.contains(ROLLED_BACK_DEV_INSTRUCTIONS))
.count();
assert_eq!(before_rollback_developer_count, 1);
assert_eq!(
requests[1]
.message_input_texts("user")
.iter()
.filter(|text| text.contains(PRETURN_CONTEXT_DIFF_CWD))
.count(),
1
);
let after_rollback_developer_count = requests[2]
.message_input_texts("developer")
.iter()
.filter(|text| text.contains(ROLLED_BACK_DEV_INSTRUCTIONS))
.count();
assert_eq!(after_rollback_developer_count, 1);
let after_rollback_user_texts = requests[2].message_input_texts("user");
assert_eq!(
after_rollback_user_texts
.iter()
.filter(|text| text.contains(PRETURN_CONTEXT_DIFF_CWD))
.count(),
1
);
assert_eq!(
after_rollback_user_texts.last().map(String::as_str),
Some(FOLLOWUP_USER)
);
insta::assert_snapshot!(
"rollback_followup_turn_trims_context_updates",
context_snapshot::format_labeled_requests_snapshot(
"rollback trims pre-turn override context updates before the follow-up request",
&[
("rolled-back turn request", &requests[1]),
("follow-up request after rollback", &requests[2]),
],
&ContextSnapshotOptions::default()
.strip_capability_instructions()
.render_mode(ContextSnapshotRenderMode::KindWithTextPrefix { max_chars: 96 }),
)
);
Ok(())
}
fn normalize_line_endings(value: &mut Value) {
match value {
Value::String(text) if text.contains('\r') => {
codex-rs/core/tests/suite/guardian_authorization.rs
+1 -11
View File
@@ -217,8 +217,7 @@ async fn guardian_revalidates_owning_session_before_allow(
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_authorization_revision_survives_compaction_not_user_input_or_rollback()
-> Result<()> {
async fn guardian_authorization_revision_survives_compaction_not_user_input() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
@@ -285,14 +284,5 @@ async fn guardian_authorization_revision_survives_compaction_not_user_input_or_r
.await;
assert_eq!(test.codex.guardian_authorization_version().await, expected);
test.codex.ensure_rollout_materialized().await;
test.codex
.submit(Op::ThreadRollback { num_turns: 1 })
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::ThreadRolledBack(_))
})
.await;
assert_ne!(test.codex.guardian_authorization_version().await, expected);
Ok(())
}
codex-rs/core/tests/suite/guardian_history.rs
+33 -8
View File
@@ -1,4 +1,4 @@
//! Exercises retained review history through compaction, resume, fork, eviction, and rollback.
//! Exercises retained review history through compaction, resume, fork, eviction, and legacy rollback replay.
use anyhow::Result;
use base64::Engine;
@@ -17,6 +17,7 @@ use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::request_user_input::RequestUserInputAnswer;
use codex_protocol::request_user_input::RequestUserInputResponse;
use codex_protocol::user_input::UserInput;
@@ -310,14 +311,15 @@ async fn guardian_history_uses_deltas_between_eviction_batches() -> Result<()> {
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn guardian_history_survives_compaction_and_eviction_but_not_rollback() -> Result<()> {
async fn guardian_history_survives_compaction_and_eviction_but_not_legacy_rollback_replay()
-> Result<()> {
skip_if_no_network!(Ok(()));
skip_if_wine_exec!(
Ok(()),
"Guardian approval actions require host-native paths"
);
let server = start_mock_server().await;
let test = test_codex()
let mut test = test_codex()
.with_config(|config| {
config.features.enable(Feature::TokenBudget).unwrap();
config
@@ -499,12 +501,35 @@ async fn guardian_history_survives_compaction_and_eviction_but_not_rollback() ->
);
test.codex.ensure_rollout_materialized().await;
test.codex
.submit(Op::ThreadRollback { num_turns: 2 })
.append_rollout_items(&[RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
ThreadRolledBackEvent { num_turns: 2 },
))])
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::ThreadRolledBack(_))
})
.await;
test.codex.shutdown_and_wait().await?;
let thread_id = test.session_configured.thread_id;
test.thread_manager.remove_thread(&thread_id).await;
let model_context = test
.thread_store
.load_latest_model_context(LoadThreadHistoryParams {
thread_id,
include_archived: false,
})
.await?;
test.codex = test
.thread_manager
.resume_thread_with_history(
test.config.clone(),
InitialHistory::Resumed(ResumedHistory {
conversation_id: thread_id,
history: Arc::new(model_context.items),
rollout_path: None,
}),
test.thread_manager.auth_manager(),
/*parent_trace*/ None,
ClientMcpExtensions::default(),
)
.await?
.thread;
} else {
assert!(!transcript.contains(">>> TRUSTED USER ANSWERS START"));
assert!(!transcript.contains("Do not publish anything."));
codex-rs/core/tests/suite/guardian_retained_context.rs
+35 -59
View File
@@ -1,5 +1,5 @@
//! Retained-instruction and answer lifecycles through real sessions and durable checkpoints.
//! Real user input covers steering, compaction and rollback; legacy replay fixtures also use
//! Real user input covers steering and compaction; legacy rollback replay fixtures use
//! the public rollout append API. Resume and child forks use production paths.
use std::collections::HashMap;
@@ -36,6 +36,7 @@ use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::request_user_input::RequestUserInputAnswer;
use codex_protocol::request_user_input::RequestUserInputResponse;
use codex_protocol::user_input::UserInput;
@@ -63,7 +64,7 @@ use wiremock::MockServer;
use wiremock::matchers::header;
// Keep child completion out of the parent's history and wait for parent turn cleanup
// before checking rollback boundaries.
// before checking inherited authorization.
#[derive(Default)]
struct ForkTestLifecycle {
entered: Notify,
@@ -104,7 +105,7 @@ async fn wait_for_thread_idle(thread: &CodexThread) {
.get_or_init(ThreadIdle::default);
tokio::time::timeout(Duration::from_secs(10), idle.0.notified())
.await
.expect("thread should become idle before rollback");
.expect("thread should become idle");
}
async fn record_answer(
@@ -220,7 +221,7 @@ async fn compact_and_assert_answers(
expected: &[VerifiedAnswer],
) -> Result<RetainedContext> {
// Inspect live state by persisting a real compaction checkpoint, not a private getter.
// Repeating this after resume/rollback also catches checkpoint resurrection.
// Repeating this after legacy rollback replay also catches checkpoint resurrection.
thread.submit(Op::Compact).await?;
wait_for_event(thread, |event| matches!(event, EventMsg::TurnComplete(_))).await;
thread.flush_rollout().await?;
@@ -250,11 +251,11 @@ enum InstructionSize {
Oversized,
}
#[test_case(ThreadHistoryMode::Legacy, true, InstructionSize::Normal; "enabled legacy rollback")]
#[test_case(ThreadHistoryMode::Legacy, true, InstructionSize::Normal; "enabled legacy rollback replay")]
#[test_case(ThreadHistoryMode::Paginated, true, InstructionSize::Normal; "enabled paginated resume")]
#[test_case(ThreadHistoryMode::Legacy, false, InstructionSize::Normal; "disabled legacy rollback")]
#[test_case(ThreadHistoryMode::Legacy, false, InstructionSize::Normal; "disabled legacy rollback replay")]
#[test_case(ThreadHistoryMode::Paginated, false, InstructionSize::Normal; "disabled paginated resume")]
#[test_case(ThreadHistoryMode::Legacy, true, InstructionSize::Oversized; "oversized instruction rollback")]
#[test_case(ThreadHistoryMode::Legacy, true, InstructionSize::Oversized; "oversized instruction rollback replay")]
#[test_case(ThreadHistoryMode::Paginated, true, InstructionSize::Oversized; "oversized instruction resume")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn retained_instructions_keep_identity_across_compaction_and_resume(
@@ -272,8 +273,7 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume(
"Project detail. ".repeat(2_000)
),
};
// Paginated history supports checkpoint resume, but not the full-history read
// used by ThreadRollback. Exercise rollback on its supported legacy path.
// Legacy rollouts can contain rollback markers; paginated histories use checkpoint resume.
let rollback_counts: &[usize] = match history_mode {
ThreadHistoryMode::Legacy => &[1, 0],
ThreadHistoryMode::Paginated => &[],
@@ -496,11 +496,12 @@ async fn retained_instructions_keep_identity_across_compaction_and_resume(
serde_json::Value::Null
}
);
thread.submit(Op::ThreadRollback { num_turns: 1 }).await?;
wait_for_event(&thread, |event| {
matches!(event, EventMsg::ThreadRolledBack(_))
})
.await;
thread
.append_rollout_items(&[RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
ThreadRolledBackEvent { num_turns: 1 },
))])
.await?;
thread = resume(&test, &thread).await?;
expected["user_messages"]
.as_array_mut()
.expect("expected retained user messages")
@@ -772,7 +773,7 @@ async fn disabled_capture_stays_incomplete_after_compaction_and_enabled_resume()
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn retained_answers_rollback_only_the_steered_instruction() -> Result<()> {
async fn legacy_rollback_replay_retains_only_surviving_steered_answers() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let test = test_codex()
@@ -907,11 +908,12 @@ async fn retained_answers_rollback_only_the_steered_instruction() -> Result<()>
let mut thread = resume(&test, &test.codex).await?;
compact_and_assert_answers(&test, &thread, &answers).await?;
for expected in [&answers[..1], &[]] {
thread.submit(Op::ThreadRollback { num_turns: 1 }).await?;
wait_for_event(&thread, |event| {
matches!(event, EventMsg::ThreadRolledBack(_))
})
.await;
thread
.append_rollout_items(&[RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
ThreadRolledBackEvent { num_turns: 1 },
))])
.await?;
thread = resume(&test, &thread).await?;
compact_and_assert_answers(&test, &thread, expected).await?;
thread = resume(&test, &thread).await?;
compact_and_assert_answers(&test, &thread, expected).await?;
@@ -922,7 +924,7 @@ async fn retained_answers_rollback_only_the_steered_instruction() -> Result<()>
#[derive(Clone, Copy)]
enum LifecycleBoundary {
Rollback,
LegacyRollbackReplay,
ChildFork,
}
@@ -1221,8 +1223,8 @@ async fn forked_parent_instructions_do_not_become_local_authorization(
sse(vec![ev_completed("parent-fork-completion")]),
)
.await;
// Hold the child at a real tool boundary until rollback is checked. Its completion
// notification must not add a parent message or consume a parent mock response.
// Pause the child while checking inherited authorization. Its completion notification
// must not add a parent message or consume a parent mock response.
let child_requests = mount_sse_once_match(
&server,
move |request: &wiremock::Request| {
@@ -1274,36 +1276,9 @@ async fn forked_parent_instructions_do_not_become_local_authorization(
thread_context_enabled.then_some((0, true)),
"inherited conversation must not populate child-local authorization",
);
let root_before = child.guardian_root_snapshot().await.context("live root")?;
let root_snapshot = child.guardian_root_snapshot().await.context("live root")?;
assert!(
root_before
.messages
.contains(&codex_core::GuardianRootMessage::User(
PARENT_GRANT.to_owned()
))
);
wait_for_thread_idle(&test.codex).await;
test.codex
.submit(Op::ThreadRollback { num_turns: 1 })
.await?;
wait_for_event(&test.codex, |event| {
if let EventMsg::Error(error) = event {
panic!("rollback failed: {error:?}");
}
matches!(event, EventMsg::ThreadRolledBack(_))
})
.await;
let root_after = child
.guardian_root_snapshot()
.await
.context("live root after rollback")?;
assert_ne!(
root_before.authorization_version,
root_after.authorization_version
);
assert!(
!root_after
root_snapshot
.messages
.contains(&codex_core::GuardianRootMessage::User(
PARENT_GRANT.to_owned()
@@ -1368,7 +1343,7 @@ async fn forked_parent_instructions_do_not_become_local_authorization(
Ok(())
}
#[test_case(ThreadHistoryMode::Legacy, LifecycleBoundary::Rollback; "legacy rollback")]
#[test_case(ThreadHistoryMode::Legacy, LifecycleBoundary::LegacyRollbackReplay; "legacy rollback replay")]
#[test_case(ThreadHistoryMode::Legacy, LifecycleBoundary::ChildFork; "legacy child fork")]
#[test_case(ThreadHistoryMode::Paginated, LifecycleBoundary::ChildFork; "paginated child fork")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -1455,15 +1430,16 @@ async fn retained_answers_cross_real_session_boundaries(
let expected = [before.clone(), after];
match boundary {
LifecycleBoundary::Rollback => {
LifecycleBoundary::LegacyRollbackReplay => {
// First remove the suffix answer, then the source retained only in the checkpoint.
let mut thread = thread;
for expected in [std::slice::from_ref(&before), &[]] {
thread.submit(Op::ThreadRollback { num_turns: 1 }).await?;
wait_for_event(&thread, |event| {
matches!(event, EventMsg::ThreadRolledBack(_))
})
.await;
thread
.append_rollout_items(&[RolloutItem::EventMsg(EventMsg::ThreadRolledBack(
ThreadRolledBackEvent { num_turns: 1 },
))])
.await?;
thread = resume(&test, &thread).await?;
compact_and_assert_answers(&test, &thread, expected).await?;
thread = resume(&test, &thread).await?;
compact_and_assert_answers(&test, &thread, expected).await?;
codex-rs/core/tests/suite/model_switching.rs
-239
View File
@@ -3,12 +3,7 @@ use codex_config::types::Personality;
use codex_core::CodexThread;
use codex_core::ForkSnapshot;
use codex_core::TurnInputRequest;
use codex_core::config::Config;
use codex_core::config::Constrained;
use codex_extension_api::ExtensionFuture;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_extension_api::ThreadIdleInput;
use codex_extension_api::ThreadLifecycleContributor;
use codex_features::Feature;
use codex_history::RolloutItem;
use codex_login::CodexAuth;
@@ -64,9 +59,7 @@ use pretty_assertions::assert_eq;
use serde_json::json;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use test_case::test_case;
use tokio::sync::Notify;
use wiremock::MockServer;
fn read_only_user_turn(test: &TestCodex, items: Vec<UserInput>, model: String) -> TurnInputRequest {
@@ -303,131 +296,6 @@ async fn first_turn_after_empty_prefix_fork_preserves_inherited_base_instruction
Ok(())
}
#[derive(Clone, Copy)]
enum RollbackFollowup {
StartupModel,
SwitchedModel,
ColdResume,
}
#[derive(Default)]
struct RollbackReady {
idle: Notify,
}
impl ThreadLifecycleContributor<Config> for RollbackReady {
fn on_thread_idle<'a>(&'a self, _input: ThreadIdleInput<'a>) -> ExtensionFuture<'a, ()> {
Box::pin(async move {
self.idle.notify_one();
})
}
}
impl RollbackReady {
async fn wait(&self) {
// TurnComplete is delivered before the active turn is cleared. Rollback requires
// the later thread-idle callback so it cannot race with turn cleanup.
tokio::time::timeout(Duration::from_secs(10), self.idle.notified())
.await
.expect("thread should become idle before rollback");
}
}
#[test_case(RollbackFollowup::StartupModel; "return to startup model")]
#[test_case(RollbackFollowup::SwitchedModel; "retry switched model")]
#[test_case(RollbackFollowup::ColdResume; "retry switched model after cold resume")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn rollback_first_turn_model_change_removes_its_instructions(
followup: RollbackFollowup,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let response_mock = mount_sse_sequence(
&server,
vec![sse_completed("resp-first"), sse_completed("resp-followup")],
)
.await;
let initial_model = "gpt-5.6-terra";
let switched_model = "gpt-5.5";
let rollback_ready = Arc::new(RollbackReady::default());
let mut extensions = ExtensionRegistryBuilder::new();
extensions.thread_lifecycle_contributor(rollback_ready.clone());
let mut builder = test_codex()
.with_model_info_override(initial_model, configure_model_switching_fixture)
.with_extensions(Arc::new(extensions.build()));
let test = builder.build_with_auto_env(&server).await?;
submit_model_turn(
&test.codex,
switched_model,
ThreadSettingsOverrides::default(),
)
.await?;
rollback_ready.wait().await;
test.codex
.submit(Op::ThreadRollback { num_turns: 1 })
.await?;
wait_for_event(&test.codex, |ev| {
if let EventMsg::Error(error) = ev {
panic!("rollback failed: {error:?}");
}
matches!(ev, EventMsg::ThreadRolledBack(_))
})
.await;
let test = match followup {
RollbackFollowup::ColdResume => {
let mut resume_builder = test_codex()
.with_model_info_override(initial_model, configure_model_switching_fixture)
.with_model(switched_model);
resume_builder.restart(&server, &test).await?
}
RollbackFollowup::StartupModel | RollbackFollowup::SwitchedModel => test,
};
let followup_model = match followup {
RollbackFollowup::StartupModel => initial_model,
RollbackFollowup::SwitchedModel | RollbackFollowup::ColdResume => switched_model,
};
submit_model_turn(
&test.codex,
followup_model,
ThreadSettingsOverrides::default(),
)
.await?;
let request = &response_mock.requests()[1];
assert_eq!(request.body_json()["model"], followup_model);
let misaligned_messages = request
.inputs_of_type("message")
.into_iter()
.filter(|message| {
message["internal_chat_message_metadata_passthrough"]["content_item_kinds"]
.as_array()
.is_some_and(|kinds| {
message["content"]
.as_array()
.is_none_or(|content| content.len() != kinds.len())
})
})
.collect::<Vec<_>>();
assert_eq!(misaligned_messages, Vec::<serde_json::Value>::new());
let model_switch_count = request
.message_input_texts("developer")
.iter()
.filter(|text| text.contains("<model_switch>"))
.count();
assert_eq!(
model_switch_count,
usize::from(followup_model == switched_model),
"rolled-back model instructions must not survive or be duplicated"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn model_change_appends_model_instructions_developer_message() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -1359,113 +1227,6 @@ async fn model_change_from_generated_image_to_text_preserves_prior_generated_ima
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn thread_rollback_after_generated_image_drops_entire_image_turn_history() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = MockServer::start().await;
let image_model_slug = "test-image-model";
let image_model = test_model_info(
image_model_slug,
"Test Image Model",
"supports image input",
default_input_modalities(),
);
mount_models_once(
&server,
ModelsResponse {
models: vec![image_model],
},
)
.await;
let responses = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp-1"),
ev_image_generation_call("ig_rollback", "completed", "lobster", "Zm9v"),
ev_completed_with_tokens("resp-1", /*total_tokens*/ 10),
]),
sse_completed("resp-2"),
],
)
.await;
let rollback_ready = Arc::new(RollbackReady::default());
let mut extensions = ExtensionRegistryBuilder::new();
extensions.thread_lifecycle_contributor(rollback_ready.clone());
let mut builder = test_codex()
.with_extensions(Arc::new(extensions.build()))
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(move |config| {
config.model = Some(image_model_slug.to_string());
});
let test = builder.build(&server).await?;
let models_manager = test.thread_manager.get_models_manager();
let _ = models_manager
.list_models(
RefreshStrategy::OnlineIfUncached,
codex_core::test_support::default_http_client_factory(),
)
.await;
test.codex
.start_or_steer_turn(read_only_user_turn(
&test,
vec![UserInput::Text {
text: "generate a lobster".to_string(),
text_elements: Vec::new(),
}],
image_model_slug.to_string(),
))
.await?;
wait_for_event(&test.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
rollback_ready.wait().await;
test.codex
.submit(Op::ThreadRollback { num_turns: 1 })
.await?;
wait_for_event(&test.codex, |ev| {
if let EventMsg::Error(error) = ev {
panic!("rollback failed: {error:?}");
}
matches!(ev, EventMsg::ThreadRolledBack(_))
})
.await;
test.codex
.start_or_steer_turn(read_only_user_turn(
&test,
vec![UserInput::Text {
text: "after rollback".to_string(),
text_elements: Vec::new(),
}],
image_model_slug.to_string(),
))
.await?;
wait_for_event(&test.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let requests = responses.requests();
assert_eq!(requests.len(), 2, "expected two model requests");
let second_request = requests.last().expect("expected second request");
assert!(
!second_request
.message_input_texts("user")
.iter()
.any(|text| text == "generate a lobster"),
"rollback should remove the rolled-back image-generation user turn"
);
assert!(
second_request
.inputs_of_type("image_generation_call")
.is_empty(),
"rollback should remove the generated image call with the rolled-back turn"
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn model_switch_to_smaller_model_updates_token_context_window() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/core/tests/suite/reasoning_effort_override.rs
+1 -149
View File
@@ -7,10 +7,6 @@ use codex_core::SuspendTurnOutcome;
use codex_core::TurnInputRequest;
use codex_core::TurnInputSubmission;
use codex_core::config::Config;
use codex_extension_api::ExtensionFuture;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_extension_api::ThreadIdleInput;
use codex_extension_api::ThreadLifecycleContributor;
use codex_features::Feature;
use codex_login::CodexAuth;
use codex_protocol::models::ResponseItem;
@@ -32,8 +28,6 @@ use serde_json::Value;
use std::sync::Arc;
use std::time::Duration;
use test_case::test_case;
use tokio::sync::Notify;
use tokio::time::timeout;
use wiremock::ResponseTemplate;
fn override_builder() -> TestCodexBuilder {
@@ -378,8 +372,6 @@ enum PrewarmStartup {
New,
Resume,
Fork,
ResumeThenRollback,
ForkThenRollback,
}
#[test_case(true, PrewarmStartup::New; "new thread feature enabled")]
@@ -388,8 +380,6 @@ enum PrewarmStartup {
#[test_case(false, PrewarmStartup::Resume; "resumed thread feature disabled")]
#[test_case(true, PrewarmStartup::Fork; "forked thread feature enabled")]
#[test_case(false, PrewarmStartup::Fork; "forked thread feature disabled")]
#[test_case(true, PrewarmStartup::ResumeThenRollback; "resumed thread rollback before first turn")]
#[test_case(true, PrewarmStartup::ForkThenRollback; "forked thread rollback before first turn")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reasoning_effort_override_websocket_prewarm_preserves_baseline(
feature_enabled: bool,
@@ -443,10 +433,7 @@ async fn reasoning_effort_override_websocket_prewarm_preserves_baseline(
.map(String::as_str),
Some("previous turn"),
);
if matches!(
startup,
PrewarmStartup::Fork | PrewarmStartup::ForkThenRollback
) {
if matches!(startup, PrewarmStartup::Fork) {
previous.codex.shutdown_and_wait().await?;
let mut config = previous.config.clone();
configure_prewarm(&mut config);
@@ -475,23 +462,6 @@ async fn reasoning_effort_override_websocket_prewarm_preserves_baseline(
.await?;
assert_eq!(warmup.body_json()["generate"], false);
assert_eq!(warmup.body_json()["reasoning"]["effort"], "medium");
if matches!(
startup,
PrewarmStartup::ResumeThenRollback | PrewarmStartup::ForkThenRollback
) {
// Observing the warmup request guarantees that Medium is pinned before rollback.
test.codex
.submit(Op::ThreadRollback { num_turns: 1 })
.await?;
let rollback = wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::ThreadRolledBack(_) | EventMsg::Error(_))
})
.await;
assert!(
matches!(rollback, EventMsg::ThreadRolledBack(_)),
"rollback failed: {rollback:?}",
);
}
submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
@@ -765,124 +735,6 @@ async fn reasoning_effort_override_disabled_on_resume_retires_update_at_compacti
Ok(())
}
#[test_case(1, ReasoningEffort::High; "later turn keep changed selection")]
#[test_case(1, ReasoningEffort::Medium; "later turn match restored settings")]
#[test_case(2, ReasoningEffort::High; "through first turn")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reasoning_effort_override_rollback_reestablishes_selected_effort(
num_turns: u32,
effort: ReasoningEffort,
) -> anyhow::Result<()> {
skip_if_no_network!(Ok(()));
#[derive(Default)]
struct ThreadIdle {
ready: Notify,
}
impl ThreadLifecycleContributor<Config> for ThreadIdle {
fn on_thread_idle<'a>(&'a self, _input: ThreadIdleInput<'a>) -> ExtensionFuture<'a, ()> {
Box::pin(async move {
self.ready.notify_one();
})
}
}
let idle = Arc::new(ThreadIdle::default());
let mut extensions = ExtensionRegistryBuilder::new();
extensions.thread_lifecycle_contributor(idle.clone());
let server = responses::start_mock_server().await;
let first = responses::mount_sse_once(
&server,
responses::sse(vec![responses::ev_completed("first")]),
)
.await;
let removed = responses::mount_sse_once(
&server,
responses::sse(vec![responses::ev_completed("removed")]),
)
.await;
let replacement = responses::mount_sse_once(
&server,
responses::sse(vec![responses::ev_completed("replacement")]),
)
.await;
let test = override_builder()
.with_extensions(Arc::new(extensions.build()))
.build_with_auto_env(&server)
.await?;
test.submit_text_turn("first turn").await?;
timeout(Duration::from_secs(/*secs*/ 10), idle.ready.notified()).await?;
submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
effort: Some(Some(ReasoningEffort::High)),
..Default::default()
},
)
.await?;
test.submit_text_turn("removed turn").await?;
// Completion can be delivered before the active slot is cleared. Wait for
// actual idle so this test exercises effort replay, not rollback admission.
timeout(Duration::from_secs(/*secs*/ 10), idle.ready.notified()).await?;
test.codex.submit(Op::ThreadRollback { num_turns }).await?;
let rollback = wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::ThreadRolledBack(_) | EventMsg::Error(_))
})
.await;
let EventMsg::ThreadRolledBack(rollback) = rollback else {
panic!("rollback failed: {rollback:?}");
};
assert_eq!(rollback.num_turns, num_turns);
submit_thread_settings(
&test.codex,
ThreadSettingsOverrides {
effort: Some(Some(effort.clone())),
..Default::default()
},
)
.await?;
test.submit_text_turn("replacement turn").await?;
timeout(Duration::from_secs(/*secs*/ 10), idle.ready.notified()).await?;
assert_eq!(
effort_updates(&first.single_request()),
[effort_update(ReasoningEffort::Medium)]
);
assert_eq!(
effort_updates(&removed.single_request()),
[
effort_update(ReasoningEffort::Medium),
effort_update(ReasoningEffort::High)
]
);
let request = replacement.single_request();
assert_eq!(request.body_json()["reasoning"]["effort"], effort.as_str());
let mut expected = if num_turns == 1 {
vec![effort_update(ReasoningEffort::Medium)]
} else {
Vec::new()
};
expected.push(effort_update(effort.clone()));
assert_eq!(effort_updates(&request), expected);
let input = request.input();
assert!(
!input
.iter()
.any(|item| item["content"][0]["text"] == "removed turn")
);
let tail = responses::strip_response_item_ids_from_json(responses::strip_metadata_from_json(
Value::Array(input[input.len() - 2..].to_vec()),
));
assert_eq!(
tail,
serde_json::json!([
message("user", "replacement turn"),
effort_update(effort.clone()),
])
);
Ok(())
}
#[test_case(ReasoningEffort::Medium; "same selection")]
#[test_case(ReasoningEffort::High; "changed selection")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
codex-rs/core/tests/suite/rollout_budget.rs
-56
View File
@@ -1,14 +1,12 @@
use anyhow::Result;
use codex_core::TurnInputRequest;
use codex_core::config::RolloutBudgetConfig;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_features::Feature;
use codex_model_provider_info::built_in_model_providers;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::user_input::UserInput;
use core_test_support::ThreadIdle;
use core_test_support::responses::ResponsesRequest;
use core_test_support::responses::ev_assistant_message;
use core_test_support::responses::ev_completed;
@@ -24,7 +22,6 @@ use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use test_case::test_case;
use tokio::time::timeout;
@@ -468,56 +465,3 @@ async fn restates_the_current_remainder_after_compaction() -> Result<()> {
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn restates_the_current_remainder_after_rollback() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let responses = mount_sse_sequence(
&server,
vec![
sse(vec![
ev_response_created("resp-1"),
ev_completed_with_tokens("resp-1", /*total_tokens*/ 30),
]),
sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]),
],
)
.await;
let mut extensions = ExtensionRegistryBuilder::new();
extensions.thread_lifecycle_contributor(Arc::new(ThreadIdle));
let test = test_codex()
.with_extensions(Arc::new(extensions.build()))
.with_config(|config| {
config.rollout_budget = Some(RolloutBudgetConfig {
reminder_at_remaining_tokens: vec![50],
..rollout_budget()
});
})
.build(&server)
.await?;
test.submit_turn("rolled-back turn").await?;
ThreadIdle::wait(&test.codex).await;
test.codex
.submit(Op::ThreadRollback { num_turns: 1 })
.await?;
wait_for_event(&test.codex, |event| {
if let EventMsg::Error(error) = event {
panic!("rollback failed: {error:?}");
}
matches!(event, EventMsg::ThreadRolledBack(_))
})
.await;
test.submit_turn("turn after rollback").await?;
let requests = responses.requests();
assert_eq!(
rollout_budget_texts(&requests[1]),
vec![rollout_budget_message(/*remaining_tokens*/ 70)],
"rollback should rearm the current budget reminder without refunding usage"
);
Ok(())
}
codex-rs/core/tests/suite/snapshots/all__suite__compact_resume_fork__rollback_followup_turn_trims_context_updates.snap
-23
View File
@@ -1,23 +0,0 @@
---
source: core/tests/suite/compact_resume_fork.rs
expression: "context_snapshot::format_labeled_requests_snapshot(\"rollback trims pre-turn override context updates before the follow-up request\",\n&[(\"rolled-back turn request\", &requests[1]),\n(\"follow-up request after rollback\", &requests[2]),],\n&ContextSnapshotOptions::default().strip_capability_instructions().render_mode(ContextSnapshotRenderMode::KindWithTextPrefix\n{ max_chars: 96 }),)"
---
Scenario: rollback trims pre-turn override context updates before the follow-up request
## rolled-back turn request
00:message/developer:<PERMISSIONS_INSTRUCTIONS>
01:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
02:message/user:turn 1 user
03:message/assistant:turn 1 assistant
04:message/developer:<collaboration_mode>ROLLED_BACK_DEV_INSTRUCTIONS</collaboration_mode>
05:message/user:<ENVIRONMENT_CONTEXT:cwd=PRETURN_CONTEXT_DIFF_CWD>
06:message/user:turn 2 user
## follow-up request after rollback
00:message/developer:<PERMISSIONS_INSTRUCTIONS>
01:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
02:message/user:turn 1 user
03:message/assistant:turn 1 assistant
04:message/developer:<collaboration_mode>ROLLED_BACK_DEV_INSTRUCTIONS</collaboration_mode>
05:message/user:<ENVIRONMENT_CONTEXT:cwd=PRETURN_CONTEXT_DIFF_CWD>
06:message/user:follow-up user
codex-rs/core/tests/suite/snapshots/all__suite__compact_resume_fork__rollback_past_compaction_shapes.snap
-26
View File
@@ -1,26 +0,0 @@
---
source: core/tests/suite/compact_resume_fork.rs
expression: "context_snapshot::format_labeled_requests_snapshot(\"rollback past compaction replay after rollback\",\n&[(\"compaction request\", &requests[1]), (\"before rollback\", &requests[2]),\n(\"after rollback\", &requests[3]),],\n&ContextSnapshotOptions::default().strip_capability_instructions().render_mode(ContextSnapshotRenderMode::KindWithTextPrefix\n{ max_chars: 64 }),)"
---
Scenario: rollback past compaction replay after rollback
## compaction request
00:message/developer:<PERMISSIONS_INSTRUCTIONS>
01:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
02:message/user:hello world
03:message/assistant:FIRST_REPLY
04:message/user:<SUMMARIZATION_PROMPT>
## before rollback
00:message/user:hello world
01:message/user:<COMPACTION_SUMMARY>\nSUMMARY_ONLY_CONTEXT
02:message/developer:<PERMISSIONS_INSTRUCTIONS>
03:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
04:message/user:EDITED_AFTER_COMPACT
## after rollback
00:message/user:hello world
01:message/user:<COMPACTION_SUMMARY>\nSUMMARY_ONLY_CONTEXT
02:message/developer:<PERMISSIONS_INSTRUCTIONS>
03:message/user:<ENVIRONMENT_CONTEXT:cwd=<CWD>>
04:message/user:AFTER_ROLLBACK
codex-rs/protocol/src/protocol.rs
+3 -8
View File
@@ -743,12 +743,6 @@ pub enum Op {
/// model.
SetThreadMemoryMode { mode: ThreadMemoryMode },
/// Request Codex to drop the last N user turns from in-memory context.
///
/// This does not attempt to revert local filesystem changes. Clients are
/// responsible for undoing any edits on disk.
ThreadRollback { num_turns: u32 },
/// Request a code review from the agent.
Review { review_request: ReviewRequest },
@@ -962,7 +956,6 @@ impl Op {
Self::ReloadUserConfig => "reload_user_config",
Self::Compact => "compact",
Self::SetThreadMemoryMode { .. } => "set_thread_memory_mode",
Self::ThreadRollback { .. } => "thread_rollback",
Self::Review { .. } => "review",
Self::ApproveGuardianDeniedAction { .. } => "approve_guardian_denied_action",
Self::Shutdown => "shutdown",
@@ -1405,7 +1398,8 @@ pub enum EventMsg {
/// Conversation history was compacted (either automatically or manually).
ContextCompacted(ContextCompactedEvent),
/// Conversation history was rolled back by dropping the last N user turns.
/// Legacy persisted marker for dropping the last N user turns.
/// Retained for replay of existing rollouts; live rollback operations are unsupported.
ThreadRolledBack(ThreadRolledBackEvent),
/// Agent has started a turn.
@@ -1888,6 +1882,7 @@ pub enum CodexErrorInfo {
ActiveTurnNotSteerable {
turn_kind: NonSteerableTurnKind,
},
// Retained to deserialize errors recorded in legacy rollouts.
ThreadRollbackFailed,
Other,
}
codex-rs/rollout/src/model_context.rs
+1 -1
View File
@@ -95,7 +95,7 @@ impl ModelContextScan {
self.active_segment.saw_compaction = true;
}
RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => {
// Paginated threads reject rollback. Keep old rollouts correct rather than
// Rollback markers only occur in legacy history. Keep old rollouts correct rather than
// duplicating rollback survival semantics in this bounded selector.
self.must_scan_to_start = true;
}
codex-rs/thread-store/src/store.rs
+1 -1
View File
@@ -144,7 +144,7 @@ pub trait ThreadStore: Any + Send + Sync {
/// already-durable thread data.
fn discard_thread(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()>;
/// Loads persisted history for resume, fork, rollback, and memory jobs.
/// Loads persisted history for resume, fork, and memory jobs.
fn load_history(
&self,
params: LoadThreadHistoryParams,
codex-rs/thread-store/src/types.rs
+1 -1
View File
@@ -153,7 +153,7 @@ pub struct AppendThreadItemsParams {
pub items: Vec<RolloutItem>,
}
/// Parameters for loading persisted history for resume, fork, rollback, and memory jobs.
/// Parameters for loading persisted history for resume, fork, and memory jobs.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct LoadThreadHistoryParams {
/// Thread id to load.
codex-rs/tui/src/app.rs
+1 -2
View File
@@ -607,8 +607,7 @@ pub(crate) struct App {
pub(crate) backtrack: crate::app_backtrack::BacktrackState,
/// When set, the next draw rebuilds terminal scrollback from the retained transcript cells.
///
/// This is used after a confirmed thread rollback to ensure scrollback reflects the trimmed
/// transcript cells.
/// This keeps scrollback consistent with the retained transcript after backtracking.
pub(crate) backtrack_render_pending: bool,
pub(crate) feedback: codex_feedback::CodexFeedback,
feedback_audience: FeedbackAudience,
codex-rs/tui/src/pager_overlay.rs
+1 -1
View File
@@ -695,7 +695,7 @@ impl TranscriptOverlay {
/// Replace committed transcript cells while keeping any cached in-progress output that is
/// currently shown at the end of the overlay.
///
/// This is used when existing history is trimmed (for example after rollback) so the
/// This is used when existing history is replaced or trimmed so the
/// transcript overlay immediately reflects the same committed cells as the main transcript.
pub(crate) fn replace_cells(&mut self, cells: Vec<Arc<dyn HistoryCell>>) {
let follow_bottom = self.view.is_scrolled_to_bottom();
sdk/python/src/openai_codex/generated/v2_all.py
+1 -39
View File
@@ -5833,21 +5833,6 @@ class ThreadRevertedNotification(BaseModel):
thread_id: Annotated[str, Field(alias="threadId")]
class ThreadRollbackParams(BaseModel):
model_config = ConfigDict(
populate_by_name=True,
)
num_turns: Annotated[
int,
Field(
alias="numTurns",
description="The number of turns to drop from the end of the thread. Must be >= 1.\n\nThis only modifies the thread's history and does not revert local file changes that have been made by the agent. Clients are responsible for reverting these changes.",
ge=0,
),
]
thread_id: Annotated[str, Field(alias="threadId")]
class ThreadSearchSortKey(Enum):
created_at = "created_at"
updated_at = "updated_at"
@@ -6815,15 +6800,6 @@ class ThreadApproveGuardianDeniedActionRequest(BaseModel):
params: ThreadApproveGuardianDeniedActionParams
class ThreadRollbackRequest(BaseModel):
model_config = ConfigDict(
populate_by_name=True,
)
id: RequestId
method: Annotated[Literal["thread/rollback"], Field(title="Thread/rollbackRequestMethod")]
params: ThreadRollbackParams
class ThreadRevertRequest(BaseModel):
model_config = ConfigDict(
populate_by_name=True,
@@ -11767,7 +11743,7 @@ class Thread(BaseModel):
turns: Annotated[
list[Turn],
Field(
description="Only populated on `thread/resume`, `thread/rollback`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list."
description="Only populated on `thread/resume`, `thread/fork`, and `thread/read` (when `includeTurns` is true) responses. For all other responses and notifications returning a Thread, the turns field will be an empty list."
),
]
updated_at: Annotated[
@@ -11934,18 +11910,6 @@ class ThreadRevertResponse(BaseModel):
] = None
class ThreadRollbackResponse(BaseModel):
model_config = ConfigDict(
populate_by_name=True,
)
thread: Annotated[
Thread,
Field(
description="The updated thread after applying the rollback, with `turns` populated.\n\nThe ThreadItems stored in each Turn are lossy since we explicitly do not persist all agent interactions, such as command executions. This is the same behavior as `thread/resume`."
),
]
class ThreadSearchResult(BaseModel):
model_config = ConfigDict(
populate_by_name=True,
@@ -12169,7 +12133,6 @@ class ClientRequest(
| ThreadCompactStartRequest
| ThreadShellCommandRequest
| ThreadApproveGuardianDeniedActionRequest
| ThreadRollbackRequest
| ThreadRevertRequest
| ThreadListRequest
| ThreadSectionListRequest
@@ -12277,7 +12240,6 @@ class ClientRequest(
| ThreadCompactStartRequest
| ThreadShellCommandRequest
| ThreadApproveGuardianDeniedActionRequest
| ThreadRollbackRequest
| ThreadRevertRequest
| ThreadListRequest
| ThreadSectionListRequest