Skip to content

Support paginated thread forks #35220

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
copyberry merged 1 commit into from
Jul 24, 2026
Merged

Support paginated thread forks #35220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions codex-rs/Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions codex-rs/app-server/src/external_agent_migration/session_importer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -324,6 +324,7 @@ impl ExternalAgentSessionImporter {
selected_capability_roots: Vec::new(),
multi_agent_version: Some(MultiAgentVersion::V1),
history_mode: ThreadHistoryMode::Legacy,
history_base: None,
subagent_history_start_ordinal: None,
initial_window_id: uuid::Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
Expand Down
199 changes: 146 additions & 53 deletions codex-rs/app-server/src/request_processors/thread_processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3999,9 +3999,7 @@ impl ThreadRequestProcessor {
/*include_history*/ false,
)
.await?;
if matches!(source_thread.history_mode, ThreadHistoryMode::Paginated) {
return Err(method_not_found("paginated_threads is not supported yet"));
}
let paginated_source = matches!(source_thread.history_mode, ThreadHistoryMode::Paginated);
if last_turn_id.is_some() && before_turn_id.is_some() {
return Err(invalid_request(
"`beforeTurnId` cannot be combined with `lastTurnId`",
Expand All @@ -4012,35 +4010,69 @@ impl ThreadRequestProcessor {
"`deferGoalContinuation` cannot be combined with `ephemeral`",
));
}
let mut source_thread = self
.read_stored_thread_for_resume(&thread_id, path.as_ref(), /*include_history*/ true)
.await?;
let source_thread_id = source_thread.thread_id;
let source_thread_name = source_thread
.name
.as_deref()
.and_then(codex_core::util::normalize_thread_name);
let source_history_items = source_thread
.history
.take()
.map(|history| history.items)
.ok_or_else(|| {
internal_error(format!(
"thread {source_thread_id} did not include persisted history"
))
})?;
let source_history_items = Arc::new(source_history_items);
let history_items = match (last_turn_id.as_deref(), before_turn_id.as_deref()) {
(Some(last_turn_id), None) => Arc::new(
truncate_rollout_after_turn_id(&source_history_items, last_turn_id)
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
),
(None, Some(before_turn_id)) => Arc::new(
truncate_rollout_before_turn_id(&source_history_items, before_turn_id)
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
),
(None, None) => Arc::clone(&source_history_items),
(Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"),
let prepared_fork = if paginated_source {
if ephemeral {
return Err(invalid_request(
"ephemeral paginated thread/fork is not supported",
));
}
let boundary = match (last_turn_id.as_deref(), before_turn_id.as_deref()) {
(Some(turn_id), None) => {
codex_thread_store::ForkBoundary::ThroughTurn(turn_id.to_string())
}
(None, Some(turn_id)) => {
codex_thread_store::ForkBoundary::BeforeTurn(turn_id.to_string())
}
(None, None) => codex_thread_store::ForkBoundary::Latest,
(Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"),
};
Some(
self.thread_store
.prepare_fork(codex_thread_store::PrepareForkParams {
thread_id: source_thread_id,
boundary,
})
.await
.map_err(|err| match err {
ThreadStoreError::InvalidRequest { message } => invalid_request(message),
ThreadStoreError::ThreadNotFound { thread_id } => {
invalid_request(format!("no rollout found for thread id {thread_id}"))
}
ThreadStoreError::Unsupported { .. } => {
method_not_found("paginated_threads is not supported yet")
}
err => internal_error(format!("failed to prepare paginated fork: {err}")),
})?,
)
} else {
None
};
let source_history_items = if let Some(prepared_fork) = prepared_fork.as_ref() {
Arc::clone(&prepared_fork.model_context)
} else {
let mut source_thread = self
.read_stored_thread_for_resume(
&thread_id,
path.as_ref(),
/*include_history*/ true,
)
.await?;
Arc::new(
source_thread
.history
.take()
.map(|history| history.items)
.ok_or_else(|| {
internal_error(format!(
"thread {source_thread_id} did not include persisted history"
))
})?,
)
};
let history_cwd = Some(source_thread.cwd.clone());

Expand All @@ -4066,11 +4098,6 @@ impl ThreadRequestProcessor {
} else {
Some(cli_overrides)
};
let thread_history = InitialHistory::Resumed(ResumedHistory {
conversation_id: source_thread_id,
history: Arc::clone(&history_items),
rollout_path: source_thread.rollout_path.clone(),
});
let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots);
let mut typesafe_overrides = self.build_thread_config_overrides(
model,
Expand All @@ -4087,8 +4114,37 @@ impl ThreadRequestProcessor {
/*personality*/ None,
);
typesafe_overrides.ephemeral = ephemeral.then_some(true);
let latest_context = if paginated_source
&& typesafe_overrides.approvals_reviewer.is_none()
&& !request_overrides
.as_ref()
.is_some_and(|overrides| overrides.contains_key("approvals_reviewer"))
{
if let Ok(parent) = self.thread_manager.get_thread(source_thread_id).await {
typesafe_overrides.approvals_reviewer =
Some(parent.config_snapshot().await.approvals_reviewer);
None
} else if last_turn_id.is_some() || before_turn_id.is_some() {
Some(
self.thread_store
.load_latest_model_context(StoreLoadThreadHistoryParams {
thread_id: source_thread_id,
include_archived: true,
})
.await
.map_err(thread_store_resume_read_error)?
.items,
)
} else {
None
}
} else {
None
};
merge_persisted_approvals_reviewer(
&source_history_items,
latest_context
.as_deref()
.unwrap_or_else(|| source_history_items.as_ref()),
request_overrides.as_ref(),
&mut typesafe_overrides,
);
Expand All @@ -4101,30 +4157,64 @@ impl ThreadRequestProcessor {
let goals_enabled = config.features.enabled(Feature::Goals);

let fallback_model_provider = config.model_provider_id.clone();
let parent_trace = self.request_trace_context(&request_id).await;
let thread_source = thread_source.map(Into::into);

let (history_items, new_thread) = if let Some(prepared_fork) = prepared_fork {
let history_items = Arc::clone(&prepared_fork.model_context);
let new_thread = self
.thread_manager
.fork_prepared_thread(
config,
prepared_fork,
thread_source,
parent_trace,
supports_openai_form_elicitation,
)
.await;
(history_items, new_thread)
} else {
let history_items = match (last_turn_id.as_deref(), before_turn_id.as_deref()) {
(Some(last_turn_id), None) => Arc::new(
truncate_rollout_after_turn_id(&source_history_items, last_turn_id)
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
),
(None, Some(before_turn_id)) => Arc::new(
truncate_rollout_before_turn_id(&source_history_items, before_turn_id)
.map_err(|err| core_thread_write_error("truncate thread for fork", err))?,
),
(None, None) => Arc::clone(&source_history_items),
(Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"),
};
let new_thread = self
.thread_manager
.fork_thread_from_history(
ForkSnapshot::Interrupted,
config,
InitialHistory::Resumed(ResumedHistory {
conversation_id: source_thread_id,
history: Arc::clone(&history_items),
rollout_path: source_thread.rollout_path.clone(),
}),
thread_source,
parent_trace,
supports_openai_form_elicitation,
)
.await;
(history_items, new_thread)
};
let NewThread {
thread_id,
thread: forked_thread,
session_configured,
..
} = self
.thread_manager
.fork_thread_from_history(
ForkSnapshot::Interrupted,
config,
thread_history,
thread_source.map(Into::into),
self.request_trace_context(&request_id).await,
supports_openai_form_elicitation,
)
.await
.map_err(|err| match err.details() {
CodexErrorDetails::Io(_) | CodexErrorDetails::Json(_) => {
invalid_request(format!("failed to load thread {source_thread_id}: {err}"))
}
CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()),
_ => internal_error(format!("error forking thread: {err}")),
})?;
} = new_thread.map_err(|err| match err.details() {
CodexErrorDetails::Io(_) | CodexErrorDetails::Json(_) => {
invalid_request(format!("failed to load thread {source_thread_id}: {err}"))
}
CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()),
_ => internal_error(format!("error forking thread: {err}")),
})?;

Self::set_app_server_client_info(
forked_thread.as_ref(),
Expand Down Expand Up @@ -4196,12 +4286,12 @@ impl ThreadRequestProcessor {
// pathless, so they rebuild their visible history from the copied source history instead.
let mut thread = if session_configured.rollout_path.is_some() {
let stored_thread = self
.read_stored_thread_for_new_fork(thread_id, include_turns)
.read_stored_thread_for_new_fork(thread_id, include_turns && !paginated_source)
.await?;
self.stored_thread_to_api_thread(
stored_thread,
fallback_model_provider.as_str(),
include_turns,
include_turns && !paginated_source,
)
} else {
let mut thread = build_thread_from_snapshot(
Expand All @@ -4222,6 +4312,9 @@ impl ThreadRequestProcessor {
}
thread
};
if paginated_source && include_turns {
thread.turns = self.paginated_thread_full_turns(thread_id).await?;
}
if let Some(name) = source_thread_name {
set_thread_name_from_title(&mut thread, name);
}
Expand Down
1 change: 1 addition & 0 deletions codex-rs/app-server/tests/suite/conversation_summary.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,7 @@ async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() ->
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: Default::default(),
history_base: None,
subagent_history_start_ordinal: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
Expand Down
1 change: 1 addition & 0 deletions codex-rs/app-server/tests/suite/v2/remote_thread_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,6 +210,7 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist
selected_capability_roots: Vec::new(),
multi_agent_version: None,
history_mode: Default::default(),
history_base: None,
subagent_history_start_ordinal: None,
initial_window_id: Uuid::now_v7().to_string(),
metadata: ThreadPersistenceMetadata {
Expand Down
Loading
Loading