Split resume route ownership
This commit is contained in:
Meta Vens
parent
9c0e24854f
commit
83fa7a46c2
7 changed files with 805 additions and 797 deletions
|
|
@ -51,8 +51,8 @@ remembers doctrine; the branch remembers the blade.
|
|||
- Thread administrative mutation routes now live in `codex_message_processor/thread_admin_routes.rs`: archive/unarchive, elicitation counters, set-name, memory mode, memory reset, metadata update, rollback, compact, background terminal cleanup, shell command submission, and guardian approval. The root processor is about 5,750 lines.
|
||||
- Thread read/list projection routes now live in `codex_message_processor/thread_read_routes.rs`: persisted/live thread listing, loaded-thread listing, `thread/read`, read-list filter normalization/tests, Epiphany state attachment for reads, rollout turn paging, and read-side rollout field hydration.
|
||||
- Thread-store rollout-path reads are now an explicit `ThreadStore::read_thread_by_rollout_path` capability with typed params. App-server read routes no longer downcast `Arc<dyn ThreadStore>` to `LocalThreadStore`; local and remote stores define the capability/rejection boundary, and the local rollout-path read test exercises the trait method.
|
||||
- Thread resume/fork orchestration now lives in `codex_message_processor/thread_resume_routes.rs`; it owns resume from history/rollout, running-thread resume response replay, persisted resume metadata hydration, fork response construction, and the shared thread-title attachment helper. The root processor is about 4,336 lines.
|
||||
- Running-thread resume response composition and resume override metadata helpers now live in `codex_message_processor/thread_resume_routes.rs`; root no longer owns `handle_pending_thread_resume_request`, resume mismatch collection, or persisted resume metadata merging.
|
||||
- Thread resume/fork orchestration has been split by lifecycle contract. `thread_resume_routes.rs` owns cold resume from history/rollout plus persisted resume metadata; `running_thread_resume_routes.rs` owns rejoining already loaded threads through listener replay; `thread_fork_routes.rs` owns new-thread materialization from a source rollout. The root processor is about 1,012 lines and mostly owns JSON-RPC dispatch, shared construction/error helpers, and shutdown lifecycle.
|
||||
- The fork and running-resume splits are backed by focused app-server integration tests: `thread_fork`, `thread_resume_rejects_history_when_thread_is_running`, `thread_resume_rejects_mismatched_path_when_thread_is_running`, and `thread_resume_rejoins_running_thread_even_with_override_mismatch`.
|
||||
- Thread start route handling and its async lifecycle task now live in `codex_message_processor/thread_start_routes.rs`; root no longer owns `thread_start_task`.
|
||||
- Thread config helpers now live in `codex_message_processor/thread_config.rs`: instruction-source loading, thread config override construction, config-load error shaping, and dynamic-tool validation. The pure dynamic-tool validation contract has module-local unit tests.
|
||||
- Turn control routes now live in `codex_message_processor/turn_routes.rs`: turn start, thread item injection, app-server client info setting, turn steer, turn interrupt, mode normalization, and input-limit validation with module-local unit tests.
|
||||
|
|
|
|||
Binary file not shown.
File diff suppressed because one or more lines are too long
|
|
@ -8,8 +8,10 @@ mod listener_lifecycle;
|
|||
mod mcp_routes;
|
||||
mod realtime_routes;
|
||||
mod review_routes;
|
||||
mod running_thread_resume_routes;
|
||||
mod thread_admin_routes;
|
||||
mod thread_config;
|
||||
mod thread_fork_routes;
|
||||
mod thread_projection;
|
||||
mod thread_read_routes;
|
||||
mod thread_resume_routes;
|
||||
|
|
@ -20,10 +22,10 @@ use self::auth_routes::ActiveLogin;
|
|||
pub(crate) use self::epiphany_automation::maybe_run_epiphany_coordinator_automation_for_turn_boundary;
|
||||
pub(crate) use self::epiphany_automation::maybe_run_epiphany_pre_compaction_checkpoint_intervention_for_token_count;
|
||||
pub(super) use self::listener_lifecycle::{EnsureConversationListenerResult, ListenerTaskContext};
|
||||
pub(crate) use self::running_thread_resume_routes::handle_pending_thread_resume_request;
|
||||
pub(crate) use self::thread_config::{config_load_error, validate_dynamic_tools};
|
||||
pub(crate) use self::thread_projection::*;
|
||||
pub(super) use self::thread_read_routes::ThreadReadViewError;
|
||||
pub(crate) use self::thread_resume_routes::handle_pending_thread_resume_request;
|
||||
use crate::bespoke_event_handling::apply_bespoke_event_handling;
|
||||
use crate::bespoke_event_handling::maybe_emit_hook_prompt_item_completed;
|
||||
use crate::command_exec::CommandExecManager;
|
||||
|
|
|
|||
451
vendor/codex/codex-rs/app-server/src/codex_message_processor/running_thread_resume_routes.rs
vendored
Normal file
451
vendor/codex/codex-rs/app-server/src/codex_message_processor/running_thread_resume_routes.rs
vendored
Normal file
|
|
@ -0,0 +1,451 @@
|
|||
use super::*;
|
||||
|
||||
impl CodexMessageProcessor {
|
||||
pub(super) async fn resume_running_thread(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
params: &ThreadResumeParams,
|
||||
) -> bool {
|
||||
if let Ok(existing_thread_id) = ThreadId::from_string(¶ms.thread_id)
|
||||
&& let Ok(existing_thread) = self.thread_manager.get_thread(existing_thread_id).await
|
||||
{
|
||||
if params.history.is_some() {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!(
|
||||
"cannot resume thread {existing_thread_id} with history while it is already running"
|
||||
),
|
||||
)
|
||||
.await;
|
||||
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
|
||||
{
|
||||
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()
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return true;
|
||||
}
|
||||
|
||||
let thread_state = self
|
||||
.thread_state_manager
|
||||
.thread_state(existing_thread_id)
|
||||
.await;
|
||||
if let Err(error) = self
|
||||
.ensure_listener_task_running(
|
||||
existing_thread_id,
|
||||
existing_thread.clone(),
|
||||
thread_state.clone(),
|
||||
ApiVersion::V2,
|
||||
)
|
||||
.await
|
||||
{
|
||||
self.outgoing.send_error(request_id, error).await;
|
||||
return true;
|
||||
}
|
||||
|
||||
let config_snapshot = existing_thread.config_snapshot().await;
|
||||
let mismatch_details = collect_resume_override_mismatches(params, &config_snapshot);
|
||||
if !mismatch_details.is_empty() {
|
||||
tracing::warn!(
|
||||
"thread/resume overrides ignored for running thread {}: {}",
|
||||
existing_thread_id,
|
||||
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
|
||||
{
|
||||
Ok(thread) => thread,
|
||||
Err(message) => {
|
||||
self.send_internal_error(request_id, message).await;
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
let listener_command_tx = {
|
||||
let thread_state = thread_state.lock().await;
|
||||
thread_state.listener_command_tx()
|
||||
};
|
||||
let Some(listener_command_tx) = listener_command_tx else {
|
||||
let err = JSONRPCErrorError {
|
||||
code: INTERNAL_ERROR_CODE,
|
||||
message: format!(
|
||||
"failed to enqueue running thread resume for thread {existing_thread_id}: thread listener is not running"
|
||||
),
|
||||
data: None,
|
||||
};
|
||||
self.outgoing.send_error(request_id, err).await;
|
||||
return true;
|
||||
};
|
||||
|
||||
let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse(
|
||||
Box::new(crate::thread_state::PendingThreadResumeRequest {
|
||||
request_id: request_id.clone(),
|
||||
rollout_path: rollout_path.clone(),
|
||||
config_snapshot,
|
||||
instruction_sources,
|
||||
thread_summary,
|
||||
}),
|
||||
);
|
||||
if listener_command_tx.send(command).is_err() {
|
||||
let err = JSONRPCErrorError {
|
||||
code: INTERNAL_ERROR_CODE,
|
||||
message: format!(
|
||||
"failed to enqueue running thread resume for thread {existing_thread_id}: thread listener command channel is closed"
|
||||
),
|
||||
data: None,
|
||||
};
|
||||
self.outgoing.send_error(request_id, err).await;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
false
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[expect(
|
||||
clippy::await_holding_invalid_type,
|
||||
reason = "running-thread resume subscription must be serialized against pending unloads"
|
||||
)]
|
||||
pub(crate) async fn handle_pending_thread_resume_request(
|
||||
conversation_id: ThreadId,
|
||||
conversation: &Arc<CodexThread>,
|
||||
_codex_home: &Path,
|
||||
thread_state_manager: &ThreadStateManager,
|
||||
thread_state: &Arc<Mutex<ThreadState>>,
|
||||
thread_watch_manager: &ThreadWatchManager,
|
||||
outgoing: &Arc<OutgoingMessageSender>,
|
||||
pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
|
||||
pending: crate::thread_state::PendingThreadResumeRequest,
|
||||
) {
|
||||
let active_turn = {
|
||||
let state = thread_state.lock().await;
|
||||
state.active_turn_snapshot()
|
||||
};
|
||||
tracing::debug!(
|
||||
thread_id = %conversation_id,
|
||||
request_id = ?pending.request_id,
|
||||
active_turn_present = active_turn.is_some(),
|
||||
active_turn_id = ?active_turn.as_ref().map(|turn| turn.id.as_str()),
|
||||
active_turn_status = ?active_turn.as_ref().map(|turn| &turn.status),
|
||||
"composing running thread resume response"
|
||||
);
|
||||
let has_live_in_progress_turn =
|
||||
matches!(conversation.agent_status().await, AgentStatus::Running)
|
||||
|| active_turn
|
||||
.as_ref()
|
||||
.is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress));
|
||||
|
||||
let request_id = pending.request_id;
|
||||
let connection_id = request_id.connection_id;
|
||||
let mut thread = pending.thread_summary;
|
||||
if let Err(message) = populate_thread_turns(
|
||||
&mut thread,
|
||||
ThreadTurnSource::RolloutPath(pending.rollout_path.as_path()),
|
||||
active_turn.as_ref(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
outgoing
|
||||
.send_error(
|
||||
request_id,
|
||||
JSONRPCErrorError {
|
||||
code: INTERNAL_ERROR_CODE,
|
||||
message,
|
||||
data: None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
let thread_status = thread_watch_manager
|
||||
.loaded_status_for_thread(&thread.id)
|
||||
.await;
|
||||
|
||||
set_thread_status_and_interrupt_stale_turns(
|
||||
&mut thread,
|
||||
thread_status,
|
||||
has_live_in_progress_turn,
|
||||
);
|
||||
|
||||
{
|
||||
let pending_thread_unloads = pending_thread_unloads.lock().await;
|
||||
if pending_thread_unloads.contains(&conversation_id) {
|
||||
drop(pending_thread_unloads);
|
||||
outgoing
|
||||
.send_error(
|
||||
request_id,
|
||||
JSONRPCErrorError {
|
||||
code: INVALID_REQUEST_ERROR_CODE,
|
||||
message: format!(
|
||||
"thread {conversation_id} is closing; retry thread/resume after the thread is closed"
|
||||
),
|
||||
data: None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
if !thread_state_manager
|
||||
.try_add_connection_to_thread(conversation_id, connection_id)
|
||||
.await
|
||||
{
|
||||
tracing::debug!(
|
||||
thread_id = %conversation_id,
|
||||
connection_id = ?connection_id,
|
||||
"skipping running thread resume for closed connection"
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
let ThreadConfigSnapshot {
|
||||
model,
|
||||
model_provider_id,
|
||||
service_tier,
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
sandbox_policy,
|
||||
permission_profile,
|
||||
cwd,
|
||||
reasoning_effort,
|
||||
..
|
||||
} = pending.config_snapshot;
|
||||
let instruction_sources = pending.instruction_sources;
|
||||
let permission_profile =
|
||||
thread_response_permission_profile(&sandbox_policy, permission_profile);
|
||||
|
||||
let response = ThreadResumeResponse {
|
||||
thread,
|
||||
model,
|
||||
model_provider: model_provider_id,
|
||||
service_tier,
|
||||
cwd,
|
||||
instruction_sources,
|
||||
approval_policy: approval_policy.into(),
|
||||
approvals_reviewer: approvals_reviewer.into(),
|
||||
sandbox: sandbox_policy.into(),
|
||||
permission_profile,
|
||||
reasoning_effort,
|
||||
};
|
||||
let token_usage_thread = response.thread.clone();
|
||||
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_path(
|
||||
pending.rollout_path.as_path(),
|
||||
&token_usage_thread,
|
||||
)
|
||||
.await;
|
||||
outgoing.send_response(request_id, response).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(
|
||||
outgoing,
|
||||
connection_id,
|
||||
conversation_id,
|
||||
&token_usage_thread,
|
||||
conversation.as_ref(),
|
||||
token_usage_turn_id,
|
||||
)
|
||||
.await;
|
||||
outgoing
|
||||
.replay_requests_to_connection_for_thread(connection_id, conversation_id)
|
||||
.await;
|
||||
}
|
||||
|
||||
fn collect_resume_override_mismatches(
|
||||
request: &ThreadResumeParams,
|
||||
config_snapshot: &ThreadConfigSnapshot,
|
||||
) -> Vec<String> {
|
||||
let mut mismatch_details = Vec::new();
|
||||
|
||||
if let Some(requested_model) = request.model.as_deref()
|
||||
&& requested_model != config_snapshot.model
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"model requested={requested_model} active={}",
|
||||
config_snapshot.model
|
||||
));
|
||||
}
|
||||
if let Some(requested_provider) = request.model_provider.as_deref()
|
||||
&& requested_provider != config_snapshot.model_provider_id
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"model_provider requested={requested_provider} active={}",
|
||||
config_snapshot.model_provider_id
|
||||
));
|
||||
}
|
||||
if let Some(requested_service_tier) = request.service_tier.as_ref()
|
||||
&& requested_service_tier != &config_snapshot.service_tier
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"service_tier requested={requested_service_tier:?} active={:?}",
|
||||
config_snapshot.service_tier
|
||||
));
|
||||
}
|
||||
if let Some(requested_cwd) = request.cwd.as_deref() {
|
||||
let requested_cwd_path = std::path::PathBuf::from(requested_cwd);
|
||||
if requested_cwd_path != config_snapshot.cwd.as_path() {
|
||||
mismatch_details.push(format!(
|
||||
"cwd requested={} active={}",
|
||||
requested_cwd_path.display(),
|
||||
config_snapshot.cwd.display()
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_approval) = request.approval_policy.as_ref() {
|
||||
let active_approval: AskForApproval = config_snapshot.approval_policy.into();
|
||||
if requested_approval != &active_approval {
|
||||
mismatch_details.push(format!(
|
||||
"approval_policy requested={requested_approval:?} active={active_approval:?}"
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_review_policy) = request.approvals_reviewer.as_ref() {
|
||||
let active_review_policy: codex_app_server_protocol::ApprovalsReviewer =
|
||||
config_snapshot.approvals_reviewer.into();
|
||||
if requested_review_policy != &active_review_policy {
|
||||
mismatch_details.push(format!(
|
||||
"approvals_reviewer requested={requested_review_policy:?} active={active_review_policy:?}"
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_sandbox) = request.sandbox.as_ref() {
|
||||
let sandbox_matches = matches!(
|
||||
(requested_sandbox, &config_snapshot.sandbox_policy),
|
||||
(
|
||||
SandboxMode::ReadOnly,
|
||||
codex_protocol::protocol::SandboxPolicy::ReadOnly { .. }
|
||||
) | (
|
||||
SandboxMode::WorkspaceWrite,
|
||||
codex_protocol::protocol::SandboxPolicy::WorkspaceWrite { .. }
|
||||
) | (
|
||||
SandboxMode::DangerFullAccess,
|
||||
codex_protocol::protocol::SandboxPolicy::DangerFullAccess
|
||||
) | (
|
||||
SandboxMode::DangerFullAccess,
|
||||
codex_protocol::protocol::SandboxPolicy::ExternalSandbox { .. }
|
||||
)
|
||||
);
|
||||
if !sandbox_matches {
|
||||
mismatch_details.push(format!(
|
||||
"sandbox requested={requested_sandbox:?} active={:?}",
|
||||
config_snapshot.sandbox_policy
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_permission_profile) = request.permission_profile.as_ref() {
|
||||
let requested_permission_profile =
|
||||
codex_protocol::models::PermissionProfile::from(requested_permission_profile.clone());
|
||||
if requested_permission_profile != config_snapshot.permission_profile {
|
||||
mismatch_details.push(format!(
|
||||
"permission_profile requested={requested_permission_profile:?} active={:?}",
|
||||
config_snapshot.permission_profile
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_personality) = request.personality.as_ref()
|
||||
&& config_snapshot.personality.as_ref() != Some(requested_personality)
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"personality requested={requested_personality:?} active={:?}",
|
||||
config_snapshot.personality
|
||||
));
|
||||
}
|
||||
|
||||
if request.config.is_some() {
|
||||
mismatch_details
|
||||
.push("config overrides were provided and ignored while running".to_string());
|
||||
}
|
||||
if request.base_instructions.is_some() {
|
||||
mismatch_details
|
||||
.push("baseInstructions override was provided and ignored while running".to_string());
|
||||
}
|
||||
if request.developer_instructions.is_some() {
|
||||
mismatch_details.push(
|
||||
"developerInstructions override was provided and ignored while running".to_string(),
|
||||
);
|
||||
}
|
||||
if request.persist_extended_history {
|
||||
mismatch_details.push(
|
||||
"persistExtendedHistory override was provided and ignored while running".to_string(),
|
||||
);
|
||||
}
|
||||
|
||||
mismatch_details
|
||||
}
|
||||
341
vendor/codex/codex-rs/app-server/src/codex_message_processor/thread_fork_routes.rs
vendored
Normal file
341
vendor/codex/codex-rs/app-server/src/codex_message_processor/thread_fork_routes.rs
vendored
Normal file
|
|
@ -0,0 +1,341 @@
|
|||
use super::*;
|
||||
|
||||
impl CodexMessageProcessor {
|
||||
pub(super) async fn thread_fork(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
params: ThreadForkParams,
|
||||
) {
|
||||
let ThreadForkParams {
|
||||
thread_id,
|
||||
path,
|
||||
model,
|
||||
model_provider,
|
||||
service_tier,
|
||||
cwd,
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
sandbox,
|
||||
permission_profile,
|
||||
config: cli_overrides,
|
||||
base_instructions,
|
||||
developer_instructions,
|
||||
ephemeral,
|
||||
persist_extended_history,
|
||||
} = params;
|
||||
if sandbox.is_some() && permission_profile.is_some() {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
"`permissionProfile` cannot be combined with `sandbox`".to_string(),
|
||||
)
|
||||
.await;
|
||||
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(),
|
||||
)
|
||||
.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;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let history_cwd =
|
||||
read_history_cwd_from_state_db(&self.config, source_thread_id, rollout_path.as_path())
|
||||
.await;
|
||||
|
||||
// Persist Windows sandbox mode.
|
||||
let mut cli_overrides = cli_overrides.unwrap_or_default();
|
||||
if cfg!(windows) {
|
||||
match WindowsSandboxLevel::from_config(&self.config) {
|
||||
WindowsSandboxLevel::Elevated => {
|
||||
cli_overrides
|
||||
.insert("windows.sandbox".to_string(), serde_json::json!("elevated"));
|
||||
}
|
||||
WindowsSandboxLevel::RestrictedToken => {
|
||||
cli_overrides.insert(
|
||||
"windows.sandbox".to_string(),
|
||||
serde_json::json!("unelevated"),
|
||||
);
|
||||
}
|
||||
WindowsSandboxLevel::Disabled => {}
|
||||
}
|
||||
}
|
||||
let request_overrides = if cli_overrides.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(cli_overrides)
|
||||
};
|
||||
let mut typesafe_overrides = self.build_thread_config_overrides(
|
||||
model,
|
||||
model_provider,
|
||||
service_tier,
|
||||
cwd,
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
sandbox,
|
||||
permission_profile,
|
||||
base_instructions,
|
||||
developer_instructions,
|
||||
/*personality*/ None,
|
||||
);
|
||||
typesafe_overrides.ephemeral = ephemeral.then_some(true);
|
||||
// Derive a Config using the same logic as new conversation, honoring overrides if provided.
|
||||
let config = match self
|
||||
.config_manager
|
||||
.load_for_cwd(request_overrides, typesafe_overrides, history_cwd)
|
||||
.await
|
||||
{
|
||||
Ok(config) => config,
|
||||
Err(err) => {
|
||||
self.outgoing
|
||||
.send_error(request_id, config_load_error(&err))
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let fallback_model_provider = config.model_provider_id.clone();
|
||||
let instruction_sources = Self::instruction_sources_from_config(&config).await;
|
||||
|
||||
let NewThread {
|
||||
thread_id,
|
||||
thread: forked_thread,
|
||||
session_configured,
|
||||
..
|
||||
} = match self
|
||||
.thread_manager
|
||||
.fork_thread(
|
||||
ForkSnapshot::Interrupted,
|
||||
config,
|
||||
rollout_path.clone(),
|
||||
persist_extended_history,
|
||||
self.request_trace_context(&request_id).await,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(thread) => thread,
|
||||
Err(err) => {
|
||||
match err {
|
||||
CodexErr::Io(_) | CodexErr::Json(_) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("failed to load rollout `{}`: {err}", rollout_path.display()),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
CodexErr::InvalidRequest(message) => {
|
||||
self.send_invalid_request_error(request_id, message).await;
|
||||
}
|
||||
_ => {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!("error forking thread: {err}"),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Auto-attach a conversation listener when forking a thread.
|
||||
Self::log_listener_attach_result(
|
||||
self.ensure_conversation_listener(
|
||||
thread_id,
|
||||
request_id.connection_id,
|
||||
/*raw_events_enabled*/ false,
|
||||
ApiVersion::V2,
|
||||
)
|
||||
.await,
|
||||
thread_id,
|
||||
request_id.connection_id,
|
||||
"thread",
|
||||
);
|
||||
|
||||
// Persistent forks materialize their own rollout immediately. Ephemeral forks stay
|
||||
// pathless, so they rebuild their visible history from the copied source rollout 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
|
||||
{
|
||||
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) => {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!(
|
||||
"failed to load rollout `{}` for thread {thread_id}: {err}",
|
||||
fork_rollout_path.display()
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let config_snapshot = forked_thread.config_snapshot().await;
|
||||
// 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());
|
||||
if let Err(message) = populate_thread_turns(
|
||||
&mut thread,
|
||||
ThreadTurnSource::HistoryItems(&history_items),
|
||||
/*active_turn*/ None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
self.send_internal_error(request_id, message).await;
|
||||
return;
|
||||
}
|
||||
thread
|
||||
};
|
||||
|
||||
if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref()
|
||||
&& 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;
|
||||
}
|
||||
|
||||
thread.epiphany_state = live_thread_epiphany_state(forked_thread.as_ref()).await;
|
||||
|
||||
self.thread_watch_manager
|
||||
.upsert_thread_silently(thread.clone())
|
||||
.await;
|
||||
|
||||
thread.status = resolve_thread_status(
|
||||
self.thread_watch_manager
|
||||
.loaded_status_for_thread(&thread.id)
|
||||
.await,
|
||||
/*has_in_progress_turn*/ false,
|
||||
);
|
||||
let permission_profile = thread_response_permission_profile(
|
||||
&session_configured.sandbox_policy,
|
||||
forked_thread.config_snapshot().await.permission_profile,
|
||||
);
|
||||
|
||||
let response = ThreadForkResponse {
|
||||
thread: thread.clone(),
|
||||
model: session_configured.model,
|
||||
model_provider: session_configured.model_provider_id,
|
||||
service_tier: session_configured.service_tier,
|
||||
cwd: session_configured.cwd,
|
||||
instruction_sources,
|
||||
approval_policy: session_configured.approval_policy.into(),
|
||||
approvals_reviewer: session_configured.approvals_reviewer.into(),
|
||||
sandbox: session_configured.sandbox_policy.into(),
|
||||
permission_profile,
|
||||
reasoning_effort: session_configured.reasoning_effort,
|
||||
};
|
||||
if self.config.features.enabled(Feature::GeneralAnalytics) {
|
||||
self.analytics_events_client.track_response(
|
||||
request_id.connection_id.0,
|
||||
ClientResponse::ThreadFork {
|
||||
request_id: request_id.request_id.clone(),
|
||||
response: response.clone(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
let connection_id = request_id.connection_id;
|
||||
let token_usage_thread = response.thread.clone();
|
||||
let token_usage_turn_id = if let Some(turn_id) =
|
||||
latest_token_usage_turn_id_for_thread_path(&token_usage_thread).await
|
||||
{
|
||||
Some(turn_id)
|
||||
} else {
|
||||
latest_token_usage_turn_id_from_rollout_path(
|
||||
rollout_path.as_path(),
|
||||
&token_usage_thread,
|
||||
)
|
||||
.await
|
||||
};
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
// Mirror the resume contract for forks: the new thread is usable as soon
|
||||
// as the response arrives, so restored usage must follow immediately.
|
||||
send_thread_token_usage_update_to_connection(
|
||||
&self.outgoing,
|
||||
connection_id,
|
||||
thread_id,
|
||||
&token_usage_thread,
|
||||
forked_thread.as_ref(),
|
||||
token_usage_turn_id,
|
||||
)
|
||||
.await;
|
||||
|
||||
let notif = ThreadStartedNotification { thread };
|
||||
self.outgoing
|
||||
.send_server_notification(ServerNotification::ThreadStarted(notif))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
|
@ -266,182 +266,6 @@ impl CodexMessageProcessor {
|
|||
Some(persisted_metadata)
|
||||
}
|
||||
|
||||
async fn resume_running_thread(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
params: &ThreadResumeParams,
|
||||
) -> bool {
|
||||
if let Ok(existing_thread_id) = ThreadId::from_string(¶ms.thread_id)
|
||||
&& let Ok(existing_thread) = self.thread_manager.get_thread(existing_thread_id).await
|
||||
{
|
||||
if params.history.is_some() {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!(
|
||||
"cannot resume thread {existing_thread_id} with history while it is already running"
|
||||
),
|
||||
)
|
||||
.await;
|
||||
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
|
||||
{
|
||||
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()
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return true;
|
||||
}
|
||||
|
||||
let thread_state = self
|
||||
.thread_state_manager
|
||||
.thread_state(existing_thread_id)
|
||||
.await;
|
||||
if let Err(error) = self
|
||||
.ensure_listener_task_running(
|
||||
existing_thread_id,
|
||||
existing_thread.clone(),
|
||||
thread_state.clone(),
|
||||
ApiVersion::V2,
|
||||
)
|
||||
.await
|
||||
{
|
||||
self.outgoing.send_error(request_id, error).await;
|
||||
return true;
|
||||
}
|
||||
|
||||
let config_snapshot = existing_thread.config_snapshot().await;
|
||||
let mismatch_details = collect_resume_override_mismatches(params, &config_snapshot);
|
||||
if !mismatch_details.is_empty() {
|
||||
tracing::warn!(
|
||||
"thread/resume overrides ignored for running thread {}: {}",
|
||||
existing_thread_id,
|
||||
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
|
||||
{
|
||||
Ok(thread) => thread,
|
||||
Err(message) => {
|
||||
self.send_internal_error(request_id, message).await;
|
||||
return true;
|
||||
}
|
||||
};
|
||||
|
||||
let listener_command_tx = {
|
||||
let thread_state = thread_state.lock().await;
|
||||
thread_state.listener_command_tx()
|
||||
};
|
||||
let Some(listener_command_tx) = listener_command_tx else {
|
||||
let err = JSONRPCErrorError {
|
||||
code: INTERNAL_ERROR_CODE,
|
||||
message: format!(
|
||||
"failed to enqueue running thread resume for thread {existing_thread_id}: thread listener is not running"
|
||||
),
|
||||
data: None,
|
||||
};
|
||||
self.outgoing.send_error(request_id, err).await;
|
||||
return true;
|
||||
};
|
||||
|
||||
let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse(
|
||||
Box::new(crate::thread_state::PendingThreadResumeRequest {
|
||||
request_id: request_id.clone(),
|
||||
rollout_path: rollout_path.clone(),
|
||||
config_snapshot,
|
||||
instruction_sources,
|
||||
thread_summary,
|
||||
}),
|
||||
);
|
||||
if listener_command_tx.send(command).is_err() {
|
||||
let err = JSONRPCErrorError {
|
||||
code: INTERNAL_ERROR_CODE,
|
||||
message: format!(
|
||||
"failed to enqueue running thread resume for thread {existing_thread_id}: thread listener command channel is closed"
|
||||
),
|
||||
data: None,
|
||||
};
|
||||
self.outgoing.send_error(request_id, err).await;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
false
|
||||
}
|
||||
|
||||
async fn resume_thread_from_history(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
|
|
@ -576,616 +400,6 @@ impl CodexMessageProcessor {
|
|||
set_thread_name_from_title(thread, title);
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn thread_fork(
|
||||
&self,
|
||||
request_id: ConnectionRequestId,
|
||||
params: ThreadForkParams,
|
||||
) {
|
||||
let ThreadForkParams {
|
||||
thread_id,
|
||||
path,
|
||||
model,
|
||||
model_provider,
|
||||
service_tier,
|
||||
cwd,
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
sandbox,
|
||||
permission_profile,
|
||||
config: cli_overrides,
|
||||
base_instructions,
|
||||
developer_instructions,
|
||||
ephemeral,
|
||||
persist_extended_history,
|
||||
} = params;
|
||||
if sandbox.is_some() && permission_profile.is_some() {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
"`permissionProfile` cannot be combined with `sandbox`".to_string(),
|
||||
)
|
||||
.await;
|
||||
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(),
|
||||
)
|
||||
.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;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
let history_cwd =
|
||||
read_history_cwd_from_state_db(&self.config, source_thread_id, rollout_path.as_path())
|
||||
.await;
|
||||
|
||||
// Persist Windows sandbox mode.
|
||||
let mut cli_overrides = cli_overrides.unwrap_or_default();
|
||||
if cfg!(windows) {
|
||||
match WindowsSandboxLevel::from_config(&self.config) {
|
||||
WindowsSandboxLevel::Elevated => {
|
||||
cli_overrides
|
||||
.insert("windows.sandbox".to_string(), serde_json::json!("elevated"));
|
||||
}
|
||||
WindowsSandboxLevel::RestrictedToken => {
|
||||
cli_overrides.insert(
|
||||
"windows.sandbox".to_string(),
|
||||
serde_json::json!("unelevated"),
|
||||
);
|
||||
}
|
||||
WindowsSandboxLevel::Disabled => {}
|
||||
}
|
||||
}
|
||||
let request_overrides = if cli_overrides.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(cli_overrides)
|
||||
};
|
||||
let mut typesafe_overrides = self.build_thread_config_overrides(
|
||||
model,
|
||||
model_provider,
|
||||
service_tier,
|
||||
cwd,
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
sandbox,
|
||||
permission_profile,
|
||||
base_instructions,
|
||||
developer_instructions,
|
||||
/*personality*/ None,
|
||||
);
|
||||
typesafe_overrides.ephemeral = ephemeral.then_some(true);
|
||||
// Derive a Config using the same logic as new conversation, honoring overrides if provided.
|
||||
let config = match self
|
||||
.config_manager
|
||||
.load_for_cwd(request_overrides, typesafe_overrides, history_cwd)
|
||||
.await
|
||||
{
|
||||
Ok(config) => config,
|
||||
Err(err) => {
|
||||
self.outgoing
|
||||
.send_error(request_id, config_load_error(&err))
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
let fallback_model_provider = config.model_provider_id.clone();
|
||||
let instruction_sources = Self::instruction_sources_from_config(&config).await;
|
||||
|
||||
let NewThread {
|
||||
thread_id,
|
||||
thread: forked_thread,
|
||||
session_configured,
|
||||
..
|
||||
} = match self
|
||||
.thread_manager
|
||||
.fork_thread(
|
||||
ForkSnapshot::Interrupted,
|
||||
config,
|
||||
rollout_path.clone(),
|
||||
persist_extended_history,
|
||||
self.request_trace_context(&request_id).await,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(thread) => thread,
|
||||
Err(err) => {
|
||||
match err {
|
||||
CodexErr::Io(_) | CodexErr::Json(_) => {
|
||||
self.send_invalid_request_error(
|
||||
request_id,
|
||||
format!("failed to load rollout `{}`: {err}", rollout_path.display()),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
CodexErr::InvalidRequest(message) => {
|
||||
self.send_invalid_request_error(request_id, message).await;
|
||||
}
|
||||
_ => {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!("error forking thread: {err}"),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
// Auto-attach a conversation listener when forking a thread.
|
||||
Self::log_listener_attach_result(
|
||||
self.ensure_conversation_listener(
|
||||
thread_id,
|
||||
request_id.connection_id,
|
||||
/*raw_events_enabled*/ false,
|
||||
ApiVersion::V2,
|
||||
)
|
||||
.await,
|
||||
thread_id,
|
||||
request_id.connection_id,
|
||||
"thread",
|
||||
);
|
||||
|
||||
// Persistent forks materialize their own rollout immediately. Ephemeral forks stay
|
||||
// pathless, so they rebuild their visible history from the copied source rollout 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
|
||||
{
|
||||
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) => {
|
||||
self.send_internal_error(
|
||||
request_id,
|
||||
format!(
|
||||
"failed to load rollout `{}` for thread {thread_id}: {err}",
|
||||
fork_rollout_path.display()
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
let config_snapshot = forked_thread.config_snapshot().await;
|
||||
// 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());
|
||||
if let Err(message) = populate_thread_turns(
|
||||
&mut thread,
|
||||
ThreadTurnSource::HistoryItems(&history_items),
|
||||
/*active_turn*/ None,
|
||||
)
|
||||
.await
|
||||
{
|
||||
self.send_internal_error(request_id, message).await;
|
||||
return;
|
||||
}
|
||||
thread
|
||||
};
|
||||
|
||||
if let Some(fork_rollout_path) = session_configured.rollout_path.as_ref()
|
||||
&& 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;
|
||||
}
|
||||
|
||||
thread.epiphany_state = live_thread_epiphany_state(forked_thread.as_ref()).await;
|
||||
|
||||
self.thread_watch_manager
|
||||
.upsert_thread_silently(thread.clone())
|
||||
.await;
|
||||
|
||||
thread.status = resolve_thread_status(
|
||||
self.thread_watch_manager
|
||||
.loaded_status_for_thread(&thread.id)
|
||||
.await,
|
||||
/*has_in_progress_turn*/ false,
|
||||
);
|
||||
let permission_profile = thread_response_permission_profile(
|
||||
&session_configured.sandbox_policy,
|
||||
forked_thread.config_snapshot().await.permission_profile,
|
||||
);
|
||||
|
||||
let response = ThreadForkResponse {
|
||||
thread: thread.clone(),
|
||||
model: session_configured.model,
|
||||
model_provider: session_configured.model_provider_id,
|
||||
service_tier: session_configured.service_tier,
|
||||
cwd: session_configured.cwd,
|
||||
instruction_sources,
|
||||
approval_policy: session_configured.approval_policy.into(),
|
||||
approvals_reviewer: session_configured.approvals_reviewer.into(),
|
||||
sandbox: session_configured.sandbox_policy.into(),
|
||||
permission_profile,
|
||||
reasoning_effort: session_configured.reasoning_effort,
|
||||
};
|
||||
if self.config.features.enabled(Feature::GeneralAnalytics) {
|
||||
self.analytics_events_client.track_response(
|
||||
request_id.connection_id.0,
|
||||
ClientResponse::ThreadFork {
|
||||
request_id: request_id.request_id.clone(),
|
||||
response: response.clone(),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
let connection_id = request_id.connection_id;
|
||||
let token_usage_thread = response.thread.clone();
|
||||
let token_usage_turn_id = if let Some(turn_id) =
|
||||
latest_token_usage_turn_id_for_thread_path(&token_usage_thread).await
|
||||
{
|
||||
Some(turn_id)
|
||||
} else {
|
||||
latest_token_usage_turn_id_from_rollout_path(
|
||||
rollout_path.as_path(),
|
||||
&token_usage_thread,
|
||||
)
|
||||
.await
|
||||
};
|
||||
self.outgoing.send_response(request_id, response).await;
|
||||
// Mirror the resume contract for forks: the new thread is usable as soon
|
||||
// as the response arrives, so restored usage must follow immediately.
|
||||
send_thread_token_usage_update_to_connection(
|
||||
&self.outgoing,
|
||||
connection_id,
|
||||
thread_id,
|
||||
&token_usage_thread,
|
||||
forked_thread.as_ref(),
|
||||
token_usage_turn_id,
|
||||
)
|
||||
.await;
|
||||
|
||||
let notif = ThreadStartedNotification { thread };
|
||||
self.outgoing
|
||||
.send_server_notification(ServerNotification::ThreadStarted(notif))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
#[expect(
|
||||
clippy::await_holding_invalid_type,
|
||||
reason = "running-thread resume subscription must be serialized against pending unloads"
|
||||
)]
|
||||
pub(crate) async fn handle_pending_thread_resume_request(
|
||||
conversation_id: ThreadId,
|
||||
conversation: &Arc<CodexThread>,
|
||||
_codex_home: &Path,
|
||||
thread_state_manager: &ThreadStateManager,
|
||||
thread_state: &Arc<Mutex<ThreadState>>,
|
||||
thread_watch_manager: &ThreadWatchManager,
|
||||
outgoing: &Arc<OutgoingMessageSender>,
|
||||
pending_thread_unloads: &Arc<Mutex<HashSet<ThreadId>>>,
|
||||
pending: crate::thread_state::PendingThreadResumeRequest,
|
||||
) {
|
||||
let active_turn = {
|
||||
let state = thread_state.lock().await;
|
||||
state.active_turn_snapshot()
|
||||
};
|
||||
tracing::debug!(
|
||||
thread_id = %conversation_id,
|
||||
request_id = ?pending.request_id,
|
||||
active_turn_present = active_turn.is_some(),
|
||||
active_turn_id = ?active_turn.as_ref().map(|turn| turn.id.as_str()),
|
||||
active_turn_status = ?active_turn.as_ref().map(|turn| &turn.status),
|
||||
"composing running thread resume response"
|
||||
);
|
||||
let has_live_in_progress_turn =
|
||||
matches!(conversation.agent_status().await, AgentStatus::Running)
|
||||
|| active_turn
|
||||
.as_ref()
|
||||
.is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress));
|
||||
|
||||
let request_id = pending.request_id;
|
||||
let connection_id = request_id.connection_id;
|
||||
let mut thread = pending.thread_summary;
|
||||
if let Err(message) = populate_thread_turns(
|
||||
&mut thread,
|
||||
ThreadTurnSource::RolloutPath(pending.rollout_path.as_path()),
|
||||
active_turn.as_ref(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
outgoing
|
||||
.send_error(
|
||||
request_id,
|
||||
JSONRPCErrorError {
|
||||
code: INTERNAL_ERROR_CODE,
|
||||
message,
|
||||
data: None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
let thread_status = thread_watch_manager
|
||||
.loaded_status_for_thread(&thread.id)
|
||||
.await;
|
||||
|
||||
set_thread_status_and_interrupt_stale_turns(
|
||||
&mut thread,
|
||||
thread_status,
|
||||
has_live_in_progress_turn,
|
||||
);
|
||||
|
||||
{
|
||||
let pending_thread_unloads = pending_thread_unloads.lock().await;
|
||||
if pending_thread_unloads.contains(&conversation_id) {
|
||||
drop(pending_thread_unloads);
|
||||
outgoing
|
||||
.send_error(
|
||||
request_id,
|
||||
JSONRPCErrorError {
|
||||
code: INVALID_REQUEST_ERROR_CODE,
|
||||
message: format!(
|
||||
"thread {conversation_id} is closing; retry thread/resume after the thread is closed"
|
||||
),
|
||||
data: None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
if !thread_state_manager
|
||||
.try_add_connection_to_thread(conversation_id, connection_id)
|
||||
.await
|
||||
{
|
||||
tracing::debug!(
|
||||
thread_id = %conversation_id,
|
||||
connection_id = ?connection_id,
|
||||
"skipping running thread resume for closed connection"
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
let ThreadConfigSnapshot {
|
||||
model,
|
||||
model_provider_id,
|
||||
service_tier,
|
||||
approval_policy,
|
||||
approvals_reviewer,
|
||||
sandbox_policy,
|
||||
permission_profile,
|
||||
cwd,
|
||||
reasoning_effort,
|
||||
..
|
||||
} = pending.config_snapshot;
|
||||
let instruction_sources = pending.instruction_sources;
|
||||
let permission_profile =
|
||||
thread_response_permission_profile(&sandbox_policy, permission_profile);
|
||||
|
||||
let response = ThreadResumeResponse {
|
||||
thread,
|
||||
model,
|
||||
model_provider: model_provider_id,
|
||||
service_tier,
|
||||
cwd,
|
||||
instruction_sources,
|
||||
approval_policy: approval_policy.into(),
|
||||
approvals_reviewer: approvals_reviewer.into(),
|
||||
sandbox: sandbox_policy.into(),
|
||||
permission_profile,
|
||||
reasoning_effort,
|
||||
};
|
||||
let token_usage_thread = response.thread.clone();
|
||||
let token_usage_turn_id = latest_token_usage_turn_id_from_rollout_path(
|
||||
pending.rollout_path.as_path(),
|
||||
&token_usage_thread,
|
||||
)
|
||||
.await;
|
||||
outgoing.send_response(request_id, response).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(
|
||||
outgoing,
|
||||
connection_id,
|
||||
conversation_id,
|
||||
&token_usage_thread,
|
||||
conversation.as_ref(),
|
||||
token_usage_turn_id,
|
||||
)
|
||||
.await;
|
||||
outgoing
|
||||
.replay_requests_to_connection_for_thread(connection_id, conversation_id)
|
||||
.await;
|
||||
}
|
||||
|
||||
fn collect_resume_override_mismatches(
|
||||
request: &ThreadResumeParams,
|
||||
config_snapshot: &ThreadConfigSnapshot,
|
||||
) -> Vec<String> {
|
||||
let mut mismatch_details = Vec::new();
|
||||
|
||||
if let Some(requested_model) = request.model.as_deref()
|
||||
&& requested_model != config_snapshot.model
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"model requested={requested_model} active={}",
|
||||
config_snapshot.model
|
||||
));
|
||||
}
|
||||
if let Some(requested_provider) = request.model_provider.as_deref()
|
||||
&& requested_provider != config_snapshot.model_provider_id
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"model_provider requested={requested_provider} active={}",
|
||||
config_snapshot.model_provider_id
|
||||
));
|
||||
}
|
||||
if let Some(requested_service_tier) = request.service_tier.as_ref()
|
||||
&& requested_service_tier != &config_snapshot.service_tier
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"service_tier requested={requested_service_tier:?} active={:?}",
|
||||
config_snapshot.service_tier
|
||||
));
|
||||
}
|
||||
if let Some(requested_cwd) = request.cwd.as_deref() {
|
||||
let requested_cwd_path = std::path::PathBuf::from(requested_cwd);
|
||||
if requested_cwd_path != config_snapshot.cwd.as_path() {
|
||||
mismatch_details.push(format!(
|
||||
"cwd requested={} active={}",
|
||||
requested_cwd_path.display(),
|
||||
config_snapshot.cwd.display()
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_approval) = request.approval_policy.as_ref() {
|
||||
let active_approval: AskForApproval = config_snapshot.approval_policy.into();
|
||||
if requested_approval != &active_approval {
|
||||
mismatch_details.push(format!(
|
||||
"approval_policy requested={requested_approval:?} active={active_approval:?}"
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_review_policy) = request.approvals_reviewer.as_ref() {
|
||||
let active_review_policy: codex_app_server_protocol::ApprovalsReviewer =
|
||||
config_snapshot.approvals_reviewer.into();
|
||||
if requested_review_policy != &active_review_policy {
|
||||
mismatch_details.push(format!(
|
||||
"approvals_reviewer requested={requested_review_policy:?} active={active_review_policy:?}"
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_sandbox) = request.sandbox.as_ref() {
|
||||
let sandbox_matches = matches!(
|
||||
(requested_sandbox, &config_snapshot.sandbox_policy),
|
||||
(
|
||||
SandboxMode::ReadOnly,
|
||||
codex_protocol::protocol::SandboxPolicy::ReadOnly { .. }
|
||||
) | (
|
||||
SandboxMode::WorkspaceWrite,
|
||||
codex_protocol::protocol::SandboxPolicy::WorkspaceWrite { .. }
|
||||
) | (
|
||||
SandboxMode::DangerFullAccess,
|
||||
codex_protocol::protocol::SandboxPolicy::DangerFullAccess
|
||||
) | (
|
||||
SandboxMode::DangerFullAccess,
|
||||
codex_protocol::protocol::SandboxPolicy::ExternalSandbox { .. }
|
||||
)
|
||||
);
|
||||
if !sandbox_matches {
|
||||
mismatch_details.push(format!(
|
||||
"sandbox requested={requested_sandbox:?} active={:?}",
|
||||
config_snapshot.sandbox_policy
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_permission_profile) = request.permission_profile.as_ref() {
|
||||
let requested_permission_profile =
|
||||
codex_protocol::models::PermissionProfile::from(requested_permission_profile.clone());
|
||||
if requested_permission_profile != config_snapshot.permission_profile {
|
||||
mismatch_details.push(format!(
|
||||
"permission_profile requested={requested_permission_profile:?} active={:?}",
|
||||
config_snapshot.permission_profile
|
||||
));
|
||||
}
|
||||
}
|
||||
if let Some(requested_personality) = request.personality.as_ref()
|
||||
&& config_snapshot.personality.as_ref() != Some(requested_personality)
|
||||
{
|
||||
mismatch_details.push(format!(
|
||||
"personality requested={requested_personality:?} active={:?}",
|
||||
config_snapshot.personality
|
||||
));
|
||||
}
|
||||
|
||||
if request.config.is_some() {
|
||||
mismatch_details
|
||||
.push("config overrides were provided and ignored while running".to_string());
|
||||
}
|
||||
if request.base_instructions.is_some() {
|
||||
mismatch_details
|
||||
.push("baseInstructions override was provided and ignored while running".to_string());
|
||||
}
|
||||
if request.developer_instructions.is_some() {
|
||||
mismatch_details.push(
|
||||
"developerInstructions override was provided and ignored while running".to_string(),
|
||||
);
|
||||
}
|
||||
if request.persist_extended_history {
|
||||
mismatch_details.push(
|
||||
"persistExtendedHistory override was provided and ignored while running".to_string(),
|
||||
);
|
||||
}
|
||||
|
||||
mismatch_details
|
||||
}
|
||||
|
||||
fn merge_persisted_resume_metadata(
|
||||
|
|
|
|||
Reference in a new issue