openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-23 18:08:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Avoid holding metadata permit during cold resume config load (#42773)

Browse Source 
## Why

Cold `thread/resume` configuration loading can call back into the host. Holding the global metadata permit during that work blocks unrelated thread metadata updates and MCP requests.

## What changed

- Release the metadata permit while loading configuration for a cold resume.
- Reacquire the permit and reload the resolved thread history before startup.
- Reload configuration when persisted metadata, settings, or the history working directory changed during the unlocked interval.
- Rejoin a thread that another request resumed concurrently, and avoid reopening threads that were archived or deleted.

## Testing

Add coverage for unrelated metadata and MCP requests during a blocked config load, concurrent resume mutations, and the extra history read required for non-local thread stores.

GitOrigin-RevId: 8cbd3868eeff84968765b1e3dbc7590ebe5fa892
This commit is contained in:
Chris Hayduk
2026-09-04 14:23:27 +00:00
committed by copyberry
parent 99d66aa1c5
commit c9fac4dd5a
5 changed files with 740 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_processor.rs
+101 -46
View File
@@ -23,6 +23,7 @@ use codex_protocol::error::CodexErrorDetails;
use codex_protocol::mcp::ClientMcpExtensions;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_thread_store::PersistContext;
use std::ops::ControlFlow;
pub(super) const THREAD_LIST_DEFAULT_LIMIT: usize = 25;
pub(super) const THREAD_LIST_MAX_LIMIT: usize = 100;
@@ -85,6 +86,19 @@ struct ThreadListFilters {
relation_filter: Option<StoreThreadRelationFilter>,
}
// Persisted inputs that can change while loading configuration without the metadata permit.
#[derive(PartialEq)]
struct ResumeConfigState {
history_cwd: Option<PathBuf>,
persisted_metadata: Option<ThreadMetadata>,
persisted_settings: Option<PersistedResumeSettings>,
}
struct PreparedResumeConfig {
state: ResumeConfigState,
config: Config,
}
struct ThreadRevertRuntimeSnapshot {
config: Config,
settings: CodexThreadSettingsOverrides,
@@ -541,15 +555,20 @@ impl ThreadRequestProcessor {
app_server_client_version: Option<String>,
client_mcp_extensions: ClientMcpExtensions,
) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> {
self.thread_resume_inner(
request_id,
params,
app_server_client_name,
app_server_client_version,
client_mcp_extensions,
)
.await
.map(|()| None)
let mut prepared_config = None;
while self
.thread_resume_inner(
request_id.clone(),
&params,
app_server_client_name.clone(),
app_server_client_version.clone(),
client_mcp_extensions.clone(),
&mut prepared_config,
)
.await?
.is_continue()
{}
Ok(None)
}
pub(crate) async fn thread_fork(
@@ -3587,11 +3606,12 @@ impl ThreadRequestProcessor {
async fn thread_resume_inner(
&self,
request_id: ConnectionRequestId,
params: ThreadResumeParams,
params: &ThreadResumeParams,
app_server_client_name: Option<String>,
app_server_client_version: Option<String>,
client_mcp_extensions: ClientMcpExtensions,
) -> Result<(), JSONRPCErrorError> {
prepared_config: &mut Option<PreparedResumeConfig>,
) -> Result<ControlFlow<()>, JSONRPCErrorError> {
if let Ok(thread_id) = ThreadId::from_string(&params.thread_id)
&& self
.pending_thread_unloads
@@ -3607,7 +3627,7 @@ impl ThreadRequestProcessor {
)),
)
.await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
if params.sandbox.is_some() && params.permissions.is_some() {
@@ -3617,7 +3637,7 @@ impl ThreadRequestProcessor {
invalid_request("`permissions` cannot be combined with `sandbox`"),
)
.await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
let redact_resume_payloads =
should_redact_thread_resume_payloads(app_server_client_name.as_deref());
@@ -3626,24 +3646,24 @@ impl ThreadRequestProcessor {
Ok(permit) => permit,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
};
let stored_thread_from_running_probe = match self
.resume_running_thread(
&request_id,
&params,
params,
app_server_client_name.clone(),
app_server_client_version.clone(),
/*cold_resume_history*/ None,
)
.await
{
Ok(RunningThreadResumeResult::Handled) => return Ok(()),
Ok(RunningThreadResumeResult::Handled) => return Ok(ControlFlow::Break(())),
Ok(RunningThreadResumeResult::NotRunning(stored_thread)) => stored_thread,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
};
@@ -3666,7 +3686,7 @@ impl ThreadRequestProcessor {
personality,
exclude_turns,
initial_turns_page,
} = params;
} = params.clone();
let include_turns = !exclude_turns;
let resume_result = if let Some(history) = history {
@@ -3697,14 +3717,26 @@ impl ThreadRequestProcessor {
Ok(value) => value,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
};
if let InitialHistory::Resumed(resumed) = &thread_history
&& self
.pending_thread_unloads
.lock()
.await
.contains(&resumed.conversation_id)
{
return Err(invalid_request(format!(
"thread {} is closing; retry thread/resume after the thread is closed",
resumed.conversation_id
)));
}
let paginated_thread_id = resume_source_thread.as_ref().and_then(|thread| {
matches!(thread.history_mode, ThreadHistoryMode::Paginated).then_some(thread.thread_id)
});
let paginated_resume = paginated_thread_id.is_some();
if paginated_resume && include_turns {
if paginated_resume && include_turns && prepared_config.is_none() {
self.send_deprecation_notice(
request_id.connection_id,
PAGINATED_FULL_HISTORY_DEPRECATION_SUMMARY,
@@ -3750,7 +3782,7 @@ impl ThreadRequestProcessor {
)
.await?
{
RunningThreadResumeResult::Handled => Ok(()),
RunningThreadResumeResult::Handled => Ok(ControlFlow::Break(())),
RunningThreadResumeResult::NotRunning(_) => Err(invalid_request(
"cannot resume an unloaded multi-agent v2 sub-agent through its parent; resume the parent first, or use thread/read to inspect it",
)),
@@ -3803,7 +3835,7 @@ impl ThreadRequestProcessor {
)),
)
.await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
};
typesafe_overrides.approval_policy = Some(approval_policy);
@@ -3818,24 +3850,47 @@ impl ThreadRequestProcessor {
)
.await;
// Derive a Config using the same logic as new conversation, honoring overrides if provided.
let mut config = match self
.config_manager
.load_for_cwd(request_overrides, typesafe_overrides, history_cwd)
.await
{
Ok(config) => config,
Err(err) => {
let error = config_load_error(&err);
self.outgoing.send_error(request_id, error).await;
return Ok(());
}
};
if !has_explicit_model_resume_override
let clear_reasoning_effort = !has_explicit_model_resume_override
&& persisted_metadata
.as_ref()
.is_some_and(|metadata| metadata.reasoning_effort.is_none())
{
.is_some_and(|metadata| metadata.reasoning_effort.is_none());
let config_state = ResumeConfigState {
history_cwd: history_cwd.clone(),
persisted_metadata,
persisted_settings: match &thread_history {
InitialHistory::Resumed(resumed) => {
latest_persisted_resume_settings(&resumed.history)
}
InitialHistory::New | InitialHistory::Cleared | InitialHistory::Forked(_) => None,
},
};
let mut config = match prepared_config.take() {
Some(prepared) if prepared.state == config_state => prepared.config,
_ => {
// Config loading can call back into Desktop; release the permit during host work.
drop(_thread_list_state_permit);
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 Ok(ControlFlow::Break(()));
}
};
*prepared_config = Some(PreparedResumeConfig {
state: config_state,
config,
});
// Recheck the thread and history after a possible archive, delete, or aliased resume.
return Ok(ControlFlow::Continue(()));
}
};
if clear_reasoning_effort {
config.model_reasoning_effort = None;
}
@@ -3866,7 +3921,7 @@ impl ThreadRequestProcessor {
.await
{
self.outgoing.send_error(request_id, err).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
let instruction_sources = codex_thread.legacy_instruction_sources().await;
let SessionConfiguredEvent { rollout_path, .. } = session_configured;
@@ -3874,7 +3929,7 @@ impl ThreadRequestProcessor {
let error =
internal_error(format!("rollout path missing for thread {thread_id}"));
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
};
// Paginated JSONL is canonical, but its SQLite projection can lag after a
// previous write failure. Persist after reopening the live writer so legacy
@@ -3887,14 +3942,14 @@ impl ThreadRequestProcessor {
.map_err(thread_store_resume_read_error)
{
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
let materialized_turns = if paginated_resume && include_turns {
match self.paginated_thread_full_turns(thread_id).await {
Ok(turns) => Some(turns),
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
}
} else {
@@ -3929,7 +3984,7 @@ impl ThreadRequestProcessor {
self.outgoing
.send_error(request_id, internal_error(message))
.await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
};
thread.thread_source = codex_thread
@@ -3965,7 +4020,7 @@ impl ThreadRequestProcessor {
Ok(cursors) => cursors,
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
}
} else {
@@ -3992,7 +4047,7 @@ impl ThreadRequestProcessor {
Ok(page) => Some(page),
Err(error) => {
self.outgoing.send_error(request_id, error).await;
return Ok(());
return Ok(ControlFlow::Break(()));
}
}
} else {
@@ -4071,7 +4126,7 @@ impl ThreadRequestProcessor {
self.outgoing.send_error(request_id, error).await;
}
}
Ok(())
Ok(ControlFlow::Break(()))
}
async fn load_and_apply_persisted_resume_metadata(