mirror of
https://github.com/openai/codex.git
synced 2026-09-20 12:47:38 +00:00
17baabd01b
## Why `CodexThread` retained the full `SessionConfiguredEvent`, including initial replay messages, and cloned it even for callers that only needed thread or session IDs. ## What changed Store `ThreadStartupMetadata` without replay history and expose it by reference through `startup_metadata()`. Update callers to read IDs from this metadata. When rejoining a running thread, rebuild the startup response using the supplied resume history for `initial_messages`. ## Testing Extend the resume test to verify that rejoining reuses the loaded thread, preserves startup metadata, and still returns the original replay messages as a prefix. GitOrigin-RevId: 9d656063088c28f091d29e5862e4b9d894b7642c
1040 lines
38 KiB
Rust
1040 lines
38 KiB
Rust
use crate::agent::AgentStatus;
|
|
use crate::config::ConstraintResult;
|
|
use crate::context::ContextualUserFragment;
|
|
use crate::context::GuardianReviewEvidence;
|
|
use crate::elicitation::ElicitationRegistration;
|
|
use crate::session::SessionIo;
|
|
use crate::session::SessionSettingsUpdate;
|
|
use crate::session::new_submission_id;
|
|
use crate::session::session::Session;
|
|
use crate::session::step_settings::StepSettingsUpdate;
|
|
use crate::thread_startup_metadata::ThreadStartupMetadata;
|
|
use codex_diagnostics::Gauge;
|
|
use codex_diagnostics::GaugeGuard;
|
|
use codex_exec_server::SelectedCapabilityRootsStatus;
|
|
use codex_extension_api::ConversationHistorySnapshot;
|
|
use codex_extension_api::ThreadIdleCause;
|
|
use codex_features::Feature;
|
|
use codex_history::RolloutItem;
|
|
use codex_otel::SessionTelemetry;
|
|
use codex_otel::current_span_w3c_trace_context;
|
|
use codex_protocol::ThreadId;
|
|
use codex_protocol::config_types::ApprovalsReviewer;
|
|
use codex_protocol::config_types::CollaborationMode;
|
|
use codex_protocol::config_types::Personality;
|
|
use codex_protocol::config_types::ReasoningSummary;
|
|
use codex_protocol::config_types::WindowsSandboxLevel;
|
|
use codex_protocol::error::CodexErr;
|
|
use codex_protocol::error::Result as CodexResult;
|
|
use codex_protocol::mcp::CallToolResult;
|
|
use codex_protocol::mcp::ClientMcpExtensions;
|
|
use codex_protocol::models::ActivePermissionProfile;
|
|
use codex_protocol::models::PermissionProfile;
|
|
use codex_protocol::models::ProfileWorkspaceRoot;
|
|
use codex_protocol::models::ResponseItem;
|
|
use codex_protocol::openai_models::ReasoningEffort;
|
|
use codex_protocol::protocol::AskForApproval;
|
|
use codex_protocol::protocol::EnvironmentConfig;
|
|
use codex_protocol::protocol::EnvironmentConfigState;
|
|
use codex_protocol::protocol::Event;
|
|
use codex_protocol::protocol::MultiAgentVersion;
|
|
use codex_protocol::protocol::Op;
|
|
use codex_protocol::protocol::SandboxPolicy;
|
|
use codex_protocol::protocol::SessionSource;
|
|
use codex_protocol::protocol::Submission;
|
|
use codex_protocol::protocol::ThreadHistoryMode;
|
|
use codex_protocol::protocol::ThreadMemoryMode;
|
|
use codex_protocol::protocol::ThreadSettingsSnapshot;
|
|
use codex_protocol::protocol::ThreadSource;
|
|
use codex_protocol::protocol::TokenUsageInfo;
|
|
use codex_protocol::protocol::TurnEnvironmentSelection;
|
|
use codex_protocol::protocol::TurnEnvironmentSelections;
|
|
use codex_protocol::protocol::W3cTraceContext;
|
|
use codex_protocol::turn_input::RecoverTurnRequest;
|
|
use codex_protocol::turn_input::StartIfIdleSubmission;
|
|
use codex_protocol::turn_input::SteerSubmission;
|
|
use codex_protocol::turn_input::SuspendTurnOutcome;
|
|
use codex_protocol::turn_input::TurnInputMode;
|
|
use codex_protocol::turn_input::TurnInputRequest;
|
|
use codex_protocol::turn_input::TurnInputSubmission;
|
|
use codex_protocol::turn_input::TurnStartOptions;
|
|
use codex_thread_store::PersistContext;
|
|
use codex_thread_store::StoredThread;
|
|
use codex_thread_store::StoredThreadHistory;
|
|
use codex_thread_store::ThreadMetadataPatch;
|
|
use codex_thread_store::ThreadStoreError;
|
|
use codex_thread_store::ThreadStoreResult;
|
|
use codex_utils_absolute_path::AbsolutePathBuf;
|
|
use codex_utils_path_uri::LegacyAppPathString;
|
|
use codex_utils_path_uri::PathUri;
|
|
use rmcp::model::ReadResourceRequestParams;
|
|
use std::path::PathBuf;
|
|
use std::sync::Arc;
|
|
use tokio::sync::Mutex;
|
|
use tokio::sync::oneshot;
|
|
use tokio::sync::watch;
|
|
use tokio_util::sync::CancellationToken;
|
|
|
|
use codex_rollout::state_db::StateDbHandle;
|
|
|
|
static LIVE_THREADS: Gauge = Gauge::new("core.threads.live");
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct ThreadConfigSnapshot {
|
|
pub model: String,
|
|
pub model_provider_id: String,
|
|
pub service_tier: Option<String>,
|
|
pub approval_policy: AskForApproval,
|
|
pub approvals_reviewer: ApprovalsReviewer,
|
|
pub permission_profile: PermissionProfile,
|
|
/// Resolved Full Access across the thread and every selected environment.
|
|
pub full_access: bool,
|
|
pub active_permission_profile: Option<ActivePermissionProfile>,
|
|
pub environments: TurnEnvironmentSelections,
|
|
pub workspace_roots: Vec<AbsolutePathBuf>,
|
|
pub profile_workspace_roots: Vec<ProfileWorkspaceRoot>,
|
|
pub ephemeral: bool,
|
|
pub reasoning_effort: Option<ReasoningEffort>,
|
|
pub reasoning_summary: Option<ReasoningSummary>,
|
|
pub personality: Option<Personality>,
|
|
pub collaboration_mode: CollaborationMode,
|
|
pub session_source: SessionSource,
|
|
pub history_mode: ThreadHistoryMode,
|
|
pub forked_from_thread_id: Option<ThreadId>,
|
|
pub parent_thread_id: Option<ThreadId>,
|
|
pub thread_source: Option<ThreadSource>,
|
|
pub originator: String,
|
|
pub disabled_plugin_ids: Vec<String>,
|
|
}
|
|
|
|
impl ThreadConfigSnapshot {
|
|
pub fn cwd(&self) -> &AbsolutePathBuf {
|
|
&self.environments.legacy_fallback_cwd
|
|
}
|
|
|
|
pub fn environment_selections(&self) -> &[TurnEnvironmentSelection] {
|
|
&self.environments.environments
|
|
}
|
|
|
|
/// Whether the primary environment has resolved configuration, if one is selected.
|
|
pub fn is_primary_environment_configured(&self) -> bool {
|
|
self.environment_selections()
|
|
.first()
|
|
.is_none_or(|selection| {
|
|
matches!(
|
|
selection.config,
|
|
EnvironmentConfigState::FromThread | EnvironmentConfigState::Ready(_)
|
|
)
|
|
})
|
|
}
|
|
|
|
pub fn sandbox_policy(&self) -> SandboxPolicy {
|
|
codex_sandboxing::compatibility_sandbox_policy_for_permission_profile(
|
|
&self.permission_profile,
|
|
self.cwd().as_path(),
|
|
)
|
|
}
|
|
}
|
|
|
|
/// Thread settings overrides that app-server validates before starting a turn.
|
|
#[derive(Clone, Default)]
|
|
pub struct CodexThreadSettingsOverrides {
|
|
pub environments: Option<TurnEnvironmentSelections>,
|
|
pub runtime_workspace_roots: Option<Vec<AbsolutePathBuf>>,
|
|
pub profile_workspace_roots: Option<Vec<ProfileWorkspaceRoot>>,
|
|
pub approval_policy: Option<AskForApproval>,
|
|
pub approvals_reviewer: Option<ApprovalsReviewer>,
|
|
pub sandbox_policy: Option<SandboxPolicy>,
|
|
pub permission_profile: Option<PermissionProfile>,
|
|
pub active_permission_profile: Option<ActivePermissionProfile>,
|
|
pub windows_sandbox_level: Option<WindowsSandboxLevel>,
|
|
pub model: Option<String>,
|
|
pub effort: Option<Option<ReasoningEffort>>,
|
|
pub summary: Option<ReasoningSummary>,
|
|
pub service_tier: Option<Option<String>>,
|
|
pub collaboration_mode: Option<CollaborationMode>,
|
|
pub personality: Option<Personality>,
|
|
pub disabled_plugin_ids: Option<Vec<String>>,
|
|
}
|
|
|
|
pub use codex_guardian_context::GuardianRootMessage;
|
|
|
|
/// Authorization state that changes on genuine user input or history resets.
|
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
|
pub struct GuardianAuthorizationVersion {
|
|
/// User-message/reset revision, preserved across compaction and internal context.
|
|
pub user_message_revision: u64,
|
|
/// Successful host answers captured by the temporary legacy path.
|
|
pub user_input_response_count: usize,
|
|
/// False when required retained answers or root instructions are unavailable.
|
|
pub retained_context_complete: bool,
|
|
}
|
|
|
|
/// Bounded root conversation and authorization state from one history snapshot.
|
|
#[derive(Debug, Eq, PartialEq)]
|
|
pub struct GuardianRootSnapshot {
|
|
pub authorization_version: GuardianAuthorizationVersion,
|
|
pub messages: Vec<GuardianRootMessage>,
|
|
pub trusted_skill_paths: Vec<String>,
|
|
}
|
|
|
|
pub struct CodexThread {
|
|
pub(crate) session: Arc<Session>,
|
|
pub(crate) io: SessionIo,
|
|
// Registration source controls live access and lifecycle hooks. Managed Guardian
|
|
// reviewers keep their existing subagent identity inside the session.
|
|
pub(crate) session_source: SessionSource,
|
|
startup_metadata: ThreadStartupMetadata,
|
|
rollout_path: Option<PathBuf>,
|
|
out_of_band_elicitations: Mutex<OutOfBandElicitations>,
|
|
_diagnostics_guard: GaugeGuard,
|
|
}
|
|
|
|
#[derive(Default)]
|
|
struct OutOfBandElicitations {
|
|
count: i64,
|
|
registration: Option<ElicitationRegistration>,
|
|
}
|
|
|
|
#[derive(Debug, Eq, PartialEq)]
|
|
pub struct BackgroundTerminalInfo {
|
|
pub item_id: String,
|
|
pub process_id: String,
|
|
pub command: String,
|
|
pub cwd: PathUri,
|
|
}
|
|
|
|
/// Conduit for the bidirectional stream of messages that compose a thread
|
|
/// (formerly called a conversation) in Codex.
|
|
impl CodexThread {
|
|
pub(crate) fn new(
|
|
session: Arc<Session>,
|
|
io: SessionIo,
|
|
startup_metadata: ThreadStartupMetadata,
|
|
rollout_path: Option<PathBuf>,
|
|
session_source: SessionSource,
|
|
) -> Self {
|
|
Self {
|
|
session,
|
|
io,
|
|
session_source,
|
|
startup_metadata,
|
|
rollout_path,
|
|
out_of_band_elicitations: Mutex::new(OutOfBandElicitations::default()),
|
|
_diagnostics_guard: LIVE_THREADS.track(),
|
|
}
|
|
}
|
|
|
|
pub async fn submit(&self, op: Op) -> CodexResult<String> {
|
|
self.io.submit(op).await
|
|
}
|
|
|
|
/// Returns the session telemetry handle for thread-scoped production instrumentation.
|
|
pub fn session_telemetry(&self) -> SessionTelemetry {
|
|
self.session.services.session_telemetry.clone()
|
|
}
|
|
|
|
/// Whether analytics is enabled for this thread after configuration and host overrides.
|
|
pub fn analytics_enabled(&self) -> bool {
|
|
self.session.services.analytics_events_client.is_enabled()
|
|
}
|
|
|
|
/// Returns extension-owned data attached to this thread runtime.
|
|
pub fn thread_extension_data(&self) -> &codex_extension_api::ExtensionData {
|
|
&self.session.services.thread_extension_data
|
|
}
|
|
|
|
pub async fn shutdown_and_wait(&self) -> CodexResult<()> {
|
|
self.io.shutdown_and_wait().await
|
|
}
|
|
|
|
/// Wait until the underlying session loop has terminated.
|
|
pub async fn wait_until_terminated(&self) {
|
|
self.io.session_loop_termination.clone().await;
|
|
}
|
|
|
|
pub(crate) async fn emit_thread_ready_lifecycle(&self) {
|
|
let contributors = self
|
|
.session
|
|
.services
|
|
.extensions
|
|
.thread_lifecycle_contributors();
|
|
// Hook-free reviewers must reach their owner without suspending after registration.
|
|
// Otherwise cancellation can strand the registered thread before cleanup is installed.
|
|
if contributors.is_empty() {
|
|
return;
|
|
}
|
|
let config = self.config().await;
|
|
for contributor in contributors {
|
|
contributor
|
|
.on_thread_ready(codex_extension_api::ThreadReadyInput {
|
|
config: config.as_ref(),
|
|
session_source: &self.session_source,
|
|
session_store: &self.session.services.session_extension_data,
|
|
thread_store: &self.session.services.thread_extension_data,
|
|
})
|
|
.await;
|
|
}
|
|
}
|
|
|
|
pub(crate) async fn emit_thread_resume_lifecycle(&self) {
|
|
for contributor in self
|
|
.session
|
|
.services
|
|
.extensions
|
|
.thread_lifecycle_contributors()
|
|
{
|
|
contributor
|
|
.on_thread_resume(codex_extension_api::ThreadResumeInput {
|
|
session_store: &self.session.services.session_extension_data,
|
|
thread_store: &self.session.services.thread_extension_data,
|
|
})
|
|
.await;
|
|
}
|
|
}
|
|
|
|
pub async fn emit_thread_idle_lifecycle_if_idle(&self, cause: ThreadIdleCause) {
|
|
self.session.emit_thread_idle_lifecycle_if_idle(cause).await;
|
|
}
|
|
|
|
#[doc(hidden)]
|
|
pub async fn ensure_rollout_materialized(&self) {
|
|
self.session
|
|
.ensure_rollout_materialized(PersistContext::Standard)
|
|
.await;
|
|
}
|
|
|
|
#[doc(hidden)]
|
|
pub async fn flush_rollout(&self) -> std::io::Result<()> {
|
|
self.session.flush_rollout().await
|
|
}
|
|
|
|
pub async fn submit_with_trace(
|
|
&self,
|
|
op: Op,
|
|
trace: Option<W3cTraceContext>,
|
|
) -> CodexResult<String> {
|
|
self.io
|
|
.submit_with_trace(
|
|
op, trace, /*parent_turn_id*/ None, /*root_turn_id*/ None,
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Submits turn input without requiring the caller to inspect thread state.
|
|
///
|
|
/// The result describes whether Core started a turn, steered an active
|
|
/// turn, or declined it without recording or enqueueing the input.
|
|
/// User input and named standalone function-call outputs are accepted.
|
|
pub async fn start_or_steer_turn(
|
|
&self,
|
|
request: TurnInputRequest,
|
|
) -> CodexResult<TurnInputSubmission> {
|
|
self.submit_turn_input_with_mode(request, TurnInputMode::StartOrSteer)
|
|
.await
|
|
}
|
|
|
|
/// Starts a regular turn only when the thread is idle.
|
|
///
|
|
/// Core declines the input without recording or enqueueing it when idle
|
|
/// work cannot start.
|
|
pub async fn start_turn_if_idle(
|
|
&self,
|
|
request: TurnInputRequest,
|
|
) -> CodexResult<StartIfIdleSubmission> {
|
|
match self
|
|
.submit_turn_input_with_mode(request, TurnInputMode::StartIfIdle)
|
|
.await?
|
|
{
|
|
TurnInputSubmission::Started { turn_id } => {
|
|
Ok(StartIfIdleSubmission::Started { turn_id })
|
|
}
|
|
TurnInputSubmission::NotSubmitted { reason } => {
|
|
Ok(StartIfIdleSubmission::NotSubmitted { reason })
|
|
}
|
|
TurnInputSubmission::Steered { .. } => {
|
|
unreachable!("start-if-idle submission cannot steer")
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Starts a new internal continuation turn when idle, including in Plan mode.
|
|
/// Rejects if a newer task has started, even if it has already finished.
|
|
/// The input must be a response item; it is never treated as user authorization.
|
|
pub async fn continue_turn_if_idle(
|
|
&self,
|
|
request: TurnInputRequest,
|
|
expected_previous_turn_id: String,
|
|
) -> CodexResult<TurnInputSubmission> {
|
|
self.submit_turn_input_with_mode(
|
|
request,
|
|
TurnInputMode::ContinueIfIdle {
|
|
expected_previous_turn_id,
|
|
},
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Resumes an interrupted regular turn only when the thread is idle.
|
|
///
|
|
/// Recovery starts no new user input and preserves the turn ID that was
|
|
/// already recorded for the interrupted turn.
|
|
pub async fn recover_turn_if_idle(
|
|
&self,
|
|
request: RecoverTurnRequest,
|
|
) -> CodexResult<StartIfIdleSubmission> {
|
|
self.session
|
|
.services
|
|
.agent_control
|
|
.ensure_execution_capacity_for_turn_start(self)
|
|
.await?;
|
|
let RecoverTurnRequest {
|
|
turn_id,
|
|
thread_settings,
|
|
trace,
|
|
cyber_access_program,
|
|
} = request;
|
|
let root_turn_id = self
|
|
.session
|
|
.reference_context_item()
|
|
.await
|
|
.filter(|context| context.turn_id.as_deref() == Some(turn_id.as_str()))
|
|
.and_then(|context| context.root_turn_id);
|
|
let start_options = TurnStartOptions {
|
|
cyber_access_program,
|
|
root_turn_id,
|
|
..Default::default()
|
|
};
|
|
match self
|
|
.io
|
|
.submit_recover_turn(thread_settings, start_options, trace, turn_id)
|
|
.await?
|
|
{
|
|
TurnInputSubmission::Started { turn_id } => {
|
|
Ok(StartIfIdleSubmission::Started { turn_id })
|
|
}
|
|
TurnInputSubmission::NotSubmitted { reason } => {
|
|
Ok(StartIfIdleSubmission::NotSubmitted { reason })
|
|
}
|
|
TurnInputSubmission::Steered { .. } => {
|
|
unreachable!("recovered turn submission cannot steer")
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Stops the active unfinished root turn without recording TurnAborted or
|
|
/// TurnComplete, so another worker can recover its original turn ID.
|
|
///
|
|
/// Suspension is refused while a currently loaded descendant exists. Past
|
|
/// descendants do not prevent recovery, and concurrent descendant admission
|
|
/// is not sealed. Queued user input and outstanding approval, elicitation,
|
|
/// or server-request waiters remain best effort and may be discarded.
|
|
///
|
|
/// The session processes an accepted request even if its caller disconnects.
|
|
/// Callers must not transfer ownership until suspension succeeds, which
|
|
/// requires stopping execution, flushing history, and closing its writer.
|
|
pub async fn suspend_turn_and_shutdown(&self) -> CodexResult<SuspendTurnOutcome> {
|
|
if self.session_source.is_non_root_agent() {
|
|
return Err(CodexErr::UnsupportedOperation(
|
|
"turn suspension requires the owning root thread".to_string(),
|
|
));
|
|
}
|
|
|
|
// The session owns accepted suspension, so dropping this caller cannot interrupt
|
|
// cancellation, persistence, or writer shutdown halfway through a handoff.
|
|
let (reply, result) = oneshot::channel();
|
|
self.io
|
|
.tx_sub
|
|
.send(Submission {
|
|
id: new_submission_id(),
|
|
op: Op::SuspendTurnAndShutdown { reply },
|
|
trace: current_span_w3c_trace_context(),
|
|
parent_turn_id: None,
|
|
root_turn_id: None,
|
|
})
|
|
.await
|
|
.map_err(|_| CodexErr::Fatal("thread session has stopped".to_string()))?;
|
|
let outcome = result
|
|
.await
|
|
.map_err(|_| CodexErr::Fatal("thread suspension reply was lost".to_string()))??;
|
|
if matches!(&outcome, SuspendTurnOutcome::Suspended { .. }) {
|
|
self.io.session_loop_termination.clone().await;
|
|
}
|
|
Ok(outcome)
|
|
}
|
|
|
|
/// Steers only if `expected_turn_id` is still the active regular turn.
|
|
pub async fn steer_turn(
|
|
&self,
|
|
request: TurnInputRequest,
|
|
expected_turn_id: String,
|
|
) -> CodexResult<SteerSubmission> {
|
|
match self
|
|
.submit_turn_input_with_mode(request, TurnInputMode::Steer { expected_turn_id })
|
|
.await?
|
|
{
|
|
TurnInputSubmission::Steered { turn_id } => Ok(SteerSubmission::Steered { turn_id }),
|
|
TurnInputSubmission::NotSubmitted { reason } => {
|
|
Ok(SteerSubmission::NotSubmitted { reason })
|
|
}
|
|
TurnInputSubmission::Started { .. } => {
|
|
unreachable!("steer-only submission cannot start a turn")
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn submit_turn_input_with_mode(
|
|
&self,
|
|
request: TurnInputRequest,
|
|
mode: TurnInputMode,
|
|
) -> CodexResult<TurnInputSubmission> {
|
|
if !matches!(mode, TurnInputMode::Steer { .. }) {
|
|
self.session
|
|
.services
|
|
.agent_control
|
|
.ensure_execution_capacity_for_turn_start(self)
|
|
.await?;
|
|
}
|
|
self.io.submit_turn_input(request, mode).await
|
|
}
|
|
|
|
/// Persist whether this thread is eligible for future memory generation.
|
|
pub async fn set_thread_memory_mode(&self, mode: ThreadMemoryMode) -> anyhow::Result<()> {
|
|
self.session.set_thread_memory_mode(mode).await
|
|
}
|
|
|
|
/// Injects model-visible items into the currently active turn.
|
|
///
|
|
/// This is the thread-level bridge to `Session::inject_if_running` for
|
|
/// callers that only hold a `CodexThread`.
|
|
/// It returns the unchanged items when this thread has no active turn.
|
|
pub async fn inject_if_running(
|
|
&self,
|
|
items: Vec<ResponseItem>,
|
|
) -> Result<(), Vec<ResponseItem>> {
|
|
self.session.inject_if_running(items).await
|
|
}
|
|
|
|
/// Captures a regular turn only after its input is recorded. The caller must flush the rollout.
|
|
pub async fn interrupted_turn(
|
|
&self,
|
|
) -> Option<(String, TurnStartOptions, TurnEnvironmentSelection)> {
|
|
self.session.interrupted_turn().await
|
|
}
|
|
|
|
/// Returns the trusted root when the expected turn is currently active.
|
|
pub async fn active_turn_root(&self, expected_turn_id: &str) -> Option<String> {
|
|
let active = self.session.active_turn.lock().await;
|
|
let task = active.as_ref()?.task.as_ref()?;
|
|
if task.turn_context.sub_id != expected_turn_id {
|
|
return None;
|
|
}
|
|
task.turn_context.turn_metadata_state.root_turn_id()
|
|
}
|
|
|
|
pub async fn set_app_server_client_info(
|
|
&self,
|
|
app_server_client_name: Option<String>,
|
|
app_server_client_version: Option<String>,
|
|
mcp_elicitations_auto_deny: bool,
|
|
) -> ConstraintResult<()> {
|
|
self.session
|
|
.set_app_server_client_info(
|
|
app_server_client_name,
|
|
app_server_client_version,
|
|
mcp_elicitations_auto_deny,
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Preview persistent thread settings overrides without committing them.
|
|
pub async fn preview_thread_settings_overrides(
|
|
&self,
|
|
overrides: CodexThreadSettingsOverrides,
|
|
) -> ConstraintResult<ThreadConfigSnapshot> {
|
|
let updates = Self::thread_settings_update(overrides);
|
|
self.session.preview_settings(&updates).await
|
|
}
|
|
|
|
/// Restores thread-owned mutable settings captured from another loaded runtime.
|
|
///
|
|
/// Runtime replacement uses this after resume so clients keep their current thread settings
|
|
/// rather than reverting to the original layer-backed config.
|
|
pub async fn restore_thread_settings(
|
|
&self,
|
|
settings: CodexThreadSettingsOverrides,
|
|
) -> ConstraintResult<()> {
|
|
let updates = Self::thread_settings_update(settings);
|
|
self.session.update_settings(updates).await.map(|_| ())
|
|
}
|
|
|
|
/// Persists current settings without emitting a live settings event.
|
|
///
|
|
/// Serializes snapshot capture and persistence with accepted settings updates.
|
|
pub async fn checkpoint_thread_settings(&self) -> ThreadStoreResult<()> {
|
|
self.session.checkpoint_thread_settings().await
|
|
}
|
|
|
|
fn thread_settings_update(overrides: CodexThreadSettingsOverrides) -> SessionSettingsUpdate {
|
|
let CodexThreadSettingsOverrides {
|
|
environments,
|
|
runtime_workspace_roots,
|
|
profile_workspace_roots,
|
|
approval_policy,
|
|
approvals_reviewer,
|
|
sandbox_policy,
|
|
permission_profile,
|
|
active_permission_profile,
|
|
windows_sandbox_level,
|
|
model,
|
|
effort,
|
|
summary,
|
|
service_tier,
|
|
collaboration_mode,
|
|
personality,
|
|
disabled_plugin_ids,
|
|
} = overrides;
|
|
SessionSettingsUpdate {
|
|
step_settings: StepSettingsUpdate {
|
|
model,
|
|
effort,
|
|
collaboration_mode,
|
|
reasoning_summary: summary,
|
|
service_tier,
|
|
personality,
|
|
approval_policy,
|
|
approvals_reviewer,
|
|
},
|
|
environments,
|
|
runtime_workspace_roots,
|
|
profile_workspace_roots,
|
|
sandbox_policy,
|
|
permission_profile,
|
|
active_permission_profile,
|
|
windows_sandbox_level,
|
|
disabled_plugin_ids,
|
|
..Default::default()
|
|
}
|
|
}
|
|
|
|
pub async fn next_event(&self) -> CodexResult<Event> {
|
|
self.io.next_event().await
|
|
}
|
|
|
|
/// Returns the event count for a finite drain before transferring the receiver.
|
|
///
|
|
/// The caller must own the only event reader until it consumes this many events.
|
|
/// Events queued after this snapshot remain for the next reader.
|
|
pub fn queued_event_count(&self) -> usize {
|
|
self.io.rx_event.len()
|
|
}
|
|
|
|
pub async fn agent_status(&self) -> AgentStatus {
|
|
self.io.agent_status().await
|
|
}
|
|
|
|
pub async fn list_background_terminals(&self) -> Vec<BackgroundTerminalInfo> {
|
|
self.session.list_background_terminals().await
|
|
}
|
|
|
|
pub async fn terminate_background_terminal(&self, process_id: i32) -> bool {
|
|
self.session.terminate_background_terminal(process_id).await
|
|
}
|
|
|
|
pub(crate) fn subscribe_status(&self) -> watch::Receiver<AgentStatus> {
|
|
self.io.agent_status.clone()
|
|
}
|
|
|
|
/// Returns the complete token usage snapshot currently cached for this thread.
|
|
///
|
|
/// This accessor is intentionally narrower than direct session access: it lets
|
|
/// app-server lifecycle paths replay restored usage after resume or fork without
|
|
/// exposing broader session mutation authority. A caller that only reads
|
|
/// `total_token_usage` would drop last-turn usage and make the v2
|
|
/// `thread/tokenUsage/updated` payload incomplete.
|
|
pub async fn token_usage_info(&self) -> Option<TokenUsageInfo> {
|
|
self.session.token_usage_info().await
|
|
}
|
|
|
|
/// Records a context fragment without creating a new user turn boundary.
|
|
pub(crate) async fn inject_fragment_without_turn(&self, fragment: impl ContextualUserFragment) {
|
|
let item = ContextualUserFragment::into(fragment);
|
|
self.session
|
|
.inject_no_new_turn(vec![item], /*current_turn_context*/ None)
|
|
.await;
|
|
}
|
|
|
|
/// Record raw Responses API items without starting a new turn.
|
|
pub async fn inject_response_items(&self, items: Vec<ResponseItem>) -> CodexResult<()> {
|
|
self.inject_response_items_for_turn(items).await?;
|
|
self.session.flush_rollout().await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Record raw Responses API items immediately before admitting a user turn.
|
|
///
|
|
/// The caller must submit the associated user input while retaining its
|
|
/// thread-operation lock. The subsequent turn persistence includes both
|
|
/// these items and the user input, without an independent rollout flush.
|
|
pub async fn inject_response_items_for_turn(
|
|
&self,
|
|
items: Vec<ResponseItem>,
|
|
) -> CodexResult<()> {
|
|
if items.is_empty() {
|
|
return Err(CodexErr::InvalidRequest(
|
|
"items must not be empty".to_string(),
|
|
));
|
|
}
|
|
|
|
let turn_context = self.session.new_default_turn().await;
|
|
if self.session.reference_context_item().await.is_none() {
|
|
// This history-only API runs without run_turn, so it owns its initial step.
|
|
let step_context = self
|
|
.session
|
|
.capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
|
|
.await?;
|
|
self.session
|
|
.record_context_updates_and_set_reference_context_item(step_context.as_ref())
|
|
.await?;
|
|
}
|
|
self.session
|
|
.inject_client_response_items(items, turn_context.as_ref())
|
|
.await;
|
|
Ok(())
|
|
}
|
|
|
|
pub fn rollout_path(&self) -> Option<PathBuf> {
|
|
self.rollout_path.clone()
|
|
}
|
|
|
|
/// Returns startup metadata without the one-time initial message replay.
|
|
pub fn startup_metadata(&self) -> &ThreadStartupMetadata {
|
|
&self.startup_metadata
|
|
}
|
|
|
|
pub(crate) fn is_running(&self) -> bool {
|
|
!self.io.tx_sub.is_closed()
|
|
}
|
|
|
|
pub async fn guardian_trunk_rollout_path(&self) -> Option<PathBuf> {
|
|
self.session
|
|
.guardian_review_session()?
|
|
.trunk()
|
|
.await?
|
|
.rollout_path()
|
|
.await
|
|
}
|
|
|
|
pub async fn load_history(
|
|
&self,
|
|
include_archived: bool,
|
|
) -> ThreadStoreResult<StoredThreadHistory> {
|
|
let live_thread = self
|
|
.session
|
|
.live_thread_for_persistence("load history")
|
|
.map_err(|err| ThreadStoreError::Internal {
|
|
message: err.to_string(),
|
|
})?;
|
|
live_thread.load_history(include_archived).await
|
|
}
|
|
|
|
pub async fn read_thread(
|
|
&self,
|
|
include_archived: bool,
|
|
include_history: bool,
|
|
) -> ThreadStoreResult<StoredThread> {
|
|
let live_thread = self
|
|
.session
|
|
.live_thread_for_persistence("read thread")
|
|
.map_err(|err| ThreadStoreError::Internal {
|
|
message: err.to_string(),
|
|
})?;
|
|
live_thread
|
|
.read_thread(include_archived, include_history)
|
|
.await
|
|
}
|
|
|
|
pub async fn update_thread_metadata(
|
|
&self,
|
|
patch: ThreadMetadataPatch,
|
|
include_archived: bool,
|
|
) -> ThreadStoreResult<StoredThread> {
|
|
let live_thread = self
|
|
.session
|
|
.live_thread_for_persistence("update thread metadata")
|
|
.map_err(|err| ThreadStoreError::Internal {
|
|
message: err.to_string(),
|
|
})?;
|
|
live_thread.update_metadata(patch, include_archived).await
|
|
}
|
|
|
|
/// Appends rollout items through the live thread so derived metadata stays in sync.
|
|
pub async fn append_rollout_items(&self, items: &[RolloutItem]) -> ThreadStoreResult<()> {
|
|
let live_thread = self
|
|
.session
|
|
.live_thread_for_persistence("append rollout items")
|
|
.map_err(|err| ThreadStoreError::Internal {
|
|
message: err.to_string(),
|
|
})?;
|
|
live_thread.append_items(items).await
|
|
}
|
|
|
|
pub fn state_db(&self) -> Option<StateDbHandle> {
|
|
self.session.state_db()
|
|
}
|
|
|
|
pub async fn config_snapshot(&self) -> ThreadConfigSnapshot {
|
|
self.session.thread_config_snapshot().await
|
|
}
|
|
|
|
/// Returns the active turn's reviewer, including live updates, or the thread default.
|
|
pub async fn approvals_reviewer_for_turn(&self, turn_id: &str) -> ApprovalsReviewer {
|
|
if let Some((turn, settings, _)) = self
|
|
.session
|
|
.active_turn_context_and_strict_auto_review()
|
|
.await
|
|
&& turn.sub_id == turn_id
|
|
{
|
|
settings.approvals_reviewer()
|
|
} else {
|
|
self.config_snapshot().await.approvals_reviewer
|
|
}
|
|
}
|
|
|
|
/// Returns thread-owned settings suitable for rollout persistence and resume.
|
|
pub async fn thread_settings_snapshot(&self) -> ThreadSettingsSnapshot {
|
|
self.session.thread_settings_snapshot().await
|
|
}
|
|
|
|
/// Captures thread-owned settings and environment selections for runtime restoration.
|
|
pub async fn restorable_thread_settings(&self) -> CodexThreadSettingsOverrides {
|
|
self.session.restorable_thread_settings().await
|
|
}
|
|
|
|
/// Returns the MCP extensions declared by the client that created this runtime.
|
|
pub fn client_mcp_extensions(&self) -> ClientMcpExtensions {
|
|
self.session.services.client_mcp_extensions.clone()
|
|
}
|
|
|
|
/// Returns the files that supplied the thread's loaded model instructions.
|
|
pub async fn instruction_sources(&self) -> Vec<PathUri> {
|
|
self.session.instruction_sources().await
|
|
}
|
|
|
|
/// Returns loaded instruction sources rendered as legacy app-server path strings.
|
|
pub async fn legacy_instruction_sources(&self) -> Vec<LegacyAppPathString> {
|
|
self.instruction_sources()
|
|
.await
|
|
.into_iter()
|
|
.map(Into::into)
|
|
.collect()
|
|
}
|
|
|
|
pub async fn config(&self) -> Arc<crate::config::Config> {
|
|
self.session.get_config().await
|
|
}
|
|
|
|
/// Observes this thread's published MCP connections that match the requested config.
|
|
pub async fn mcp_connection_statuses(
|
|
&self,
|
|
config: &codex_mcp::McpConfig,
|
|
) -> std::collections::HashMap<String, codex_protocol::mcp::McpServerConnectionStatus> {
|
|
self.session
|
|
.services
|
|
.mcp_runtime
|
|
.connection_statuses(config)
|
|
.await
|
|
}
|
|
|
|
/// Resolves MCP configuration and environment bindings from the same config snapshot.
|
|
pub async fn runtime_mcp_config_and_context(
|
|
&self,
|
|
config: &crate::config::Config,
|
|
) -> (codex_mcp::McpConfig, codex_mcp::McpRuntimeContext) {
|
|
self.session.runtime_mcp_config_and_context(config).await
|
|
}
|
|
|
|
/// Captures the exact MCP config and environment bindings for the current thread state.
|
|
pub async fn current_mcp_config_and_runtime_context(
|
|
&self,
|
|
) -> (Arc<codex_mcp::McpConfig>, codex_mcp::McpRuntimeContext) {
|
|
let config = self.session.get_config().await;
|
|
let (mcp_config, runtime_context) = self.runtime_mcp_config_and_context(&config).await;
|
|
(Arc::new(mcp_config), runtime_context)
|
|
}
|
|
|
|
pub fn multi_agent_version(&self) -> Option<MultiAgentVersion> {
|
|
self.session.multi_agent_version()
|
|
}
|
|
|
|
/// Shares an immutable view of the live parent model context and retained host facts.
|
|
pub async fn conversation_history_snapshot(&self) -> Arc<dyn ConversationHistorySnapshot> {
|
|
self.session.conversation_history_snapshot().await
|
|
}
|
|
|
|
/// Returns the current user-authorization revision for Guardian.
|
|
pub async fn guardian_authorization_version(&self) -> GuardianAuthorizationVersion {
|
|
let history = self.conversation_history_snapshot().await;
|
|
self.thread_extension_data()
|
|
.get_or_init(GuardianReviewEvidence::default)
|
|
.authorization_version(history.as_ref())
|
|
}
|
|
|
|
/// Returns bounded root conversation evidence and its authorization version atomically.
|
|
pub async fn guardian_root_snapshot(&self) -> Option<GuardianRootSnapshot> {
|
|
self.session
|
|
.services
|
|
.agent_control
|
|
.root_user_authorization(self.session.thread_id)
|
|
.await
|
|
}
|
|
|
|
/// Refresh the thread's layer-backed user config state from a caller-supplied
|
|
/// config snapshot. Thread-scoped layers and session-static settings remain
|
|
/// unchanged.
|
|
pub async fn refresh_runtime_config(&self, next_config: crate::config::Config) {
|
|
self.session.refresh_runtime_config(next_config).await;
|
|
}
|
|
|
|
/// Refresh MCP configuration and managed requirements without reloading unrelated settings.
|
|
pub async fn refresh_mcp_config(&self, next_config: crate::config::Config) {
|
|
self.session.refresh_mcp_config(next_config).await;
|
|
}
|
|
|
|
/// Refreshes this thread's Apps tools before returning their runtime state.
|
|
pub async fn refresh_codex_apps_tools(
|
|
&self,
|
|
) -> anyhow::Result<codex_mcp::CodexAppsToolSnapshot> {
|
|
self.session.refresh_codex_apps_tools().await
|
|
}
|
|
|
|
pub async fn environment_selections(&self) -> Vec<TurnEnvironmentSelection> {
|
|
self.session.services.turn_environments.selections()
|
|
}
|
|
|
|
/// Installs resolved environment configuration and capability roots on this thread.
|
|
pub async fn environment_ready(
|
|
&self,
|
|
selection: &TurnEnvironmentSelection,
|
|
config: EnvironmentConfig,
|
|
) -> CodexResult<()> {
|
|
self.session.environment_ready(selection, config).await
|
|
}
|
|
|
|
/// Fails this thread's pending environment without affecting other attached threads.
|
|
pub async fn environment_failed(
|
|
&self,
|
|
selection: &TurnEnvironmentSelection,
|
|
error: String,
|
|
) -> CodexResult<()> {
|
|
self.session.environment_failed(selection, error).await
|
|
}
|
|
|
|
/// Passively inspects the selected capability roots whose environments are ready now.
|
|
pub fn inspect_selected_capability_roots(&self) -> SelectedCapabilityRootsStatus {
|
|
self.session.inspect_selected_capability_roots()
|
|
}
|
|
|
|
pub async fn read_mcp_resource(
|
|
&self,
|
|
server: &str,
|
|
params: ReadResourceRequestParams,
|
|
) -> anyhow::Result<serde_json::Value> {
|
|
self.session.refresh_mcp_if_dirty().await;
|
|
let result = self
|
|
.session
|
|
.services
|
|
.mcp_runtime
|
|
.latest_read_resource(server, params)
|
|
.await?;
|
|
|
|
Ok(serde_json::to_value(result)?)
|
|
}
|
|
|
|
/// Reads an app resource using the current authority of its originating tool call.
|
|
pub async fn read_mcp_resource_for_call(
|
|
&self,
|
|
call_id: &str,
|
|
uri: &str,
|
|
) -> anyhow::Result<serde_json::Value> {
|
|
self.session.refresh_mcp_if_dirty().await;
|
|
let result = self
|
|
.session
|
|
.services
|
|
.mcp_runtime
|
|
.read_resource_for_call(self.session.thread_id, call_id, uri)
|
|
.await?;
|
|
|
|
Ok(serde_json::to_value(result)?)
|
|
}
|
|
|
|
pub async fn start_mcp_event_stream(
|
|
&self,
|
|
name: &str,
|
|
arguments: serde_json::Value,
|
|
meta: Option<serde_json::Value>,
|
|
) -> anyhow::Result<codex_mcp::McpEventStream> {
|
|
let meta = match meta.as_ref() {
|
|
Some(serde_json::Value::Object(meta)) => Some(meta),
|
|
Some(other) => {
|
|
anyhow::bail!("MCP event request _meta must be a JSON object, got {other}")
|
|
}
|
|
None => None,
|
|
};
|
|
let _ = self.session.services.auth_manager.auth().await;
|
|
self.session.refresh_mcp_if_dirty().await;
|
|
codex_mcp::McpResourceClient::new(Arc::clone(&self.session.services.mcp_runtime))
|
|
.open_event_stream(name, &arguments, meta)
|
|
.await
|
|
}
|
|
|
|
pub async fn call_mcp_tool(
|
|
&self,
|
|
server: &str,
|
|
tool: &str,
|
|
arguments: Option<serde_json::Value>,
|
|
meta: Option<serde_json::Value>,
|
|
) -> anyhow::Result<CallToolResult> {
|
|
self.session.refresh_mcp_if_dirty().await;
|
|
self.session
|
|
.services
|
|
.mcp_runtime
|
|
.latest_call_tool(
|
|
server, tool, /*environment_id*/ None, arguments, meta,
|
|
/*requested_timeout*/ None, /*wait_for_server*/ true,
|
|
)
|
|
.await
|
|
}
|
|
|
|
pub fn enabled(&self, feature: Feature) -> bool {
|
|
self.session.enabled(feature)
|
|
}
|
|
|
|
pub async fn increment_out_of_band_elicitation_count(&self) -> CodexResult<i64> {
|
|
let mut elicitations = self.out_of_band_elicitations.lock().await;
|
|
let incremented = elicitations.count.checked_add(1).ok_or_else(|| {
|
|
CodexErr::Fatal("out-of-band elicitation count overflowed".to_string())
|
|
})?;
|
|
if elicitations.count == 0 {
|
|
elicitations.registration = Some(self.session.services.elicitations.register());
|
|
}
|
|
elicitations.count = incremented;
|
|
Ok(incremented)
|
|
}
|
|
|
|
pub async fn decrement_out_of_band_elicitation_count(&self) -> CodexResult<i64> {
|
|
let mut elicitations = self.out_of_band_elicitations.lock().await;
|
|
if elicitations.count == 0 {
|
|
return Err(CodexErr::InvalidRequest(
|
|
"out-of-band elicitation count is already zero".to_string(),
|
|
));
|
|
}
|
|
|
|
elicitations.count -= 1;
|
|
if elicitations.count == 0 {
|
|
elicitations.registration = None;
|
|
}
|
|
Ok(elicitations.count)
|
|
}
|
|
}
|