openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Route Guardian reviewers through ThreadManager for inline parents (#45518)

Browse Source 
## Why

Inline delegates have no entry in the thread registry, so reviewer creation cannot depend on looking up the parent or waiting for its thread-ready notification.

## What changed

- Capture parent identity, authentication, shared agent control, originator, and inherited instructions in `StartThreadOptions` so `ThreadManager` can start a child without a registered parent.
- Route Guardian reviewer creation through this path, remove the standalone fallback and readiness gate, and require a Guardian extension host.
- Install explicit reviewer hosts in unit tests and the Guardian reviewer extension in the integration test harness, using `ExtensionRegistry::to_builder()` to preserve existing contributors.

## Testing

Extend the thread-manager regression test to remove the parent from the registry before starting a child, then verify inherited lineage, originator, session identity, and authentication, and exclusion from the public thread list.

GitOrigin-RevId: 468ded6fdce7520cb39d55c1a884dcfa5aaea2d9
This commit is contained in:
jif
2026-09-14 21:23:06 +00:00
committed by copyberry
parent 280c7e1e56
commit 7c73903be2
18 changed files with 318 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+1
View File
@@ -5498,6 +5498,7 @@ dependencies = [
"codex-exec-server",
"codex-extension-api",
"codex-features",
"codex-guardian-v2",
"codex-home",
"codex-hooks",
"codex-http-client",
codex-rs/core/src/codex_delegate.rs
-3
View File
@@ -81,11 +81,8 @@ pub(crate) async fn run_codex_thread_interactive(
} else {
Arc::clone(&parent_session.services.extensions)
};
// Inline delegates never register with ThreadManager or receive on_thread_ready.
// Bind their standalone spawn path before inherited extensions run.
let mut thread_extension_init = codex_extension_api::ExtensionDataInit::default();
thread_extension_init.insert(isolation);
thread_extension_init.insert(crate::guardian::GuardianReviewSessionHost::default());
let (session, io) = Session::spawn(SessionSpawnArgs {
startup: None,
config,
codex-rs/core/src/guardian/mod.rs
+2
View File
@@ -19,6 +19,8 @@ mod review;
mod review_session;
mod reviewer_config;
mod runtime;
#[cfg(test)]
pub(crate) mod test_host;
use codex_protocol::items::ModelInvocationContext;
use std::sync::Arc;
codex-rs/core/src/guardian/review_session.rs
+4 -12
View File
@@ -2,10 +2,10 @@
//! The extension owns review policy and pooling; this module binds runtime operations
//! to the captured parent action, environments, authorization and context snapshots.
#[path = "review_session_factory.rs"]
mod factory;
pub(crate) use factory::prewarm_guardian_review_session;
pub(crate) use factory::run_guardian_review_session;
#[path = "review_session_setup.rs"]
mod setup;
pub(crate) use setup::prewarm_guardian_review_session;
pub(crate) use setup::run_guardian_review_session;
#[path = "review_session_threads.rs"]
mod managed_threads;
@@ -55,7 +55,6 @@ use tokio_util::sync::CancellationToken;
use tracing::warn;
use crate::agents_md_manager::SessionInstructions;
use crate::codex_delegate::run_codex_thread_interactive;
use crate::config::Config;
use crate::config::Constrained;
use crate::config::ManagedFeatures;
@@ -70,7 +69,6 @@ use crate::image_preparation::ImagePreparationMode;
use crate::image_preparation::ImageResizeNoticeMode;
use crate::image_preparation::prepare_response_items;
use crate::image_preparation::unified_image_budget_enabled;
use crate::session::GitEnrichmentPolicy;
use crate::session::SessionIo;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
@@ -141,12 +139,6 @@ impl GuardianReviewSessionHost {
managed_threads: Some(managed_threads::ManagedReviewerThreads::new(manager)),
}
}
pub fn mark_ready(&self) {
if let Some(threads) = &self.managed_threads {
threads.mark_ready();
}
}
}
pub(crate) type GuardianReviewSessionManager =
codex-rs/core/src/guardian/review_session_factory.rs → codex-rs/core/src/guardian/review_session_setup.rs
+95 -79
View File
@@ -7,7 +7,7 @@ use codex_guardian_reviewer::ReviewerRequest;
use codex_guardian_reviewer::ReviewerSessionFactory;
use codex_guardian_reviewer::SessionDisposition;
pub(super) struct PreparedSession {
pub(super) struct PreparedGuardianContext {
parent: Arc<Session>,
context: GuardianReviewContext,
config: Config,
@@ -17,7 +17,7 @@ pub(super) struct PreparedSession {
host: Arc<GuardianReviewSessionHost>,
}
impl PreparedSession {
impl PreparedGuardianContext {
async fn prepare(
parent: Arc<Session>,
context: GuardianReviewContext,
@@ -57,10 +57,8 @@ impl PreparedSession {
}
}
impl ReviewerSessionFactory for PreparedSession {
type Session = GuardianReviewSession;
fn context(&self, previous: Option<&GuardianReviewSession>) -> GuardianReviewSessionReuseKey {
impl PreparedGuardianContext {
fn reuse_key(&self, previous: Option<&GuardianReviewSession>) -> GuardianReviewSessionReuseKey {
let mut key = self.key.clone();
if self.context_policy != ReviewContextPolicy::ThreadOwned
&& self.parent_compaction.is_none()
@@ -73,6 +71,70 @@ impl ReviewerSessionFactory for PreparedSession {
key
}
/// Returns captured parent inputs; Guardian selects the agent's identity and lifecycle.
async fn thread_options(
&self,
initial_history: Option<InitialHistory>,
) -> crate::StartThreadOptions {
let initial_history = initial_history.or_else(|| {
self.parent_compaction
.clone()
.map(|item| InitialHistory::Forked(vec![RolloutItem::ResponseItem(item.into())]))
});
let mut config = self.config.clone();
config.model_provider.supports_websockets &= self
.parent
.services
.model_client
.responses_websocket_enabled();
crate::StartThreadOptions {
internal_parent: Some(crate::thread_manager::InternalSessionParent {
thread_id: self.parent.thread_id(),
auth_manager: Arc::clone(&self.parent.services.auth_manager),
agent_control: self.parent.services.agent_control.clone(),
originator: self.context.turn().originator.clone(),
inherited_instructions: Some(self.parent.inherited_instructions().await),
}),
initial_history: initial_history.unwrap_or(InitialHistory::New),
environments: Some(self.context.environments().to_selections()),
inherited_environments: Some(self.context.environments().clone()),
client_mcp_extensions: self.parent.services.client_mcp_extensions.clone(),
..crate::StartThreadOptions::new(config)
}
}
/// Binds context bookkeeping to an agent that Guardian has already started.
async fn bind_session(
&self,
session: Arc<Session>,
io: SessionIo,
context: GuardianReviewSessionReuseKey,
state: GuardianReviewState,
cancellation: CancellationToken,
) -> GuardianReviewSession {
let inherited = session.inherited_instructions().await;
let context = GuardianReviewSessionReuseKey {
user_instructions: inherited.user,
thread_instructions: inherited.thread,
..context
};
GuardianReviewSession {
session,
io,
cancel_token: cancellation,
reuse_key: context,
state: Mutex::new(state),
}
}
}
impl ReviewerSessionFactory for PreparedGuardianContext {
type Session = GuardianReviewSession;
fn context(&self, previous: Option<&GuardianReviewSession>) -> GuardianReviewSessionReuseKey {
self.reuse_key(previous)
}
async fn spawn(
&self,
context: GuardianReviewSessionReuseKey,
@@ -80,87 +142,41 @@ impl ReviewerSessionFactory for PreparedSession {
snapshot: Option<GuardianReviewForkSnapshot>,
cancellation: CancellationToken,
) -> anyhow::Result<GuardianReviewSession> {
let mut config = self.config.clone();
if matches!(kind, GuardianReviewSessionKind::EphemeralForked) {
config.ephemeral = true;
}
let (initial_history, conversation, last_admitted_node_repl_response_sequence) =
match snapshot {
Some(snapshot) => {
let (conversation, history) = ConversationState::fork(snapshot);
(
Some(history.initial_history),
conversation,
history.last_admitted_node_repl_response_sequence,
)
}
None => (
self.parent_compaction.clone().map(|item| {
InitialHistory::Forked(vec![RolloutItem::ResponseItem(item.into())])
}),
ConversationState::default(),
0,
),
};
let (session, io) = match &self.host.managed_threads {
Some(threads) => {
threads
.spawn(
&self.parent,
&self.context,
config,
cancellation.clone(),
initial_history,
)
.await?
}
None => {
Box::pin(run_codex_thread_interactive(
config,
Arc::clone(&self.parent.services.auth_manager),
self.parent.services.models_manager.clone(),
Arc::clone(&self.parent),
Arc::clone(self.context.turn()),
self.context.environments().clone(),
cancellation.clone(),
SubAgentSource::Other(GUARDIAN_REVIEWER_NAME.to_owned()),
codex_extension_api::SessionIsolation::Isolated,
initial_history,
GitEnrichmentPolicy::Skip,
codex_sandboxing::WindowsSandboxProxySettingsMode::Preserve,
))
.await?
}
};
let inherited = session.inherited_instructions().await;
let context = GuardianReviewSessionReuseKey {
user_instructions: inherited.user,
thread_instructions: inherited.thread,
..context
};
Ok(GuardianReviewSession {
session,
io,
cancel_token: cancellation,
reuse_key: context,
state: Mutex::new(GuardianReviewState {
conversation,
last_admitted_node_repl_response_sequence,
pending_node_repl_evidence_admission: None,
let (conversation, history) = snapshot.map(ConversationState::fork).unzip();
let state = GuardianReviewState {
conversation: conversation.unwrap_or_default(),
last_admitted_node_repl_response_sequence: history.as_ref().map_or(0, |history| {
history.last_admitted_node_repl_response_sequence
}),
})
pending_node_repl_evidence_admission: None,
};
let mut options = self
.thread_options(history.map(|history| history.initial_history))
.await;
if matches!(kind, GuardianReviewSessionKind::EphemeralForked) {
options.config.ephemeral = true;
}
let threads = self.host.managed_threads.as_ref().ok_or_else(|| {
anyhow::anyhow!("Guardian extension is not installed for this thread")
})?;
let (session, io) = threads
.spawn(&self.parent, options, cancellation.clone())
.await?;
Ok(self
.bind_session(session, io, context, state, cancellation)
.await)
}
}
pub(super) struct PreparedReview {
factory: PreparedSession,
factory: PreparedGuardianContext,
params: GuardianReviewSessionParams,
}
impl ReviewerRequest for PreparedReview {
type Factory = PreparedSession;
type Factory = PreparedGuardianContext;
fn factory(&self) -> &PreparedSession {
fn factory(&self) -> &PreparedGuardianContext {
&self.factory
}
fn deadline(&self) -> tokio::time::Instant {
@@ -212,7 +228,7 @@ pub(crate) async fn run_guardian_review_session(
pub(super) async fn prepare_review(
params: GuardianReviewSessionParams,
) -> anyhow::Result<PreparedReview> {
let factory = PreparedSession::prepare(
let factory = PreparedGuardianContext::prepare(
Arc::clone(&params.parent_session),
params.parent_context.clone(),
params.spawn_config.clone(),
@@ -233,7 +249,7 @@ pub(crate) fn prewarm_guardian_review_session(
let context = GuardianReviewContext::from(turn);
let config = guardian_review_session_config(&parent, &context).await?;
let history = parent.clone_history().await;
let factory = PreparedSession::prepare(
let factory = PreparedGuardianContext::prepare(
Arc::clone(&parent),
context,
config.spawn_config,
codex-rs/core/src/guardian/review_session_tests.rs
+2 -4
View File
@@ -76,7 +76,7 @@ async fn run_review_preserves_evidence_during_parent_compaction() {
let manager = GuardianReviewSessionManager::default();
prewarm_test_session(&manager, reviewer).await;
// Capture the review context, then compact the parent before the reviewer builds its prompt.
let prepared = factory::prepare_review(params).await.unwrap();
let prepared = setup::prepare_review(params).await.unwrap();
let checkpoint: ResponseItem = serde_json::from_value(serde_json::json!({
"type": "compaction", "id": "cmp_new", "encrypted_content": "new-checkpoint"
}))
@@ -312,9 +312,7 @@ async fn spawned_guardian_reuse_key_matches_inherited_instructions() {
thread_instructions: latest.clone(),
..stale_key.clone()
};
let prepared = factory::prepare_review(params)
.await
.expect("prepare review");
let prepared = setup::prepare_review(params).await.expect("prepare review");
let review = prepared
.factory()
.spawn(
codex-rs/core/src/guardian/review_session_threads.rs
+12 -46
View File
@@ -1,5 +1,5 @@
//! Adapts the existing reviewer lifecycle to the extension's ThreadManager.
//! Parent registration gates spawning; prompts, reuse and review outcomes stay shared.
//! Captured parent inputs support inline delegates; prompts and review outcomes stay shared.
use std::sync::Arc;
use std::sync::Weak;
@@ -11,15 +11,12 @@ use codex_protocol::protocol::InternalSessionSource;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadSource;
use tokio::sync::watch;
use tokio_util::sync::CancellationToken;
use super::GUARDIAN_REVIEWER_NAME;
use super::GuardianReviewContext;
use crate::StartThreadOptions;
use crate::ThreadManager;
use crate::codex_delegate::forward_session_io;
use crate::config::Config;
use crate::config::Constrained;
use crate::session::SessionIo;
use crate::session::emit_subagent_session_started;
@@ -27,72 +24,41 @@ use crate::session::session::Session;
pub(super) struct ManagedReviewerThreads {
manager: Weak<ThreadManager>,
ready: watch::Sender<bool>,
}
impl ManagedReviewerThreads {
pub(super) fn new(manager: Weak<ThreadManager>) -> Self {
Self {
manager,
ready: watch::channel(/*init*/ false).0,
}
}
pub(super) fn mark_ready(&self) {
self.ready.send_replace(true);
Self { manager }
}
pub(super) async fn spawn(
&self,
parent: &Arc<Session>,
context: &GuardianReviewContext,
mut config: Config,
mut options: StartThreadOptions,
cancel: CancellationToken,
history: Option<InitialHistory>,
) -> anyhow::Result<(Arc<Session>, SessionIo)> {
let mut ready = self.ready.subscribe();
tokio::select! {
result = ready.wait_for(|ready| *ready) => { result?; }
_ = cancel.cancelled() => anyhow::bail!("guardian reviewer is shutting down"),
}
let manager = self
.manager
.upgrade()
.ok_or_else(|| anyhow::anyhow!("thread manager is no longer available"))?;
// Match the standalone delegate's settings before using the managed spawn path.
if config.permissions.approval_policy.value() != AskForApproval::Never {
if options.config.permissions.approval_policy.value() != AskForApproval::Never {
anyhow::bail!("Codex delegates require approval policy `never`");
}
config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never);
config.model_provider.supports_websockets &=
parent.services.model_client.responses_websocket_enabled();
let mut thread_extension_init = codex_extension_api::ExtensionDataInit::default();
thread_extension_init.insert(codex_extension_api::SessionIsolation::Isolated);
let options = StartThreadOptions {
thread_extension_init,
session_source: Some(SessionSource::Internal(InternalSessionSource::Guardian)),
thread_source: Some(ThreadSource::GuardianReview),
environments: Some(context.environments().to_selections()),
inherited_environments: Some(context.environments().clone()),
client_mcp_extensions: parent.services.client_mcp_extensions.clone(),
..StartThreadOptions::new(config)
};
let history = match history.unwrap_or(InitialHistory::New) {
options.config.permissions.approval_policy = Constrained::allow_only(AskForApproval::Never);
options.session_source = Some(SessionSource::Internal(InternalSessionSource::Guardian));
options.thread_source = Some(ThreadSource::GuardianReview);
options
.thread_extension_init
.insert(codex_extension_api::SessionIsolation::Isolated);
options.initial_history = match options.initial_history {
InitialHistory::Forked(history) => InitialHistory::Forked(history),
InitialHistory::New | InitialHistory::Cleared => InitialHistory::New,
InitialHistory::Resumed(_) => {
anyhow::bail!("guardian review forks cannot resume an existing thread")
}
};
// Generic internal sessions start isolated. Reviewers explicitly inherit
// the parent's applied instructions, just like standalone delegates.
let spawned = manager
.spawn_internal_session_with_history(
parent.thread_id(),
options,
history,
Some(parent.inherited_instructions().await),
)
.start_thread(options)
.or_cancel(&cancel)
.await
.map_err(codex_protocol::error::CodexErr::from)??;
codex-rs/core/src/guardian/test_host.rs
+35
View File
@@ -0,0 +1,35 @@
//! Explicit ThreadManager host for core's context-adapter unit tests.
//! Application-level tests install the real Guardian extension instead.
use std::sync::Arc;
use codex_home::CodexHomeUserInstructionsProvider;
use codex_protocol::protocol::SessionSource;
use crate::config::Config;
use crate::session::session::Session;
pub(crate) fn install(session: &Session, config: &Config) {
let manager = Arc::new(crate::ThreadManager::new(
config,
Arc::clone(&session.services.auth_manager),
Arc::clone(&session.services.models_manager),
crate::CodexAppsToolsCache::default(),
SessionSource::Exec,
session.services.turn_environments.environment_manager(),
codex_extension_api::empty_extension_registry(),
Arc::new(CodexHomeUserInstructionsProvider::new(
config.codex_home.clone(),
)),
/*analytics_events_client*/ None,
crate::passthrough_image_store(),
Arc::clone(&session.services.thread_store),
/*agent_graph_store*/ None,
session.installation_id.clone(),
/*attestation_provider*/ None,
/*external_time_provider*/ None,
));
let host = super::GuardianReviewSessionHost::with_thread_manager(Arc::downgrade(&manager));
session.services.thread_extension_data.insert(host);
session.services.thread_extension_data.insert(manager);
}
codex-rs/core/src/guardian/tests.rs
+5
View File
@@ -178,6 +178,7 @@ async fn guardian_test_session_turn_and_rx(
.expect("session should be uniquely owned")
.services
.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
let turn_mut = Arc::get_mut(&mut turn).expect("turn should be uniquely owned");
turn_mut.config = Arc::clone(&config);
turn_mut.provider =
@@ -240,6 +241,7 @@ async fn guardian_test_session_and_turn_with_base_url(
config.model_provider.clone(),
);
session.services.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
turn.config = Arc::clone(&config);
turn.provider = create_model_provider(config.model_provider.clone(), turn.auth_manager.clone());
@@ -1973,6 +1975,7 @@ async fn guardian_request_model_for_auto_review(
.expect("session should be unique")
.services
.models_manager = Arc::new(models_manager);
crate::guardian::test_host::install(&session, &turn.config);
}
}
update_turn_settings_for_test(
@@ -2188,6 +2191,7 @@ async fn guardian_review_request_layout_matches_model_visible_request_snapshot()
config.model_provider.clone(),
);
session.services.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
let memory_extension = Arc::new(GuardianMemoryContextProbe);
let mut extensions = codex_extension_api::ExtensionRegistryBuilder::<Config>::new();
extensions.thread_lifecycle_contributor(memory_extension.clone());
@@ -3018,6 +3022,7 @@ async fn guardian_review_surfaces_responses_api_errors_in_rejection_reason() ->
.expect("session should be uniquely owned")
.services
.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
let turn_mut = Arc::get_mut(&mut turn).expect("turn should be uniquely owned");
turn_mut.config = Arc::clone(&config);
turn_mut.provider =
codex-rs/core/src/lib.rs
+1
View File
@@ -121,6 +121,7 @@ mod thread_manager;
pub(crate) mod web_search;
pub(crate) mod windows_sandbox_read_grants;
pub use thread_manager::ForkSnapshot;
pub use thread_manager::InternalSessionParent;
pub use thread_manager::NewThread;
pub use thread_manager::StartThreadOptions;
pub use thread_manager::ThreadManager;
codex-rs/core/src/session/tests.rs
+2
View File
@@ -6167,6 +6167,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
skills_snapshot,
);
session.mark_mcp_runtime_dirty();
crate::guardian::test_host::install(&session, &turn_context.config);
(session, turn_context)
}
@@ -8344,6 +8345,7 @@ where
skills_snapshot,
));
session.mark_mcp_runtime_dirty();
crate::guardian::test_host::install(&session, &turn_context.config);
(session, turn_context, rx_event)
}
codex-rs/core/src/session/tests/guardian_tests.rs
+4
View File
@@ -223,6 +223,7 @@ async fn request_permissions_routes_to_guardian_when_reviewer_is_enabled() {
config.model_provider.clone(),
);
session.services.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
turn_context_raw.config = Arc::clone(&config);
turn_context_raw.provider = create_model_provider(
config.model_provider.clone(),
@@ -423,6 +424,7 @@ async fn request_permissions_guardian_review_stops_when_cancelled() {
.expect("single session ref")
.services
.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
turn_context_raw.config = Arc::clone(&config);
turn_context_raw.provider = create_model_provider(
config.model_provider.clone(),
@@ -549,6 +551,7 @@ async fn guardian_allows_exec_command_additional_permissions_requests_past_polic
config.model_provider.clone(),
);
session.services.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
turn_context_raw.config = Arc::clone(&config);
turn_context_raw.provider = create_model_provider(
config.model_provider.clone(),
@@ -668,6 +671,7 @@ async fn strict_auto_review_turn_grant_forces_guardian_for_exec_command_policy_s
config.model_provider.clone(),
);
session.services.models_manager = models_manager;
crate::guardian::test_host::install(&session, &config);
turn_context_raw.config = Arc::clone(&config);
turn_context_raw.provider = create_model_provider(
config.model_provider.clone(),
codex-rs/core/src/thread_manager.rs
+57 -15
View File
@@ -232,6 +232,18 @@ pub struct ThreadManager {
_test_codex_home_guard: Option<TempCodexHomeGuard>,
}
/// Captured parent identity for an internal child, including inline parents that
/// do not have a public entry in the thread manager. Authentication and the shared
/// agent budget remain bound to the parent that created this value.
#[derive(Clone)]
pub struct InternalSessionParent {
pub(crate) thread_id: ThreadId,
pub(crate) auth_manager: Arc<AuthManager>,
pub(crate) agent_control: AgentControl,
pub(crate) originator: String,
pub(crate) inherited_instructions: Option<SessionInstructions>,
}
pub struct StartThreadOptions {
pub config: Config,
/// Host-owned provider for this root thread's additional instructions.
@@ -249,6 +261,8 @@ pub struct StartThreadOptions {
pub thread_instructions_provider: Option<Arc<dyn ThreadInstructionsProvider>>,
pub allow_provider_model_fallback: bool,
pub initial_history: InitialHistory,
/// Parent captured for `start_thread`, including a parent not in the live registry.
pub internal_parent: Option<InternalSessionParent>,
pub history_mode: Option<ThreadHistoryMode>,
pub session_source: Option<SessionSource>,
pub thread_source: Option<ThreadSource>,
@@ -275,6 +289,7 @@ impl StartThreadOptions {
thread_instructions_provider: None,
allow_provider_model_fallback: false,
initial_history: InitialHistory::New,
internal_parent: None,
history_mode: None,
session_source: None,
thread_source: None,
@@ -298,6 +313,7 @@ struct ThreadSpawnRequest {
auth_manager: Arc<AuthManager>,
agent_control: AgentControl,
parent_thread_id: Option<ThreadId>,
parent_originator: Option<String>,
forked_from_thread_id: Option<ThreadId>,
fork_persistence: ForkPersistence,
inherited_environments: Option<TurnEnvironmentSnapshot>,
@@ -318,6 +334,7 @@ impl ThreadSpawnRequest {
auth_manager,
agent_control,
parent_thread_id: None,
parent_originator: None,
forked_from_thread_id: None,
fork_persistence: ForkPersistence::Copied,
inherited_environments: None,
@@ -1044,17 +1061,15 @@ impl ThreadManager {
));
}
let parent = self.get_thread(parent_thread_id).await?;
let forked_from_thread_id = history.forked_from_id();
options.initial_history = history;
let mut request = ThreadSpawnRequest::new(
options,
Arc::clone(&parent.session.services.auth_manager),
parent.session.services.agent_control.clone(),
);
request.parent_thread_id = Some(parent_thread_id);
request.forked_from_thread_id = forked_from_thread_id;
request.inherited_instructions = inherited_instructions;
Box::pin(self.state.spawn_thread(request)).await
options.internal_parent = Some(InternalSessionParent {
thread_id: parent_thread_id,
auth_manager: Arc::clone(&parent.session.services.auth_manager),
agent_control: parent.session.services.agent_control.clone(),
originator: parent.config_snapshot().await.originator,
inherited_instructions,
});
self.start_thread(options).await
}
/// Allocates a thread ID before startup so a caller can associate host-owned state with it.
@@ -1068,7 +1083,6 @@ impl ThreadManager {
forked_from_thread_id: Option<ThreadId>,
startup: Option<Arc<crate::session::startup::SessionStartup>>,
) -> CodexResult<NewThread> {
let agent_control = self.agent_control_for_config(&options.config);
let (resumed_session_source, resumed_thread_source) = options
.initial_history
.get_resumed_session_sources()
@@ -1080,9 +1094,33 @@ impl ThreadManager {
.unwrap_or(resumed_session_source),
);
options.thread_source = options.thread_source.take().or(resumed_thread_source);
let mut request =
ThreadSpawnRequest::new(options, Arc::clone(&self.state.auth_manager), agent_control);
request.forked_from_thread_id = forked_from_thread_id;
let parent = options.internal_parent.take();
if parent.is_some()
&& (!matches!(options.session_source, Some(SessionSource::Internal(_)))
|| matches!(options.initial_history, InitialHistory::Resumed(_)))
{
return Err(CodexErr::InvalidRequest(
"a captured internal parent requires a new or forked internal session".to_owned(),
));
}
let mut request = if let Some(parent) = parent {
let mut request =
ThreadSpawnRequest::new(options, parent.auth_manager, parent.agent_control);
request.parent_thread_id = Some(parent.thread_id);
request.parent_originator = Some(parent.originator);
request.inherited_instructions = parent.inherited_instructions;
request.forked_from_thread_id = request.options.initial_history.forked_from_id();
request
} else {
let agent_control = self.agent_control_for_config(&options.config);
let mut request = ThreadSpawnRequest::new(
options,
Arc::clone(&self.state.auth_manager),
agent_control,
);
request.forked_from_thread_id = forked_from_thread_id;
request
};
request.startup = startup;
Box::pin(self.state.spawn_thread(request)).await
}
@@ -1748,6 +1786,7 @@ impl ThreadManagerState {
session_source: &SessionSource,
parent_thread_id: Option<ThreadId>,
forked_from_thread_id: Option<ThreadId>,
parent_originator: Option<String>,
) -> String {
let persisted_originator = initial_history.get_session_originator();
let inherited_originator = match initial_history {
@@ -1777,7 +1816,7 @@ impl ThreadManagerState {
metrics_service_name,
env_originator,
persisted_originator,
inherited_originator,
inherited_originator.or(parent_originator),
originator().value,
)
}
@@ -1938,6 +1977,7 @@ impl ThreadManagerState {
auth_manager,
agent_control,
parent_thread_id,
parent_originator,
forked_from_thread_id,
fork_persistence,
inherited_environments,
@@ -1950,6 +1990,7 @@ impl ThreadManagerState {
thread_instructions_provider,
allow_provider_model_fallback,
initial_history,
internal_parent: _,
history_mode,
session_source,
thread_source,
@@ -2084,6 +2125,7 @@ impl ThreadManagerState {
&session_source,
parent_thread_id,
forked_from_thread_id,
parent_originator,
)
.await;
let source_changed_during_startup = Arc::new(AtomicBool::new(false));
codex-rs/core/src/thread_manager_tests.rs
+41
View File
@@ -1476,6 +1476,47 @@ async fn spawn_internal_session_preserves_parent_lineage_without_forking_history
.is_none()
);
// Captured lineage also works for an inline parent with no registry entry.
manager.remove_thread(&parent.thread_id).await;
let child = manager
.start_thread(StartThreadOptions {
internal_parent: Some(InternalSessionParent {
thread_id: parent.thread_id,
auth_manager: Arc::clone(&parent.thread.session.services.auth_manager),
agent_control: parent.thread.session.services.agent_control.clone(),
originator: reviewer_config.originator.clone(),
inherited_instructions: None,
}),
session_source: Some(SessionSource::Internal(
InternalSessionSource::MemoryConsolidation,
)),
..StartThreadOptions::new(parent.thread.session.get_config().await.as_ref().clone())
})
.await
.expect("start child without a parent registry entry");
let child_config = child.thread.config_snapshot().await;
assert_eq!(
(
child_config.parent_thread_id,
child_config.originator,
child.thread.session.session_id(),
),
(
Some(parent.thread_id),
reviewer_config.originator,
parent.thread.session.session_id(),
)
);
assert!(Arc::ptr_eq(
&child.thread.session.services.auth_manager,
&parent.thread.session.services.auth_manager
));
assert!(manager.list_thread_ids().await.is_empty());
parent
.thread
.shutdown_and_wait()
.await
.expect("stop parent");
manager
.shutdown_all_threads_bounded(Duration::from_secs(10))
.await;
codex-rs/core/tests/common/Cargo.toml
+1
View File
@@ -20,6 +20,7 @@ codex-config = { workspace = true }
codex-core = { workspace = true }
codex-extension-api = { workspace = true }
codex-exec-server = { workspace = true }
codex-guardian-v2 = { workspace = true }
codex-home = { workspace = true }
codex-features = { workspace = true }
codex-hooks = { workspace = true }
codex-rs/core/tests/common/test_codex.rs
+34 -28
View File
@@ -719,39 +719,45 @@ impl TestCodexBuilder {
.models_manager
.clone()
.unwrap_or_else(|| codex_core::build_models_manager(&config, auth_manager.clone()));
let thread_manager = ThreadManager::new(
&config,
auth_manager.clone(),
models_manager,
codex_core::CodexAppsToolsCache::default(),
SessionSource::Exec,
Arc::clone(&environment_manager),
Arc::clone(&self.extensions),
user_instructions_provider,
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
Arc::clone(&thread_store),
codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,
/*attestation_provider*/ None,
/*external_time_provider*/ self.external_time_provider.clone(),
);
let code_mode_host_program = self
.code_mode_host_program
.take()
.or_else(|| codex_utils_cargo_bin::cargo_bin("codex-code-mode-host").ok());
let thread_manager = if config.features.enabled(Feature::CodeModeHost)
&& let Some(code_mode_host_program) = code_mode_host_program
{
codex_core::test_support::with_code_mode_host_program(
thread_manager,
code_mode_host_program,
let thread_manager = Arc::new_cyclic(|manager| {
let mut extensions = self.extensions.to_builder();
let guardian = Arc::new(codex_guardian_v2::GuardianReviewerExtension::new(
manager.clone(),
));
extensions.thread_lifecycle_contributor(guardian);
let thread_manager = ThreadManager::new(
&config,
)
} else {
thread_manager
};
let thread_manager = Arc::new(thread_manager);
auth_manager.clone(),
models_manager,
codex_core::CodexAppsToolsCache::default(),
SessionSource::Exec,
Arc::clone(&environment_manager),
Arc::new(extensions.build()),
user_instructions_provider,
/*analytics_events_client*/ None,
codex_core::passthrough_image_store(),
Arc::clone(&thread_store),
codex_core::local_agent_graph_store_from_state_db(state_db.as_ref()),
installation_id,
/*attestation_provider*/ None,
/*external_time_provider*/ self.external_time_provider.clone(),
);
if config.features.enabled(Feature::CodeModeHost)
&& let Some(code_mode_host_program) = code_mode_host_program
{
codex_core::test_support::with_code_mode_host_program(
thread_manager,
code_mode_host_program,
&config,
)
} else {
thread_manager
}
});
let user_shell_override = self.user_shell_override.clone();
let client_mcp_extensions = || {
ClientMcpExtensions::new(
codex-rs/ext/extension-api/src/registry.rs
+22
View File
@@ -162,6 +162,28 @@ pub struct ExtensionRegistry<C: Sync> {
}
impl<C: Sync> ExtensionRegistry<C> {
/// Copies the registered contributors into a builder for host-specific additions.
pub fn to_builder(&self) -> ExtensionRegistryBuilder<C> {
ExtensionRegistryBuilder {
registry: Self {
event_sink: self.event_sink.clone(),
turn_start_admission: self.turn_start_admission.clone(),
thread_lifecycle_contributors: self.thread_lifecycle_contributors.clone(),
turn_lifecycle_contributors: self.turn_lifecycle_contributors.clone(),
config_contributors: self.config_contributors.clone(),
token_usage_contributors: self.token_usage_contributors.clone(),
skill_invocation_contributors: self.skill_invocation_contributors.clone(),
context_contributors: self.context_contributors.clone(),
mcp_server_contributors: self.mcp_server_contributors.clone(),
turn_input_contributors: self.turn_input_contributors.clone(),
tool_contributors: self.tool_contributors.clone(),
tool_lifecycle_contributors: self.tool_lifecycle_contributors.clone(),
turn_item_contributors: self.turn_item_contributors.clone(),
approval_review_contributors: self.approval_review_contributors.clone(),
},
}
}
/// Acquires the host's turn-start permit, or an empty permit for ungated hosts.
/// A missing permit rejects the start before Core consumes pending input.
pub fn admit_turn_start(&self) -> Option<Box<dyn Send>> {
codex-rs/ext/guardian-v2/src/sync_reviewer/mod.rs
-12
View File
@@ -10,7 +10,6 @@ use codex_core::guardian_review::GuardianReviewSessionHost;
use codex_extension_api::ExtensionFuture;
use codex_extension_api::ExtensionRegistryBuilder;
use codex_extension_api::ThreadLifecycleContributor;
use codex_extension_api::ThreadReadyInput;
use codex_extension_api::ThreadStartInput;
/// Owns reviewer state through the same thread manager as the parent conversation.
@@ -39,17 +38,6 @@ impl ThreadLifecycleContributor<Config> for GuardianExtension {
});
})
}
fn on_thread_ready<'a>(
&'a self,
input: ThreadReadyInput<'a, Config>,
) -> ExtensionFuture<'a, ()> {
Box::pin(async move {
if let Some(sessions) = input.thread_store.get::<GuardianReviewSessionHost>() {
sessions.mark_ready();
}
})
}
}
pub fn install(