openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 17:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
7827cd886db8d129e06bc843185d3d2b0d5cef52
codex/codex-rs/app-server/tests/suite/v2/thread_fork.rs
T
jif 1e4ba692b8 Preserve thread metadata before rollout files are materialized (#55380) 
## Why

A live thread can have updated SQLite metadata before its rollout file exists.
Reading metadata from the missing file loses creation details such as fork and
parent relationships and triggers unnecessary filesystem discovery and retries.

## What changed

- Read deferred `SessionMeta` through the rollout writer, ordered with persistence.
  Return no in-memory metadata once the rollout is materialized or resumed.
- Combine that creation metadata with SQLite metadata for reads that exclude
  history, and use it to determine the canonical history mode during updates.
- Keep untouched threads without a stored record unavailable for fork and resume.

## Testing

Add coverage for metadata reads and updates before persistence in both legacy and
paginated history modes, including equality with metadata read after persistence.
Verify the recorder exposes creation metadata only while deferred, and expand
fork/resume rejection tests across history modes and resume turn inclusion.

GitOrigin-RevId: 8d293cc8f87156409f195130437d93858a4099b9
2026-10-19 21:02:19 +00:00

3089 lines
106 KiB
Rust
Raw Blame History

use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::MockResponsesConfig;
use app_test_support::TestAppServer;
use app_test_support::create_fake_paginated_rollout;
use app_test_support::create_fake_rollout;
use app_test_support::create_fake_rollout_with_token_usage;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use app_test_support::rollout_path;
use app_test_support::to_response;
use app_test_support::write_chatgpt_auth;
use app_test_support::write_models_cache;
use codex_app_server_protocol::ActivePermissionProfile;
use codex_app_server_protocol::ApprovalsReviewer;
use codex_app_server_protocol::AskForApproval;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::DeprecationNoticeNotification;
use codex_app_server_protocol::DynamicToolFunctionSpec;
use codex_app_server_protocol::DynamicToolSpec;
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::SandboxMode;
use codex_app_server_protocol::SandboxPolicy;
use codex_app_server_protocol::ServerNotification;
use codex_app_server_protocol::SessionSource;
use codex_app_server_protocol::ThreadAttachmentAddParams;
use codex_app_server_protocol::ThreadAttachmentAddResponse;
use codex_app_server_protocol::ThreadAttachmentListParams;
use codex_app_server_protocol::ThreadAttachmentListResponse;
use codex_app_server_protocol::ThreadAttachmentRemoveParams;
use codex_app_server_protocol::ThreadAttachmentRemoveResponse;
use codex_app_server_protocol::ThreadClosedNotification;
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::ThreadListParams;
use codex_app_server_protocol::ThreadListResponse;
use codex_app_server_protocol::ThreadReadParams;
use codex_app_server_protocol::ThreadReadResponse;
use codex_app_server_protocol::ThreadResumeParams;
use codex_app_server_protocol::ThreadResumeResponse;
use codex_app_server_protocol::ThreadSearchOccurrencesParams;
use codex_app_server_protocol::ThreadSearchOccurrencesResponse;
use codex_app_server_protocol::ThreadSource;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStartedNotification;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadStatusChangedNotification;
use codex_app_server_protocol::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadTurnsListResponse;
use codex_app_server_protocol::ThreadUnsubscribeParams;
use codex_app_server_protocol::ThreadUnsubscribeResponse;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStatus;
use codex_app_server_protocol::UserInput;
use codex_config::types::AuthCredentialsStoreMode;
use codex_core::test_support::all_model_presets;
use codex_features::Feature;
use codex_login::REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR;
use codex_protocol::ThreadId;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::items::TurnItem as CoreTurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::MultiAgentVersion;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::UserMessageEvent;
use codex_rollout::RolloutItem;
use codex_rollout::RolloutLine;
use codex_rollout::append_rollout_item_to_path;
use codex_rollout::append_thread_name;
use codex_rollout::read_session_meta_line;
use codex_state::StateRuntime;
use codex_utils_absolute_path::test_support::PathExt;
use core_test_support::responses;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use tempfile::TempDir;
use test_case::test_case;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::mount_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
#[cfg(not(windows))]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
async fn list_threads(mcp: &mut TestAppServer) -> Result<ThreadListResponse> {
let list_id = mcp
.send_thread_list_request(ThreadListParams {
excluded_thread_ids: None,
originators: None,
cursor: None,
limit: Some(50),
sort_key: None,
sort_direction: None,
model_providers: None,
source_kinds: None,
archived: None,
section_id: None,
project_id: None,
cwd: None,
use_state_db_only: false,
search_term: None,
parent_thread_id: None,
ancestor_thread_id: None,
})
.await?;
let list_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(list_id)),
)
.await??;
to_response::<ThreadListResponse>(list_resp)
}
#[tokio::test]
async fn thread_fork_creates_new_thread_and_emits_started() -> 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 preview = "Saved user message";
let conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
preview,
Some("mock_provider"),
/*git_info*/ None,
)?;
let original_path = codex_home
.path()
.join("sessions")
.join("2025")
.join("01")
.join("05")
.join(format!(
"rollout-2025-01-05T12-00-00-{conversation_id}.jsonl"
));
assert!(
original_path.exists(),
"expected original rollout to exist at {}",
original_path.display()
);
let mut session_meta = read_session_meta_line(&original_path).await?;
session_meta.meta.multi_agent_version = Some(MultiAgentVersion::V1);
append_rollout_item_to_path(&original_path, &RolloutItem::SessionMeta(session_meta)).await?;
let original_contents = std::fs::read_to_string(&original_path)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
thread_source: Some(ThreadSource::User),
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let fork_result = fork_resp.result.clone();
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
// Wire contract: thread title field is `name`, serialized as null when unset.
let thread_json = fork_result
.get("thread")
.and_then(Value::as_object)
.expect("thread/fork result.thread must be an object");
assert_eq!(
thread_json.get("sessionId").and_then(Value::as_str),
Some(thread.session_id.as_str()),
"forked threads should serialize `sessionId` on the thread object"
);
assert_eq!(
thread_json.get("name"),
Some(&Value::Null),
"forked threads do not inherit a name; expected `name: null`"
);
assert_eq!(
fork_result.get("sessionId"),
None,
"thread/fork should not serialize a top-level `sessionId`"
);
let after_contents = std::fs::read_to_string(&original_path)?;
assert_eq!(
after_contents, original_contents,
"fork should not mutate the original rollout file"
);
assert_ne!(thread.id, conversation_id);
assert_eq!(thread.session_id, thread.id);
assert_eq!(thread.forked_from_id, Some(conversation_id.clone()));
assert_eq!(thread.preview, preview);
assert_eq!(thread.model_provider, "mock_provider");
assert_eq!(thread.status, ThreadStatus::Idle);
let thread_path = thread.path.clone().expect("thread path");
assert!(thread_path.as_path().is_absolute());
assert_ne!(thread_path.as_path(), original_path);
assert!(thread.cwd.as_path().is_absolute());
assert_eq!(thread.source, SessionSource::VsCode);
assert_eq!(thread.thread_source, Some(ThreadSource::User));
assert_eq!(thread.name, None);
assert_eq!(
thread.turns.len(),
1,
"expected forked thread to include one turn"
);
let turn = &thread.turns[0];
assert_eq!(turn.status, TurnStatus::Interrupted);
assert_eq!(turn.items.len(), 1, "expected user message item");
match &turn.items[0] {
ThreadItem::UserMessage { content, .. } => {
assert_eq!(
content,
&vec![UserInput::Text {
text: preview.to_string(),
text_elements: Vec::new(),
}]
);
}
other => panic!("expected user message item, got {other:?}"),
}
// A corresponding thread/started notification should arrive.
let deadline = tokio::time::Instant::now() + DEFAULT_READ_TIMEOUT;
let notif = loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
let message = timeout(remaining, mcp.read_next_message()).await??;
let JSONRPCMessage::Notification(notif) = message else {
continue;
};
if notif.method == "thread/status/changed" {
let status_changed: ThreadStatusChangedNotification =
serde_json::from_value(notif.params.expect("params must be present"))?;
if status_changed.thread_id == thread.id {
anyhow::bail!(
"thread/fork should introduce the thread without a preceding thread/status/changed"
);
}
continue;
}
if notif.method == "thread/started" {
break notif;
}
};
let started_params = notif.params.clone().expect("params must be present");
let started_thread_json = started_params
.get("thread")
.and_then(Value::as_object)
.expect("thread/started params.thread must be an object");
assert_eq!(
started_thread_json.get("name"),
Some(&Value::Null),
"thread/started must serialize `name: null` when unset"
);
assert_eq!(
started_thread_json.get("turns"),
Some(&json!([])),
"thread/started must not emit copied fork turns"
);
assert_eq!(
started_thread_json
.get("threadSource")
.and_then(Value::as_str),
Some("user"),
"thread/started should preserve the caller-supplied fork origin"
);
let started: ThreadStartedNotification =
serde_json::from_value(notif.params.expect("params must be present"))?;
let mut expected_started_thread = thread;
expected_started_thread.turns.clear();
assert_eq!(started.thread, expected_started_thread);
Ok(())
}
#[tokio::test]
async fn thread_fork_preserves_persisted_approvals_reviewer() -> Result<()> {
assert_thread_fork_preserves_persisted_approvals_reviewer(ThreadHistoryMode::Legacy).await
}
#[tokio::test]
async fn paginated_thread_fork_preserves_persisted_approvals_reviewer() -> Result<()> {
assert_thread_fork_preserves_persisted_approvals_reviewer(ThreadHistoryMode::Paginated).await
}
#[tokio::test]
async fn thread_fork_preserves_persisted_permission_profile_and_honors_overrides() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] {
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri())
.with_root_config("default_permissions = \":danger-full-access\"")
.with_extra_config("[permissions.dev]\nextends = \":read-only\"")
.write(codex_home.path())?;
let source_thread_id = {
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_managed_config()
.build_initialized()
.await?;
let start_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams {
history_mode: Some(history_mode),
approval_policy: Some(AskForApproval::OnRequest),
permissions: Some("dev".to_string()),
..Default::default()
})
.await?;
let ThreadStartResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(start_id)).await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "persist permission profile".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
}),
)
.await??;
thread.id
};
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_managed_config()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id.clone(),
..Default::default()
})
.await?;
let ThreadForkResponse {
approval_policy,
sandbox,
active_permission_profile,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert!(matches!(sandbox, SandboxPolicy::ReadOnly { .. }));
assert_eq!(approval_policy, AskForApproval::OnRequest);
assert_eq!(
active_permission_profile,
Some(ActivePermissionProfile {
id: "dev".to_string(),
extends: Some(":read-only".to_string()),
})
);
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id.clone(),
approval_policy: Some(AskForApproval::Never),
sandbox: Some(SandboxMode::DangerFullAccess),
..Default::default()
})
.await?;
let ThreadForkResponse {
approval_policy,
sandbox,
active_permission_profile,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert!(matches!(sandbox, SandboxPolicy::DangerFullAccess));
assert_eq!(approval_policy, AskForApproval::Never);
assert_eq!(active_permission_profile, None);
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id,
permissions: Some(":workspace".to_string()),
..Default::default()
})
.await?;
let ThreadForkResponse {
sandbox,
active_permission_profile,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert!(matches!(sandbox, SandboxPolicy::WorkspaceWrite { .. }));
assert_eq!(
active_permission_profile,
Some(ActivePermissionProfile::new(":workspace"))
);
}
Ok(())
}
async fn assert_thread_fork_preserves_persisted_approvals_reviewer(
history_mode: ThreadHistoryMode,
) -> 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 (source_thread_id, source_turn_id) = {
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let start_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams {
history_mode: Some(history_mode),
permissions: Some(":workspace".to_string()),
..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 turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "materialize this thread".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(turn_id)),
)
.await??;
let TurnStartResponse { turn } = to_response(turn_resp)?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let second_turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
input: vec![UserInput::Text {
text: "switch to auto-review".to_string(),
text_elements: Vec::new(),
}],
approval_policy: Some(AskForApproval::OnRequest),
approvals_reviewer: Some(ApprovalsReviewer::AutoReview),
permissions: Some(":read-only".to_string()),
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(second_turn_id)),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
if matches!(history_mode, ThreadHistoryMode::Paginated) {
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: thread.id.clone(),
last_turn_id: Some(turn.id.clone()),
..Default::default()
})
.await?;
let ThreadForkResponse {
approval_policy,
approvals_reviewer,
active_permission_profile,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(approval_policy, AskForApproval::OnRequest);
assert_eq!(approvals_reviewer, ApprovalsReviewer::AutoReview);
assert_eq!(
active_permission_profile,
Some(ActivePermissionProfile::new(":read-only"))
);
}
(thread.id, turn.id)
};
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id.clone(),
last_turn_id: Some(source_turn_id.clone()),
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse {
approval_policy,
approvals_reviewer,
active_permission_profile,
..
} = to_response(fork_resp)?;
assert_eq!(approval_policy, AskForApproval::OnRequest);
assert_eq!(approvals_reviewer, ApprovalsReviewer::AutoReview);
assert_eq!(
active_permission_profile,
Some(ActivePermissionProfile::new(":read-only"))
);
if matches!(history_mode, ThreadHistoryMode::Paginated) {
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id,
last_turn_id: Some(source_turn_id),
approval_policy: Some(AskForApproval::Never),
approvals_reviewer: Some(ApprovalsReviewer::User),
..Default::default()
})
.await?;
let ThreadForkResponse {
approval_policy,
approvals_reviewer,
active_permission_profile,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(approval_policy, AskForApproval::Never);
assert_eq!(approvals_reviewer, ApprovalsReviewer::User);
assert_eq!(
active_permission_profile,
Some(ActivePermissionProfile::new(":read-only"))
);
}
Ok(())
}
#[tokio::test]
async fn thread_fork_at_last_turn_id_keeps_only_terminal_prefix() -> Result<()> {
assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix(ThreadHistoryMode::Legacy).await
}
#[tokio::test]
async fn paginated_thread_fork_at_named_boundaries_keeps_only_terminal_prefix() -> Result<()> {
assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix(ThreadHistoryMode::Paginated)
.await
}
async fn assert_thread_fork_at_named_boundary_keeps_only_terminal_prefix(
history_mode: ThreadHistoryMode,
) -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let start_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams {
history_mode: Some(history_mode),
..Default::default()
})
.await?;
let ThreadStartResponse {
thread: source_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(start_id)).await??;
let source_thread_id = source_thread.id.clone();
let source_path = source_thread.path.expect("source thread path");
let mut turn_ids = Vec::new();
for text in ["first", "second", "third"] {
let turn_request_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: source_thread_id.clone(),
client_user_message_id: None,
input: vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let TurnStartResponse { turn } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(turn_request_id)).await??;
turn_ids.push(turn.id);
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
}
let original_contents = std::fs::read_to_string(source_path.as_path())?;
// Membership added after the requested history cutoff must still be inherited.
let attachment: ThreadAttachmentAddResponse = mcp
.request(|request_id| ClientRequest::ThreadAttachmentAdd {
request_id,
params: ThreadAttachmentAddParams {
thread_id: source_thread_id.clone(),
attachment_type: "pull_request".to_string(),
identity_key: "openai/codex#123".to_string(),
payload: json!({"url": "https://github.com/openai/codex/pull/123"}),
},
})
.await?;
let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?;
sqlx::query(
"CREATE TRIGGER fail_fork_attachment_copy BEFORE INSERT ON thread_attachments \
BEGIN SELECT RAISE(FAIL, 'attachment copy failure'); END",
)
.execute(&pool)
.await?;
let fork_without_attachments: ThreadForkResponse = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: source_thread_id.clone(),
last_turn_id: Some(turn_ids[1].clone()),
..Default::default()
},
})
.await?;
assert_eq!(
fork_without_attachments
.thread
.turns
.iter()
.map(|turn| turn.id.clone())
.collect::<Vec<_>>(),
turn_ids[..2]
);
assert!(
fork_without_attachments
.thread
.path
.as_ref()
.expect("durable fork path")
.exists()
);
let without_attachments: ThreadAttachmentListResponse = mcp
.request(|request_id| ClientRequest::ThreadAttachmentList {
request_id,
params: ThreadAttachmentListParams {
thread_id: fork_without_attachments.thread.id.clone(),
cursor: None,
limit: None,
},
})
.await?;
assert!(without_attachments.data.is_empty());
assert_eq!(
sqlx::query_scalar::<_, String>("SELECT id FROM threads WHERE id = ?")
.bind(&fork_without_attachments.thread.id)
.fetch_one(&pool)
.await?,
fork_without_attachments.thread.id
);
sqlx::query("DROP TRIGGER fail_fork_attachment_copy")
.execute(&pool)
.await?;
pool.close().await;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id.clone(),
last_turn_id: Some(turn_ids[1].clone()),
..Default::default()
})
.await?;
let ThreadForkResponse {
thread: forked_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
let started = loop {
let notification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/started"),
)
.await??;
let started: ThreadStartedNotification =
serde_json::from_value(notification.params.expect("params must be present"))?;
if started.thread.id == forked_thread.id {
break started;
}
};
assert!(started.thread.turns.is_empty());
assert_eq!(
started.thread.forked_from_id,
Some(source_thread_id.clone())
);
let copied: ThreadAttachmentListResponse = mcp
.request(|request_id| ClientRequest::ThreadAttachmentList {
request_id,
params: ThreadAttachmentListParams {
thread_id: started.thread.id,
cursor: None,
limit: None,
},
})
.await?;
assert_eq!(copied.data.len(), 1);
assert_ne!(copied.data[0].id, attachment.attachment.id);
assert_eq!(
copied.data[0],
codex_app_server_protocol::ThreadAttachment {
id: copied.data[0].id.clone(),
created_at: copied.data[0].created_at,
..attachment.attachment.clone()
}
);
let _: ThreadAttachmentRemoveResponse = mcp
.request(|request_id| ClientRequest::ThreadAttachmentRemove {
request_id,
params: ThreadAttachmentRemoveParams {
thread_id: forked_thread.id.clone(),
attachment_type: attachment.attachment.attachment_type.clone(),
identity_key: attachment.attachment.identity_key.clone(),
},
})
.await?;
let parent_attachments: ThreadAttachmentListResponse = mcp
.request(|request_id| ClientRequest::ThreadAttachmentList {
request_id,
params: ThreadAttachmentListParams {
thread_id: source_thread_id.clone(),
cursor: None,
limit: None,
},
})
.await?;
assert_eq!(parent_attachments.data, vec![attachment.attachment]);
assert_eq!(
forked_thread
.turns
.iter()
.map(|turn| turn.id.clone())
.collect::<Vec<_>>(),
turn_ids[..2]
);
assert!(
forked_thread
.turns
.iter()
.all(|turn| turn.status == TurnStatus::Completed)
);
assert_eq!(forked_thread.forked_from_id, Some(source_thread_id.clone()));
if history_mode == ThreadHistoryMode::Legacy {
assert_eq!(forked_thread.preview, "first");
}
assert_eq!(
std::fs::read_to_string(source_path.as_path())?,
original_contents,
"forking at a turn must not mutate the source rollout"
);
let forked_path = forked_thread.path.clone().expect("forked thread path");
let forked_contents = std::fs::read_to_string(forked_path.as_path())?;
if history_mode == ThreadHistoryMode::Paginated {
assert!(
read_session_meta_line(forked_path.as_path())
.await?
.meta
.history_base
.is_some()
);
assert!(!forked_contents.contains(turn_ids[1].as_str()));
} else {
assert!(forked_contents.contains(turn_ids[1].as_str()));
}
assert!(!forked_contents.contains(turn_ids[2].as_str()));
if history_mode == ThreadHistoryMode::Paginated {
let before_fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id,
before_turn_id: Some(turn_ids[2].clone()),
..Default::default()
})
.await?;
let ThreadForkResponse {
thread: before_fork,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(before_fork_id)).await??;
assert_eq!(
before_fork
.turns
.iter()
.map(|turn| turn.id.clone())
.collect::<Vec<_>>(),
turn_ids[..2]
);
let completed = timeout(
DEFAULT_READ_TIMEOUT,
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: forked_thread.id.clone(),
input: vec![UserInput::Text {
text: "private child prompt".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
}),
)
.await??;
let ThreadForkResponse {
thread: ephemeral_fork,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: forked_thread.id.clone(),
before_turn_id: Some(completed.turn.id),
ephemeral: true,
exclude_turns: true,
..Default::default()
},
})
.await?;
assert_eq!(ephemeral_fork.preview, "first");
}
// Resuming a fork must not copy its parent's attachments again after an explicit removal.
mcp.shutdown_gracefully().await?;
let mut resumed = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let _: ThreadResumeResponse = resumed
.request(|request_id| ClientRequest::ThreadResume {
request_id,
params: ThreadResumeParams {
thread_id: forked_thread.id.clone(),
..Default::default()
},
})
.await?;
let after_resume: ThreadAttachmentListResponse = resumed
.request(|request_id| ClientRequest::ThreadAttachmentList {
request_id,
params: ThreadAttachmentListParams {
thread_id: forked_thread.id,
cursor: None,
limit: None,
},
})
.await?;
assert!(after_resume.data.is_empty());
Ok(())
}
#[tokio::test]
async fn thread_fork_defers_inherited_active_goal_until_next_turn() -> Result<()> {
let server = create_mock_responses_server_sequence_unchecked(vec![
responses::sse(vec![
responses::ev_response_created("first-source-turn"),
responses::ev_completed("first-source-turn"),
]),
responses::sse(vec![
responses::ev_response_created("second-source-turn"),
responses::ev_completed("second-source-turn"),
]),
responses::sse(vec![
responses::ev_response_created("explicit-fork-turn"),
responses::ev_completed_with_tokens("explicit-fork-turn", /*total_tokens*/ 20),
]),
responses::sse(vec![
responses::ev_response_created("goal-continuation"),
responses::ev_completed_with_tokens("goal-continuation", /*total_tokens*/ 100),
]),
])
.await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let config_path = codex_home.path().join("config.toml");
let config = std::fs::read_to_string(&config_path)?;
std::fs::write(
&config_path,
format!("{config}\n[features]\ngoals = true\n"),
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_managed_config()
.build_initialized()
.await?;
let start_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams::default())
.await?;
let ThreadStartResponse {
thread: source_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(start_id)).await??;
let source_thread_id = ThreadId::from_string(&source_thread.id)?;
let mut turn_ids = Vec::new();
for text in ["first", "second"] {
let completed = timeout(
DEFAULT_READ_TIMEOUT,
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: source_thread.id.clone(),
input: vec![UserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}],
..Default::default()
}),
)
.await??;
turn_ids.push(completed.turn.id);
}
// Stop the source before its active goal exists so a late idle hook cannot continue it.
timeout(DEFAULT_READ_TIMEOUT, mcp.shutdown_gracefully()).await??;
drop(mcp);
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_managed_config()
.build_initialized()
.await?;
let state_db = StateRuntime::init(
codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()),
"mock_provider".into(),
)
.await?;
let source_goal = state_db
.thread_goals()
.replace_thread_goal(
source_thread_id,
"continue after the retry",
codex_state::ThreadGoalStatus::Active,
/*token_budget*/ Some(150),
)
.await?;
state_db
.thread_goals()
.account_thread_goal_usage(
source_thread_id,
/*time_delta_seconds*/ 11,
/*token_delta*/ 37,
codex_state::GoalAccountingMode::ActiveOnly,
Some(source_goal.goal_id.as_str()),
)
.await?;
let source_goal = state_db
.thread_goals()
.get_thread_goal(source_thread_id)
.await?
.expect("source goal");
let mut forked_threads = Vec::new();
for (last_turn_id, before_turn_id, expected_turn_count) in [
(None, None, 2),
(Some(turn_ids[0].clone()), None, 1),
(None, Some(turn_ids[0].clone()), 0),
] {
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread.id.clone(),
last_turn_id,
before_turn_id,
defer_goal_continuation: true,
..Default::default()
})
.await?;
let ThreadForkResponse {
thread: forked_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
let forked_thread_id = ThreadId::from_string(&forked_thread.id)?;
assert_eq!(forked_thread.turns.len(), expected_turn_count);
let mut expected_goal = source_goal.clone();
expected_goal.thread_id = forked_thread_id;
assert_eq!(
state_db
.thread_goals()
.get_thread_goal(forked_thread_id)
.await?,
Some(expected_goal)
);
assert!(
state_db
.thread_goals()
.has_thread_goal_continuation_deferral(forked_thread_id)
.await?
);
forked_threads.push(forked_thread);
}
assert_eq!(
state_db
.thread_goals()
.get_thread_goal(source_thread_id)
.await?,
Some(source_goal.clone())
);
assert!(
!mcp.pending_notification_methods()
.iter()
.any(|method| method == "turn/started"),
"deferred goal should not start a turn while forking"
);
assert_eq!(
server
.received_requests()
.await
.expect("wiremock requests")
.iter()
.filter(|request| request.url.path().ends_with("/responses"))
.count(),
2,
"deferred goal should not issue a model request while forking"
);
let forked_thread = forked_threads.pop().expect("empty-prefix fork");
let forked_thread_id = ThreadId::from_string(&forked_thread.id)?;
drop(mcp);
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_managed_config()
.build_initialized()
.await?;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: forked_thread.id.clone(),
..Default::default()
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
assert!(
!mcp.pending_notification_methods()
.iter()
.any(|method| method == "turn/started"),
"deferred goal should remain deferred after app-server restart"
);
timeout(
DEFAULT_READ_TIMEOUT,
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: forked_thread.id,
input: vec![UserInput::Text {
text: "retry the interrupted prompt".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
}),
)
.await??;
assert!(
!state_db
.thread_goals()
.has_thread_goal_continuation_deferral(forked_thread_id)
.await?,
"first explicit turn should consume the deferred-goal marker"
);
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/started"),
)
.await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let forked_goal = state_db
.thread_goals()
.get_thread_goal(forked_thread_id)
.await?
.expect("forked goal");
assert_eq!(forked_goal.goal_id, source_goal.goal_id);
assert_eq!(forked_goal.objective, source_goal.objective);
assert_eq!(forked_goal.token_budget, Some(150));
assert_eq!(forked_goal.tokens_used, 157);
assert!(forked_goal.time_used_seconds >= source_goal.time_used_seconds);
assert_eq!(
forked_goal.status,
codex_state::ThreadGoalStatus::BudgetLimited
);
assert_eq!(
state_db
.thread_goals()
.get_thread_goal(source_thread_id)
.await?,
Some(source_goal)
);
Ok(())
}
#[tokio::test]
async fn thread_fork_inherits_explicit_source_name_from_session_index() -> 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 conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let source_thread_id = ThreadId::from_string(&conversation_id)?;
let source_name = "Renamed parent thread";
append_thread_name(codex_home.path(), source_thread_id, source_name).await?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
let ThreadListResponse { data, .. } = list_threads(&mut mcp).await?;
let listed = data
.iter()
.find(|candidate| candidate.id == thread.id)
.expect("thread/list should include the forked thread");
assert_eq!(listed.name.as_deref(), Some(source_name));
Ok(())
}
#[tokio::test]
async fn thread_fork_can_load_source_by_path() -> 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 preview = "Saved user message";
let conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
preview,
Some("mock_provider"),
/*git_info*/ None,
)?;
let original_path = codex_home
.path()
.join("sessions")
.join("2025")
.join("01")
.join("05")
.join(format!(
"rollout-2025-01-05T12-00-00-{conversation_id}.jsonl"
));
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: "not-a-valid-thread-id".to_string(),
path: Some(original_path),
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_ne!(thread.id, conversation_id);
assert_eq!(thread.forked_from_id, Some(conversation_id));
assert_eq!(thread.preview, preview);
assert_eq!(thread.model_provider, "mock_provider");
assert_eq!(thread.turns.len(), 1, "expected copied fork history");
Ok(())
}
#[tokio::test]
async fn thread_fork_can_cut_before_unfinished_stored_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let filename_ts = "2025-01-05T12-00-00";
let conversation_id = create_fake_rollout(
codex_home.path(),
filename_ts,
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let source_path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
let unfinished_turn_id = "unfinished-turn";
append_rollout_item_to_path(
&source_path,
&RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_attribution: None,
turn_id: unfinished_turn_id.to_string(),
root_turn_id: None,
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
})),
)
.await?;
append_rollout_item_to_path(
&source_path,
&RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "Unfinished user message".to_string(),
..Default::default()
})),
)
.await?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let read_id = mcp
.send_thread_read_request(ThreadReadParams {
thread_id: conversation_id.clone(),
include_turns: true,
})
.await?;
let ThreadReadResponse {
thread: source_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(read_id)).await??;
assert_eq!(source_thread.turns.len(), 2);
assert_eq!(source_thread.turns[1].id, unfinished_turn_id);
assert_eq!(source_thread.turns[1].status, TurnStatus::Interrupted);
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
before_turn_id: Some(unfinished_turn_id.to_string()),
..Default::default()
})
.await?;
let ThreadForkResponse {
thread: forked_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(forked_thread.turns.len(), 1);
assert_eq!(forked_thread.preview, "Saved user message");
Ok(())
}
#[tokio::test]
async fn thread_fork_emits_restored_token_usage_before_next_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let conversation_id = create_fake_rollout_with_token_usage(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
thread_source: Some(ThreadSource::User),
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await??;
let parsed: ServerNotification = note.try_into()?;
let ServerNotification::ThreadTokenUsageUpdated(notification) = parsed else {
panic!("expected thread/tokenUsage/updated notification");
};
assert_eq!(notification.thread_id, thread.id);
assert_eq!(notification.turn_id, thread.turns[0].id);
assert_eq!(notification.token_usage.total.total_tokens, 150);
assert_eq!(notification.token_usage.total.input_tokens, 120);
assert_eq!(notification.token_usage.total.cached_input_tokens, 20);
assert_eq!(notification.token_usage.total.output_tokens, 30);
assert_eq!(notification.token_usage.total.reasoning_output_tokens, 10);
assert_eq!(notification.token_usage.last.total_tokens, 90);
assert_eq!(notification.token_usage.model_context_window, Some(200_000));
Ok(())
}
#[tokio::test]
async fn thread_fork_can_exclude_turns_and_skip_restored_token_usage() -> 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 conversation_id = create_fake_rollout_with_token_usage(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
exclude_turns: true,
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(thread.forked_from_id, Some(conversation_id));
assert_eq!(thread.preview, "Saved user message");
assert!(thread.turns.is_empty());
let note = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await;
assert!(
note.is_err(),
"excludeTurns=true should not replay token usage"
);
Ok(())
}
#[tokio::test]
async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri())
.with_root_config(&format!(r#"chatgpt_base_url = "{}""#, server.uri()))
.write(codex_home.path())?;
mount_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.without_managed_config()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
thread_source: Some(ThreadSource::User),
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(
event,
&thread.id,
&thread.session_id,
"codex",
"mock-model",
"forked",
"user",
);
assert_eq!(
event["event_params"]["forked_from_thread_id"],
thread
.forked_from_id
.as_deref()
.expect("forked thread has a source thread")
);
Ok(())
}
#[test_case(ThreadHistoryMode::Legacy; "legacy")]
#[test_case(ThreadHistoryMode::Paginated; "paginated")]
#[tokio::test]
async fn thread_fork_rejects_unmaterialized_thread(history_mode: ThreadHistoryMode) -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let start_id = mcp
.send_thread_start_request_with_auto_env(ThreadStartParams {
model: Some("mock-model".to_string()),
history_mode: Some(history_mode),
..Default::default()
})
.await?;
let ThreadStartResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(start_id)).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: thread.id,
..Default::default()
})
.await?;
let fork_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert!(
fork_err
.error
.message
.contains("no rollout found for thread id"),
"unexpected fork error: {}",
fork_err.error.message
);
Ok(())
}
#[tokio::test]
async fn thread_fork_creates_reference_backed_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 conversation_id = create_fake_paginated_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let source_path = rollout_path(
codex_home.path(),
"2025-01-05T12-00-00",
conversation_id.as_str(),
);
for item in [
RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_attribution: None,
turn_id: "turn-1".to_string(),
root_turn_id: None,
trace_id: None,
started_at: Some(10),
model_context_window: None,
collaboration_mode_kind: Default::default(),
})),
RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: "turn-1".to_string(),
last_agent_message: None,
error: None,
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
time_to_first_token_ms: None,
})),
] {
append_rollout_item_to_path(source_path.as_path(), &item).await?;
}
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
config: Some([("thread_unload_delay_secs".into(), json!(0))].into()),
..Default::default()
})
.await?;
let ThreadForkResponse {
thread: forked_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(forked_thread.forked_from_id, Some(conversation_id.clone()));
assert_eq!(forked_thread.turns.len(), 1);
let forked_thread_id = forked_thread.id.clone();
let forked_path = forked_thread.path.expect("forked rollout path");
assert!(!std::fs::read_to_string(forked_path.as_path())?.contains("Saved user message"));
let meta = read_session_meta_line(forked_path.as_path()).await?;
let history_base = meta.meta.history_base.expect("history base");
assert_eq!(
history_base.thread_id,
ThreadId::from_string(conversation_id.as_str())?
);
let turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: forked_thread_id.clone(),
input: vec![UserInput::Text {
text: "Continue from the fork".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(turn_id)).await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let requests = server.received_requests().await.expect("wiremock requests");
let response_request = requests
.iter()
.find(|request| request.url.path().ends_with("/responses"))
.expect("forked turn response request");
let request_body = response_request.body_json::<Value>()?;
let turn_metadata: Value = serde_json::from_str(
request_body["client_metadata"]["x-codex-turn-metadata"]
.as_str()
.expect("forked turn metadata"),
)?;
assert_eq!(
turn_metadata["forked_from_thread_id"].as_str(),
Some(conversation_id.as_str())
);
assert_eq!(
turn_metadata["forked_from_ordinal_exclusive"].as_u64(),
Some(history_base.end_ordinal_exclusive)
);
let model_input = request_body["input"]
.as_array()
.expect("response input array");
let model_input = serde_json::to_string(model_input)?;
assert!(model_input.contains("Saved user message"));
assert!(model_input.contains("Continue from the fork"));
// excludeTurns only controls response hydration; it must not change the inherited prefix.
let exclude_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
exclude_turns: true,
..Default::default()
})
.await?;
let ThreadForkResponse {
thread: excluded_turns_thread,
..
} = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(exclude_id)).await??;
assert!(excluded_turns_thread.turns.is_empty());
let excluded_turns_path = excluded_turns_thread.path.expect("forked rollout path");
let excluded_turns_meta = read_session_meta_line(excluded_turns_path.as_path()).await?;
assert_eq!(excluded_turns_meta.meta.history_base, Some(history_base));
let ThreadForkResponse {
thread: nested_thread,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: forked_thread_id.clone(),
exclude_turns: true,
..ThreadForkParams::default()
},
})
.await?;
assert_eq!(nested_thread.forked_from_id, Some(forked_thread_id.clone()));
assert_eq!(nested_thread.history_mode, ThreadHistoryMode::Paginated);
assert!(nested_thread.turns.is_empty());
let nested_path = nested_thread.path.expect("nested fork rollout path");
let nested_meta = read_session_meta_line(nested_path.as_path()).await?;
assert_eq!(
nested_meta
.meta
.history_base
.expect("nested fork history base")
.thread_id,
ThreadId::from_string(forked_thread_id.as_str())?
);
let _: ThreadUnsubscribeResponse = mcp
.request(|request_id| ClientRequest::ThreadUnsubscribe {
request_id,
params: ThreadUnsubscribeParams {
thread_id: forked_thread_id.clone(),
},
})
.await?;
let closed: ThreadClosedNotification =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("thread/closed")).await??;
assert_eq!(
closed,
ThreadClosedNotification {
thread_id: forked_thread_id
}
);
// The nested fork did not inherit its parent's unload delay.
let _: ThreadUnsubscribeResponse = mcp
.request(|request_id| ClientRequest::ThreadUnsubscribe {
request_id,
params: ThreadUnsubscribeParams {
thread_id: nested_thread.id,
},
})
.await?;
assert!(
timeout(
std::time::Duration::from_millis(250),
mcp.read_stream_until_notification_message("thread/closed"),
)
.await
.is_err()
);
Ok(())
}
#[cfg(unix)]
#[tokio::test]
async fn thread_fork_accepts_resumed_rollout_under_symlinked_sessions_root() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
let external = TempDir::new()?;
let external_sessions = external.path().join("sessions");
std::fs::create_dir_all(external_sessions.as_path())?;
std::os::unix::fs::symlink(
external_sessions.as_path(),
codex_home.path().join("sessions"),
)?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let conversation_id = create_fake_paginated_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let source_path = rollout_path(
codex_home.path(),
"2025-01-05T12-00-00",
conversation_id.as_str(),
);
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id.clone(),
path: Some(source_path),
..Default::default()
})
.await?;
let _: ThreadResumeResponse =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(resume_id)).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(thread.forked_from_id, Some(conversation_id));
Ok(())
}
#[tokio::test]
async fn thread_fork_warns_for_paginated_full_history_hydration() -> 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 conversation_id = create_fake_paginated_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
..Default::default()
})
.await?;
let _: DeprecationNoticeNotification = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_notification("deprecationNotice"),
)
.await??;
let _: ThreadForkResponse = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
mcp.clear_message_buffer();
let metadata_fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
exclude_turns: true,
..Default::default()
})
.await?;
let _: ThreadForkResponse =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(metadata_fork_id)).await??;
assert!(
!mcp.pending_notification_methods()
.contains(&"deprecationNotice".to_string())
);
mcp.clear_message_buffer();
let invalid_fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
last_turn_id: Some("turn-1".to_string()),
before_turn_id: Some("turn-1".to_string()),
..Default::default()
})
.await?;
let _: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(invalid_fork_id)),
)
.await??;
assert!(
!mcp.pending_notification_methods()
.contains(&"deprecationNotice".to_string())
);
Ok(())
}
#[tokio::test]
async fn thread_fork_freezes_active_paginated_turn_as_interrupted() -> Result<()> {
assert_thread_fork_freezes_active_paginated_turn_as_interrupted(MultiAgentVersion::V1).await
}
#[tokio::test]
async fn thread_fork_persists_developer_interruption_marker_for_multi_agent_v2() -> Result<()> {
assert_thread_fork_freezes_active_paginated_turn_as_interrupted(MultiAgentVersion::V2).await
}
async fn assert_thread_fork_freezes_active_paginated_turn_as_interrupted(
multi_agent_version: MultiAgentVersion,
) -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
let config = MockResponsesConfig::new(&server.uri());
let (config, expected_marker_role, thread_source) = match multi_agent_version {
MultiAgentVersion::V2 => (
config.enable_feature(Feature::MultiAgentV2),
"developer",
Some(ThreadSource::Subagent),
),
MultiAgentVersion::V1 => (config, "user", None),
MultiAgentVersion::Disabled => unreachable!("interruption markers require agent support"),
};
config.write(codex_home.path())?;
let source_thread_id = create_fake_paginated_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let source_path = rollout_path(codex_home.path(), "2025-01-05T12-00-00", &source_thread_id);
let source_id = ThreadId::from_string(source_thread_id.as_str())?;
let user_response_item = |id: &str| {
RolloutItem::ResponseItem(
ResponseItem::Message {
status: None,
encrypted_content: None,
id: None,
role: "user".to_string(),
content: vec![ContentItem::InputText {
text: format!("{id} model input"),
}],
phase: None,
internal_chat_message_metadata_passthrough: None,
}
.into(),
)
};
let completed_user_item = |id: &str, completed_at_ms| {
RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id: source_id,
turn_id: "active-turn".to_string(),
item: CoreTurnItem::UserMessage(UserMessageItem {
id: id.to_string(),
client_id: None,
content: vec![codex_protocol::user_input::UserInput::Text {
text: format!("{id} needle"),
text_elements: Vec::new(),
}],
}),
started_at_ms: Some(0),
completed_at_ms,
}))
};
append_rollout_item_to_path(
source_path.as_path(),
&RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_attribution: None,
turn_id: "active-turn".to_string(),
root_turn_id: Some("causal-root".to_string()),
trace_id: None,
started_at: Some(10),
model_context_window: None,
collaboration_mode_kind: Default::default(),
})),
)
.await?;
append_rollout_item_to_path(source_path.as_path(), &user_response_item("before-fork")).await?;
append_rollout_item_to_path(
source_path.as_path(),
&completed_user_item("before-fork", /*completed_at_ms*/ 1),
)
.await?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let ThreadForkResponse {
thread: ephemeral_fork,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: source_thread_id.clone(),
thread_source: thread_source.clone(),
ephemeral: true,
exclude_turns: true,
..Default::default()
},
})
.await?;
let invalid_fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: source_thread_id.clone(),
last_turn_id: Some("active-turn".to_string()),
..Default::default()
})
.await?;
let invalid_fork = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(invalid_fork_id)),
)
.await??;
assert_eq!(
invalid_fork.error.message,
"lastTurnId 'active-turn' identifies an in-progress turn"
);
let ThreadForkResponse {
thread: forked_thread,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: source_thread_id.clone(),
thread_source,
..Default::default()
},
})
.await?;
let forked_thread_id = forked_thread.id.clone();
let forked_path = forked_thread.path.expect("forked rollout path");
let history_base = read_session_meta_line(forked_path.as_path())
.await?
.meta
.history_base
.expect("fork history base");
let child_rollout = std::fs::read_to_string(forked_path.as_path())?
.lines()
.map(codex_rollout::parse_rollout_line)
.collect::<Result<Vec<_>, _>>()?;
assert!(matches!(
child_rollout.as_slice(),
[
RolloutLine { item: RolloutItem::SessionMeta(_), .. },
RolloutLine {
item: RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(_)),
..
},
RolloutLine {
item: RolloutItem::ResponseItem(response_item),
..
},
RolloutLine {
item: RolloutItem::EventMsg(EventMsg::TurnAborted(aborted)),
..
},
] if matches!(
&response_item.item,
codex_protocol::models::ResponseItem::Message { role, .. }
if role == expected_marker_role
) && aborted.turn_id.as_deref() == Some("active-turn")
));
append_rollout_item_to_path(source_path.as_path(), &user_response_item("after-fork")).await?;
append_rollout_item_to_path(
source_path.as_path(),
&completed_user_item("after-fork", /*completed_at_ms*/ 2),
)
.await?;
append_rollout_item_to_path(
source_path.as_path(),
&RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
root_turn_id: None,
turn_id: "active-turn".to_string(),
last_agent_message: None,
error: None,
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
time_to_first_token_ms: None,
})),
)
.await?;
let _: TurnStartResponse = mcp
.request(|request_id| ClientRequest::TurnStart {
request_id,
params: TurnStartParams {
thread_id: ephemeral_fork.id,
input: vec![UserInput::Text {
text: "Continue in an ephemeral fork".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let requests = server.received_requests().await.expect("response requests");
let input = requests
.iter()
.rev()
.find(|request| request.url.path().ends_with("/responses"))
.expect("ephemeral fork model request")
.body_json::<Value>()?["input"]
.clone();
let serialized_input = serde_json::to_string(&input)?;
assert!(serialized_input.contains("before-fork model input"));
assert!(!serialized_input.contains("after-fork model input"));
assert!(input.as_array().is_some_and(|items| {
items.iter().any(|item| {
item["role"] == expected_marker_role
&& item["content"].as_array().is_some_and(|content| {
content.iter().any(|fragment| {
fragment["text"]
.as_str()
.is_some_and(|text| text.contains("<turn_aborted>"))
})
})
})
}));
let ThreadTurnsListResponse { data: turns, .. } = mcp
.request(|request_id| ClientRequest::ThreadTurnsList {
request_id,
params: ThreadTurnsListParams {
thread_id: forked_thread_id.clone(),
cursor: None,
limit: None,
sort_direction: None,
items_view: None,
},
})
.await?;
assert_eq!(turns.len(), 1);
assert_eq!(turns[0].id, "active-turn");
assert_eq!(turns[0].root_turn_id.as_deref(), Some("causal-root"));
assert_eq!(turns[0].status, TurnStatus::Interrupted);
assert_eq!(turns[0].items.len(), 1);
assert!(matches!(
&turns[0].items[0],
ThreadItem::UserMessage { id, .. } if id == "before-fork"
));
let search: ThreadSearchOccurrencesResponse = mcp
.request(|request_id| ClientRequest::ThreadSearchOccurrences {
request_id,
params: ThreadSearchOccurrencesParams {
thread_id: forked_thread_id.clone(),
search_term: "needle".to_string(),
cursor: None,
limit: Some(1),
},
})
.await?;
assert_eq!(search.data.len(), 1);
assert_eq!(search.data[0].item_id, "before-fork");
assert!(search.next_cursor.is_none());
let searched_turns: ThreadTurnsListResponse = mcp
.request(|request_id| ClientRequest::ThreadTurnsList {
request_id,
params: ThreadTurnsListParams {
thread_id: forked_thread_id.clone(),
cursor: Some(search.data[0].turn_cursor.clone()),
limit: Some(1),
sort_direction: None,
items_view: None,
},
})
.await?;
assert_eq!(searched_turns.data, turns);
let ThreadForkResponse {
thread: nested_fork,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: forked_thread_id.clone(),
last_turn_id: Some("active-turn".to_string()),
..Default::default()
},
})
.await?;
let ThreadTurnsListResponse {
data: nested_turns, ..
} = mcp
.request(|request_id| ClientRequest::ThreadTurnsList {
request_id,
params: ThreadTurnsListParams {
thread_id: nested_fork.id,
cursor: None,
limit: None,
sort_direction: None,
items_view: None,
},
})
.await?;
assert_eq!(nested_turns, turns);
let ThreadForkResponse {
thread: nested_before,
..
} = mcp
.request(|request_id| ClientRequest::ThreadFork {
request_id,
params: ThreadForkParams {
thread_id: forked_thread_id.clone(),
before_turn_id: Some("active-turn".to_string()),
..Default::default()
},
})
.await?;
assert!(nested_before.turns.is_empty());
drop(mcp);
let mut resumed_app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let ThreadResumeResponse {
thread: resumed_thread,
..
} = resumed_app_server
.request(|request_id| ClientRequest::ThreadResume {
request_id,
params: ThreadResumeParams {
thread_id: forked_thread_id,
..Default::default()
},
})
.await?;
let mut expected_resumed_turns = turns;
for turn in &mut expected_resumed_turns {
turn.items_view = TurnItemsView::Full;
}
assert_eq!(resumed_thread.turns, expected_resumed_turns);
let _: TurnStartResponse = resumed_app_server
.request(|request_id| ClientRequest::TurnStart {
request_id,
params: TurnStartParams {
thread_id: resumed_thread.id,
input: vec![UserInput::Text {
text: "Continue after cold resume".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
},
})
.await?;
timeout(
DEFAULT_READ_TIMEOUT,
resumed_app_server.read_stream_until_notification_message("turn/completed"),
)
.await??;
let requests = server.received_requests().await.expect("response requests");
let request_body = requests
.iter()
.rev()
.find(|request| request.url.path().ends_with("/responses"))
.expect("cold-resumed model request")
.body_json::<Value>()?;
let turn_metadata: Value = serde_json::from_str(
request_body["client_metadata"]["x-codex-turn-metadata"]
.as_str()
.expect("cold-resumed turn metadata"),
)?;
assert_eq!(
turn_metadata["forked_from_thread_id"].as_str(),
Some(source_thread_id.as_str())
);
assert_eq!(
turn_metadata["forked_from_ordinal_exclusive"].as_u64(),
Some(history_base.end_ordinal_exclusive)
);
let model_input = request_body["input"].as_array().expect("model input");
assert!(model_input.iter().any(|item| {
item["role"] == expected_marker_role
&& item["content"].as_array().is_some_and(|content| {
content.iter().any(|fragment| {
fragment["text"]
.as_str()
.is_some_and(|text| text.to_ascii_lowercase().contains("interrupt"))
})
})
}));
let serialized_input = serde_json::to_string(model_input)?;
assert!(serialized_input.contains("Saved user message"));
assert!(serialized_input.contains("before-fork model input"));
assert!(!serialized_input.contains("after-fork model input"));
assert!(serialized_input.contains("Continue after cold resume"));
Ok(())
}
#[tokio::test]
async fn thread_fork_with_empty_path_uses_thread_id() -> 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 conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
path: Some(std::path::PathBuf::new()),
thread_source: Some(ThreadSource::User),
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(
thread.forked_from_id.as_deref(),
Some(conversation_id.as_str())
);
Ok(())
}
#[tokio::test]
async fn thread_fork_surfaces_cloud_config_bundle_load_errors() -> Result<()> {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/backend-api/wham/config/bundle"))
.respond_with(
ResponseTemplate::new(401)
.insert_header("content-type", "text/html")
.set_body_string("<html>nope</html>"),
)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/oauth/token"))
.respond_with(ResponseTemplate::new(401).set_body_json(json!({
"error": { "code": "refresh_token_invalidated" }
})))
.mount(&server)
.await;
let codex_home = TempDir::new()?;
let model_server = create_mock_responses_server_repeating_assistant("Done").await;
let chatgpt_base_url = format!("{}/backend-api", server.uri());
MockResponsesConfig::new(&model_server.uri())
.with_root_config(&format!(r#"chatgpt_base_url = "{chatgpt_base_url}""#))
.write(codex_home.path())?;
write_chatgpt_auth(
codex_home.path(),
ChatGptAuthFixture::new("chatgpt-token")
.refresh_token("stale-refresh-token")
.plan_type("business")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123")
.account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
let conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let refresh_token_url = format!("{}/oauth/token", server.uri());
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.with_env_overrides(&[
("OPENAI_API_KEY", None),
(
REFRESH_TOKEN_URL_OVERRIDE_ENV_VAR,
Some(refresh_token_url.as_str()),
),
])
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let fork_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert!(
fork_err
.error
.message
.contains("failed to load configuration"),
"unexpected fork error: {}",
fork_err.error.message
);
assert_eq!(
fork_err.error.data,
Some(json!({
"reason": "cloudConfigBundle",
"errorCode": "Auth",
"action": "relogin",
"statusCode": 401,
"detail": "Your access token could not be refreshed because your refresh token was revoked. Please log out and sign in again.",
}))
);
Ok(())
}
#[tokio::test]
async fn thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()> {
assert_thread_fork_ephemeral_remains_pathless_and_omits_listing(ThreadHistoryMode::Legacy).await
}
#[tokio::test]
async fn paginated_thread_fork_ephemeral_remains_pathless_and_omits_listing() -> Result<()> {
assert_thread_fork_ephemeral_remains_pathless_and_omits_listing(ThreadHistoryMode::Paginated)
.await
}
async fn assert_thread_fork_ephemeral_remains_pathless_and_omits_listing(
history_mode: ThreadHistoryMode,
) -> 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 preview = "Saved user message";
let create_rollout = match history_mode {
ThreadHistoryMode::Legacy => create_fake_rollout,
ThreadHistoryMode::Paginated => create_fake_paginated_rollout,
};
let conversation_id = create_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
preview,
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
if history_mode == ThreadHistoryMode::Paginated {
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
ephemeral: true,
..Default::default()
})
.await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert_eq!(
error.error.message,
"ephemeral paginated thread/fork requires `excludeTurns: true`"
);
}
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id.clone(),
ephemeral: true,
exclude_turns: history_mode == ThreadHistoryMode::Paginated,
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let fork_result = fork_resp.result.clone();
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let fork_thread_id = thread.id.clone();
assert!(
thread.ephemeral,
"ephemeral forks should be marked explicitly"
);
assert_eq!(
thread.path, None,
"ephemeral forks should not expose a path"
);
assert_eq!(thread.preview, preview);
assert_eq!(thread.status, ThreadStatus::Idle);
assert_eq!(thread.name, None);
if history_mode == ThreadHistoryMode::Paginated {
assert!(thread.turns.is_empty());
} else {
assert_eq!(thread.turns.len(), 1, "expected copied fork history");
let turn = &thread.turns[0];
assert_eq!(turn.status, TurnStatus::Completed);
assert_eq!(turn.items.len(), 1, "expected user message item");
match &turn.items[0] {
ThreadItem::UserMessage { content, .. } => {
assert_eq!(
content,
&vec![UserInput::Text {
text: preview.to_string(),
text_elements: Vec::new(),
}]
);
}
other => panic!("expected user message item, got {other:?}"),
}
}
let thread_json = fork_result
.get("thread")
.and_then(Value::as_object)
.expect("thread/fork result.thread must be an object");
assert_eq!(
thread_json.get("ephemeral").and_then(Value::as_bool),
Some(true),
"ephemeral forks should serialize `ephemeral: true`"
);
let deadline = tokio::time::Instant::now() + DEFAULT_READ_TIMEOUT;
let notif = loop {
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
let message = timeout(remaining, mcp.read_next_message()).await??;
let JSONRPCMessage::Notification(notif) = message else {
continue;
};
if notif.method == "thread/status/changed" {
let status_changed: ThreadStatusChangedNotification =
serde_json::from_value(notif.params.expect("params must be present"))?;
if status_changed.thread_id == fork_thread_id {
anyhow::bail!(
"thread/fork should introduce the thread without a preceding thread/status/changed"
);
}
continue;
}
if notif.method == "thread/started" {
break notif;
}
};
let started_params = notif.params.clone().expect("params must be present");
let started_thread_json = started_params
.get("thread")
.and_then(Value::as_object)
.expect("thread/started params.thread must be an object");
assert_eq!(
started_thread_json
.get("ephemeral")
.and_then(Value::as_bool),
Some(true),
"thread/started should serialize `ephemeral: true` for ephemeral forks"
);
assert_eq!(
started_thread_json.get("turns"),
Some(&json!([])),
"thread/started must not emit copied ephemeral fork turns"
);
let started: ThreadStartedNotification =
serde_json::from_value(notif.params.expect("params must be present"))?;
let mut expected_started_thread = thread;
expected_started_thread.turns.clear();
assert_eq!(started.thread, expected_started_thread);
let ThreadListResponse { data, .. } = list_threads(&mut mcp).await?;
assert!(
data.iter().all(|candidate| candidate.id != fork_thread_id),
"ephemeral forks should not appear in thread/list"
);
assert!(
data.iter().any(|candidate| candidate.id == conversation_id),
"persistent source thread should remain listed"
);
let turn_id = mcp
.send_turn_start_request(TurnStartParams {
thread_id: fork_thread_id.clone(),
client_user_message_id: None,
input: vec![UserInput::Text {
text: "continue".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse = timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(turn_id)).await??;
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_notification_message("turn/completed"),
)
.await??;
let requests = server.received_requests().await.expect("response requests");
let model_input = requests
.iter()
.find(|request| request.url.path().ends_with("/responses"))
.expect("ephemeral fork model request")
.body_json::<Value>()?["input"]
.to_string();
assert!(model_input.contains(preview));
assert!(model_input.contains("continue"));
let ThreadListResponse { data, .. } = list_threads(&mut mcp).await?;
assert!(data.iter().all(|thread| thread.id != fork_thread_id));
Ok(())
}
#[tokio::test]
async fn thread_fork_rejects_incompatible_boundaries_and_ephemeral_goal_deferral() -> 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 thread_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
for (params, expected_message) in [
(
ThreadForkParams {
thread_id: thread_id.clone(),
last_turn_id: Some("turn-1".to_string()),
before_turn_id: Some("turn-2".to_string()),
..Default::default()
},
"`beforeTurnId` cannot be combined with `lastTurnId`",
),
(
ThreadForkParams {
thread_id: thread_id.clone(),
ephemeral: true,
defer_goal_continuation: true,
..Default::default()
},
"`deferGoalContinuation` cannot be combined with `ephemeral`",
),
] {
let fork_id = mcp.send_thread_fork_request(params).await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert_eq!(error.error.message, expected_message);
}
Ok(())
}
#[tokio::test]
async fn pathless_ephemeral_thread_rejects_codex_home_path_after_reload() -> 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 parent_thread_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Parent message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let side_thread_id = {
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let fork_id = app_server
.send_thread_fork_request(ThreadForkParams {
thread_id: parent_thread_id,
ephemeral: true,
..Default::default()
})
.await?;
let ThreadForkResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(fork_id)).await??;
assert!(thread.ephemeral);
assert_eq!(thread.path, None);
let turn_id = app_server
.send_turn_start_request(TurnStartParams {
thread_id: thread.id.clone(),
client_user_message_id: None,
input: vec![UserInput::Text {
text: "continue".to_string(),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let _: TurnStartResponse =
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(turn_id)).await??;
timeout(
DEFAULT_READ_TIMEOUT,
app_server.read_stream_until_notification_message("turn/completed"),
)
.await??;
thread.id
};
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.without_auto_env()
.build_initialized()
.await?;
let codex_home_path = codex_home.path().to_path_buf();
let resume_id = app_server
.send_thread_resume_request(ThreadResumeParams {
thread_id: side_thread_id.clone(),
path: Some(codex_home_path.clone()),
..Default::default()
})
.await?;
let resume_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
app_server.read_stream_until_error_message(RequestId::Integer(resume_id)),
)
.await??;
assert!(
resume_err.error.message.contains("path is a directory"),
"unexpected resume error: {}",
resume_err.error.message
);
assert!(
!resume_err.error.message.contains("Is a directory"),
"resume should reject the directory before rollout reading: {}",
resume_err.error.message
);
let fork_id = app_server
.send_thread_fork_request(ThreadForkParams {
thread_id: side_thread_id,
path: Some(codex_home_path),
..Default::default()
})
.await?;
let fork_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
app_server.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert!(
fork_err.error.message.contains("path is a directory"),
"unexpected fork error: {}",
fork_err.error.message
);
assert!(
!fork_err.error.message.contains("Is a directory"),
"fork should reject the directory before rollout reading: {}",
fork_err.error.message
);
Ok(())
}
#[test_case(ThreadHistoryMode::Legacy; "legacy")]
#[test_case(ThreadHistoryMode::Paginated; "paginated")]
#[tokio::test]
async fn inheriting_fork_preserves_parent_config(history_mode: ThreadHistoryMode) -> Result<()> {
let server = responses::start_mock_server().await;
let mock = responses::mount_sse_sequence(
&server,
vec![
responses::sse(vec![
responses::ev_reasoning_item("reasoning", &["A summary"], &["Private reasoning"]),
responses::ev_assistant_message("answer", "Done"),
responses::ev_completed("initial"),
]),
responses::sse(vec![responses::ev_completed("child")]),
responses::sse(vec![responses::ev_completed("parent")]),
],
)
.await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
write_models_cache(codex_home.path()).await?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let model = all_model_presets()
.iter()
.find(|model| model.show_in_picker && !model.service_tiers.is_empty())
.expect("a bundled model supports service tiers");
let tier = &model.service_tiers[0].id;
let parent = mcp
.start_thread(ThreadStartParams {
history_mode: Some(history_mode),
base_instructions: Some("Retain these base instructions.".into()),
developer_instructions: Some("Retain this system policy context.".into()),
dynamic_tools: Some(vec![DynamicToolSpec::Function(DynamicToolFunctionSpec {
name: "parent_tool".into(),
description: "The parent's client-provided tool description.".into(),
input_schema: json!({"type": "object", "properties": {}}),
defer_loading: false,
})]),
..Default::default()
})
.await?;
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: parent.thread.id.clone(),
input: vec![UserInput::Text {
text: "Initial turn".into(),
text_elements: vec![],
}],
service_tier: Some(Some(tier.clone())),
collaboration_mode: Some(CollaborationMode {
mode: ModeKind::Plan,
settings: Settings {
model: model.id.clone(),
reasoning_effort: Some(ReasoningEffort::High),
developer_instructions: Some("Keep the parent's planning instructions.".into()),
},
}),
disabled_plugin_ids: Some(vec!["unused@plugin".into()]),
..Default::default()
})
.await?;
for (delay, message) in [
(
json!(-1),
"`thread_unload_delay_secs` must be a non-negative integer",
),
(json!(u64::MAX), "`thread_unload_delay_secs` is too large"),
] {
let request_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: parent.thread.id.clone(),
experimental_prediction_mode: true,
ephemeral: true,
config: Some([("thread_unload_delay_secs".into(), delay)].into()),
..Default::default()
})
.await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
)
.await??;
assert_eq!(error.error.message, message);
}
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: parent.thread.id.clone(),
experimental_prediction_mode: true,
ephemeral: true,
exclude_turns: true,
config: Some([("thread_unload_delay_secs".into(), json!(0))].into()),
..Default::default()
})
.await?;
let child: ThreadForkResponse =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_response(fork_id)).await??;
assert_eq!(
(
&child.model,
&child.reasoning_effort,
&child.service_tier,
&child.disabled_plugin_ids
),
(
&model.id,
&Some(ReasoningEffort::High),
&Some(tier.clone()),
&vec!["unused@plugin".to_string()]
),
);
assert!(child.thread.path.is_none());
for id in [&child.thread.id, &parent.thread.id] {
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: id.clone(),
input: vec![UserInput::Text {
text: "Continue identically".into(),
text_elements: vec![],
}],
..Default::default()
})
.await?;
}
let requests = mock.requests();
assert_eq!(requests.len(), 3);
let comparable = |index: usize| {
let body = requests[index].body_json();
responses::strip_response_item_ids_from_json(json!({
"input": body["input"],
"instructions": body["instructions"],
"tools": body["tools"],
"model": body["model"],
"reasoning": body["reasoning"],
"service_tier": body["service_tier"],
}))
};
assert_eq!(comparable(1), comparable(2));
assert_eq!(
requests[1].body_json()["prompt_cache_key"],
requests[2].body_json()["prompt_cache_key"]
);
assert!(
requests[1].body_json()["tools"]
.to_string()
.contains("parent_tool")
);
let _: ThreadUnsubscribeResponse = mcp
.request(|request_id| ClientRequest::ThreadUnsubscribe {
request_id,
params: ThreadUnsubscribeParams {
thread_id: child.thread.id.clone(),
},
})
.await?;
let closed: ThreadClosedNotification =
timeout(DEFAULT_READ_TIMEOUT, mcp.read_notification("thread/closed")).await??;
assert_eq!(
closed,
ThreadClosedNotification {
thread_id: child.thread.id
}
);
Ok(())
}
#[test_case(ThreadHistoryMode::Legacy, false; "fresh persistent legacy")]
#[test_case(ThreadHistoryMode::Legacy, true; "ephemeral legacy")]
#[test_case(ThreadHistoryMode::Paginated, false; "fresh persistent paginated")]
#[test_case(ThreadHistoryMode::Paginated, true; "ephemeral paginated")]
#[tokio::test]
async fn inheriting_fork_requires_persisted_parent(
history_mode: ThreadHistoryMode,
ephemeral: bool,
) -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
write_models_cache(codex_home.path()).await?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let parent = mcp
.start_thread(ThreadStartParams {
history_mode: Some(history_mode),
ephemeral: Some(ephemeral),
..Default::default()
})
.await?;
if ephemeral {
mcp.start_turn_and_wait_for_completion(TurnStartParams {
thread_id: parent.thread.id.clone(),
input: vec![UserInput::Text {
text: "Complete an ephemeral turn".into(),
text_elements: vec![],
}],
..Default::default()
})
.await?;
}
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: parent.thread.id.clone(),
experimental_prediction_mode: true,
ephemeral: true,
exclude_turns: true,
..Default::default()
})
.await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert!(
error
.error
.message
.contains("no rollout found for thread id"),
"unexpected fork error: {}",
error.error.message
);
if let Some(path) = parent.thread.path {
assert!(!path.exists(), "forking should not materialize the parent");
}
Ok(())
}
#[tokio::test]
async fn inheriting_fork_requires_loaded_parent() -> Result<()> {
let codex_home = TempDir::new()?;
let thread_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id,
experimental_prediction_mode: true,
ephemeral: true,
..Default::default()
})
.await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert_eq!(
error.error.message,
"`experimentalPredictionMode` requires a loaded parent",
);
Ok(())
}
#[tokio::test]
async fn inheriting_fork_requires_ephemeral_child() -> Result<()> {
let codex_home = TempDir::new()?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: "not-loaded".into(),
experimental_prediction_mode: true,
..Default::default()
})
.await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert_eq!(
error.error.message,
"`experimentalPredictionMode` requires `ephemeral: true`",
);
Ok(())
}
#[tokio::test]
async fn inheriting_fork_rejects_configuration_overrides() -> Result<()> {
let codex_home = TempDir::new()?;
let mut mcp = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
for (key, value) in [
("model", json!("other-model")),
("modelProvider", json!("other-provider")),
("serviceTier", Value::Null),
("cwd", json!(codex_home.path())),
("runtimeWorkspaceRoots", json!([])),
("approvalPolicy", json!("never")),
("approvalsReviewer", json!("user")),
("sandbox", json!("read-only")),
("permissions", json!(":read-only")),
("baseInstructions", json!("override")),
("developerInstructions", json!("override")),
("config", json!({"model": "other-model"})),
("config", json!({"model_reasoning_effort": "low"})),
] {
let mut params = json!({
"threadId": "not-loaded", "experimentalPredictionMode": true, "ephemeral": true,
});
params[key] = value;
let request = mcp.send_raw_request("thread/fork", Some(params)).await?;
let error = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(request)),
)
.await??;
assert_eq!(
error.error.message,
"`experimentalPredictionMode` cannot be combined with configuration overrides",
"override: {key}"
);
}
Ok(())
}
Reference in New Issue View Git Blame Copy Permalink