agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Migrate fork and resume reads to thread store (#18900)

Browse Source 
- Route cold thread/resume and thread/fork source loading through
ThreadStore reads instead of direct rollout path operations
- Keep lookups that explicitly specify a rollout-path using the local
thread store methods but return an invalid-request error for remote
ThreadStore configurations
- Add some additional unit tests for code path coverage
This commit is contained in:
Tom
2026-04-24 13:51:37 -07:00
committed by GitHub
Unverified
parent 13e0ec1614
commit 0a9b559c0b
18 changed files with 966 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+303 -331
View File
@@ -332,6 +332,7 @@ use codex_protocol::protocol::McpServerRefreshConfig;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RateLimitSnapshot as CoreRateLimitSnapshot;
use codex_protocol::protocol::RealtimeVoicesList;
use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::ReviewDelivery as CoreReviewDelivery;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::ReviewTarget as CoreReviewTarget;
@@ -354,6 +355,7 @@ use codex_state::log_db::LogDbLayer;
use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams;
use codex_thread_store::ListThreadsParams as StoreListThreadsParams;
use codex_thread_store::LocalThreadStore;
use codex_thread_store::ReadThreadByRolloutPathParams as StoreReadThreadByRolloutPathParams;
use codex_thread_store::ReadThreadParams as StoreReadThreadParams;
use codex_thread_store::RemoteThreadStore;
use codex_thread_store::SortDirection as StoreSortDirection;
@@ -404,7 +406,6 @@ use crate::thread_state::ThreadState;
use crate::thread_state::ThreadStateManager;
use token_usage_replay::latest_token_usage_turn_id_for_thread_path;
use token_usage_replay::latest_token_usage_turn_id_from_rollout_items;
use token_usage_replay::latest_token_usage_turn_id_from_rollout_path;
use token_usage_replay::send_thread_token_usage_update_to_connection;
const THREAD_LIST_DEFAULT_LIMIT: usize = 25;
@@ -662,9 +663,7 @@ pub(crate) struct CodexMessageProcessorArgs {
fn configured_thread_store(config: &Config) -> Arc<dyn ThreadStore> {
match config.experimental_thread_store_endpoint.as_deref() {
Some(endpoint) => Arc::new(RemoteThreadStore::new(endpoint)),
None => Arc::new(LocalThreadStore::new(
codex_rollout::RolloutConfig::from_view(config),
)),
None => Arc::new(configured_local_thread_store(config)),
}
}
@@ -675,6 +674,10 @@ fn environment_selection_error_message(err: CodexErr) -> String {
}
}
fn configured_local_thread_store(config: &Config) -> LocalThreadStore {
LocalThreadStore::new(codex_rollout::RolloutConfig::from_view(config))
}
impl CodexMessageProcessor {
async fn instruction_sources_from_config(config: &Config) -> Vec<AbsolutePathBuf> {
codex_core::AgentsMdManager::new(config)
@@ -4509,22 +4512,22 @@ impl CodexMessageProcessor {
} = params;
let include_turns = !exclude_turns;
let thread_history = if let Some(history) = history {
let (thread_history, resume_source_thread) = if let Some(history) = history {
let Some(thread_history) = self
.resume_thread_from_history(request_id.clone(), history.as_slice())
.await
else {
return;
};
thread_history
(thread_history, None)
} else {
let Some(thread_history) = self
let Some((thread_history, stored_thread)) = self
.resume_thread_from_rollout(request_id.clone(), &thread_id, path.as_ref())
.await
else {
return;
};
thread_history
(thread_history, Some(stored_thread))
};
let history_cwd = thread_history.session_cwd();
@@ -4541,13 +4544,12 @@ impl CodexMessageProcessor {
developer_instructions,
personality,
);
let persisted_resume_metadata = self
.load_and_apply_persisted_resume_metadata(
&thread_history,
&mut request_overrides,
&mut typesafe_overrides,
)
.await;
self.load_and_apply_persisted_resume_metadata(
&thread_history,
&mut request_overrides,
&mut typesafe_overrides,
)
.await;
// Derive a Config using the same logic as new conversation, honoring overrides if provided.
let config = match self
@@ -4612,7 +4614,7 @@ impl CodexMessageProcessor {
codex_thread.as_ref(),
&response_history,
rollout_path.as_path(),
persisted_resume_metadata.as_ref(),
resume_source_thread,
include_turns,
)
.await
@@ -4738,77 +4740,58 @@ impl CodexMessageProcessor {
return true;
}
let rollout_path = if let Some(path) = existing_thread.rollout_path() {
if path.exists() {
path
} else {
match find_thread_path_by_id_str(
&self.config.codex_home,
&existing_thread_id.to_string(),
)
.await
{
Ok(Some(path)) => path,
Ok(None) => {
self.send_invalid_request_error(
request_id,
format!("no rollout found for thread id {existing_thread_id}"),
)
.await;
return true;
}
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {existing_thread_id}: {err}"),
)
.await;
return true;
}
}
}
} else {
match find_thread_path_by_id_str(
&self.config.codex_home,
&existing_thread_id.to_string(),
)
.await
{
Ok(Some(path)) => path,
Ok(None) => {
self.send_invalid_request_error(
request_id,
format!("no rollout found for thread id {existing_thread_id}"),
)
.await;
return true;
}
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {existing_thread_id}: {err}"),
)
.await;
return true;
}
}
};
if let Some(requested_path) = params.path.as_ref()
&& requested_path != &rollout_path
if let (Some(requested_path), Some(active_path)) = (
params.path.as_ref(),
existing_thread.rollout_path().as_ref(),
) && requested_path != active_path
{
self.send_invalid_request_error(
request_id,
format!(
"cannot resume running thread {existing_thread_id} with mismatched path: requested `{}`, active `{}`",
requested_path.display(),
rollout_path.display()
active_path.display()
),
)
.await;
return true;
}
let Some(source_thread) = self
.read_stored_thread_for_resume(
request_id.clone(),
&params.thread_id,
params.path.as_ref(),
/*include_history*/ true,
)
.await
else {
return true;
};
if source_thread.thread_id != existing_thread_id {
self.send_invalid_request_error(
request_id,
format!(
"cannot resume running thread {existing_thread_id} from source thread {}",
source_thread.thread_id
),
)
.await;
return true;
}
let Some(history_items) = source_thread
.history
.as_ref()
.map(|history| history.items.clone())
else {
self.send_internal_error(
request_id,
format!("thread {existing_thread_id} did not include persisted history"),
)
.await;
return true;
};
let thread_state = self
.thread_state_manager
.thread_state(existing_thread_id)
@@ -4835,18 +4818,15 @@ impl CodexMessageProcessor {
mismatch_details.join("; ")
);
}
let mut config_for_instruction_sources = self.config.as_ref().clone();
config_for_instruction_sources.cwd = config_snapshot.cwd.clone();
let instruction_sources =
Self::instruction_sources_from_config(&config_for_instruction_sources).await;
let thread_summary = match load_thread_summary_for_rollout(
&self.config,
existing_thread_id,
rollout_path.as_path(),
config_snapshot.model_provider_id.as_str(),
/*persisted_metadata*/ None,
)
.await
let mut summary_source_thread = source_thread;
summary_source_thread.history = None;
let thread_summary = match self
.stored_thread_to_api_thread(
summary_source_thread,
config_snapshot.model_provider_id.as_str(),
/*include_turns*/ false,
)
.await
{
Ok(thread) => thread,
Err(message) => {
@@ -4854,6 +4834,10 @@ impl CodexMessageProcessor {
return true;
}
};
let mut config_for_instruction_sources = self.config.as_ref().clone();
config_for_instruction_sources.cwd = config_snapshot.cwd.clone();
let instruction_sources =
Self::instruction_sources_from_config(&config_for_instruction_sources).await;
let listener_command_tx = {
let thread_state = thread_state.lock().await;
@@ -4874,7 +4858,7 @@ impl CodexMessageProcessor {
let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse(
Box::new(crate::thread_state::PendingThreadResumeRequest {
request_id: request_id.clone(),
rollout_path: rollout_path.clone(),
history_items,
config_snapshot,
instruction_sources,
thread_summary,
@@ -4920,57 +4904,133 @@ impl CodexMessageProcessor {
request_id: ConnectionRequestId,
thread_id: &str,
path: Option<&PathBuf>,
) -> Option<InitialHistory> {
let rollout_path = if let Some(path) = path {
path.clone()
) -> Option<(InitialHistory, StoredThread)> {
match self
.read_stored_thread_for_resume(
request_id.clone(),
thread_id,
path,
/*include_history*/ true,
)
.await
{
Some(stored_thread) => self
.stored_thread_to_initial_history(request_id, &stored_thread)
.await
.map(|history| (history, stored_thread)),
None => None,
}
}
async fn read_stored_thread_for_resume(
&self,
request_id: ConnectionRequestId,
thread_id: &str,
path: Option<&PathBuf>,
include_history: bool,
) -> Option<StoredThread> {
let result = if let Some(path) = path {
self.thread_store
.read_thread_by_rollout_path(StoreReadThreadByRolloutPathParams {
rollout_path: path.clone(),
include_archived: true,
include_history,
})
.await
} else {
let existing_thread_id = match ThreadId::from_string(thread_id) {
Ok(id) => id,
Err(err) => {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("invalid thread id: {err}"),
data: None,
};
self.outgoing.send_error(request_id, error).await;
self.send_invalid_request_error(
request_id,
format!("invalid thread id: {err}"),
)
.await;
return None;
}
};
match find_thread_path_by_id_str(
&self.config.codex_home,
&existing_thread_id.to_string(),
)
.await
{
Ok(Some(path)) => path,
Ok(None) => {
self.send_invalid_request_error(
request_id,
format!("no rollout found for thread id {existing_thread_id}"),
)
.await;
return None;
}
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {existing_thread_id}: {err}"),
)
.await;
return None;
}
}
let params = StoreReadThreadParams {
thread_id: existing_thread_id,
include_archived: true,
include_history,
};
self.thread_store.read_thread(params).await
};
match RolloutRecorder::get_rollout_history(&rollout_path).await {
Ok(initial_history) => Some(initial_history),
match result {
Ok(thread) => Some(thread),
Err(err) => {
self.send_invalid_request_error(
self.outgoing
.send_error(request_id, thread_store_resume_read_error(err))
.await;
None
}
}
}
async fn stored_thread_to_initial_history(
&self,
request_id: ConnectionRequestId,
stored_thread: &StoredThread,
) -> Option<InitialHistory> {
let thread_id = stored_thread.thread_id;
let history = match stored_thread.history.as_ref() {
Some(history) => history.items.clone(),
None => {
self.send_internal_error(
request_id,
format!("failed to load rollout `{}`: {err}", rollout_path.display()),
format!("thread {thread_id} did not include persisted history"),
)
.await;
return None;
}
};
Some(InitialHistory::Resumed(ResumedHistory {
conversation_id: thread_id,
history,
rollout_path: stored_thread.rollout_path.clone(),
}))
}
async fn stored_thread_to_api_thread(
&self,
stored_thread: StoredThread,
fallback_provider: &str,
include_turns: bool,
) -> std::result::Result<Thread, String> {
let (mut thread, history) =
thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd);
if include_turns && let Some(history) = history {
populate_thread_turns(
&mut thread,
ThreadTurnSource::HistoryItems(&history.items),
/*active_turn*/ None,
)
.await?;
}
Ok(thread)
}
async fn read_stored_thread_for_new_fork(
&self,
request_id: ConnectionRequestId,
thread_store: &dyn ThreadStore,
thread_id: ThreadId,
include_history: bool,
) -> Option<StoredThread> {
match thread_store
.read_thread(StoreReadThreadParams {
thread_id,
include_archived: true,
include_history,
})
.await
{
Ok(thread) => Some(thread),
Err(err) => {
self.outgoing
.send_error(request_id, thread_store_resume_read_error(err))
.await;
None
}
}
@@ -4982,20 +5042,42 @@ impl CodexMessageProcessor {
thread: &CodexThread,
thread_history: &InitialHistory,
rollout_path: &Path,
persisted_resume_metadata: Option<&ThreadMetadata>,
resume_source_thread: Option<StoredThread>,
include_turns: bool,
) -> std::result::Result<Thread, String> {
let config_snapshot = thread.config_snapshot().await;
let thread = match thread_history {
InitialHistory::Resumed(resumed) => {
load_thread_summary_for_rollout(
&self.config,
resumed.conversation_id,
resumed.rollout_path.as_path(),
config_snapshot.model_provider_id.as_str(),
persisted_resume_metadata,
)
.await
let fallback_provider = config_snapshot.model_provider_id.as_str();
if let Some(mut stored_thread) = resume_source_thread {
stored_thread.history = None;
Ok(thread_from_stored_thread(
stored_thread,
fallback_provider,
&self.config.cwd,
)
.0)
} else {
match self
.thread_store
.read_thread(StoreReadThreadParams {
thread_id: resumed.conversation_id,
include_archived: true,
include_history: false,
})
.await
{
Ok(stored_thread) => Ok(thread_from_stored_thread(
stored_thread,
fallback_provider,
&self.config.cwd,
)
.0),
Err(read_err) => {
Err(format!("failed to read thread from store: {read_err}"))
}
}
}
}
InitialHistory::Forked(items) => {
let mut thread = build_thread_from_snapshot(
@@ -5061,50 +5143,31 @@ impl CodexMessageProcessor {
return;
}
let (rollout_path, source_thread_id) = if let Some(path) = path {
(path, None)
} else {
let existing_thread_id = match ThreadId::from_string(&thread_id) {
Ok(id) => id,
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("invalid thread id: {err}"),
)
.await;
return;
}
};
match find_thread_path_by_id_str(
&self.config.codex_home,
&existing_thread_id.to_string(),
let Some(source_thread) = self
.read_stored_thread_for_resume(
request_id.clone(),
&thread_id,
path.as_ref(),
/*include_history*/ true,
)
.await
{
Ok(Some(p)) => (p, Some(existing_thread_id)),
Ok(None) => {
self.send_invalid_request_error(
request_id,
format!("no rollout found for thread id {existing_thread_id}"),
)
.await;
return;
}
Err(err) => {
self.send_invalid_request_error(
request_id,
format!("failed to locate thread id {existing_thread_id}: {err}"),
)
.await;
return;
}
}
else {
return;
};
let history_cwd =
read_history_cwd_from_state_db(&self.config, source_thread_id, rollout_path.as_path())
.await;
let source_thread_id = source_thread.thread_id;
let Some(history_items) = source_thread
.history
.as_ref()
.map(|history| history.items.clone())
else {
self.send_internal_error(
request_id,
format!("thread {source_thread_id} did not include persisted history"),
)
.await;
return;
};
let history_cwd = Some(source_thread.cwd.clone());
// Persist Windows sandbox mode.
let mut cli_overrides = cli_overrides.unwrap_or_default();
@@ -5159,6 +5222,7 @@ impl CodexMessageProcessor {
let fallback_model_provider = config.model_provider_id.clone();
let instruction_sources = Self::instruction_sources_from_config(&config).await;
let fork_thread_store = configured_thread_store(&config);
let NewThread {
thread_id,
@@ -5167,10 +5231,14 @@ impl CodexMessageProcessor {
..
} = match self
.thread_manager
.fork_thread(
.fork_thread_from_history(
ForkSnapshot::Interrupted,
config,
rollout_path.clone(),
InitialHistory::Resumed(ResumedHistory {
conversation_id: source_thread_id,
history: history_items.clone(),
rollout_path: source_thread.rollout_path.clone(),
}),
persist_extended_history,
self.request_trace_context(&request_id).await,
)
@@ -5182,7 +5250,7 @@ impl CodexMessageProcessor {
CodexErr::Io(_) | CodexErr::Json(_) => {
self.send_invalid_request_error(
request_id,
format!("failed to load rollout `{}`: {err}", rollout_path.display()),
format!("failed to load thread {source_thread_id}: {err}"),
)
.await;
}
@@ -5216,25 +5284,33 @@ impl CodexMessageProcessor {
);
// Persistent forks materialize their own rollout immediately. Ephemeral forks stay
// pathless, so they rebuild their visible history from the copied source rollout instead.
// pathless, so they rebuild their visible history from the copied source history instead.
let mut thread = if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref() {
match read_summary_from_rollout(
fork_rollout_path.as_path(),
fallback_model_provider.as_str(),
)
.await
let Some(stored_thread) = self
.read_stored_thread_for_new_fork(
request_id.clone(),
fork_thread_store.as_ref(),
thread_id,
include_turns,
)
.await
else {
return;
};
match self
.stored_thread_to_api_thread(
stored_thread,
fallback_model_provider.as_str(),
include_turns,
)
.await
{
Ok(summary) => {
let mut thread = summary_to_thread(summary, &self.config.cwd);
thread.forked_from_id =
forked_from_id_from_rollout(fork_rollout_path.as_path()).await;
thread
}
Err(err) => {
Ok(thread) => thread,
Err(message) => {
self.send_internal_error(
request_id,
format!(
"failed to load rollout `{}` for thread {thread_id}: {err}",
"failed to load rollout `{}` for thread {thread_id}: {message}",
fork_rollout_path.display()
),
)
@@ -5247,30 +5323,8 @@ impl CodexMessageProcessor {
// forked thread names do not inherit the source thread name
let mut thread =
build_thread_from_snapshot(thread_id, &config_snapshot, /*path*/ None);
let history_items = match read_rollout_items_from_rollout(rollout_path.as_path()).await
{
Ok(items) => items,
Err(err) => {
self.send_internal_error(
request_id,
format!(
"failed to load source rollout `{}` for thread {thread_id}: {err}",
rollout_path.display()
),
)
.await;
return;
}
};
thread.preview = preview_from_rollout_items(&history_items);
thread.forked_from_id = source_thread_id
.or_else(|| {
history_items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.id),
_ => None,
})
})
.map(|id| id.to_string());
thread.forked_from_id = Some(source_thread_id.to_string());
if include_turns
&& let Err(message) = populate_thread_turns(
&mut thread,
@@ -5285,19 +5339,6 @@ impl CodexMessageProcessor {
thread
};
if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref()
&& include_turns
&& let Err(message) = populate_thread_turns(
&mut thread,
ThreadTurnSource::RolloutPath(fork_rollout_path.as_path()),
/*active_turn*/ None,
)
.await
{
self.send_internal_error(request_id, message).await;
return;
}
self.thread_watch_manager
.upsert_thread_silently(thread.clone())
.await;
@@ -5346,11 +5387,10 @@ impl CodexMessageProcessor {
{
Some(turn_id)
} else {
latest_token_usage_turn_id_from_rollout_path(
rollout_path.as_path(),
latest_token_usage_turn_id_from_rollout_items(
&history_items,
token_usage_thread.turns.as_slice(),
)
.await
};
// Mirror the resume contract for forks: the new thread is usable as soon
// as the response arrives, so restored usage must follow immediately.
@@ -5393,14 +5433,13 @@ impl CodexMessageProcessor {
.as_any()
.downcast_ref::<LocalThreadStore>()
else {
let error = JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message:
"rollout path queries are only supported with the local thread store"
.to_string(),
data: None,
};
return self.outgoing.send_error(request_id, error).await;
self.send_invalid_request_error(
request_id,
"rollout path queries are only supported with the local thread store"
.to_string(),
)
.await;
return;
};
local_thread_store
@@ -8812,7 +8851,7 @@ async fn handle_pending_thread_resume_request(
if pending.include_turns
&& let Err(message) = populate_thread_turns(
&mut thread,
ThreadTurnSource::RolloutPath(pending.rollout_path.as_path()),
ThreadTurnSource::HistoryItems(&pending.history_items),
active_turn.as_ref(),
)
.await
@@ -8904,11 +8943,10 @@ async fn handle_pending_thread_resume_request(
// Match cold resume: metadata-only resume should attach the listener without
// paying the cost of turn reconstruction for historical usage replay.
if let Some(token_usage_thread) = token_usage_thread {
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_path(
pending.rollout_path.as_path(),
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_items(
&pending.history_items,
token_usage_thread.turns.as_slice(),
)
.await;
);
// Rejoining a loaded thread has the same UI contract as a cold resume, but
// uses the live conversation state instead of reconstructing a new session.
send_thread_token_usage_update_to_connection(
@@ -8927,7 +8965,6 @@ async fn handle_pending_thread_resume_request(
}
enum ThreadTurnSource<'a> {
RolloutPath(&'a Path),
HistoryItems(&'a [RolloutItem]),
}
@@ -8937,18 +8974,6 @@ async fn populate_thread_turns(
active_turn: Option<&Turn>,
) -> std::result::Result<(), String> {
let mut turns = match turn_source {
ThreadTurnSource::RolloutPath(rollout_path) => {
read_rollout_items_from_rollout(rollout_path)
.await
.map(|items| build_turns_from_rollout_items(&items))
.map_err(|err| {
format!(
"failed to load rollout `{}` for thread {}: {err}",
rollout_path.display(),
thread.id
)
})?
}
ThreadTurnSource::HistoryItems(items) => build_turns_from_rollout_items(items),
};
if let Some(active_turn) = active_turn {
@@ -9372,36 +9397,6 @@ fn validate_dynamic_tools(tools: &[ApiDynamicToolSpec]) -> Result<(), String> {
Ok(())
}
async fn read_history_cwd_from_state_db(
config: &Config,
thread_id: Option<ThreadId>,
rollout_path: &Path,
) -> Option<PathBuf> {
if let Some(state_db_ctx) = get_state_db(config).await
&& let Some(thread_id) = thread_id
&& let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await
{
return Some(metadata.cwd);
}
match read_session_meta_line(rollout_path).await {
Ok(meta_line) => Some(meta_line.meta.cwd),
Err(err) => {
let rollout_path = rollout_path.display();
warn!("failed to read session metadata from rollout {rollout_path}: {err}");
None
}
}
}
async fn read_summary_from_state_db_by_thread_id(
config: &Config,
thread_id: ThreadId,
) -> Option<ConversationSummary> {
let state_db_ctx = open_state_db_for_direct_thread_lookup(config).await;
read_summary_from_state_db_context_by_thread_id(state_db_ctx.as_ref(), thread_id).await
}
async fn read_summary_from_state_db_context_by_thread_id(
state_db_ctx: Option<&StateDbHandle>,
thread_id: ThreadId,
@@ -9495,6 +9490,26 @@ fn thread_store_list_error(err: ThreadStoreError) -> JSONRPCErrorError {
}
}
fn thread_store_resume_read_error(err: ThreadStoreError) -> JSONRPCErrorError {
match err {
ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message,
data: None,
},
ThreadStoreError::ThreadNotFound { thread_id } => JSONRPCErrorError {
code: INVALID_REQUEST_ERROR_CODE,
message: format!("no rollout found for thread id {thread_id}"),
data: None,
},
err => JSONRPCErrorError {
code: INTERNAL_ERROR_CODE,
message: format!("failed to read thread: {err}"),
data: None,
},
}
}
fn conversation_summary_thread_id_read_error(
conversation_id: ThreadId,
err: ThreadStoreError,
@@ -9914,45 +9929,6 @@ fn map_git_info(git_info: &CoreGitInfo) -> ConversationGitInfo {
}
}
async fn load_thread_summary_for_rollout(
config: &Config,
thread_id: ThreadId,
rollout_path: &Path,
fallback_provider: &str,
persisted_metadata: Option<&ThreadMetadata>,
) -> std::result::Result<Thread, String> {
let mut thread = read_summary_from_rollout(rollout_path, fallback_provider)
.await
.map(|summary| summary_to_thread(summary, &config.cwd))
.map_err(|err| {
format!(
"failed to load rollout `{}` for thread {thread_id}: {err}",
rollout_path.display()
)
})?;
thread.forked_from_id = forked_from_id_from_rollout(rollout_path).await;
if let Some(persisted_metadata) = persisted_metadata {
merge_mutable_thread_metadata(
&mut thread,
summary_to_thread(
summary_from_thread_metadata(persisted_metadata),
&config.cwd,
),
);
} else if let Some(summary) = read_summary_from_state_db_by_thread_id(config, thread_id).await {
merge_mutable_thread_metadata(&mut thread, summary_to_thread(summary, &config.cwd));
}
let title = if let Some(metadata) = persisted_metadata {
non_empty_title(metadata)
} else {
title_from_state_db(config, thread_id).await
};
if let Some(title) = title {
set_thread_name_from_title(&mut thread, title);
}
Ok(thread)
}
async fn forked_from_id_from_rollout(path: &Path) -> Option<String> {
read_session_meta_line(path)
.await
@@ -9961,10 +9937,6 @@ async fn forked_from_id_from_rollout(path: &Path) -> Option<String> {
.map(|thread_id| thread_id.to_string())
}
fn merge_mutable_thread_metadata(thread: &mut Thread, persisted_thread: Thread) {
thread.git_info = persisted_thread.git_info;
}
fn preview_from_rollout_items(items: &[RolloutItem]) -> String {
items
.iter()
codex-rs/app-server/src/thread_state.rs
+2 -2
View File
@@ -8,10 +8,10 @@ use codex_core::CodexThread;
use codex_core::ThreadConfigSnapshot;
use codex_protocol::ThreadId;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Weak;
use tokio::sync::Mutex;
@@ -27,7 +27,7 @@ type PendingInterruptQueue = Vec<(
pub(crate) struct PendingThreadResumeRequest {
pub(crate) request_id: ConnectionRequestId,
pub(crate) rollout_path: PathBuf,
pub(crate) history_items: Vec<RolloutItem>,
pub(crate) config_snapshot: ThreadConfigSnapshot,
pub(crate) instruction_sources: Vec<AbsolutePathBuf>,
pub(crate) thread_summary: codex_app_server_protocol::Thread,
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+111
View File
@@ -32,6 +32,7 @@ use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use std::path::Path;
use std::path::PathBuf;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
@@ -49,6 +50,7 @@ use super::analytics::wait_for_analytics_payload;
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);
const INTERNAL_ERROR_CODE: i64 = -32603;
#[tokio::test]
async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
@@ -195,6 +197,88 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
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()?;
create_config_toml(codex_home.path(), &server.uri())?;
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 = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).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 fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
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_by_path_uses_remote_thread_store_error() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_remote_thread_store(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: "not-a-valid-thread-id".to_string(),
path: Some(PathBuf::from("sessions/2025/01/05/rollout.jsonl")),
..Default::default()
})
.await?;
let fork_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(fork_id)),
)
.await??;
assert_eq!(fork_err.error.code, INTERNAL_ERROR_CODE);
assert_eq!(
fork_err.error.message,
"failed to read thread: thread-store internal error: remote thread store does not support read_thread_by_rollout_path"
);
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;
@@ -678,6 +762,33 @@ stream_max_retries = 0
)
}
fn create_config_toml_with_remote_thread_store(
codex_home: &Path,
server_uri: &str,
) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
format!(
r#"
model = "mock-model"
approval_policy = "never"
sandbox_mode = "read-only"
experimental_thread_store_endpoint = "http://127.0.0.1:1"
model_provider = "mock_provider"
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}
fn create_config_toml_with_chatgpt_base_url(
codex_home: &Path,
server_uri: &str,
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+142 -11
View File
@@ -2,6 +2,7 @@ use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::McpProcess;
use app_test_support::create_apply_patch_sse_response;
use app_test_support::create_fake_rollout;
use app_test_support::create_fake_rollout_with_text_elements;
use app_test_support::create_fake_rollout_with_token_usage;
use app_test_support::create_final_assistant_message_sse_response;
@@ -87,6 +88,7 @@ use super::analytics::wait_for_analytics_payload;
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);
const INTERNAL_ERROR_CODE: i64 = -32603;
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
async fn wait_for_responses_request_count(
@@ -324,6 +326,37 @@ async fn thread_resume_can_skip_turns_for_metadata_only_resume() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_resume_by_path_uses_remote_thread_store_error() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_remote_thread_store(codex_home.path(), &server.uri())?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: "ignored-when-path-is-present".to_string(),
path: Some(PathBuf::from("sessions/2025/01/05/rollout.jsonl")),
..Default::default()
})
.await?;
let resume_err: JSONRPCError = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_error_message(RequestId::Integer(resume_id)),
)
.await??;
assert_eq!(resume_err.error.code, INTERNAL_ERROR_CODE);
assert_eq!(
resume_err.error.message,
"failed to read thread: thread-store internal error: remote thread store does not support read_thread_by_rollout_path"
);
Ok(())
}
#[tokio::test]
async fn thread_resume_emits_restored_token_usage_before_next_turn() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -978,6 +1011,22 @@ async fn thread_resume_without_overrides_does_not_change_updated_at_or_mtime() -
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let read_id = mcp
.send_thread_read_request(ThreadReadParams {
thread_id: thread_id.clone(),
include_turns: false,
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse {
thread: before_resume,
..
} = to_response::<ThreadReadResponse>(read_resp)?;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread_id.clone(),
@@ -991,7 +1040,7 @@ async fn thread_resume_without_overrides_does_not_change_updated_at_or_mtime() -
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(thread.updated_at, rollout.expected_updated_at);
assert_eq!(thread.updated_at, before_resume.updated_at);
assert_eq!(thread.status, ThreadStatus::Idle);
let after_modified = std::fs::metadata(&rollout.rollout_file_path)?.modified()?;
@@ -1842,13 +1891,11 @@ async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Re
mut mcp,
thread_id,
rollout_file_path,
updated_at,
} = start_materialized_thread_and_restart(codex_home.path(), "materialize").await?;
let expected_updated_at_rfc3339 = "2025-01-07T00:00:00Z";
set_rollout_mtime(rollout_file_path.as_path(), expected_updated_at_rfc3339)?;
let before_modified = std::fs::metadata(&rollout_file_path)?.modified()?;
let expected_updated_at = chrono::DateTime::parse_from_rfc3339(expected_updated_at_rfc3339)?
.with_timezone(&Utc)
.timestamp();
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
@@ -1867,7 +1914,7 @@ async fn thread_resume_with_overrides_defers_updated_at_until_turn_start() -> Re
..
} = to_response::<ThreadResumeResponse>(resume_resp)?;
assert_eq!(resumed_thread.updated_at, expected_updated_at);
assert_eq!(resumed_thread.updated_at, updated_at);
assert_eq!(resumed_thread.status, ThreadStatus::Idle);
let after_resume_modified = std::fs::metadata(&rollout_file_path)?.modified()?;
@@ -2098,6 +2145,49 @@ async fn thread_resume_prefers_path_over_thread_id() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_resume_can_load_source_by_external_path() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
let external_home = TempDir::new()?;
create_config_toml(codex_home.path(), &server.uri())?;
let thread_id = create_fake_rollout(
external_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"external path history",
Some("mock_provider"),
/*git_info*/ None,
)?;
let thread_path = rollout_path(external_home.path(), "2025-01-05T12-00-00", &thread_id);
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: "not-a-valid-thread-id".to_string(),
path: Some(thread_path.clone()),
..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: resumed, ..
} = to_response::<ThreadResumeResponse>(resume_resp)?;
let expected_thread_path = std::fs::canonicalize(&thread_path)?;
assert_eq!(resumed.id, thread_id);
assert_eq!(resumed.path, Some(expected_thread_path));
assert_eq!(resumed.preview, "external path history");
assert_eq!(resumed.status, ThreadStatus::Idle);
Ok(())
}
#[tokio::test]
async fn thread_resume_supports_history_and_overrides() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -2151,6 +2241,7 @@ struct RestartedThreadFixture {
mcp: McpProcess,
thread_id: String,
rollout_file_path: PathBuf,
updated_at: i64,
}
async fn start_materialized_thread_and_restart(
@@ -2194,10 +2285,24 @@ async fn start_materialized_thread_and_restart(
)
.await??;
let read_id = first_mcp
.send_thread_read_request(ThreadReadParams {
thread_id: thread.id.clone(),
include_turns: false,
})
.await?;
let read_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
first_mcp.read_stream_until_response_message(RequestId::Integer(read_id)),
)
.await??;
let ThreadReadResponse { thread, .. } = to_response::<ThreadReadResponse>(read_resp)?;
let thread_id = thread.id;
let rollout_file_path = thread
.path
.ok_or_else(|| anyhow::anyhow!("thread path missing from thread/start response"))?;
let updated_at = thread.updated_at;
drop(first_mcp);
@@ -2208,6 +2313,7 @@ async fn start_materialized_thread_and_restart(
mcp: second_mcp,
thread_id,
rollout_file_path: rollout_file_path.to_path_buf(),
updated_at,
})
}
@@ -2357,6 +2463,37 @@ stream_max_retries = 0
)
}
fn create_config_toml_with_remote_thread_store(
codex_home: &std::path::Path,
server_uri: &str,
) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");
std::fs::write(
config_toml,
format!(
r#"
model = "gpt-5.3-codex"
approval_policy = "never"
sandbox_mode = "read-only"
experimental_thread_store_endpoint = "http://127.0.0.1:1"
model_provider = "mock_provider"
[features]
personality = true
general_analytics = true
[model_providers.mock_provider]
name = "Mock provider for test"
base_url = "{server_uri}/v1"
wire_api = "responses"
request_max_retries = 0
stream_max_retries = 0
"#
),
)
}
fn create_config_toml_with_chatgpt_base_url(
codex_home: &std::path::Path,
server_uri: &str,
@@ -2443,7 +2580,6 @@ struct RolloutFixture {
conversation_id: String,
rollout_file_path: PathBuf,
before_modified: std::time::SystemTime,
expected_updated_at: i64,
}
fn setup_rollout_fixture(codex_home: &Path, server_uri: &str) -> Result<RolloutFixture> {
@@ -2465,14 +2601,9 @@ fn setup_rollout_fixture(codex_home: &Path, server_uri: &str) -> Result<RolloutF
let rollout_file_path = rollout_path(codex_home, filename_ts, &conversation_id);
set_rollout_mtime(rollout_file_path.as_path(), expected_updated_at_rfc3339)?;
let before_modified = std::fs::metadata(&rollout_file_path)?.modified()?;
let expected_updated_at = chrono::DateTime::parse_from_rfc3339(expected_updated_at_rfc3339)?
.with_timezone(&Utc)
.timestamp();
Ok(RolloutFixture {
conversation_id,
rollout_file_path,
before_modified,
expected_updated_at,
})
}
codex-rs/core/src/session/rollout_reconstruction_tests.rs
+12 -12
View File
@@ -88,7 +88,7 @@ async fn record_initial_history_resumed_bare_turn_context_does_not_hydrate_previ
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -162,7 +162,7 @@ async fn record_initial_history_resumed_hydrates_previous_turn_settings_from_lif
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -695,7 +695,7 @@ async fn record_initial_history_resumed_rollback_skips_only_user_turns() {
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -769,7 +769,7 @@ async fn record_initial_history_resumed_rollback_drops_incomplete_user_turn_comp
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -798,7 +798,7 @@ async fn record_initial_history_resumed_bare_turn_context_does_not_seed_referenc
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -821,7 +821,7 @@ async fn record_initial_history_resumed_does_not_seed_reference_context_item_aft
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -975,7 +975,7 @@ async fn record_initial_history_resumed_turn_context_after_compaction_reestablis
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -1109,7 +1109,7 @@ async fn record_initial_history_resumed_aborted_turn_without_id_clears_active_tu
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -1225,7 +1225,7 @@ async fn record_initial_history_resumed_unmatched_abort_preserves_active_turn_fo
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -1330,7 +1330,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_compaction_clea
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -1377,7 +1377,7 @@ async fn record_initial_history_resumed_trailing_incomplete_turn_preserves_turn_
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -1493,7 +1493,7 @@ async fn record_initial_history_resumed_replaced_incomplete_compacted_turn_clear
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
codex-rs/core/src/session/session.rs
+1 -1
View File
@@ -329,7 +329,7 @@ impl Session {
Arc::clone(&thread_store),
ResumeThreadParams {
thread_id: resumed_history.conversation_id,
rollout_path: Some(resumed_history.rollout_path.clone()),
rollout_path: resumed_history.rollout_path.clone(),
history: Some(resumed_history.history.clone()),
include_archived: true,
event_persistence_mode,
codex-rs/core/src/session/tests.rs
+3 -3
View File
@@ -1268,7 +1268,7 @@ async fn record_initial_history_reconstructs_resumed_transcript() {
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -1297,7 +1297,7 @@ async fn resumed_history_injects_initial_context_on_first_context_update_only()
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
@@ -1390,7 +1390,7 @@ async fn record_initial_history_seeds_token_info_from_rollout() {
.record_initial_history(InitialHistory::Resumed(ResumedHistory {
conversation_id: ThreadId::default(),
history: rollout_items,
rollout_path: PathBuf::from("/tmp/resume.jsonl"),
rollout_path: Some(PathBuf::from("/tmp/resume.jsonl")),
}))
.await;
codex-rs/core/src/thread_manager.rs
+71 -23
View File
@@ -745,30 +745,48 @@ impl ThreadManager {
{
let snapshot = snapshot.into();
let history = RolloutRecorder::get_rollout_history(&path).await?;
let snapshot_state = snapshot_turn_state(&history);
self.fork_thread_from_history(
snapshot,
config,
history,
persist_extended_history,
parent_trace,
)
.await
}
/// Fork an existing thread from already-loaded store history.
pub async fn fork_thread_from_history<S>(
&self,
snapshot: S,
config: Config,
history: InitialHistory,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread>
where
S: Into<ForkSnapshot>,
{
self.fork_thread_with_initial_history(
snapshot.into(),
config,
history,
persist_extended_history,
parent_trace,
)
.await
}
async fn fork_thread_with_initial_history(
&self,
snapshot: ForkSnapshot,
config: Config,
history: InitialHistory,
persist_extended_history: bool,
parent_trace: Option<W3cTraceContext>,
) -> CodexResult<NewThread> {
let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config);
let history = match snapshot {
ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => {
truncate_before_nth_user_message(history, nth_user_message, &snapshot_state)
}
ForkSnapshot::Interrupted => {
let history = match history {
InitialHistory::New => InitialHistory::New,
InitialHistory::Cleared => InitialHistory::Cleared,
InitialHistory::Forked(history) => InitialHistory::Forked(history),
InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history),
};
if snapshot_state.ends_mid_turn {
append_interrupted_boundary(
history,
snapshot_state.active_turn_id,
interrupted_marker,
)
} else {
history
}
}
};
let history = fork_history_from_snapshot(snapshot, history, interrupted_marker);
let thread_store = configured_thread_store(&config);
let environments = default_thread_environment_selections(
self.state.environment_manager.as_ref(),
@@ -1228,6 +1246,36 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState {
}
}
fn fork_history_from_snapshot(
snapshot: ForkSnapshot,
history: InitialHistory,
interrupted_marker: InterruptedTurnHistoryMarker,
) -> InitialHistory {
let snapshot_state = snapshot_turn_state(&history);
match snapshot {
ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => {
truncate_before_nth_user_message(history, nth_user_message, &snapshot_state)
}
ForkSnapshot::Interrupted => {
let history = match history {
InitialHistory::New => InitialHistory::New,
InitialHistory::Cleared => InitialHistory::Cleared,
InitialHistory::Forked(history) => InitialHistory::Forked(history),
InitialHistory::Resumed(resumed) => InitialHistory::Forked(resumed.history),
};
if snapshot_state.ends_mid_turn {
append_interrupted_boundary(
history,
snapshot_state.active_turn_id,
interrupted_marker,
)
} else {
history
}
}
}
}
/// Append the same persisted interrupt boundary used by the live interrupt path
/// to an existing fork snapshot after the source thread has been confirmed to
/// be mid-turn.
codex-rs/core/tests/suite/fork_thread.rs
+104 -22
View File
@@ -3,7 +3,9 @@ use codex_core::NewThread;
use codex_core::parse_turn_item;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::InitialHistory;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::ResumedHistory;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::user_input::UserInput;
@@ -66,27 +68,9 @@ async fn fork_thread_twice_drops_to_first_message() {
// GetHistory flushes before returning the path; no wait needed.
// Helper: read rollout items (excluding SessionMeta) from a JSONL path.
let read_items = |p: &std::path::Path| -> Vec<RolloutItem> {
let text = std::fs::read_to_string(p).expect("read rollout file");
let mut items: Vec<RolloutItem> = Vec::new();
for line in text.lines() {
if line.trim().is_empty() {
continue;
}
let v: serde_json::Value = serde_json::from_str(line).expect("jsonl line");
let rl: RolloutLine = serde_json::from_value(v).expect("rollout line");
match rl.item {
RolloutItem::SessionMeta(_) => {}
other => items.push(other),
}
}
items
};
// Compute expected prefixes after each fork by truncating base rollout
// strictly before the nth user input (0-based).
let base_items = read_items(&base_path);
let base_items = read_rollout_items(&base_path);
let find_user_input_positions = |items: &[RolloutItem]| -> Vec<usize> {
let mut pos = Vec::new();
for (i, it) in items.iter().enumerate() {
@@ -126,7 +110,7 @@ async fn fork_thread_twice_drops_to_first_message() {
let fork1_path = codex_fork1.rollout_path().expect("rollout path");
// GetHistory on fork1 flushed; the file is ready.
let fork1_items = read_items(&fork1_path);
let fork1_items = read_rollout_items(&fork1_path);
pretty_assertions::assert_eq!(
serde_json::to_value(&fork1_items).unwrap(),
serde_json::to_value(&expected_after_first).unwrap()
@@ -149,16 +133,114 @@ async fn fork_thread_twice_drops_to_first_message() {
let fork2_path = codex_fork2.rollout_path().expect("rollout path");
// GetHistory on fork2 flushed; the file is ready.
let fork1_items = read_items(&fork1_path);
let fork1_items = read_rollout_items(&fork1_path);
let fork1_user_inputs = find_user_input_positions(&fork1_items);
let cut_last_on_fork1 = fork1_user_inputs
.get(fork1_user_inputs.len().saturating_sub(1))
.copied()
.unwrap_or(0);
let expected_after_second: Vec<RolloutItem> = fork1_items[..cut_last_on_fork1].to_vec();
let fork2_items = read_items(&fork2_path);
let fork2_items = read_rollout_items(&fork2_path);
pretty_assertions::assert_eq!(
serde_json::to_value(&fork2_items).unwrap(),
serde_json::to_value(&expected_after_second).unwrap()
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn fork_thread_from_history_does_not_require_source_rollout_path() {
skip_if_no_network!();
let server = MockServer::start().await;
let sse = sse(vec![ev_response_created("resp"), ev_completed("resp")]);
Mock::given(method("POST"))
.and(path("/v1/responses"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("content-type", "text/event-stream")
.set_body_raw(sse, "text/event-stream"),
)
.expect(1)
.mount(&server)
.await;
let mut builder = test_codex();
let test = builder.build(&server).await.expect("create conversation");
let codex = test.codex.clone();
let thread_manager = test.thread_manager.clone();
codex
.submit(Op::UserInput {
environments: None,
items: vec![UserInput::Text {
text: "fork me from stored history".to_string(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await
.unwrap();
let _ = wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
let source_path = codex.rollout_path().expect("source rollout path");
let source_items = read_rollout_items(&source_path);
let NewThread {
thread: forked_thread,
..
} = thread_manager
.fork_thread_from_history(
ForkSnapshot::Interrupted,
test.config,
InitialHistory::Resumed(ResumedHistory {
conversation_id: test.session_configured.session_id,
history: source_items.clone(),
rollout_path: None,
}),
/*persist_extended_history*/ false,
/*parent_trace*/ None,
)
.await
.expect("fork from stored history");
let forked_path = forked_thread.rollout_path().expect("forked rollout path");
let forked_items = read_rollout_items(&forked_path);
let forked_items = forked_items
.iter()
.map(|item| serde_json::to_value(item).unwrap())
.collect::<Vec<_>>();
let source_items = source_items
.iter()
.map(|item| serde_json::to_value(item).unwrap())
.collect::<Vec<_>>();
assert!(
forked_items.starts_with(&source_items),
"forked history should start with the supplied source history"
);
}
fn read_rollout_items(path: &std::path::Path) -> Vec<RolloutItem> {
let text = match std::fs::read_to_string(path) {
Ok(text) => text,
Err(err) => panic!("failed to read rollout file {}: {err}", path.display()),
};
let mut items: Vec<RolloutItem> = Vec::new();
for line in text.lines() {
if line.trim().is_empty() {
continue;
}
let v: serde_json::Value = match serde_json::from_str(line) {
Ok(value) => value,
Err(err) => panic!("failed to parse rollout JSON line `{line}`: {err}"),
};
let rl: RolloutLine = match serde_json::from_value(v) {
Ok(line) => line,
Err(err) => panic!("failed to parse rollout line `{line}`: {err}"),
};
match rl.item {
RolloutItem::SessionMeta(_) => {}
other => items.push(other),
}
}
items
}
codex-rs/core/tests/suite/resume_warning.rs
+1 -1
View File
@@ -74,7 +74,7 @@ fn resume_history(
time_to_first_token_ms: None,
})),
],
rollout_path: rollout_path.to_path_buf(),
rollout_path: Some(rollout_path.to_path_buf()),
})
}
codex-rs/protocol/src/protocol.rs
+1 -1
View File
@@ -2640,7 +2640,7 @@ pub struct ConversationPathResponseEvent {
pub struct ResumedHistory {
pub conversation_id: ThreadId,
pub history: Vec<RolloutItem>,
pub rollout_path: PathBuf,
pub rollout_path: Option<PathBuf>,
}
#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)]
codex-rs/rollout/src/recorder.rs
+1 -1
View File
@@ -913,7 +913,7 @@ impl RolloutRecorder {
Ok(InitialHistory::Resumed(ResumedHistory {
conversation_id,
history: items,
rollout_path: path.to_path_buf(),
rollout_path: Some(path.to_path_buf()),
}))
}
codex-rs/thread-store/src/lib.rs
+1
View File
@@ -25,6 +25,7 @@ pub use types::GitInfoPatch;
pub use types::ListThreadsParams;
pub use types::LoadThreadHistoryParams;
pub use types::OptionalStringPatch;
pub use types::ReadThreadByRolloutPathParams;
pub use types::ReadThreadParams;
pub use types::ResumeThreadParams;
pub use types::SortDirection;
codex-rs/thread-store/src/local/mod.rs
+59
View File
@@ -27,6 +27,7 @@ use crate::ArchiveThreadParams;
use crate::CreateThreadParams;
use crate::ListThreadsParams;
use crate::LoadThreadHistoryParams;
use crate::ReadThreadByRolloutPathParams;
use crate::ReadThreadParams;
use crate::ResumeThreadParams;
use crate::StoredThread;
@@ -207,6 +208,19 @@ impl ThreadStore for LocalThreadStore {
read_thread::read_thread(self, params).await
}
async fn read_thread_by_rollout_path(
&self,
params: ReadThreadByRolloutPathParams,
) -> ThreadStoreResult<StoredThread> {
read_thread::read_thread_by_rollout_path(
self,
params.rollout_path,
params.include_archived,
params.include_history,
)
.await
}
async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult<ThreadPage> {
list_threads::list_threads(self, params).await
}
@@ -508,6 +522,51 @@ mod tests {
}));
}
#[tokio::test]
async fn read_thread_by_rollout_path_includes_history() {
let home = TempDir::new().expect("temp dir");
let store = LocalThreadStore::new(test_config(home.path()));
let thread_id = ThreadId::default();
store
.create_thread(create_thread_params(thread_id))
.await
.expect("create thread");
store
.append_items(AppendThreadItemsParams {
thread_id,
items: vec![user_message_item("path read")],
})
.await
.expect("append item");
store.flush_thread(thread_id).await.expect("flush thread");
let rollout_path = store
.live_rollout_path(thread_id)
.await
.expect("load rollout path");
let thread = store
.read_thread_by_rollout_path(
rollout_path,
/*include_archived*/ true,
/*include_history*/ true,
)
.await
.expect("read thread by rollout path");
assert_eq!(thread.thread_id, thread_id);
assert_eq!(
thread
.history
.expect("history")
.items
.into_iter()
.filter(|item| matches!(item, RolloutItem::EventMsg(EventMsg::UserMessage(_))))
.count(),
1
);
}
fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams {
CreateThreadParams {
thread_id,
codex-rs/thread-store/src/local/read_thread.rs
+119
View File
@@ -29,6 +29,13 @@ pub(super) async fn read_thread(
let thread_id = params.thread_id;
if let Some(metadata) = read_sqlite_metadata(store, thread_id).await
&& (params.include_archived || metadata.archived_at.is_none())
&& (!params.include_history
|| sqlite_rollout_path_can_load_history_for_thread(
store,
&metadata.rollout_path,
thread_id,
)
.await)
{
let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await;
attach_history_if_requested(&mut thread, params.include_history).await?;
@@ -46,6 +53,22 @@ pub(super) async fn read_thread(
Ok(thread)
}
async fn sqlite_rollout_path_can_load_history_for_thread(
store: &LocalThreadStore,
path: &std::path::Path,
thread_id: codex_protocol::ThreadId,
) -> bool {
if !tokio::fs::try_exists(path).await.unwrap_or(false) {
return false;
}
// SQLite metadata can outlive a moved/recreated rollout path. When history is
// requested, verify the path still resolves to the requested thread before
// trusting it as the source replay.
read_thread_from_rollout_path(store, path.to_path_buf())
.await
.is_ok_and(|thread| thread.thread_id == thread_id)
}
pub(super) async fn read_thread_by_rollout_path(
store: &LocalThreadStore,
rollout_path: std::path::PathBuf,
@@ -640,6 +663,102 @@ mod tests {
assert_eq!(history.items.len(), 1);
}
#[tokio::test]
async fn read_thread_falls_back_to_rollout_search_when_sqlite_path_is_stale() {
let home = TempDir::new().expect("temp dir");
let external = TempDir::new().expect("external temp dir");
let config = test_config(home.path());
let store = LocalThreadStore::new(config.clone());
let uuid = Uuid::from_u128(220);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let rollout_path =
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
let stale_path = external.path().join("missing-rollout.jsonl");
let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(),
config.model_provider_id.clone(),
)
.await
.expect("state db should initialize");
let mut builder = ThreadMetadataBuilder::new(
thread_id,
stale_path.clone(),
Utc::now(),
SessionSource::Cli,
);
builder.model_provider = Some("stale-sqlite-provider".to_string());
let mut metadata = builder.build(config.model_provider_id.as_str());
metadata.first_user_message = Some("stale sqlite preview".to_string());
runtime
.upsert_thread(&metadata)
.await
.expect("state db upsert should succeed");
let thread = store
.read_thread(ReadThreadParams {
thread_id,
include_archived: true,
include_history: true,
})
.await
.expect("read thread");
assert_eq!(thread.thread_id, thread_id);
assert_eq!(thread.rollout_path, Some(rollout_path));
assert_eq!(thread.preview, "Hello from user");
assert_eq!(thread.model_provider, config.model_provider_id);
let history = thread.history.expect("history should load");
assert_eq!(history.thread_id, thread_id);
assert_eq!(history.items.len(), 2);
}
#[tokio::test]
async fn read_thread_falls_back_when_sqlite_path_points_to_another_thread() {
let home = TempDir::new().expect("temp dir");
let external = TempDir::new().expect("external temp dir");
let config = test_config(home.path());
let store = LocalThreadStore::new(config.clone());
let uuid = Uuid::from_u128(221);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let rollout_path =
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
let other_uuid = Uuid::from_u128(222);
let stale_path = write_session_file(external.path(), "2025-01-04T12-00-00", other_uuid)
.expect("other session file");
let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(),
config.model_provider_id.clone(),
)
.await
.expect("state db should initialize");
let mut builder =
ThreadMetadataBuilder::new(thread_id, stale_path, Utc::now(), SessionSource::Cli);
builder.model_provider = Some("wrong-sqlite-provider".to_string());
let mut metadata = builder.build(config.model_provider_id.as_str());
metadata.first_user_message = Some("wrong sqlite preview".to_string());
runtime
.upsert_thread(&metadata)
.await
.expect("state db upsert should succeed");
let thread = store
.read_thread(ReadThreadParams {
thread_id,
include_archived: true,
include_history: true,
})
.await
.expect("read thread");
assert_eq!(thread.thread_id, thread_id);
assert_eq!(thread.rollout_path, Some(rollout_path));
assert_eq!(thread.preview, "Hello from user");
assert_eq!(thread.model_provider, config.model_provider_id);
let history = thread.history.expect("history should load");
assert_eq!(history.thread_id, thread_id);
assert_eq!(history.items.len(), 2);
}
#[tokio::test]
async fn read_thread_uses_session_meta_for_rollout_without_user_preview_or_sqlite_metadata() {
let home = TempDir::new().expect("temp dir");
codex-rs/thread-store/src/remote/mod.rs
+14
View File
@@ -9,6 +9,7 @@ use crate::ArchiveThreadParams;
use crate::CreateThreadParams;
use crate::ListThreadsParams;
use crate::LoadThreadHistoryParams;
use crate::ReadThreadByRolloutPathParams;
use crate::ReadThreadParams;
use crate::ResumeThreadParams;
use crate::StoredThread;
@@ -25,6 +26,10 @@ mod proto;
/// gRPC-backed [`ThreadStore`] implementation for deployments whose durable thread data lives
/// outside the app-server process.
///
/// This store is still a work in progress: app-server code should call the generic
/// [`ThreadStore`] methods, and unsupported remote operations will return explicit
/// `not_implemented` errors until the remote API catches up.
#[derive(Clone, Debug)]
pub struct RemoteThreadStore {
endpoint: String,
@@ -187,6 +192,15 @@ impl ThreadStore for RemoteThreadStore {
helpers::stored_thread_from_proto(thread)
}
async fn read_thread_by_rollout_path(
&self,
_params: ReadThreadByRolloutPathParams,
) -> ThreadStoreResult<StoredThread> {
Err(ThreadStoreError::Internal {
message: "remote thread store does not support read_thread_by_rollout_path".to_string(),
})
}
async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult<ThreadPage> {
list_threads::list_threads(self, params).await
}
codex-rs/thread-store/src/store.rs
+10 -2
View File
@@ -7,6 +7,7 @@ use crate::ArchiveThreadParams;
use crate::CreateThreadParams;
use crate::ListThreadsParams;
use crate::LoadThreadHistoryParams;
use crate::ReadThreadByRolloutPathParams;
use crate::ReadThreadParams;
use crate::ResumeThreadParams;
use crate::StoredThread;
@@ -18,8 +19,7 @@ use crate::UpdateThreadMetadataParams;
/// Storage-neutral thread persistence boundary.
#[async_trait]
pub trait ThreadStore: Any + Send + Sync {
/// Return this store as [`Any`] so callers at API boundaries can reject requests that only
/// make sense for a concrete store implementation.
/// Return this store as [`Any`] for implementation-owned escape hatches.
fn as_any(&self) -> &dyn Any;
/// Creates a new live thread.
@@ -56,6 +56,14 @@ pub trait ThreadStore: Any + Send + Sync {
/// Reads a thread summary and optionally its persisted history.
async fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreResult<StoredThread>;
/// Reads a rollout-backed thread by path when the store supports path-addressed lookups.
///
/// Deprecated: new callers should use [`ThreadStore::read_thread`] instead.
async fn read_thread_by_rollout_path(
&self,
params: ReadThreadByRolloutPathParams,
) -> ThreadStoreResult<StoredThread>;
/// Lists stored threads matching the supplied filters.
async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult<ThreadPage>;
codex-rs/thread-store/src/types.rs
+11
View File
@@ -96,6 +96,17 @@ pub struct ReadThreadParams {
pub include_history: bool,
}
/// Parameters for reading a local rollout-backed thread by path.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ReadThreadByRolloutPathParams {
/// Local rollout JSONL path to read.
pub rollout_path: PathBuf,
/// Whether archived threads are eligible.
pub include_archived: bool,
/// Whether persisted rollout items should be included in the response.
pub include_history: bool,
}
/// The sort key to use when listing stored threads.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
pub enum ThreadSortKey {