GameCult/Epiphany
Archived
Watch
9
0
Fork
You've already forked Epiphany
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Move resume listener helpers from processor

Browse source
This commit is contained in:
Meta Vens 2026-05-17 23:26:57 +01:00
commit 074d395f6f
7 changed files with 436 additions and 435 deletions
Download patch file Download diff file Expand all files Collapse all files

5
notes/fresh-workspace-handoff.md
View file

@ -50,10 +50,11 @@ 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 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 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`.
- Turn control routes now live in `codex_message_processor/turn_routes.rs`: turn start, thread item injection, app-server client info setting, turn steer, and turn interrupt. The root processor is about 3,803 lines.
- Thread projection now lives in `codex_message_processor/thread_projection.rs`: state-db/thread-store/rollout summary projection, `ConversationSummary` to API `Thread` mapping, thread-title attachment helpers, turn pagination, and rollout turn reconstruction. `codex_message_processor.rs` is about 1,665 lines after this extraction and mostly owns JSON-RPC dispatch, shared config/error helpers, and the remaining resume/listener helper seam.
- Listener/subscription lifecycle now lives in `codex_message_processor/listener_lifecycle.rs`: connection attach/detach, listener startup, idle unload, unsubscribe, teardown, archive preparation, listener task context, idle-unload state, and shutdown result handling.
- Thread projection now lives in `codex_message_processor/thread_projection.rs`: state-db/thread-store/rollout summary projection, `ConversationSummary` to API `Thread` mapping, thread-title attachment helpers, turn pagination, and rollout turn reconstruction. `codex_message_processor.rs` is about 1,243 lines after this extraction and mostly owns JSON-RPC dispatch, shared config/error helpers, thread config override construction, and processor lifecycle shutdown.
- Listener/subscription lifecycle now lives in `codex_message_processor/listener_lifecycle.rs`: connection attach/detach, listener startup, idle unload, unsubscribe, teardown, archive preparation, listener task context, idle-unload state, shutdown result handling, and listener command routing.
- The May 2026 foundation migration in `notes/epiphany-ideal-architecture-rebuild-plan.md` is complete enough to defend. Read `notes/epiphany-architectural-teardown.md` before touching Epiphany control-plane code, but do not treat it as an open-ended excuse to keep shaving the altar. The rebuild moved scene projection, freshness derivation, pressure policy, targeted context shards, bounded graph traversal, job/progress view derivation, planning view derivation, reorientation resume/regather verdict policy, pure CRRC recommendation policy, fixed-lane coordinator decision policy, role-board projection policy, role/reorient result interpretation, role self-persistence review policy, and role/reorient acceptance bundle policy into `epiphany-core`; typed acceptance receipts replaced summary-string identity for live accept paths, runtime links are dual-written on launch, and result read-back now prefers runtime links.
- Read `notes/epiphany-ideal-architecture-rebuild-plan.md` immediately after the teardown. It defines the smallest coherent replacement architecture: durable `EpiphanyState`, runtime-spine-owned `RuntimeState`, separate role memory/heartbeat state, `epiphany-core` as policy owner, app-server as adapter, Aquarium as reflector, typed acceptance receipts, and derived view lenses instead of a protocol verb zoo.
- Do not continue Aquarium UI, bridge, Face, or dogfood expansion until the teardown has a source-grounded cleanup slice plan. Epiphany is the foundation; patches on patches are not a purification rite, they are how the altar becomes load-bearing garbage.

BIN
state/ledgers.msgpack
View file

Binary file not shown.

19
state/map.yaml
View file

File diff suppressed because one or more lines are too long

424
vendor/codex/codex-rs/app-server/src/codex_message_processor.rs vendored
View file

@ -21,6 +21,7 @@ pub(crate) use self::epiphany_automation::maybe_run_epiphany_pre_compaction_chec
pub(super) use self::listener_lifecycle::{EnsureConversationListenerResult, ListenerTaskContext};
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;
@ -1148,429 +1149,6 @@ impl CodexMessageProcessor {
}
}
#[allow(clippy::too_many_arguments)]
async fn handle_thread_listener_command(
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>>>,
listener_command: ThreadListenerCommand,
) {
match listener_command {
ThreadListenerCommand::SendThreadResumeResponse(resume_request) => {
handle_pending_thread_resume_request(
conversation_id,
conversation,
codex_home,
thread_state_manager,
thread_state,
thread_watch_manager,
outgoing,
pending_thread_unloads,
*resume_request,
)
.await;
}
ThreadListenerCommand::ResolveServerRequest {
request_id,
completion_tx,
} => {
resolve_pending_server_request(
conversation_id,
thread_state_manager,
outgoing,
request_id,
)
.await;
let _ = completion_tx.send(());
}
}
}
#[allow(clippy::too_many_arguments)]
#[expect(
clippy::await_holding_invalid_type,
reason = "running-thread resume subscription must be serialized against pending unloads"
)]
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;
}
enum ThreadTurnSource<'a> {
RolloutPath(&'a Path),
HistoryItems(&'a [RolloutItem]),
}
async fn populate_thread_turns(
thread: &mut Thread,
turn_source: ThreadTurnSource<'_>,
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 {
merge_turn_history_with_active_turn(&mut turns, active_turn.clone());
}
thread.turns = turns;
Ok(())
}
async fn resolve_pending_server_request(
conversation_id: ThreadId,
thread_state_manager: &ThreadStateManager,
outgoing: &Arc<OutgoingMessageSender>,
request_id: RequestId,
) {
let thread_id = conversation_id.to_string();
let subscribed_connection_ids = thread_state_manager
.subscribed_connection_ids(conversation_id)
.await;
let outgoing = ThreadScopedOutgoingMessageSender::new(
outgoing.clone(),
subscribed_connection_ids,
conversation_id,
);
outgoing
.send_server_notification(ServerNotification::ServerRequestResolved(
ServerRequestResolvedNotification {
thread_id,
request_id,
},
))
.await;
}
fn merge_turn_history_with_active_turn(turns: &mut Vec<Turn>, active_turn: Turn) {
turns.retain(|turn| turn.id != active_turn.id);
turns.push(active_turn);
}
fn set_thread_status_and_interrupt_stale_turns(
thread: &mut Thread,
loaded_status: ThreadStatus,
has_live_in_progress_turn: bool,
) {
let status = resolve_thread_status(loaded_status, has_live_in_progress_turn);
if !matches!(status, ThreadStatus::Active { .. }) {
for turn in &mut thread.turns {
if matches!(turn.status, TurnStatus::InProgress) {
turn.status = TurnStatus::Interrupted;
}
}
}
thread.status = status;
}
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(
request_overrides: &mut Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: &mut ConfigOverrides,
persisted_metadata: &ThreadMetadata,
) {
if has_model_resume_override(request_overrides.as_ref(), typesafe_overrides) {
return;
}
typesafe_overrides.model = persisted_metadata.model.clone();
if let Some(reasoning_effort) = persisted_metadata.reasoning_effort {
request_overrides.get_or_insert_with(HashMap::new).insert(
"model_reasoning_effort".to_string(),
serde_json::Value::String(reasoning_effort.to_string()),
);
}
}
fn has_model_resume_override(
request_overrides: Option<&HashMap<String, serde_json::Value>>,
typesafe_overrides: &ConfigOverrides,
) -> bool {
typesafe_overrides.model.is_some()
|| typesafe_overrides.model_provider.is_some()
|| request_overrides.is_some_and(|overrides| overrides.contains_key("model"))
|| request_overrides
.is_some_and(|overrides| overrides.contains_key("model_reasoning_effort"))
}
fn cloud_requirements_load_error(err: &std::io::Error) -> Option<&CloudRequirementsLoadError> {
let mut current: Option<&(dyn std::error::Error + 'static)> = err
.get_ref()

68
vendor/codex/codex-rs/app-server/src/codex_message_processor/listener_lifecycle.rs vendored
View file

@ -663,3 +663,71 @@ impl CodexMessageProcessor {
Ok(())
}
}
async fn resolve_pending_server_request(
conversation_id: ThreadId,
thread_state_manager: &ThreadStateManager,
outgoing: &Arc<OutgoingMessageSender>,
request_id: RequestId,
) {
let thread_id = conversation_id.to_string();
let subscribed_connection_ids = thread_state_manager
.subscribed_connection_ids(conversation_id)
.await;
let outgoing = ThreadScopedOutgoingMessageSender::new(
outgoing.clone(),
subscribed_connection_ids,
conversation_id,
);
outgoing
.send_server_notification(ServerNotification::ServerRequestResolved(
ServerRequestResolvedNotification {
thread_id,
request_id,
},
))
.await;
}
#[allow(clippy::too_many_arguments)]
async fn handle_thread_listener_command(
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>>>,
listener_command: ThreadListenerCommand,
) {
match listener_command {
ThreadListenerCommand::SendThreadResumeResponse(resume_request) => {
handle_pending_thread_resume_request(
conversation_id,
conversation,
codex_home,
thread_state_manager,
thread_state,
thread_watch_manager,
outgoing,
pending_thread_unloads,
*resume_request,
)
.await;
}
ThreadListenerCommand::ResolveServerRequest {
request_id,
completion_tx,
} => {
resolve_pending_server_request(
conversation_id,
thread_state_manager,
outgoing,
request_id,
)
.await;
let _ = completion_tx.send(());
}
}
}

53
vendor/codex/codex-rs/app-server/src/codex_message_processor/thread_projection.rs vendored
View file

@ -909,6 +909,59 @@ fn parse_thread_turns_cursor(cursor: &str) -> Result<ThreadTurnsCursor, JSONRPCE
})
}
pub(crate) enum ThreadTurnSource<'a> {
RolloutPath(&'a Path),
HistoryItems(&'a [RolloutItem]),
}
pub(crate) async fn populate_thread_turns(
thread: &mut Thread,
turn_source: ThreadTurnSource<'_>,
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 {
merge_turn_history_with_active_turn(&mut turns, active_turn.clone());
}
thread.turns = turns;
Ok(())
}
fn merge_turn_history_with_active_turn(turns: &mut Vec<Turn>, active_turn: Turn) {
turns.retain(|turn| turn.id != active_turn.id);
turns.push(active_turn);
}
pub(crate) fn set_thread_status_and_interrupt_stale_turns(
thread: &mut Thread,
loaded_status: ThreadStatus,
has_live_in_progress_turn: bool,
) {
let status = resolve_thread_status(loaded_status, has_live_in_progress_turn);
if !matches!(status, ThreadStatus::Active { .. }) {
for turn in &mut thread.turns {
if matches!(turn.status, TurnStatus::InProgress) {
turn.status = TurnStatus::Interrupted;
}
}
}
thread.status = status;
}
pub(crate) fn reconstruct_thread_turns_from_rollout_items(
items: &[RolloutItem],
loaded_status: ThreadStatus,

302
vendor/codex/codex-rs/app-server/src/codex_message_processor/thread_resume_routes.rs vendored
View file

@ -915,3 +915,305 @@ impl CodexMessageProcessor {
.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(
request_overrides: &mut Option<HashMap<String, serde_json::Value>>,
typesafe_overrides: &mut ConfigOverrides,
persisted_metadata: &ThreadMetadata,
) {
if has_model_resume_override(request_overrides.as_ref(), typesafe_overrides) {
return;
}
typesafe_overrides.model = persisted_metadata.model.clone();
if let Some(reasoning_effort) = persisted_metadata.reasoning_effort {
request_overrides.get_or_insert_with(HashMap::new).insert(
"model_reasoning_effort".to_string(),
serde_json::Value::String(reasoning_effort.to_string()),
);
}
}
fn has_model_resume_override(
request_overrides: Option<&HashMap<String, serde_json::Value>>,
typesafe_overrides: &ConfigOverrides,
) -> bool {
typesafe_overrides.model.is_some()
|| typesafe_overrides.model_provider.is_some()
|| request_overrides.is_some_and(|overrides| overrides.contains_key("model"))
|| request_overrides
.is_some_and(|overrides| overrides.contains_key("model_reasoning_effort"))
}