mirror of
https://github.com/openai/codex.git
synced 2026-11-11 21:10:39 +00:00
jif
2fdf047c96
## Why Persisted multi-agent v2 children created after the client disconnects need lifecycle listeners so their runtimes can be released once idle. Listener attachment must also wait for an earlier runtime's cleanup to finish. ## What changed - Start lifecycle listeners for persisted v2 children even when no initialized clients are connected, and defer attachment until pending unload cleanup completes. - Keep ephemeral children loaded so follow-up tasks retain their model context. - Raise the default `thread_unload_delay_secs` from 60 seconds to 1800 seconds (30 minutes), updating the configuration documentation and sample. ## Testing Add regression coverage for children spawned after disconnect with zero- and one-second unload delays, including waiting for pending cleanup, keeping running threads loaded, releasing idle runtimes, and preserving an ephemeral child's earlier reply across a follow-up. Update configuration tests for the new default and explicit overrides. GitOrigin-RevId: 2fecc77b5c03237aa8fd7eacc4bd119b99d36e0a
57 lines
2.4 KiB
Rust
57 lines
2.4 KiB
Rust
//! Selects persistent root threads for managed daemon recovery using normal resume semantics.
|
|
//! Only threads successfully persisted through the existing thread store become candidates.
|
|
|
|
use super::ThreadRequestProcessor;
|
|
use codex_app_server_transport::daemon_recovery::InterruptedTurn;
|
|
use codex_app_server_transport::daemon_recovery::RecoverySnapshot;
|
|
use codex_thread_store::PersistContext;
|
|
use tracing::warn;
|
|
|
|
impl ThreadRequestProcessor {
|
|
#[expect(
|
|
clippy::await_holding_invalid_type,
|
|
reason = "snapshot selection must be serialized against pending unloads"
|
|
)]
|
|
pub(crate) async fn daemon_recovery_snapshot(&self) -> RecoverySnapshot {
|
|
let mut snapshot = RecoverySnapshot::default();
|
|
for thread_id in self.thread_manager.list_thread_ids().await {
|
|
let pending_unloads = self.pending_thread_unloads.lock().await;
|
|
if pending_unloads.contains_key(&thread_id) {
|
|
continue;
|
|
}
|
|
let Ok(thread) = self.thread_manager.get_thread(thread_id).await else {
|
|
continue;
|
|
};
|
|
let config = thread.config_snapshot().await;
|
|
if !config.ephemeral
|
|
&& config.parent_thread_id.is_none()
|
|
&& !config.session_source.is_non_root_agent()
|
|
{
|
|
let interrupted = thread.interrupted_turn().await;
|
|
if let Err(err) = self
|
|
.thread_store
|
|
.persist_thread(thread_id, PersistContext::Standard)
|
|
.await
|
|
{
|
|
warn!(%thread_id, %err, "skipping daemon restore for thread that could not be persisted");
|
|
continue;
|
|
}
|
|
snapshot.loaded.insert(thread_id.to_string());
|
|
if let Some((turn_id, options, environment)) = interrupted {
|
|
snapshot.interrupted.insert(
|
|
thread_id.to_string(),
|
|
InterruptedTurn {
|
|
turn_id,
|
|
output_schema: options.final_output_json_schema,
|
|
service_tier: options.service_tier,
|
|
cyber_access_program: options.cyber_access_program,
|
|
local_environment: Some((&environment).into()),
|
|
},
|
|
);
|
|
}
|
|
}
|
|
}
|
|
snapshot
|
|
}
|
|
}
|