| #[path = "daemon_continuation.rs"] |
| mod daemon_continuation; |
|
|
| #[path = "daemon_snapshot.rs"] |
| mod daemon_snapshot; |
|
|
| use super::persisted_resume_settings::PersistedResumeSettings; |
| use super::persisted_resume_settings::latest_persisted_resume_settings; |
| use super::thread_enrichment::enrich_loaded_threads; |
| use super::thread_fork_goal::inherit_thread_goal_snapshot; |
| use super::thread_input::can_accept_direct_input; |
| use super::thread_input::ensure_direct_input_allowed; |
| use super::*; |
| use crate::error_code::method_not_found; |
| use codex_app_server_protocol::SelectedCapabilityRoot; |
| use codex_app_server_protocol::ThreadHistoryMode as ApiThreadHistoryMode; |
| use codex_app_server_protocol::ThreadRevertParams; |
| use codex_app_server_protocol::ThreadRevertResponse; |
| use codex_app_server_protocol::ThreadRevertedNotification; |
| use codex_app_server_protocol::ThreadSection; |
| use codex_app_server_protocol::ThreadSectionAppearance; |
| use codex_app_server_protocol::ThreadSectionMoveParams; |
| use codex_app_server_protocol::ThreadSectionMoveResponse; |
| use codex_extension_api::ExtensionDataInit; |
| use codex_extension_api::ThreadIdleCause; |
| use codex_protocol::SanitizedGitUrl; |
| use codex_protocol::config_types::MultiAgentMode; |
| 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; |
| const PAGINATED_FULL_HISTORY_DEPRECATION_SUMMARY: &str = "Full-history hydration is deprecated for paginated threads; use `excludeTurns: true`, then page with `thread/turns/list` and `thread/items/list`."; |
| const PAGINATED_THREAD_READ_DEPRECATION_SUMMARY: &str = "Full-history hydration is deprecated for paginated threads; omit `includeTurns` or set it to `false`, then page with `thread/turns/list` and `thread/items/list`."; |
|
|
| async fn stage_pending_thread_metadata( |
| thread_manager: &ThreadManager, |
| thread_store: &dyn ThreadStore, |
| patch: StoreThreadMetadataPatch, |
| operation: &'static str, |
| ) -> Result<Option<ThreadId>, JSONRPCErrorError> { |
| if patch.is_empty() { |
| return Ok(None); |
| } |
| let thread_id = thread_manager.reserve_thread_id(); |
| thread_store |
| .stage_pending_thread_metadata(thread_id, patch) |
| .await |
| .map_err(|error| match error { |
| ThreadStoreError::Unsupported { .. } => { |
| method_not_found(format!("{operation} is unavailable without sqlite state")) |
| } |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| error => internal_error(format!("failed to stage {operation} metadata: {error}")), |
| })?; |
| Ok(Some(thread_id)) |
| } |
|
|
| async fn remove_pending_thread_metadata( |
| thread_store: &dyn ThreadStore, |
| thread_id: Option<ThreadId>, |
| ) { |
| let Some(thread_id) = thread_id else { |
| return; |
| }; |
| if let Err(error) = thread_store.remove_pending_thread_metadata(thread_id).await { |
| warn!("failed to remove staged thread metadata for {thread_id}: {error}"); |
| } |
| } |
|
|
| struct ThreadListFilters { |
| model_providers: Option<Vec<String>>, |
| source_kinds: Option<Vec<ThreadSourceKind>>, |
| archived: bool, |
| section_id: Option<Option<String>>, |
| project_id: StoreClearableField<String>, |
| cwd_filters: Option<Vec<PathBuf>>, |
| search_term: Option<String>, |
| use_state_db_only: bool, |
| relation_filter: Option<StoreThreadRelationFilter>, |
| } |
|
|
| |
| #[derive(PartialEq)] |
| struct ResumeConfigState { |
| history_cwd: Option<PathBuf>, |
| workspace_roots: Option<Vec<AbsolutePathBuf>>, |
| persisted_metadata: Option<ThreadMetadata>, |
| persisted_settings: Option<PersistedResumeSettings>, |
| } |
|
|
| struct PreparedResumeConfig { |
| state: ResumeConfigState, |
| config: Config, |
| } |
|
|
| struct ThreadRevertRuntimeSnapshot { |
| config: Config, |
| settings: CodexThreadSettingsOverrides, |
| client_mcp_extensions: ClientMcpExtensions, |
| } |
|
|
| 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_runtime_workspace_roots) = request.runtime_workspace_roots.as_ref() { |
| let requested_runtime_workspace_roots = requested_runtime_workspace_roots.to_vec(); |
| if requested_runtime_workspace_roots != config_snapshot.workspace_roots { |
| mismatch_details.push(format!( |
| "runtime_workspace_roots requested={requested_runtime_workspace_roots:?} active={:?}", |
| config_snapshot.workspace_roots |
| )); |
| } |
| } |
| 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 active_sandbox = config_snapshot.sandbox_policy(); |
| let sandbox_matches = matches!( |
| (requested_sandbox, &active_sandbox), |
| ( |
| 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={active_sandbox:?}" |
| )); |
| } |
| } |
| if request.permissions.is_some() { |
| mismatch_details.push(format!( |
| "permissions override was provided and ignored while running; active={:?}", |
| config_snapshot.active_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(), |
| ); |
| } |
| 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(); |
| typesafe_overrides.model_provider = Some(persisted_metadata.model_provider.clone()); |
|
|
| if let Some(reasoning_effort) = persisted_metadata.reasoning_effort.as_ref() { |
| request_overrides.get_or_insert_with(HashMap::new).insert( |
| "model_reasoning_effort".to_string(), |
| serde_json::Value::String(reasoning_effort.to_string()), |
| ); |
| } |
| } |
|
|
| fn normalize_thread_list_cwd_filters( |
| cwd: Option<ThreadListCwdFilter>, |
| ) -> Result<Option<Vec<PathBuf>>, JSONRPCErrorError> { |
| let Some(cwd) = cwd else { |
| return Ok(None); |
| }; |
|
|
| let cwds = match cwd { |
| ThreadListCwdFilter::One(cwd) => vec![cwd], |
| ThreadListCwdFilter::Many(cwds) => cwds, |
| }; |
| let mut normalized_cwds = Vec::with_capacity(cwds.len()); |
| for cwd in cwds { |
| let cwd = AbsolutePathBuf::relative_to_current_dir(cwd.as_str()) |
| .map(AbsolutePathBuf::into_path_buf) |
| .map_err(|err| { |
| invalid_params(format!("invalid thread/list cwd filter `{cwd}`: {err}")) |
| })?; |
| normalized_cwds.push(cwd); |
| } |
|
|
| Ok(Some(normalized_cwds)) |
| } |
|
|
| 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 has_permission_override( |
| request_overrides: Option<&HashMap<String, serde_json::Value>>, |
| typesafe_overrides: &ConfigOverrides, |
| ) -> bool { |
| typesafe_overrides.sandbox_mode.is_some() |
| || typesafe_overrides.permission_profile.is_some() |
| || typesafe_overrides.default_permissions.is_some() |
| || request_overrides.is_some_and(|overrides| { |
| overrides.contains_key("sandbox_mode") || overrides.contains_key("default_permissions") |
| }) |
| } |
|
|
| fn validate_dynamic_tools(tools: &[DynamicToolSpec]) -> Result<(), String> { |
| const DYNAMIC_TOOL_NAME_MAX_LEN: usize = 128; |
| const DYNAMIC_TOOL_NAMESPACE_MAX_LEN: usize = 64; |
| const DYNAMIC_TOOL_NAMESPACE_DESCRIPTION_MAX_LEN: usize = 1024; |
| const DYNAMIC_TOOL_IDENTIFIER_PATTERN: &str = "^[a-zA-Z0-9_-]+$"; |
| const RESERVED_RESPONSES_NAMESPACES: &[&str] = &[ |
| "api_tool", |
| "browser", |
| "computer", |
| "container", |
| "file_search", |
| "functions", |
| "image_gen", |
| "multi_tool_use", |
| "python", |
| "python_user_visible", |
| "submodel_delegator", |
| "terminal", |
| "tool_search", |
| "web", |
| ]; |
|
|
| fn escape_identifier_for_error(value: &str) -> String { |
| value.escape_default().to_string() |
| } |
|
|
| fn validate_dynamic_tool_identifier( |
| value: &str, |
| label: &str, |
| max_len: usize, |
| ) -> Result<(), String> { |
| if !value |
| .bytes() |
| .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-')) |
| { |
| return Err(format!( |
| "{label} must match {DYNAMIC_TOOL_IDENTIFIER_PATTERN} to match Responses API: {}", |
| escape_identifier_for_error(value), |
| )); |
| } |
| if value.chars().count() > max_len { |
| return Err(format!( |
| "{label} must be at most {max_len} characters to match Responses API: {}", |
| escape_identifier_for_error(value), |
| )); |
| } |
| Ok(()) |
| } |
|
|
| fn validate_dynamic_tool<'a>( |
| tool: &'a DynamicToolFunctionSpec, |
| namespace: Option<&str>, |
| seen: &mut HashSet<&'a str>, |
| ) -> Result<(), String> { |
| let name = tool.name.trim(); |
| if name.is_empty() { |
| return Err("dynamic tool name must not be empty".to_string()); |
| } |
| if name != tool.name { |
| return Err(format!( |
| "dynamic tool name has leading/trailing whitespace: {}", |
| escape_identifier_for_error(&tool.name), |
| )); |
| } |
| validate_dynamic_tool_identifier(name, "dynamic tool name", DYNAMIC_TOOL_NAME_MAX_LEN)?; |
| if name == "mcp" || name.starts_with("mcp__") { |
| return Err(format!("dynamic tool name is reserved: {name}")); |
| } |
| if !seen.insert(name) { |
| if let Some(namespace) = namespace { |
| return Err(format!( |
| "duplicate dynamic tool name in namespace {namespace}: {name}" |
| )); |
| } |
| return Err(format!("duplicate dynamic tool name: {name}")); |
| } |
| if tool.defer_loading && namespace.is_none() { |
| return Err(format!( |
| "deferred dynamic tool must include a namespace: {name}" |
| )); |
| } |
|
|
| if let Err(err) = codex_tools::parse_tool_input_schema(&tool.input_schema) { |
| return Err(format!( |
| "dynamic tool input schema is not supported for {name}: {err}" |
| )); |
| } |
| Ok(()) |
| } |
|
|
| let mut seen_tools = HashSet::new(); |
| let mut seen_namespaces = HashSet::new(); |
| for spec in tools { |
| match spec { |
| DynamicToolSpec::Function(tool) => { |
| validate_dynamic_tool(tool, None, &mut seen_tools)?; |
| } |
| DynamicToolSpec::Namespace(namespace) => { |
| let name = namespace.name.trim(); |
| if name.is_empty() { |
| return Err("dynamic tool namespace must not be empty".to_string()); |
| } |
| if name != namespace.name { |
| return Err(format!( |
| "dynamic tool namespace has leading/trailing whitespace: {}", |
| escape_identifier_for_error(&namespace.name), |
| )); |
| } |
| validate_dynamic_tool_identifier( |
| name, |
| "dynamic tool namespace", |
| DYNAMIC_TOOL_NAMESPACE_MAX_LEN, |
| )?; |
| if namespace.description.chars().count() |
| > DYNAMIC_TOOL_NAMESPACE_DESCRIPTION_MAX_LEN |
| { |
| return Err(format!( |
| "dynamic tool namespace description must be at most {DYNAMIC_TOOL_NAMESPACE_DESCRIPTION_MAX_LEN} characters" |
| )); |
| } |
| if name == "mcp" || name.starts_with("mcp__") { |
| return Err(format!("dynamic tool namespace is reserved: {name}")); |
| } |
| if RESERVED_RESPONSES_NAMESPACES.contains(&name) { |
| return Err(format!( |
| "dynamic tool namespace collides with a reserved Responses API namespace: {name}", |
| )); |
| } |
| if !seen_namespaces.insert(name) { |
| return Err(format!("duplicate dynamic tool namespace: {name}")); |
| } |
| if namespace.tools.is_empty() { |
| return Err(format!( |
| "dynamic tool namespace must contain at least one tool: {name}" |
| )); |
| } |
| let mut seen_namespace_tools = HashSet::new(); |
| for tool in &namespace.tools { |
| let DynamicToolNamespaceTool::Function(tool) = tool; |
| validate_dynamic_tool(tool, Some(name), &mut seen_namespace_tools)?; |
| } |
| } |
| } |
| } |
| Ok(()) |
| } |
|
|
| #[derive(Clone)] |
| pub(crate) struct ThreadRequestProcessor { |
| pub(super) auth_manager: Arc<AuthManager>, |
| pub(super) thread_manager: Arc<ThreadManager>, |
| pub(super) outgoing: Arc<OutgoingMessageSender>, |
| pub(super) arg0_paths: Arg0DispatchPaths, |
| pub(super) config: Arc<Config>, |
| pub(super) config_manager: ConfigManager, |
| pub(super) thread_store: Arc<dyn ThreadStore>, |
| pub(super) pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>, |
| pub(super) thread_state_manager: ThreadStateManager, |
| pub(super) thread_watch_manager: ThreadWatchManager, |
| pub(super) thread_list_state_permit: Arc<Semaphore>, |
| pub(super) thread_goal_processor: ThreadGoalRequestProcessor, |
| pub(super) state_db: Option<StateDbHandle>, |
| pub(super) log_db: Option<LogDbLayer>, |
| pub(super) background_tasks: TaskTracker, |
| pub(super) skills_watcher: Arc<SkillsWatcher>, |
| pub(super) turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>, |
| pub(super) initial_config_warnings: Arc<Vec<ConfigWarningNotification>>, |
| } |
|
|
| |
| pub(crate) enum ThreadResumeTarget { |
| Client(ConnectionRequestId), |
| DaemonRecovery(Option<codex_app_server_transport::daemon_recovery::InterruptedTurn>), |
| } |
|
|
| |
| enum RunningThreadResumeResult { |
| |
| Handled(tokio::sync::oneshot::Receiver<()>), |
| |
| |
| |
| |
| NotRunning(Option<Box<StoredThread>>), |
| } |
|
|
| impl ThreadRequestProcessor { |
| #[allow(clippy::too_many_arguments)] |
| pub(crate) fn new( |
| auth_manager: Arc<AuthManager>, |
| thread_manager: Arc<ThreadManager>, |
| outgoing: Arc<OutgoingMessageSender>, |
| arg0_paths: Arg0DispatchPaths, |
| config: Arc<Config>, |
| config_manager: ConfigManager, |
| thread_store: Arc<dyn ThreadStore>, |
| pending_thread_unloads: Arc<Mutex<HashSet<ThreadId>>>, |
| thread_state_manager: ThreadStateManager, |
| thread_watch_manager: ThreadWatchManager, |
| thread_list_state_permit: Arc<Semaphore>, |
| thread_goal_processor: ThreadGoalRequestProcessor, |
| state_db: Option<StateDbHandle>, |
| log_db: Option<LogDbLayer>, |
| skills_watcher: Arc<SkillsWatcher>, |
| turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>, |
| initial_config_warnings: Vec<ConfigWarningNotification>, |
| ) -> Self { |
| Self { |
| auth_manager, |
| thread_manager, |
| outgoing, |
| arg0_paths, |
| config, |
| config_manager, |
| thread_store, |
| pending_thread_unloads, |
| thread_state_manager, |
| thread_watch_manager, |
| thread_list_state_permit, |
| thread_goal_processor, |
| state_db, |
| log_db, |
| background_tasks: TaskTracker::new(), |
| skills_watcher, |
| turn_cost_worker, |
| initial_config_warnings: Arc::new(initial_config_warnings), |
| } |
| } |
|
|
| pub(crate) async fn thread_start( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadStartParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| client_mcp_extensions: ClientMcpExtensions, |
| request_context: RequestContext, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_start_inner( |
| request_id, |
| params, |
| app_server_client_name, |
| app_server_client_version, |
| client_mcp_extensions, |
| request_context, |
| ) |
| .await |
| .map(|()| None) |
| } |
|
|
| pub(crate) async fn thread_unsubscribe( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadUnsubscribeParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_unsubscribe_response_inner(params, request_id.connection_id) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_resume( |
| &self, |
| target: ThreadResumeTarget, |
| params: ThreadResumeParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| client_mcp_extensions: ClientMcpExtensions, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| let mut prepared_config = None; |
| |
| while Box::pin(self.thread_resume_inner( |
| &target, |
| ¶ms, |
| 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( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadForkParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| client_mcp_extensions: ClientMcpExtensions, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| |
| Box::pin(self.thread_fork_inner( |
| request_id, |
| params, |
| app_server_client_name, |
| app_server_client_version, |
| client_mcp_extensions, |
| )) |
| .await |
| .map(|()| None) |
| } |
|
|
| pub(crate) async fn thread_revert( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadRevertParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| let (response, thread_id) = self |
| .thread_revert_response( |
| &request_id, |
| params, |
| app_server_client_name, |
| app_server_client_version, |
| ) |
| .await?; |
| self.outgoing.send_response(request_id, response).await; |
| self.outgoing |
| .send_server_notification(ServerNotification::ThreadReverted( |
| ThreadRevertedNotification { thread_id }, |
| )) |
| .await; |
| Ok(None) |
| } |
|
|
| pub(crate) async fn thread_archive( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadArchiveParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| match self.thread_archive_inner(params).await { |
| Ok((response, archived_thread_ids)) => { |
| self.outgoing |
| .send_response(request_id.clone(), response) |
| .await; |
| for thread_id in archived_thread_ids { |
| self.outgoing |
| .send_server_notification(ServerNotification::ThreadArchived( |
| ThreadArchivedNotification { thread_id }, |
| )) |
| .await; |
| } |
| Ok(None) |
| } |
| Err(error) => Err(error), |
| } |
| } |
|
|
| pub(crate) async fn thread_increment_elicitation( |
| &self, |
| params: ThreadIncrementElicitationParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_increment_elicitation_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_decrement_elicitation( |
| &self, |
| params: ThreadDecrementElicitationParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_decrement_elicitation_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_set_name( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadSetNameParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| match self.thread_set_name_response_inner(params).await { |
| Ok((response, notification)) => { |
| self.outgoing |
| .send_response(request_id.clone(), response) |
| .await; |
| if let Some(notification) = notification { |
| self.outgoing |
| .send_server_notification(ServerNotification::ThreadNameUpdated( |
| notification, |
| )) |
| .await; |
| } |
| Ok(None) |
| } |
| Err(error) => Err(error), |
| } |
| } |
|
|
| pub(crate) async fn thread_metadata_update( |
| &self, |
| params: ThreadMetadataUpdateParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_metadata_update_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_section_move( |
| &self, |
| params: ThreadSectionMoveParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| let ThreadSectionMoveParams { |
| thread_id, |
| section_id, |
| before_thread_id, |
| } = params; |
| let thread_uuid = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
| if section_id |
| .as_deref() |
| .is_some_and(|section| section.trim().is_empty()) |
| { |
| return Err(invalid_request("sectionId must not be empty")); |
| } |
| if section_id.is_none() && before_thread_id.is_some() { |
| return Err(invalid_request( |
| "beforeThreadId requires a non-null sectionId", |
| )); |
| } |
| let before_thread_uuid = before_thread_id |
| .map(|thread_id| { |
| ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid before thread id: {err}"))) |
| }) |
| .transpose()?; |
|
|
| { |
| let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; |
| self.thread_manager |
| .move_thread_to_section(thread_uuid, section_id.as_deref(), before_thread_uuid) |
| .await |
| .map_err(|err| core_thread_write_error("move thread in section", err))?; |
| } |
|
|
| Ok(Some(ClientResponsePayload::ThreadSectionMove( |
| ThreadSectionMoveResponse {}, |
| ))) |
| } |
|
|
| pub(crate) async fn thread_memory_mode_set( |
| &self, |
| params: ThreadMemoryModeSetParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_memory_mode_set_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn memory_reset( |
| &self, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.memory_reset_response_inner() |
| .await |
| .map(|response: MemoryResetResponse| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_unarchive( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadUnarchiveParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| match self.thread_unarchive_inner(params).await { |
| Ok((response, notification)) => { |
| self.outgoing |
| .send_response(request_id.clone(), response) |
| .await; |
| self.outgoing |
| .send_server_notification(ServerNotification::ThreadUnarchived(notification)) |
| .await; |
| Ok(None) |
| } |
| Err(error) => Err(error), |
| } |
| } |
|
|
| pub(crate) async fn thread_compact_start( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadCompactStartParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_compact_start_inner(request_id, params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_background_terminals_clean( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadBackgroundTerminalsCleanParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_background_terminals_clean_inner(request_id, params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_background_terminals_list( |
| &self, |
| params: ThreadBackgroundTerminalsListParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_background_terminals_list_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_background_terminals_terminate( |
| &self, |
| params: ThreadBackgroundTerminalsTerminateParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_background_terminals_terminate_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| async fn send_deprecation_notice(&self, connection_id: ConnectionId, summary: &str) { |
| self.outgoing |
| .send_server_notification_to_connections( |
| &[connection_id], |
| ServerNotification::DeprecationNotice(DeprecationNoticeNotification { |
| summary: summary.to_string(), |
| details: None, |
| }), |
| ) |
| .await; |
| } |
|
|
| pub(crate) async fn thread_list( |
| &self, |
| params: ThreadListParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_list_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_search( |
| &self, |
| params: ThreadSearchParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_search_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_search_occurrences( |
| &self, |
| params: ThreadSearchOccurrencesParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_search_occurrences_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_loaded_list( |
| &self, |
| params: ThreadLoadedListParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_loaded_list_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_read( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadReadParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| let include_turns = params.include_turns; |
| let response = self.thread_read_response_inner(params).await?; |
| if include_turns |
| && matches!( |
| response.thread.history_mode, |
| ApiThreadHistoryMode::Paginated |
| ) |
| { |
| self.send_deprecation_notice( |
| request_id.connection_id, |
| PAGINATED_THREAD_READ_DEPRECATION_SUMMARY, |
| ) |
| .await; |
| } |
| Ok(Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_turns_list( |
| &self, |
| params: ThreadTurnsListParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_turns_list_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_items_list( |
| &self, |
| params: ThreadItemsListParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_items_list_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_timeline_list( |
| &self, |
| params: ThreadTimelineListParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| let thread_id = ThreadId::from_string(¶ms.thread_id) |
| .map_err(|error| invalid_request(format!("invalid thread id: {error}")))?; |
| let page = self |
| .thread_store |
| .list_timeline(StoreListTimelineParams { |
| thread_id, |
| cursor: params.cursor, |
| page_size: params |
| .limit |
| .map(|limit| limit as usize) |
| .unwrap_or(THREAD_ITEMS_DEFAULT_LIMIT) |
| .clamp(1, THREAD_ITEMS_MAX_LIMIT), |
| }) |
| .await |
| .map_err(paginated_history_list_error)?; |
| Ok(Some( |
| ThreadTimelineListResponse { |
| data: page.items, |
| next_cursor: page.next_cursor, |
| active_realtime_session_at_page_start: page.active_realtime_session_at_page_start, |
| } |
| .into(), |
| )) |
| } |
|
|
| pub(crate) async fn thread_shell_command( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadShellCommandParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_shell_command_inner(request_id, params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn thread_approve_guardian_denied_action( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadApproveGuardianDeniedActionParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.thread_approve_guardian_denied_action_inner(request_id, params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| pub(crate) async fn conversation_summary( |
| &self, |
| params: GetConversationSummaryParams, |
| ) -> Result<Option<ClientResponsePayload>, JSONRPCErrorError> { |
| self.get_thread_summary_response_inner(params) |
| .await |
| .map(|response| Some(response.into())) |
| } |
|
|
| async fn load_thread( |
| &self, |
| thread_id: &str, |
| ) -> Result<(ThreadId, Arc<CodexThread>), JSONRPCErrorError> { |
| |
| let thread_id = ThreadId::from_string(thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
|
|
| let thread = self |
| .thread_manager |
| .get_thread(thread_id) |
| .await |
| .map_err(|_| invalid_request(format!("thread not found: {thread_id}")))?; |
|
|
| Ok((thread_id, thread)) |
| } |
|
|
| pub(super) async fn acquire_thread_list_state_permit( |
| &self, |
| ) -> Result<SemaphorePermit<'_>, JSONRPCErrorError> { |
| self.thread_list_state_permit |
| .acquire() |
| .await |
| .map_err(|err| { |
| internal_error(format!("failed to acquire thread list state permit: {err}")) |
| }) |
| } |
|
|
| async fn set_app_server_client_info( |
| thread: &CodexThread, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| ) -> Result<(), JSONRPCErrorError> { |
| let mcp_elicitations_auto_deny = xcode_26_4_mcp_elicitations_auto_deny( |
| app_server_client_name.as_deref(), |
| app_server_client_version.as_deref(), |
| ); |
| thread |
| .set_app_server_client_info( |
| app_server_client_name, |
| app_server_client_version, |
| mcp_elicitations_auto_deny, |
| ) |
| .await |
| .map_err(|err| internal_error(format!("failed to set app server client info: {err}"))) |
| } |
|
|
| async fn finalize_thread_teardown(&self, thread_id: ThreadId) { |
| self.pending_thread_unloads.lock().await.remove(&thread_id); |
| self.outgoing |
| .cancel_requests_for_thread(thread_id, None) |
| .await; |
| self.thread_state_manager |
| .remove_thread_state(thread_id) |
| .await; |
| self.thread_watch_manager |
| .remove_thread(&thread_id.to_string()) |
| .await; |
| } |
|
|
| async fn thread_unsubscribe_response_inner( |
| &self, |
| params: ThreadUnsubscribeParams, |
| connection_id: ConnectionId, |
| ) -> Result<ThreadUnsubscribeResponse, JSONRPCErrorError> { |
| let thread_id = ThreadId::from_string(¶ms.thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
|
|
| if self.thread_manager.get_thread(thread_id).await.is_err() { |
| self.finalize_thread_teardown(thread_id).await; |
| return Ok(ThreadUnsubscribeResponse { |
| status: ThreadUnsubscribeStatus::NotLoaded, |
| }); |
| }; |
|
|
| let was_subscribed = self |
| .thread_state_manager |
| .unsubscribe_connection_from_thread(thread_id, connection_id) |
| .await; |
|
|
| let status = if was_subscribed { |
| ThreadUnsubscribeStatus::Unsubscribed |
| } else { |
| ThreadUnsubscribeStatus::NotSubscribed |
| }; |
| Ok(ThreadUnsubscribeResponse { status }) |
| } |
|
|
| async fn prepare_thread_for_archive( |
| &self, |
| thread_id: ThreadId, |
| ) -> Result<(), JSONRPCErrorError> { |
| self.prepare_thread_for_removal(thread_id, "archive").await |
| } |
|
|
| pub(super) async fn prepare_thread_for_removal( |
| &self, |
| thread_id: ThreadId, |
| operation: &str, |
| ) -> Result<(), JSONRPCErrorError> { |
| let removed_conversation = self |
| .thread_manager |
| .remove_thread_for_client(&thread_id) |
| .await |
| .map_err(|err| core_thread_write_error(operation, err))?; |
| if let Some(conversation) = removed_conversation { |
| info!("thread {thread_id} was active; shutting down"); |
| match wait_for_thread_shutdown(&conversation).await { |
| ThreadShutdownResult::Complete => {} |
| ThreadShutdownResult::SubmitFailed => { |
| error!( |
| "failed to submit Shutdown to thread {thread_id}; proceeding with {operation}" |
| ); |
| } |
| ThreadShutdownResult::TimedOut => { |
| warn!("thread {thread_id} shutdown timed out; proceeding with {operation}"); |
| } |
| } |
| } |
| self.finalize_thread_teardown(thread_id).await; |
| Ok(()) |
| } |
|
|
| fn listener_task_context(&self) -> ListenerTaskContext { |
| ListenerTaskContext { |
| thread_manager: Arc::clone(&self.thread_manager), |
| thread_state_manager: self.thread_state_manager.clone(), |
| outgoing: Arc::clone(&self.outgoing), |
| pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), |
| thread_watch_manager: self.thread_watch_manager.clone(), |
| codex_home: self.config.codex_home.to_path_buf(), |
| thread_unload_delay: self.config.thread_unload_delay, |
| skills_watcher: Arc::clone(&self.skills_watcher), |
| turn_cost_worker: self.turn_cost_worker.clone(), |
| } |
| } |
|
|
| async fn ensure_conversation_listener( |
| &self, |
| conversation_id: ThreadId, |
| connection_id: ConnectionId, |
| raw_events_enabled: bool, |
| ) -> Result<EnsureConversationListenerResult, JSONRPCErrorError> { |
| super::thread_lifecycle::ensure_conversation_listener( |
| self.listener_task_context(), |
| conversation_id, |
| connection_id, |
| raw_events_enabled, |
| ) |
| .await |
| } |
|
|
| async fn ensure_listener_task_running( |
| &self, |
| conversation_id: ThreadId, |
| conversation: Arc<CodexThread>, |
| thread_state: Arc<Mutex<ThreadState>>, |
| ) -> Result<(), JSONRPCErrorError> { |
| super::thread_lifecycle::ensure_listener_task_running( |
| self.listener_task_context(), |
| conversation_id, |
| conversation, |
| thread_state, |
| ) |
| .await |
| } |
|
|
| async fn thread_start_inner( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadStartParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| client_mcp_extensions: ClientMcpExtensions, |
| request_context: RequestContext, |
| ) -> Result<(), JSONRPCErrorError> { |
| let ThreadStartParams { |
| model, |
| model_provider, |
| allow_provider_model_fallback, |
| service_tier, |
| cwd, |
| runtime_workspace_roots, |
| approval_policy, |
| approvals_reviewer, |
| sandbox, |
| permissions, |
| config, |
| service_name, |
| base_instructions, |
| developer_instructions, |
| dynamic_tools, |
| selected_capability_roots, |
| mock_experimental_field: _mock_experimental_field, |
| experimental_raw_events, |
| personality, |
| multi_agent_mode: _multi_agent_mode, |
| ephemeral, |
| history_mode, |
| session_start_source, |
| thread_source, |
| project_id, |
| daybreak_enabled, |
| environments, |
| } = params; |
| if matches!( |
| history_mode, |
| Some(codex_app_server_protocol::ThreadHistoryMode::Paginated) |
| ) && !self.thread_store.supports_paginated_history_lists() |
| { |
| return Err(invalid_request( |
| "paginated threads require thread/turns/list and thread/items/list support", |
| )); |
| } |
| if sandbox.is_some() && permissions.is_some() { |
| return Err(invalid_request( |
| "`permissions` cannot be combined with `sandbox`", |
| )); |
| } |
| if let Some(project_id) = project_id.as_ref() { |
| if project_id.is_empty() { |
| return Err(invalid_request("projectId must not be empty")); |
| } |
| let project = self |
| .thread_store |
| .read_project(project_id.clone()) |
| .await |
| .map_err(|err| match err { |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| err => internal_error(format!("failed to read project: {err}")), |
| })?; |
| if project.is_none() { |
| return Err(invalid_request(format!("project not found: {project_id}"))); |
| } |
| } |
| let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots); |
| let environments = |
| resolve_turn_environment_selections(self.thread_manager.as_ref(), environments)?; |
| let mut typesafe_overrides = self.build_thread_config_overrides( |
| model, |
| model_provider, |
| service_tier, |
| cwd, |
| runtime_workspace_roots, |
| approval_policy, |
| approvals_reviewer, |
| sandbox, |
| permissions, |
| base_instructions, |
| developer_instructions, |
| personality, |
| ); |
| typesafe_overrides.ephemeral = ephemeral; |
| let listener_task_context = ListenerTaskContext { |
| thread_manager: Arc::clone(&self.thread_manager), |
| thread_state_manager: self.thread_state_manager.clone(), |
| outgoing: Arc::clone(&self.outgoing), |
| pending_thread_unloads: Arc::clone(&self.pending_thread_unloads), |
| thread_watch_manager: self.thread_watch_manager.clone(), |
| codex_home: self.config.codex_home.to_path_buf(), |
| thread_unload_delay: self.config.thread_unload_delay, |
| skills_watcher: Arc::clone(&self.skills_watcher), |
| turn_cost_worker: self.turn_cost_worker.clone(), |
| }; |
| let request_trace = request_context.request_trace(); |
| let config_manager = self.config_manager.clone(); |
| let thread_store = Arc::clone(&self.thread_store); |
| let initial_config_warnings = Arc::clone(&self.initial_config_warnings); |
| let outgoing = Arc::clone(&listener_task_context.outgoing); |
| let error_request_id = request_id.clone(); |
| let thread_start_task = async move { |
| if let Err(error) = Self::thread_start_task( |
| listener_task_context, |
| thread_store, |
| config_manager, |
| request_id, |
| app_server_client_name, |
| app_server_client_version, |
| client_mcp_extensions, |
| config, |
| typesafe_overrides, |
| dynamic_tools, |
| selected_capability_roots.unwrap_or_default(), |
| history_mode.map(Into::into), |
| session_start_source, |
| thread_source.map(Into::into), |
| project_id, |
| daybreak_enabled, |
| environments, |
| service_name, |
| allow_provider_model_fallback, |
| experimental_raw_events, |
| request_trace, |
| initial_config_warnings, |
| ) |
| .await |
| { |
| outgoing.send_error(error_request_id, error).await; |
| } |
| }; |
| self.background_tasks |
| .spawn(thread_start_task.instrument(request_context.span())) |
| .await |
| .map_err(|_| internal_error("thread startup task stopped before completing"))?; |
| Ok(()) |
| } |
|
|
| pub(crate) async fn drain_background_tasks(&self) { |
| self.background_tasks.close(); |
| if tokio::time::timeout(Duration::from_secs(10), self.background_tasks.wait()) |
| .await |
| .is_err() |
| { |
| warn!("timed out waiting for background tasks to shut down; proceeding"); |
| } |
| } |
|
|
| pub(crate) async fn clear_all_thread_listeners(&self) { |
| self.thread_state_manager.clear_all_listeners().await; |
| } |
|
|
| pub(crate) async fn shutdown_threads(&self) { |
| let report = self |
| .thread_manager |
| .shutdown_all_threads_bounded(Duration::from_secs(10)) |
| .await; |
| for thread_id in report.submit_failed { |
| warn!("failed to submit Shutdown to thread {thread_id}"); |
| } |
| for thread_id in report.timed_out { |
| warn!("timed out waiting for thread {thread_id} to shut down"); |
| } |
| } |
|
|
| async fn request_trace_context( |
| &self, |
| request_id: &ConnectionRequestId, |
| ) -> Option<codex_protocol::protocol::W3cTraceContext> { |
| self.outgoing.request_trace_context(request_id).await |
| } |
|
|
| async fn submit_core_op( |
| &self, |
| request_id: &ConnectionRequestId, |
| thread: &CodexThread, |
| op: Op, |
| ) -> CodexResult<String> { |
| thread |
| .submit_with_trace(op, self.request_trace_context(request_id).await) |
| .await |
| } |
|
|
| #[allow(clippy::too_many_arguments)] |
| async fn thread_start_task( |
| listener_task_context: ListenerTaskContext, |
| thread_store: Arc<dyn ThreadStore>, |
| config_manager: ConfigManager, |
| request_id: ConnectionRequestId, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| client_mcp_extensions: ClientMcpExtensions, |
| config_overrides: Option<HashMap<String, serde_json::Value>>, |
| typesafe_overrides: ConfigOverrides, |
| dynamic_tools: Option<Vec<DynamicToolSpec>>, |
| selected_capability_roots: Vec<SelectedCapabilityRoot>, |
| history_mode: Option<ThreadHistoryMode>, |
| session_start_source: Option<codex_app_server_protocol::ThreadStartSource>, |
| thread_source: Option<codex_protocol::protocol::ThreadSource>, |
| project_id: Option<String>, |
| daybreak_enabled: Option<bool>, |
| environment_selections: Option<Vec<TurnEnvironmentSelection>>, |
| service_name: Option<String>, |
| allow_provider_model_fallback: bool, |
| experimental_raw_events: bool, |
| request_trace: Option<W3cTraceContext>, |
| initial_config_warnings: Arc<Vec<ConfigWarningNotification>>, |
| ) -> Result<(), JSONRPCErrorError> { |
| let thread_start_started_at = std::time::Instant::now(); |
| let requested_cwd = typesafe_overrides.cwd.clone(); |
| let mut config = config_manager |
| .load_with_overrides(config_overrides.clone(), typesafe_overrides.clone()) |
| .await |
| .map_err(|err| config_load_error(&err))?; |
| if config.ephemeral && daybreak_enabled.is_some() { |
| return Err(invalid_request( |
| "daybreakEnabled is not supported for ephemeral threads", |
| )); |
| } |
| |
| |
| let effective_permission_profile = config.permissions.effective_permission_profile(); |
| let effective_permissions_trust_project = match &effective_permission_profile { |
| codex_protocol::models::PermissionProfile::Disabled |
| | codex_protocol::models::PermissionProfile::External { .. } => true, |
| codex_protocol::models::PermissionProfile::Managed { .. } => { |
| effective_permission_profile |
| .file_system_sandbox_policy() |
| .can_write_local_path_with_cwd(config.cwd.as_path(), config.cwd.as_path()) |
| } |
| }; |
|
|
| if requested_cwd.is_some() |
| && config.active_project.trust_level.is_none() |
| && effective_permissions_trust_project |
| { |
| let trust_target = resolve_root_git_project_for_trust(LOCAL_FS.as_ref(), &config.cwd) |
| .await |
| .unwrap_or_else(|| config.cwd.clone()); |
| let current_cli_overrides = config_manager.current_cli_overrides(); |
| let cli_overrides_with_trust; |
| let cli_overrides_for_reload = if let Err(err) = |
| codex_core::config::set_project_trust_level( |
| &listener_task_context.codex_home, |
| trust_target.as_path(), |
| TrustLevel::Trusted, |
| ) { |
| warn!( |
| "failed to persist trusted project state for {}; continuing with in-memory trust for this thread: {err}", |
| trust_target.display() |
| ); |
| let mut project = toml::map::Map::new(); |
| project.insert( |
| "trust_level".to_string(), |
| TomlValue::String("trusted".to_string()), |
| ); |
| let mut projects = toml::map::Map::new(); |
| projects.insert( |
| project_trust_key(trust_target.as_path()), |
| TomlValue::Table(project), |
| ); |
| cli_overrides_with_trust = current_cli_overrides |
| .iter() |
| .cloned() |
| .chain(std::iter::once(( |
| "projects".to_string(), |
| TomlValue::Table(projects), |
| ))) |
| .collect::<Vec<_>>(); |
| cli_overrides_with_trust.as_slice() |
| } else { |
| current_cli_overrides.as_slice() |
| }; |
|
|
| config = config_manager |
| .load_with_cli_overrides( |
| cli_overrides_for_reload, |
| config_overrides, |
| typesafe_overrides, |
| None, |
| ) |
| .await |
| .map_err(|err| config_load_error(&err))?; |
| } |
|
|
| |
| let mut config_warnings = config |
| .startup_warnings |
| .iter() |
| .map(|summary| ConfigWarningNotification { |
| summary: summary.clone(), |
| details: None, |
| path: None, |
| range: None, |
| }) |
| .collect::<Vec<_>>(); |
| if let Ok(Some(err)) = |
| codex_core::check_execpolicy_for_warnings(&config.config_layer_stack).await |
| { |
| config_warnings.push(crate::exec_policy_config_warning(&err)); |
| } |
| for notification in config_warnings { |
| if !initial_config_warnings.contains(¬ification) { |
| listener_task_context |
| .outgoing |
| .send_server_notification_to_connections( |
| &[request_id.connection_id], |
| ServerNotification::ConfigWarning(notification), |
| ) |
| .await; |
| } |
| } |
|
|
| let environments = environment_selections.unwrap_or_else(|| { |
| listener_task_context |
| .thread_manager |
| .default_environment_selections(&config.cwd, &config.workspace_roots) |
| }); |
| let dynamic_tools = dynamic_tools.unwrap_or_default(); |
| if !dynamic_tools.is_empty() { |
| validate_dynamic_tools(&dynamic_tools).map_err(invalid_request)?; |
| } |
| |
| let dynamic_tool_count: usize = dynamic_tools |
| .iter() |
| .map(|tool| match tool { |
| DynamicToolSpec::Function(_) => 1, |
| DynamicToolSpec::Namespace(namespace) => namespace.tools.len(), |
| }) |
| .sum(); |
| let history_mode = history_mode.or_else(|| { |
| (!config.ephemeral && thread_store.supports_paginated_history_lists()) |
| .then_some(ThreadHistoryMode::Paginated) |
| }); |
| let mut thread_extension_init = ExtensionDataInit::new(); |
| if !selected_capability_roots.is_empty() { |
| thread_extension_init.insert(selected_capability_roots); |
| } |
| let mut start_options = StartThreadOptions::new(config); |
| let reserved_thread_id = if start_options.config.ephemeral { |
| None |
| } else { |
| stage_pending_thread_metadata( |
| listener_task_context.thread_manager.as_ref(), |
| thread_store.as_ref(), |
| StoreThreadMetadataPatch { |
| project_id: project_id.clone().map(Some), |
| daybreak_enabled, |
| ..Default::default() |
| }, |
| "thread/start", |
| ) |
| .await? |
| }; |
| start_options.reserved_thread_id = reserved_thread_id; |
| let create_thread_started_at = std::time::Instant::now(); |
| let new_thread = listener_task_context |
| .thread_manager |
| .start_thread(StartThreadOptions { |
| allow_provider_model_fallback, |
| initial_history: match session_start_source |
| .unwrap_or(codex_app_server_protocol::ThreadStartSource::Startup) |
| { |
| codex_app_server_protocol::ThreadStartSource::Startup => InitialHistory::New, |
| codex_app_server_protocol::ThreadStartSource::Clear => InitialHistory::Cleared, |
| }, |
| history_mode, |
| thread_source, |
| dynamic_tools, |
| metrics_service_name: service_name, |
| parent_trace: request_trace, |
| environments: Some(environments), |
| thread_extension_init, |
| client_mcp_extensions, |
| ..start_options |
| }) |
| .instrument(tracing::info_span!( |
| "app_server.thread_start.create_thread", |
| otel.name = "app_server.thread_start.create_thread", |
| thread_start.dynamic_tool_count = dynamic_tool_count, |
| )) |
| .await; |
| let NewThread { |
| thread_id, |
| thread, |
| session_configured, |
| .. |
| } = match new_thread { |
| Ok(new_thread) => new_thread, |
| Err(err) => { |
| remove_pending_thread_metadata(thread_store.as_ref(), reserved_thread_id).await; |
| return Err(match err.details() { |
| CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()), |
| CodexErrorDetails::UnsupportedOperation(message) => { |
| method_not_found(message.clone()) |
| } |
| _ => internal_error(format!("error creating thread: {err}")), |
| }); |
| } |
| }; |
| let session_telemetry = thread.session_telemetry(); |
| session_telemetry.record_startup_phase( |
| "thread_start_create_thread", |
| create_thread_started_at.elapsed(), |
| Some("ready"), |
| ); |
|
|
| Self::set_app_server_client_info( |
| thread.as_ref(), |
| app_server_client_name, |
| app_server_client_version, |
| ) |
| .await?; |
|
|
| let instruction_sources = thread.legacy_instruction_sources().await; |
| let config_snapshot = thread |
| .config_snapshot() |
| .instrument(tracing::info_span!( |
| "app_server.thread_start.config_snapshot", |
| otel.name = "app_server.thread_start.config_snapshot", |
| )) |
| .await; |
| let mut thread = build_thread_from_snapshot( |
| thread_id, |
| session_configured.session_id.to_string(), |
| thread.multi_agent_version(), |
| &config_snapshot, |
| session_configured.rollout_path.clone(), |
| ); |
| thread.project_id = project_id.clone(); |
| thread.daybreak_enabled = daybreak_enabled; |
|
|
| |
| log_listener_attach_result( |
| super::thread_lifecycle::ensure_conversation_listener( |
| listener_task_context.clone(), |
| thread_id, |
| request_id.connection_id, |
| experimental_raw_events, |
| ) |
| .instrument(tracing::info_span!( |
| "app_server.thread_start.attach_listener", |
| otel.name = "app_server.thread_start.attach_listener", |
| thread_start.experimental_raw_events = experimental_raw_events, |
| )) |
| .await, |
| thread_id, |
| request_id.connection_id, |
| "thread", |
| ); |
|
|
| listener_task_context |
| .thread_watch_manager |
| .upsert_thread_silently(&thread.id) |
| .instrument(tracing::info_span!( |
| "app_server.thread_start.upsert_thread", |
| otel.name = "app_server.thread_start.upsert_thread", |
| )) |
| .await; |
|
|
| thread.status = resolve_thread_status( |
| listener_task_context |
| .thread_watch_manager |
| .loaded_status_for_thread(&thread.id) |
| .instrument(tracing::info_span!( |
| "app_server.thread_start.resolve_status", |
| otel.name = "app_server.thread_start.resolve_status", |
| )) |
| .await, |
| false, |
| ); |
|
|
| let sandbox = config_snapshot.sandbox_policy().into(); |
| let cwd = config_snapshot.cwd().clone(); |
| let active_permission_profile = |
| thread_response_active_permission_profile(config_snapshot.active_permission_profile); |
| let thread_originator = config_snapshot.originator.clone(); |
|
|
| let response = ThreadStartResponse { |
| thread: thread.clone(), |
| disabled_plugin_ids: config_snapshot.disabled_plugin_ids, |
| model: config_snapshot.model, |
| model_provider: config_snapshot.model_provider_id, |
| service_tier: config_snapshot.service_tier, |
| cwd, |
| runtime_workspace_roots: config_snapshot.workspace_roots, |
| instruction_sources, |
| approval_policy: config_snapshot.approval_policy.into(), |
| approvals_reviewer: config_snapshot.approvals_reviewer.into(), |
| sandbox, |
| active_permission_profile, |
| reasoning_effort: config_snapshot.reasoning_effort, |
| multi_agent_mode: MultiAgentMode::ExplicitRequestOnly, |
| }; |
| let notif = thread_started_notification(thread); |
| listener_task_context |
| .outgoing |
| .send_response_with_thread_originator(request_id, response, thread_originator) |
| .instrument(tracing::info_span!( |
| "app_server.thread_start.send_response", |
| otel.name = "app_server.thread_start.send_response", |
| )) |
| .await; |
|
|
| listener_task_context |
| .outgoing |
| .send_server_notification(ServerNotification::ThreadStarted(notif)) |
| .instrument(tracing::info_span!( |
| "app_server.thread_start.notify_started", |
| otel.name = "app_server.thread_start.notify_started", |
| )) |
| .await; |
| session_telemetry.record_startup_phase( |
| "thread_start_total", |
| thread_start_started_at.elapsed(), |
| Some("ready"), |
| ); |
| Ok(()) |
| } |
|
|
| #[allow(clippy::too_many_arguments)] |
| fn build_thread_config_overrides( |
| &self, |
| model: Option<String>, |
| model_provider: Option<String>, |
| service_tier: Option<Option<String>>, |
| cwd: Option<String>, |
| runtime_workspace_roots: Option<Vec<AbsolutePathBuf>>, |
| approval_policy: Option<codex_app_server_protocol::AskForApproval>, |
| approvals_reviewer: Option<codex_app_server_protocol::ApprovalsReviewer>, |
| sandbox: Option<SandboxMode>, |
| permissions: Option<String>, |
| base_instructions: Option<String>, |
| developer_instructions: Option<String>, |
| personality: Option<Personality>, |
| ) -> ConfigOverrides { |
| ConfigOverrides { |
| model, |
| model_provider, |
| service_tier, |
| cwd: cwd.map(PathBuf::from), |
| workspace_roots: runtime_workspace_roots, |
| default_permissions: permissions, |
| approval_policy: approval_policy |
| .map(codex_app_server_protocol::AskForApproval::to_core), |
| approvals_reviewer: approvals_reviewer |
| .map(codex_app_server_protocol::ApprovalsReviewer::to_core), |
| sandbox_mode: sandbox.map(SandboxMode::to_core), |
| codex_linux_sandbox_exe: self.arg0_paths.codex_linux_sandbox_exe.clone(), |
| main_execve_wrapper_exe: self.arg0_paths.main_execve_wrapper_exe.clone(), |
| base_instructions, |
| developer_instructions, |
| personality, |
| ..Default::default() |
| } |
| } |
|
|
| async fn thread_archive_inner( |
| &self, |
| params: ThreadArchiveParams, |
| ) -> Result<(ThreadArchiveResponse, Vec<String>), JSONRPCErrorError> { |
| let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; |
| self.thread_archive_response(params).await |
| } |
|
|
| async fn thread_archive_response( |
| &self, |
| params: ThreadArchiveParams, |
| ) -> Result<(ThreadArchiveResponse, Vec<String>), JSONRPCErrorError> { |
| let thread_id = ThreadId::from_string(¶ms.thread_id) |
| .map_err(|err| invalid_request(format!("invalid session id: {err}")))?; |
|
|
| let subtree_thread_ids = self.state_db_spawn_subtree_thread_ids(thread_id).await?; |
|
|
| let mut archive_thread_ids = Vec::new(); |
| match self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id, |
| include_archived: false, |
| include_history: false, |
| }) |
| .await |
| { |
| Ok(thread) => { |
| if thread.archived_at.is_none() { |
| archive_thread_ids.push(thread_id); |
| } |
| } |
| Err(err) => return Err(thread_store_mutation_error("archive", err)), |
| } |
| for descendant_thread_id in subtree_thread_ids.iter().copied().skip(1) { |
| match self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id: descendant_thread_id, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| { |
| Ok(thread) => { |
| if thread.archived_at.is_none() { |
| archive_thread_ids.push(descendant_thread_id); |
| } |
| } |
| Err(err) => { |
| warn!( |
| "failed to read spawned descendant thread {descendant_thread_id} while archiving {thread_id}: {err}" |
| ); |
| } |
| } |
| } |
|
|
| if archive_thread_ids.is_empty() { |
| return Ok((ThreadArchiveResponse {}, Vec::new())); |
| } |
|
|
| archive_thread_ids[1..].reverse(); |
| |
| self.prepare_thread_for_archive(thread_id).await?; |
| for &descendant_thread_id in subtree_thread_ids.iter().skip(1).rev() { |
| self.prepare_thread_for_archive(descendant_thread_id) |
| .await?; |
| } |
|
|
| let archived_thread_ids = self |
| .thread_store |
| .archive_threads(StoreArchiveThreadsParams { |
| thread_ids: archive_thread_ids, |
| writer_lock_thread_ids: subtree_thread_ids, |
| }) |
| .await |
| .map_err(|err| thread_store_mutation_error("archive", err))? |
| .into_iter() |
| .map(|thread_id| thread_id.to_string()) |
| .collect(); |
| Ok((ThreadArchiveResponse {}, archived_thread_ids)) |
| } |
|
|
| pub(super) async fn state_db_spawn_subtree_thread_ids( |
| &self, |
| thread_id: ThreadId, |
| ) -> Result<Vec<ThreadId>, JSONRPCErrorError> { |
| self.thread_manager |
| .list_agent_subtree_thread_ids(thread_id) |
| .await |
| .map_err(|err| { |
| internal_error(format!( |
| "failed to list spawned descendants for thread id {thread_id}: {err}" |
| )) |
| }) |
| } |
|
|
| async fn thread_increment_elicitation_inner( |
| &self, |
| params: ThreadIncrementElicitationParams, |
| ) -> Result<ThreadIncrementElicitationResponse, JSONRPCErrorError> { |
| let (_, thread) = self.load_thread(¶ms.thread_id).await?; |
| let count = thread |
| .increment_out_of_band_elicitation_count() |
| .await |
| .map_err(|err| { |
| internal_error(format!( |
| "failed to increment out-of-band elicitation counter: {err}" |
| )) |
| })?; |
| Ok(ThreadIncrementElicitationResponse { |
| count, |
| paused: count > 0, |
| }) |
| } |
|
|
| async fn thread_decrement_elicitation_inner( |
| &self, |
| params: ThreadDecrementElicitationParams, |
| ) -> Result<ThreadDecrementElicitationResponse, JSONRPCErrorError> { |
| let (_, thread) = self.load_thread(¶ms.thread_id).await?; |
| let count = thread |
| .decrement_out_of_band_elicitation_count() |
| .await |
| .map_err(|err| match err.details() { |
| CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()), |
| _ => internal_error(format!( |
| "failed to decrement out-of-band elicitation counter: {err}" |
| )), |
| })?; |
| Ok(ThreadDecrementElicitationResponse { |
| count, |
| paused: count > 0, |
| }) |
| } |
|
|
| async fn thread_set_name_response_inner( |
| &self, |
| params: ThreadSetNameParams, |
| ) -> Result<(ThreadSetNameResponse, Option<ThreadNameUpdatedNotification>), JSONRPCErrorError> |
| { |
| let ThreadSetNameParams { thread_id, name } = params; |
| let thread_id = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
| let Some(name) = codex_core::util::normalize_thread_name(&name) else { |
| return Err(invalid_request("thread name must not be empty")); |
| }; |
|
|
| let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; |
| self.thread_manager |
| .update_thread_metadata( |
| thread_id, |
| StoreThreadMetadataPatch { |
| name: Some(Some(name.clone())), |
| ..Default::default() |
| }, |
| false, |
| ) |
| .await |
| .map_err(|err| core_thread_write_error("set thread name", err))?; |
|
|
| Ok(( |
| ThreadSetNameResponse {}, |
| Some(ThreadNameUpdatedNotification { |
| thread_id: thread_id.to_string(), |
| thread_name: Some(name), |
| }), |
| )) |
| } |
|
|
| async fn thread_memory_mode_set_response_inner( |
| &self, |
| params: ThreadMemoryModeSetParams, |
| ) -> Result<ThreadMemoryModeSetResponse, JSONRPCErrorError> { |
| let ThreadMemoryModeSetParams { thread_id, mode } = params; |
| let thread_id = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
|
|
| self.thread_manager |
| .update_thread_metadata( |
| thread_id, |
| StoreThreadMetadataPatch { |
| memory_mode: Some(mode.to_core()), |
| ..Default::default() |
| }, |
| false, |
| ) |
| .await |
| .map_err(|err| core_thread_write_error("set thread memory mode", err))?; |
|
|
| Ok(ThreadMemoryModeSetResponse {}) |
| } |
|
|
| async fn memory_reset_response_inner(&self) -> Result<MemoryResetResponse, JSONRPCErrorError> { |
| let state_db = self |
| .state_db |
| .clone() |
| .ok_or_else(|| internal_error("sqlite state db unavailable for memory reset"))?; |
|
|
| state_db.clear_all_memory_data().await.map_err(|err| { |
| internal_error(format!("failed to clear memory rows in memories db: {err}")) |
| })?; |
|
|
| clear_memory_roots_contents(&self.config.codex_home) |
| .await |
| .map_err(|err| { |
| internal_error(format!( |
| "failed to clear memory directories under {}: {err}", |
| self.config.codex_home.display() |
| )) |
| })?; |
|
|
| Ok(MemoryResetResponse {}) |
| } |
|
|
| async fn thread_metadata_update_response_inner( |
| &self, |
| params: ThreadMetadataUpdateParams, |
| ) -> Result<ThreadMetadataUpdateResponse, JSONRPCErrorError> { |
| let ThreadMetadataUpdateParams { |
| thread_id, |
| project_id, |
| git_info, |
| daybreak_enabled, |
| } = params; |
| let thread_uuid = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
|
|
| if let Some(project_id) = project_id.as_ref() |
| && !project_id.is_empty() |
| { |
| let project = self |
| .thread_store |
| .read_project(project_id.clone()) |
| .await |
| .map_err(|err| match err { |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| err => internal_error(format!("failed to read project: {err}")), |
| })?; |
| if project.is_none() { |
| return Err(invalid_request(format!("project not found: {project_id}"))); |
| } |
| } |
|
|
| if git_info.is_none() && project_id.is_none() && daybreak_enabled.is_none() { |
| return Err(invalid_request( |
| "thread metadata update must include at least one field", |
| )); |
| } |
|
|
| let git_info = git_info |
| .map( |
| |ThreadMetadataGitInfoUpdateParams { |
| sha, |
| branch, |
| origin_url, |
| }| { |
| if sha.is_none() && branch.is_none() && origin_url.is_none() { |
| return Err(invalid_request("gitInfo must include at least one field")); |
| } |
|
|
| let origin_url = |
| Self::normalize_thread_metadata_git_field(origin_url, "gitInfo.originUrl")?; |
| let origin_url = match origin_url { |
| Some(Some(origin_url)) => { |
| Some(Some(SanitizedGitUrl::try_from(origin_url).map_err( |
| |_| invalid_request("gitInfo.originUrl must be a valid Git remote"), |
| )?)) |
| } |
| Some(None) => Some(None), |
| None => None, |
| }; |
|
|
| Ok(StoreGitInfoPatch { |
| sha: Self::normalize_thread_metadata_git_field(sha, "gitInfo.sha")?, |
| branch: Self::normalize_thread_metadata_git_field( |
| branch, |
| "gitInfo.branch", |
| )?, |
| origin_url, |
| }) |
| }, |
| ) |
| .transpose()?; |
|
|
| let project_update: StoreClearableField<String> = if let Some(project_id) = project_id { |
| if project_id.is_empty() { |
| Some(None) |
| } else { |
| Some(Some(project_id)) |
| } |
| } else { |
| None |
| }; |
| let updated_thread = { |
| let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; |
| let previous_project_id = if project_update.is_some() { |
| Some( |
| self.thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id: thread_uuid, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| .map_err(|err| match err { |
| ThreadStoreError::ThreadNotFound { .. } => { |
| invalid_request(format!("thread not found: {thread_id}")) |
| } |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| err => internal_error(format!("failed to read thread metadata: {err}")), |
| })? |
| .project_id, |
| ) |
| } else { |
| None |
| }; |
| let patch = StoreThreadMetadataPatch { |
| git_info, |
| project_id: project_update.clone(), |
| daybreak_enabled, |
| ..Default::default() |
| }; |
| let updated_thread = self |
| .thread_manager |
| .update_thread_metadata(thread_uuid, patch, true) |
| .await |
| .map_err(|err| core_thread_write_error("update thread metadata", err))?; |
| if let Some(project_id) = project_update.as_ref() |
| && previous_project_id.as_ref() != Some(project_id) |
| { |
| self.outgoing |
| .send_server_notification(ServerNotification::ThreadProjectUpdated( |
| ThreadProjectUpdatedNotification { |
| thread_id: thread_id.clone(), |
| project_id: project_id.clone(), |
| }, |
| )) |
| .await; |
| } |
| updated_thread |
| }; |
| let (mut thread, _) = thread_from_stored_thread( |
| updated_thread, |
| self.config.model_provider_id.as_str(), |
| &self.config.cwd, |
| ); |
| if let Ok(loaded_thread) = self.thread_manager.get_thread(thread_uuid).await { |
| thread.session_id = loaded_thread.session_configured().session_id.to_string(); |
| let config_snapshot = loaded_thread.config_snapshot().await; |
| apply_live_thread_settings(&mut thread, &config_snapshot); |
| } |
| self.attach_thread_name(thread_uuid, &mut thread).await; |
| thread.status = resolve_thread_status( |
| self.thread_watch_manager |
| .loaded_status_for_thread(&thread.id) |
| .await, |
| false, |
| ); |
|
|
| Ok(ThreadMetadataUpdateResponse { thread }) |
| } |
|
|
| fn normalize_thread_metadata_git_field( |
| value: Option<Option<String>>, |
| name: &str, |
| ) -> Result<Option<Option<String>>, JSONRPCErrorError> { |
| match value { |
| Some(Some(value)) => { |
| let value = value.trim().to_string(); |
| if value.is_empty() { |
| return Err(invalid_request(format!("{name} must not be empty"))); |
| } |
| Ok(Some(Some(value))) |
| } |
| Some(None) => Ok(Some(None)), |
| None => Ok(None), |
| } |
| } |
|
|
| async fn thread_unarchive_inner( |
| &self, |
| params: ThreadUnarchiveParams, |
| ) -> Result<(ThreadUnarchiveResponse, ThreadUnarchivedNotification), JSONRPCErrorError> { |
| let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; |
| let (response, thread_id) = self.thread_unarchive_response(params).await?; |
| Ok((response, ThreadUnarchivedNotification { thread_id })) |
| } |
|
|
| async fn thread_unarchive_response( |
| &self, |
| params: ThreadUnarchiveParams, |
| ) -> Result<(ThreadUnarchiveResponse, String), JSONRPCErrorError> { |
| let thread_id = ThreadId::from_string(¶ms.thread_id) |
| .map_err(|err| invalid_request(format!("invalid session id: {err}")))?; |
|
|
| let fallback_provider = self.config.model_provider_id.clone(); |
| let stored_thread = self |
| .thread_store |
| .unarchive_thread(StoreArchiveThreadParams { thread_id }) |
| .await |
| .map_err(|err| thread_store_mutation_error("unarchive", err))?; |
| let (mut thread, _) = |
| thread_from_stored_thread(stored_thread, fallback_provider.as_str(), &self.config.cwd); |
|
|
| if let Ok(loaded_thread) = self.thread_manager.get_thread(thread_id).await { |
| let config_snapshot = loaded_thread.config_snapshot().await; |
| apply_live_thread_settings(&mut thread, &config_snapshot); |
| } |
| thread.status = resolve_thread_status( |
| self.thread_watch_manager |
| .loaded_status_for_thread(&thread.id) |
| .await, |
| false, |
| ); |
| self.attach_thread_name(thread_id, &mut thread).await; |
| let thread_id = thread.id.clone(); |
| Ok((ThreadUnarchiveResponse { thread }, thread_id)) |
| } |
|
|
| async fn thread_revert_response( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadRevertParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| ) -> Result<(ThreadRevertResponse, String), JSONRPCErrorError> { |
| let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; |
| let ThreadRevertParams { |
| thread_id, |
| before_turn_id, |
| } = params; |
| let (thread_id, thread) = self.load_thread(&thread_id).await?; |
| ensure_direct_input_allowed(thread.as_ref()).await?; |
| let config_snapshot = thread.config_snapshot().await; |
| if !matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) { |
| return Err(invalid_request( |
| "thread/revert only supports paginated threads", |
| )); |
| } |
| let runtime_snapshot = ThreadRevertRuntimeSnapshot { |
| config: thread.config().await.as_ref().clone(), |
| settings: thread.restorable_thread_settings().await, |
| client_mcp_extensions: thread.client_mcp_extensions(), |
| }; |
|
|
| |
| |
| |
| if matches!( |
| self.ensure_conversation_listener( |
| thread_id, |
| request_id.connection_id, |
| false, |
| ) |
| .await?, |
| EnsureConversationListenerResult::ConnectionClosed |
| ) { |
| return Err(internal_error(format!( |
| "connection closed before thread {thread_id} could be reverted" |
| ))); |
| } |
| let thread_state = self.thread_state_manager.thread_state(thread_id).await; |
| let shutdown_drain_rx = thread_state.lock().await.register_shutdown_drain_waiter(); |
|
|
| match wait_for_thread_shutdown(&thread).await { |
| ThreadShutdownResult::Complete => {} |
| ThreadShutdownResult::SubmitFailed => { |
| thread_state.lock().await.take_shutdown_drain_waiter(); |
| return Err(internal_error(format!( |
| "failed to shut down thread {thread_id} before revert" |
| ))); |
| } |
| ThreadShutdownResult::TimedOut => { |
| thread_state.lock().await.take_shutdown_drain_waiter(); |
| return Err(internal_error(format!( |
| "timed out shutting down thread {thread_id} before revert" |
| ))); |
| } |
| } |
| let drain_result = tokio::time::timeout(Duration::from_secs(10), shutdown_drain_rx) |
| .await |
| .map_err(|_| { |
| internal_error(format!( |
| "timed out waiting for thread {thread_id} listener to drain shutdown events" |
| )) |
| }) |
| .and_then(|result| { |
| result.map_err(|_| { |
| internal_error(format!( |
| "thread {thread_id} listener stopped before draining shutdown events" |
| )) |
| }) |
| }); |
| if let Err(err) = drain_result { |
| thread_state.lock().await.take_shutdown_drain_waiter(); |
| return Err(err); |
| } |
| if self |
| .thread_manager |
| .remove_thread(&thread_id) |
| .await |
| .is_none() |
| { |
| return Err(internal_error(format!( |
| "thread {thread_id} disappeared before revert" |
| ))); |
| } |
| |
| |
| self.outgoing |
| .cancel_requests_for_thread(thread_id, None) |
| .await; |
|
|
| let revert_result = self |
| .thread_store |
| .revert_thread(codex_thread_store::RevertThreadParams { |
| thread_id, |
| before_turn_id, |
| multi_agent_version: thread.multi_agent_version(), |
| }) |
| .await |
| .map_err(|err| thread_store_mutation_error("revert", err)); |
| let response = self |
| .reload_paginated_thread( |
| request_id, |
| thread_id, |
| runtime_snapshot, |
| app_server_client_name, |
| app_server_client_version, |
| ) |
| .await?; |
| revert_result?; |
| Ok((response, thread_id.to_string())) |
| } |
|
|
| async fn reload_paginated_thread( |
| &self, |
| request_id: &ConnectionRequestId, |
| thread_id: ThreadId, |
| runtime_snapshot: ThreadRevertRuntimeSnapshot, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| ) -> Result<ThreadRevertResponse, JSONRPCErrorError> { |
| let ThreadRevertRuntimeSnapshot { |
| config, |
| settings, |
| client_mcp_extensions, |
| } = runtime_snapshot; |
| let thread_id_string = thread_id.to_string(); |
| let stored_thread = self |
| .read_stored_thread_for_resume( |
| thread_id_string.as_str(), |
| None, |
| false, |
| ) |
| .await?; |
| let (thread_history, resume_source_thread) = self |
| .load_resume_initial_history_from_stored_thread(stored_thread) |
| .await?; |
| let response_history = thread_history.clone(); |
| let NewThread { |
| thread_id: resumed_thread_id, |
| thread: codex_thread, |
| session_configured, |
| .. |
| } = self |
| .thread_manager |
| .resume_thread_with_history( |
| config, |
| thread_history, |
| self.auth_manager.clone(), |
| self.request_trace_context(request_id).await, |
| client_mcp_extensions, |
| ) |
| .await |
| .map_err(|err| internal_error(format!("error reloading thread after revert: {err}")))?; |
| if resumed_thread_id != thread_id { |
| return Err(internal_error(format!( |
| "thread {thread_id} reloaded as {resumed_thread_id} after revert" |
| ))); |
| } |
| codex_thread |
| .restore_thread_settings(settings) |
| .await |
| .map_err(|err| { |
| internal_error(format!( |
| "failed to restore thread settings after revert: {err}" |
| )) |
| })?; |
| |
| codex_thread |
| .checkpoint_thread_settings() |
| .await |
| .map_err(|err| { |
| internal_error(format!( |
| "failed to persist restored thread settings after revert: {err}" |
| )) |
| })?; |
| Self::set_app_server_client_info( |
| codex_thread.as_ref(), |
| app_server_client_name, |
| app_server_client_version, |
| ) |
| .await?; |
| let SessionConfiguredEvent { rollout_path, .. } = session_configured; |
| let rollout_path = rollout_path.ok_or_else(|| { |
| internal_error(format!( |
| "rollout path missing after reloading thread {thread_id}" |
| )) |
| })?; |
| |
| |
| |
| let thread_state = self.thread_state_manager.thread_state(thread_id).await; |
| self.ensure_listener_task_running(thread_id, Arc::clone(&codex_thread), thread_state) |
| .await?; |
| let mut thread = self |
| .load_thread_from_resume_source_or_send_internal( |
| thread_id, |
| codex_thread.as_ref(), |
| &response_history, |
| rollout_path.as_path(), |
| Some(resume_source_thread), |
| false, |
| ) |
| .await |
| .map_err(internal_error)?; |
| self.thread_watch_manager.upsert_thread(&thread.id).await; |
| let thread_status = self |
| .thread_watch_manager |
| .loaded_status_for_thread(&thread.id) |
| .await; |
| set_thread_status_and_interrupt_stale_turns( |
| &mut thread, |
| thread_status, |
| false, |
| ); |
| let (turns_backwards_cursor, items_backwards_cursor) = |
| Self::paginated_resume_backwards_cursors(self.thread_store.as_ref(), thread_id).await?; |
| Ok(ThreadRevertResponse { |
| thread, |
| turns_backwards_cursor, |
| items_backwards_cursor, |
| }) |
| } |
|
|
| async fn thread_compact_start_inner( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadCompactStartParams, |
| ) -> Result<ThreadCompactStartResponse, JSONRPCErrorError> { |
| let ThreadCompactStartParams { thread_id } = params; |
|
|
| let (_, thread) = self.load_thread(&thread_id).await?; |
| ensure_direct_input_allowed(thread.as_ref()).await?; |
| self.config_manager |
| .check_thread_model_provider(thread.config().await.as_ref()) |
| .await |
| .map_err(|error| config_load_error(&error))?; |
| self.submit_core_op(request_id, thread.as_ref(), Op::Compact) |
| .await |
| .map_err(|err| internal_error(format!("failed to start compaction: {err}")))?; |
| Ok(ThreadCompactStartResponse {}) |
| } |
|
|
| async fn thread_background_terminals_clean_inner( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadBackgroundTerminalsCleanParams, |
| ) -> Result<ThreadBackgroundTerminalsCleanResponse, JSONRPCErrorError> { |
| let ThreadBackgroundTerminalsCleanParams { thread_id } = params; |
|
|
| let (_, thread) = self.load_thread(&thread_id).await?; |
| self.submit_core_op(request_id, thread.as_ref(), Op::CleanBackgroundTerminals) |
| .await |
| .map_err(|err| { |
| internal_error(format!("failed to clean background terminals: {err}")) |
| })?; |
| Ok(ThreadBackgroundTerminalsCleanResponse {}) |
| } |
|
|
| async fn thread_background_terminals_list_inner( |
| &self, |
| params: ThreadBackgroundTerminalsListParams, |
| ) -> Result<ThreadBackgroundTerminalsListResponse, JSONRPCErrorError> { |
| let ThreadBackgroundTerminalsListParams { |
| thread_id, |
| cursor, |
| limit, |
| } = params; |
|
|
| let (_, thread) = self.load_thread(&thread_id).await?; |
| let terminals = thread |
| .list_background_terminals() |
| .await |
| .into_iter() |
| .map(|terminal| ThreadBackgroundTerminal { |
| item_id: terminal.item_id, |
| process_id: terminal.process_id, |
| command: terminal.command, |
| cwd: terminal.cwd.into(), |
| os_pid: None, |
| cpu_percent: None, |
| rss_kb: None, |
| }) |
| .collect::<Vec<_>>(); |
|
|
| let (data, next_cursor) = paginate_background_terminals(&terminals, cursor, limit)?; |
|
|
| Ok(ThreadBackgroundTerminalsListResponse { data, next_cursor }) |
| } |
|
|
| async fn thread_background_terminals_terminate_inner( |
| &self, |
| params: ThreadBackgroundTerminalsTerminateParams, |
| ) -> Result<ThreadBackgroundTerminalsTerminateResponse, JSONRPCErrorError> { |
| let ThreadBackgroundTerminalsTerminateParams { |
| thread_id, |
| process_id, |
| } = params; |
| let process_id = process_id.parse::<i32>().map_err(|err| { |
| invalid_request(format!("invalid background terminal process id: {err}")) |
| })?; |
|
|
| let (_, thread) = self.load_thread(&thread_id).await?; |
| let terminated = thread.terminate_background_terminal(process_id).await; |
| Ok(ThreadBackgroundTerminalsTerminateResponse { terminated }) |
| } |
|
|
| async fn thread_shell_command_inner( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadShellCommandParams, |
| ) -> Result<ThreadShellCommandResponse, JSONRPCErrorError> { |
| let ThreadShellCommandParams { |
| thread_id, |
| command, |
| timeout_ms, |
| } = params; |
| let command = command.trim().to_string(); |
| if command.is_empty() { |
| return Err(invalid_request("command must not be empty")); |
| } |
|
|
| let timeout_ms = timeout_ms |
| .map(|timeout_ms| { |
| u64::try_from(timeout_ms).map_err(|_| { |
| invalid_params(format!( |
| "thread/shellCommand timeoutMs must be non-negative, got {timeout_ms}" |
| )) |
| }) |
| }) |
| .transpose()?; |
|
|
| let (_, thread) = self.load_thread(&thread_id).await?; |
| ensure_direct_input_allowed(thread.as_ref()).await?; |
|
|
| |
| |
| if self |
| .thread_manager |
| .environment_manager() |
| .try_local_environment() |
| .is_none() |
| { |
| return Err(internal_error("local environment is not configured")); |
| } |
|
|
| self.submit_core_op( |
| request_id, |
| thread.as_ref(), |
| Op::RunUserShellCommand { |
| command, |
| timeout_ms, |
| }, |
| ) |
| .await |
| .map_err(|err| internal_error(format!("failed to start shell command: {err}")))?; |
| Ok(ThreadShellCommandResponse {}) |
| } |
|
|
| async fn thread_approve_guardian_denied_action_inner( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: ThreadApproveGuardianDeniedActionParams, |
| ) -> Result<ThreadApproveGuardianDeniedActionResponse, JSONRPCErrorError> { |
| let ThreadApproveGuardianDeniedActionParams { thread_id, event } = params; |
| let event = serde_json::from_value(event) |
| .map_err(|err| invalid_request(format!("invalid Guardian denial event: {err}")))?; |
| let (_, thread) = self.load_thread(&thread_id).await?; |
| ensure_direct_input_allowed(thread.as_ref()).await?; |
|
|
| self.submit_core_op( |
| request_id, |
| thread.as_ref(), |
| Op::ApproveGuardianDeniedAction { event }, |
| ) |
| .await |
| .map_err(|err| internal_error(format!("failed to approve Guardian denial: {err}")))?; |
| Ok(ThreadApproveGuardianDeniedActionResponse {}) |
| } |
|
|
| async fn thread_list_response_inner( |
| &self, |
| params: ThreadListParams, |
| ) -> Result<ThreadListResponse, JSONRPCErrorError> { |
| let ThreadListParams { |
| cursor, |
| limit, |
| sort_key, |
| sort_direction, |
| model_providers, |
| source_kinds, |
| originators, |
| archived, |
| section_id, |
| project_id, |
| cwd, |
| use_state_db_only, |
| search_term, |
| parent_thread_id, |
| ancestor_thread_id, |
| } = params; |
| if originators |
| .as_ref() |
| .is_some_and(|values| !values.is_empty()) |
| { |
| return Err(invalid_params( |
| "originator filtering is not supported by the local app-server", |
| )); |
| } |
| if project_id.is_some() && !self.thread_store.supports_projects() { |
| return Err(unsupported_thread_store_operation("projects")); |
| } |
| if let Some(Some(project_id)) = project_id.as_ref() { |
| if project_id.is_empty() { |
| return Err(invalid_params("projectId must not be empty")); |
| } |
| match self.thread_store.read_project(project_id.clone()).await { |
| Ok(Some(_)) => {} |
| Ok(None) => { |
| return Err(invalid_params(format!("project not found: {project_id}"))); |
| } |
| Err(ThreadStoreError::Unsupported { operation }) => { |
| return Err(unsupported_thread_store_operation(operation)); |
| } |
| Err(err) => { |
| return Err(internal_error(format!("failed to read project: {err}"))); |
| } |
| } |
| } |
| let cwd_filters = normalize_thread_list_cwd_filters(cwd)?; |
| let relation_filter = match (parent_thread_id, ancestor_thread_id) { |
| (Some(_), Some(_)) => { |
| return Err(invalid_request( |
| "parentThreadId and ancestorThreadId are mutually exclusive", |
| )); |
| } |
| (Some(parent_thread_id), None) => Some(StoreThreadRelationFilter::DirectChildrenOf( |
| ThreadId::from_string(&parent_thread_id) |
| .map_err(|err| invalid_request(format!("invalid parent thread id: {err}")))?, |
| )), |
| (None, Some(ancestor_thread_id)) => Some(StoreThreadRelationFilter::DescendantsOf( |
| ThreadId::from_string(&ancestor_thread_id) |
| .map_err(|err| invalid_request(format!("invalid ancestor thread id: {err}")))?, |
| )), |
| (None, None) => None, |
| }; |
|
|
| let requested_page_size = limit |
| .map(|value| value as usize) |
| .unwrap_or(THREAD_LIST_DEFAULT_LIMIT) |
| .clamp(1, THREAD_LIST_MAX_LIMIT); |
| let store_sort_key = match sort_key.unwrap_or(ThreadSortKey::CreatedAt) { |
| ThreadSortKey::CreatedAt => StoreThreadSortKey::CreatedAt, |
| ThreadSortKey::UpdatedAt => StoreThreadSortKey::UpdatedAt, |
| ThreadSortKey::RecencyAt => StoreThreadSortKey::RecencyAt, |
| ThreadSortKey::SectionPosition => StoreThreadSortKey::SectionPosition, |
| }; |
| let sort_direction = sort_direction.unwrap_or(match store_sort_key { |
| StoreThreadSortKey::SectionPosition => SortDirection::Asc, |
| StoreThreadSortKey::CreatedAt |
| | StoreThreadSortKey::UpdatedAt |
| | StoreThreadSortKey::RecencyAt => SortDirection::Desc, |
| }); |
| let (stored_threads, next_cursor) = self |
| .list_threads_common( |
| requested_page_size, |
| cursor, |
| store_sort_key, |
| sort_direction, |
| ThreadListFilters { |
| model_providers, |
| source_kinds, |
| archived: archived.unwrap_or(false), |
| section_id, |
| project_id, |
| cwd_filters, |
| search_term, |
| use_state_db_only, |
| relation_filter, |
| }, |
| ) |
| .await?; |
| let backwards_cursor = stored_threads.first().and_then(|thread| { |
| thread_backwards_cursor_for_sort_key(thread, store_sort_key, sort_direction) |
| }); |
| let mut data = Vec::with_capacity(stored_threads.len()); |
| let fallback_provider = self.config.model_provider_id.clone(); |
|
|
| for stored_thread in stored_threads { |
| let (thread, _) = thread_from_stored_thread( |
| stored_thread, |
| fallback_provider.as_str(), |
| &self.config.cwd, |
| ); |
| data.push(thread); |
| } |
|
|
| enrich_loaded_threads( |
| &self.thread_manager, |
| &self.thread_watch_manager, |
| data.as_mut_slice(), |
| |thread| thread, |
| ) |
| .await; |
| Ok(ThreadListResponse { |
| data, |
| next_cursor, |
| backwards_cursor, |
| }) |
| } |
|
|
| async fn thread_search_response_inner( |
| &self, |
| params: ThreadSearchParams, |
| ) -> Result<ThreadSearchResponse, JSONRPCErrorError> { |
| let ThreadSearchParams { |
| cursor, |
| limit, |
| sort_key, |
| sort_direction, |
| source_kinds, |
| archived, |
| search_term, |
| } = params; |
| let search_term = search_term.trim().to_string(); |
| let search_term = (!search_term.is_empty()) |
| .then_some(search_term) |
| .ok_or_else(|| invalid_request("thread/search requires a non-empty searchTerm"))?; |
| let requested_page_size = limit |
| .map(|value| value as usize) |
| .unwrap_or(THREAD_LIST_DEFAULT_LIMIT) |
| .clamp(1, THREAD_LIST_MAX_LIMIT); |
| let store_sort_key = match sort_key.unwrap_or(ThreadSearchSortKey::CreatedAt) { |
| ThreadSearchSortKey::CreatedAt => StoreThreadSortKey::CreatedAt, |
| ThreadSearchSortKey::UpdatedAt => StoreThreadSortKey::UpdatedAt, |
| ThreadSearchSortKey::RecencyAt => StoreThreadSortKey::RecencyAt, |
| }; |
| let store_sort_direction = sort_direction.unwrap_or(SortDirection::Desc); |
| let (allowed_sources, source_kind_filter) = compute_source_filters(source_kinds); |
| let mut cursor_obj = cursor; |
| let mut last_cursor = cursor_obj.clone(); |
| let mut remaining = requested_page_size; |
| let mut search_results = Vec::with_capacity(requested_page_size); |
| let mut next_cursor = None; |
|
|
| while remaining > 0 { |
| let page = self |
| .thread_store |
| .search_threads(StoreSearchThreadsParams { |
| page_size: remaining.min(THREAD_LIST_MAX_LIMIT), |
| cursor: cursor_obj.clone(), |
| sort_key: store_sort_key, |
| sort_direction: match store_sort_direction { |
| SortDirection::Asc => StoreSortDirection::Asc, |
| SortDirection::Desc => StoreSortDirection::Desc, |
| }, |
| allowed_sources: allowed_sources.clone(), |
| archived: archived.unwrap_or(false), |
| search_term: search_term.clone(), |
| }) |
| .await |
| .map_err(thread_store_list_error)?; |
|
|
| for result in page.items { |
| let source = with_thread_spawn_agent_metadata( |
| result.thread.source.clone(), |
| result.thread.agent_nickname.clone(), |
| result.thread.agent_role.clone(), |
| ); |
| if source_kind_filter |
| .as_ref() |
| .is_none_or(|filter| source_kind_matches(&source, filter)) |
| { |
| search_results.push(result); |
| if search_results.len() >= requested_page_size { |
| break; |
| } |
| } |
| } |
|
|
| remaining = requested_page_size.saturating_sub(search_results.len()); |
| next_cursor = page.next_cursor; |
| if remaining == 0 { |
| break; |
| } |
|
|
| let Some(cursor_val) = next_cursor.clone() else { |
| break; |
| }; |
| if last_cursor.as_ref() == Some(&cursor_val) { |
| next_cursor = None; |
| break; |
| } |
| last_cursor = Some(cursor_val.clone()); |
| cursor_obj = Some(cursor_val); |
| } |
|
|
| let backwards_cursor = search_results.first().and_then(|result| { |
| thread_backwards_cursor_for_sort_key( |
| &result.thread, |
| store_sort_key, |
| store_sort_direction, |
| ) |
| }); |
| let fallback_provider = self.config.model_provider_id.clone(); |
| let mut data = Vec::with_capacity(search_results.len()); |
| for result in search_results { |
| let (thread, _) = thread_from_stored_thread( |
| result.thread, |
| fallback_provider.as_str(), |
| &self.config.cwd, |
| ); |
| data.push(ThreadSearchResult { |
| thread, |
| snippet: result.snippet, |
| }); |
| } |
|
|
| enrich_loaded_threads( |
| &self.thread_manager, |
| &self.thread_watch_manager, |
| data.as_mut_slice(), |
| |result| &mut result.thread, |
| ) |
| .await; |
|
|
| Ok(ThreadSearchResponse { |
| data, |
| next_cursor, |
| backwards_cursor, |
| }) |
| } |
|
|
| async fn thread_loaded_list_response_inner( |
| &self, |
| params: ThreadLoadedListParams, |
| ) -> Result<ThreadLoadedListResponse, JSONRPCErrorError> { |
| let ThreadLoadedListParams { cursor, limit } = params; |
| let mut data: Vec<String> = self |
| .thread_manager |
| .list_thread_ids() |
| .await |
| .into_iter() |
| .map(|thread_id| thread_id.to_string()) |
| .collect(); |
|
|
| if data.is_empty() { |
| return Ok(ThreadLoadedListResponse { |
| data, |
| next_cursor: None, |
| }); |
| } |
|
|
| data.sort(); |
| let total = data.len(); |
| let start = match cursor { |
| Some(cursor) => { |
| let cursor = match ThreadId::from_string(&cursor) { |
| Ok(id) => id.to_string(), |
| Err(_) => return Err(invalid_request(format!("invalid cursor: {cursor}"))), |
| }; |
| match data.binary_search(&cursor) { |
| Ok(idx) => idx + 1, |
| Err(idx) => idx, |
| } |
| } |
| None => 0, |
| }; |
|
|
| let effective_limit = limit.unwrap_or(total as u32).max(1) as usize; |
| let end = start.saturating_add(effective_limit).min(total); |
| let page = data[start..end].to_vec(); |
| let next_cursor = page.last().filter(|_| end < total).cloned(); |
|
|
| Ok(ThreadLoadedListResponse { |
| data: page, |
| next_cursor, |
| }) |
| } |
|
|
| async fn thread_read_response_inner( |
| &self, |
| params: ThreadReadParams, |
| ) -> Result<ThreadReadResponse, JSONRPCErrorError> { |
| let ThreadReadParams { |
| thread_id, |
| include_turns, |
| } = params; |
|
|
| let thread_uuid = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
|
|
| let thread = self |
| .read_thread_view(thread_uuid, include_turns) |
| .await |
| .map_err(thread_read_view_error)?; |
| Ok(ThreadReadResponse { thread }) |
| } |
|
|
| |
| async fn read_thread_view( |
| &self, |
| thread_id: ThreadId, |
| include_turns: bool, |
| ) -> Result<Thread, ThreadReadViewError> { |
| |
| let pending_daybreak_enabled = self |
| .thread_store |
| .read_pending_thread_metadata(thread_id) |
| .await |
| .map_err(|err| { |
| ThreadReadViewError::Internal(format!( |
| "failed to read pending thread metadata: {err}" |
| )) |
| })? |
| .and_then(|metadata| metadata.daybreak_enabled); |
| let loaded_thread = self.thread_manager.get_thread(thread_id).await.ok(); |
| let mut thread = if include_turns { |
| if let Some(loaded_thread) = loaded_thread.as_ref() { |
| |
| |
| let persisted_thread = self |
| .load_persisted_thread_for_read(thread_id, false) |
| .await?; |
| self.load_live_thread_view( |
| thread_id, |
| include_turns, |
| loaded_thread, |
| persisted_thread, |
| ) |
| .await? |
| } else if let Some(thread) = self |
| .load_persisted_thread_for_read(thread_id, include_turns) |
| .await? |
| { |
| |
| |
| thread |
| } else { |
| return Err(ThreadReadViewError::InvalidRequest(format!( |
| "thread not loaded: {thread_id}" |
| ))); |
| } |
| } else if let Some(thread) = self |
| .load_persisted_thread_for_read(thread_id, include_turns) |
| .await? |
| { |
| if let Some(loaded_thread) = loaded_thread.as_ref() { |
| self.load_live_thread_view(thread_id, include_turns, loaded_thread, Some(thread)) |
| .await? |
| } else { |
| thread |
| } |
| } else if let Some(loaded_thread) = loaded_thread.as_ref() { |
| |
| |
| self.load_live_thread_view( |
| thread_id, |
| include_turns, |
| loaded_thread, |
| None, |
| ) |
| .await? |
| } else { |
| return Err(ThreadReadViewError::InvalidRequest(format!( |
| "thread not loaded: {thread_id}" |
| ))); |
| }; |
|
|
| thread.daybreak_enabled = thread.daybreak_enabled.or(pending_daybreak_enabled); |
|
|
| let has_live_in_progress_turn = if let Some(loaded_thread) = loaded_thread.as_ref() { |
| matches!(loaded_thread.agent_status().await, AgentStatus::Running) |
| } else { |
| false |
| }; |
|
|
| let thread_status = self |
| .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, |
| ); |
| Ok(thread) |
| } |
|
|
| async fn load_persisted_thread_for_read( |
| &self, |
| thread_id: ThreadId, |
| include_turns: bool, |
| ) -> Result<Option<Thread>, ThreadReadViewError> { |
| let fallback_provider = self.config.model_provider_id.as_str(); |
| if include_turns { |
| let Some(stored_thread) = self |
| .read_stored_thread_for_read(thread_id, false) |
| .await? |
| else { |
| return Ok(None); |
| }; |
| if matches!(stored_thread.history_mode, ThreadHistoryMode::Paginated) { |
| let (mut thread, _) = |
| thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); |
| thread.turns = self |
| .paginated_thread_full_turns(thread_id) |
| .await |
| .map_err(ThreadReadViewError::JsonRpc)?; |
| return Ok(Some(thread)); |
| } |
| } |
| let Some(stored_thread) = self |
| .read_stored_thread_for_read(thread_id, include_turns) |
| .await? |
| else { |
| return Ok(None); |
| }; |
| let (mut thread, history) = |
| thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); |
| if include_turns && let Some(history) = history { |
| thread.turns = build_legacy_api_turns_from_rollout_items(&history.items); |
| } |
| Ok(Some(thread)) |
| } |
|
|
| async fn read_stored_thread_for_read( |
| &self, |
| thread_id: ThreadId, |
| include_history: bool, |
| ) -> Result<Option<StoredThread>, ThreadReadViewError> { |
| match self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id, |
| include_archived: true, |
| include_history, |
| }) |
| .await |
| { |
| Ok(stored_thread) => Ok(Some(stored_thread)), |
| Err(ThreadStoreError::InvalidRequest { message }) |
| if message == format!("no rollout found for thread id {thread_id}") => |
| { |
| Ok(None) |
| } |
| Err(ThreadStoreError::ThreadNotFound { |
| thread_id: missing_thread_id, |
| }) if missing_thread_id == thread_id => Ok(None), |
| Err(ThreadStoreError::InvalidRequest { message }) => { |
| Err(ThreadReadViewError::InvalidRequest(message)) |
| } |
| Err(ThreadStoreError::Unsupported { operation }) => { |
| Err(ThreadReadViewError::Unsupported(operation)) |
| } |
| Err(err) => Err(ThreadReadViewError::Internal(format!( |
| "failed to read thread: {err}" |
| ))), |
| } |
| } |
|
|
| |
| async fn load_live_thread_view( |
| &self, |
| thread_id: ThreadId, |
| include_turns: bool, |
| loaded_thread: &CodexThread, |
| persisted_thread: Option<Thread>, |
| ) -> Result<Thread, ThreadReadViewError> { |
| let config_snapshot = loaded_thread.config_snapshot().await; |
| if include_turns && config_snapshot.ephemeral { |
| return Err(ThreadReadViewError::InvalidRequest( |
| "ephemeral threads do not support includeTurns".to_string(), |
| )); |
| } |
| let fallback_thread = |
| build_thread_from_loaded_snapshot(thread_id, &config_snapshot, loaded_thread); |
| let mut thread = if let Some(mut thread) = persisted_thread { |
| if thread.path.is_none() { |
| thread.path = fallback_thread.path.clone(); |
| } |
| thread.session_id.clone_from(&fallback_thread.session_id); |
| thread.ephemeral = fallback_thread.ephemeral; |
| thread.can_accept_direct_input = fallback_thread.can_accept_direct_input; |
| thread |
| } else { |
| fallback_thread |
| }; |
| apply_live_thread_settings(&mut thread, &config_snapshot); |
| self.apply_thread_read_store_fields(thread_id, &mut thread, include_turns, loaded_thread) |
| .await?; |
| Ok(thread) |
| } |
|
|
| async fn apply_thread_read_store_fields( |
| &self, |
| thread_id: ThreadId, |
| thread: &mut Thread, |
| include_turns: bool, |
| loaded_thread: &CodexThread, |
| ) -> Result<(), ThreadReadViewError> { |
| self.attach_thread_name(thread_id, thread).await; |
|
|
| if include_turns { |
| if matches!( |
| thread.history_mode, |
| codex_app_server_protocol::ThreadHistoryMode::Paginated |
| ) { |
| self.thread_store |
| .persist_thread(thread_id, PersistContext::Standard) |
| .await |
| .map_err(|err| thread_read_history_load_error(thread_id, err))?; |
| thread.turns = self |
| .paginated_thread_full_turns(thread_id) |
| .await |
| .map_err(ThreadReadViewError::JsonRpc)?; |
| return Ok(()); |
| } |
| let history = loaded_thread |
| .load_history( true) |
| .await |
| .map_err(|err| thread_read_history_load_error(thread_id, err))?; |
| thread.turns = build_legacy_api_turns_from_rollout_items(&history.items); |
| } |
|
|
| Ok(()) |
| } |
|
|
| async fn thread_turns_list_response_inner( |
| &self, |
| params: ThreadTurnsListParams, |
| ) -> Result<ThreadTurnsListResponse, JSONRPCErrorError> { |
| let ThreadTurnsListParams { |
| thread_id, |
| cursor, |
| limit, |
| sort_direction, |
| items_view, |
| } = params; |
| let thread_uuid = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
| match self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id: thread_uuid, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| { |
| Ok(thread) if thread.history_mode == ThreadHistoryMode::Paginated => { |
| return self |
| .paginated_thread_turns_list_response( |
| thread_uuid, |
| cursor, |
| limit, |
| sort_direction, |
| items_view, |
| ) |
| .await; |
| } |
| Ok(_) => {} |
| Err(ThreadStoreError::InvalidRequest { message }) |
| if message == format!("no rollout found for thread id {thread_uuid}") => {} |
| Err(ThreadStoreError::ThreadNotFound { thread_id }) if thread_id == thread_uuid => {} |
| Err(ThreadStoreError::InvalidRequest { message }) => { |
| return Err(invalid_request(message)); |
| } |
| Err(ThreadStoreError::Unsupported { operation }) => { |
| return Err(unsupported_thread_store_operation(operation)); |
| } |
| Err(err) => return Err(internal_error(format!("failed to read thread: {err}"))), |
| } |
|
|
| let items = self |
| .load_thread_turns_list_history(thread_uuid) |
| .await |
| .map_err(thread_read_view_error)?; |
| |
| |
| |
| |
| |
| let loaded_thread = self.thread_manager.get_thread(thread_uuid).await.ok(); |
| let has_live_running_thread = match loaded_thread.as_ref() { |
| Some(thread) => matches!(thread.agent_status().await, AgentStatus::Running), |
| None => false, |
| }; |
| let active_turn = if loaded_thread.is_some() { |
| |
| |
| |
| let thread_state = self.thread_state_manager.thread_state(thread_uuid).await; |
| let state = thread_state.lock().await; |
| state.active_turn_snapshot() |
| } else { |
| None |
| }; |
| build_thread_turns_page_response( |
| &items, |
| self.thread_watch_manager |
| .loaded_status_for_thread(&thread_uuid.to_string()) |
| .await, |
| has_live_running_thread, |
| active_turn, |
| ThreadTurnsPageOptions { |
| cursor: cursor.as_deref(), |
| limit, |
| sort_direction: sort_direction.unwrap_or(SortDirection::Desc), |
| items_view: items_view.unwrap_or(TurnItemsView::Summary), |
| }, |
| ) |
| } |
|
|
| async fn thread_search_occurrences_response_inner( |
| &self, |
| params: ThreadSearchOccurrencesParams, |
| ) -> Result<ThreadSearchOccurrencesResponse, JSONRPCErrorError> { |
| let ThreadSearchOccurrencesParams { |
| thread_id, |
| search_term, |
| cursor, |
| limit, |
| } = params; |
| let thread_id = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
| if search_term.trim().is_empty() { |
| return Err(invalid_request( |
| "thread/searchOccurrences requires a non-empty searchTerm", |
| )); |
| } |
| let page_size = limit |
| .map(|value| value as usize) |
| .unwrap_or(THREAD_SEARCH_OCCURRENCES_DEFAULT_LIMIT) |
| .clamp(1, THREAD_SEARCH_OCCURRENCES_MAX_LIMIT); |
| let page = self |
| .thread_store |
| .search_thread_occurrences(StoreSearchThreadOccurrencesParams { |
| thread_id, |
| search_term, |
| cursor, |
| page_size, |
| }) |
| .await |
| .map_err(|err| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| ThreadStoreError::ThreadNotFound { thread_id } => { |
| invalid_request(format!("no rollout found for thread id {thread_id}")) |
| } |
| err => internal_error(format!("failed to search thread occurrences: {err}")), |
| })?; |
| Ok(ThreadSearchOccurrencesResponse { |
| data: page |
| .items |
| .into_iter() |
| .map(|item| ThreadSearchOccurrence { |
| turn_id: item.turn_id, |
| item_id: item.item_id, |
| snippet: item.snippet, |
| snippet_match_range: ThreadSearchTextRange { |
| start: item.snippet_match_range.start, |
| end: item.snippet_match_range.end, |
| }, |
| turn_cursor: item.turn_cursor, |
| }) |
| .collect(), |
| next_cursor: page.next_cursor, |
| }) |
| } |
|
|
| async fn paginated_thread_turns_list_response( |
| &self, |
| thread_id: ThreadId, |
| cursor: Option<String>, |
| limit: Option<u32>, |
| sort_direction: Option<SortDirection>, |
| items_view: Option<TurnItemsView>, |
| ) -> Result<ThreadTurnsListResponse, JSONRPCErrorError> { |
| let items_view = items_view.unwrap_or(TurnItemsView::Summary); |
| let page_size = thread_turns_page_size(limit); |
| let sort_direction = match sort_direction.unwrap_or(SortDirection::Desc) { |
| SortDirection::Asc => StoreSortDirection::Asc, |
| SortDirection::Desc => StoreSortDirection::Desc, |
| }; |
| |
| |
| let stored_items_view = match items_view { |
| TurnItemsView::NotLoaded => StoredTurnItemsView::NotLoaded, |
| TurnItemsView::Summary => StoredTurnItemsView::Summary, |
| TurnItemsView::Full => StoredTurnItemsView::NotLoaded, |
| }; |
| let page = self |
| .thread_store |
| .list_turns(StoreListTurnsParams { |
| thread_id, |
| include_archived: true, |
| cursor, |
| page_size, |
| sort_direction, |
| items_view: stored_items_view, |
| }) |
| .await |
| .map_err(|err| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| ThreadStoreError::ThreadNotFound { thread_id } => { |
| invalid_request(format!("no rollout found for thread id {thread_id}")) |
| } |
| err => internal_error(format!("failed to list thread history: {err}")), |
| })?; |
| let mut turns = Vec::with_capacity(page.turns.len()); |
| for turn in page.turns { |
| let mut turn = stored_turn_to_api_turn(turn, items_view)?; |
| if matches!(items_view, TurnItemsView::Full) { |
| turn.items = self |
| .paginated_turn_full_items(thread_id, turn.id.as_str()) |
| .await?; |
| } |
| turns.push(turn); |
| } |
| let loaded_thread = self.thread_manager.get_thread(thread_id).await.ok(); |
| let has_live_running_thread = match loaded_thread.as_ref() { |
| Some(thread) => matches!(thread.agent_status().await, AgentStatus::Running), |
| None => false, |
| }; |
| normalize_thread_turns_status( |
| &mut turns, |
| self.thread_watch_manager |
| .loaded_status_for_thread(&thread_id.to_string()) |
| .await, |
| has_live_running_thread, |
| ); |
| Ok(ThreadTurnsListResponse { |
| data: turns, |
| next_cursor: page.next_cursor, |
| backwards_cursor: page.backwards_cursor, |
| }) |
| } |
|
|
| |
| |
| async fn paginated_turn_full_items( |
| &self, |
| thread_id: ThreadId, |
| turn_id: &str, |
| ) -> Result<Vec<ThreadItem>, JSONRPCErrorError> { |
| let mut cursor = None; |
| let mut items = Vec::new(); |
| loop { |
| let page = self |
| .thread_store |
| .list_items(StoreListItemsParams { |
| thread_id, |
| turn_id: Some(turn_id.to_string()), |
| include_archived: true, |
| cursor: cursor.clone(), |
| page_size: THREAD_ITEMS_MAX_LIMIT, |
| sort_direction: StoreSortDirection::Asc, |
| sort_key: StoreItemSortKey::CreatedAtOrdinal, |
| after_updated_at_ordinal: None, |
| }) |
| .await |
| .map_err(paginated_history_list_error)?; |
| for item in page.items { |
| items.push(deserialize_stored_thread_item(item)?); |
| } |
| let Some(next_cursor) = page.next_cursor else { |
| return Ok(items); |
| }; |
| if cursor.as_ref() == Some(&next_cursor) { |
| return Err(internal_error(format!( |
| "failed to load full turn items for {turn_id}: thread store returned a repeated cursor" |
| ))); |
| } |
| cursor = Some(next_cursor); |
| } |
| } |
|
|
| |
| |
| async fn paginated_thread_full_turns( |
| &self, |
| thread_id: ThreadId, |
| ) -> Result<Vec<Turn>, JSONRPCErrorError> { |
| let mut cursor = None; |
| let mut turns = Vec::new(); |
| loop { |
| let page = self |
| .paginated_thread_turns_list_response( |
| thread_id, |
| cursor.clone(), |
| Some(THREAD_TURNS_MAX_LIMIT as u32), |
| Some(SortDirection::Asc), |
| Some(TurnItemsView::Full), |
| ) |
| .await?; |
| turns.extend(page.data); |
| let Some(next_cursor) = page.next_cursor else { |
| return Ok(turns); |
| }; |
| if cursor.as_ref() == Some(&next_cursor) { |
| return Err(internal_error(format!( |
| "failed to load full thread turns for {thread_id}: thread store returned a repeated cursor" |
| ))); |
| } |
| cursor = Some(next_cursor); |
| } |
| } |
|
|
| async fn paginated_resume_initial_turns_page( |
| &self, |
| thread_id: ThreadId, |
| params: &ThreadResumeInitialTurnsPageParams, |
| ) -> Result<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> { |
| self.paginated_thread_turns_list_response( |
| thread_id, |
| None, |
| params.limit, |
| params.sort_direction, |
| params.items_view, |
| ) |
| .await |
| .map(Into::into) |
| } |
|
|
| async fn paginated_resume_initial_turns_page_with_active_slot( |
| &self, |
| thread_id: ThreadId, |
| params: &ThreadResumeInitialTurnsPageParams, |
| ) -> Result<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> { |
| |
| |
| |
| let page_size = thread_turns_page_size(params.limit); |
| if page_size == 1 { |
| |
| |
| let mut page = self |
| .paginated_resume_initial_turns_page(thread_id, params) |
| .await?; |
| page.next_cursor = page.backwards_cursor.clone(); |
| page.data.clear(); |
| return Ok(page); |
| } |
|
|
| let mut params = params.clone(); |
| params.limit = Some((page_size - 1) as u32); |
| self.paginated_resume_initial_turns_page(thread_id, ¶ms) |
| .await |
| } |
|
|
| pub(super) async fn paginated_resume_backwards_cursors( |
| thread_store: &dyn ThreadStore, |
| thread_id: ThreadId, |
| ) -> Result<(Option<String>, Option<String>), JSONRPCErrorError> { |
| let turns_page = thread_store |
| .list_turns(StoreListTurnsParams { |
| thread_id, |
| include_archived: true, |
| cursor: None, |
| page_size: 1, |
| sort_direction: StoreSortDirection::Desc, |
| items_view: StoredTurnItemsView::NotLoaded, |
| }) |
| .await |
| .map_err(paginated_history_list_error)?; |
| let items_page = thread_store |
| .list_items(StoreListItemsParams { |
| thread_id, |
| turn_id: None, |
| include_archived: true, |
| cursor: None, |
| page_size: 1, |
| sort_direction: StoreSortDirection::Desc, |
| sort_key: StoreItemSortKey::CreatedAtOrdinal, |
| after_updated_at_ordinal: None, |
| }) |
| .await |
| .map_err(paginated_history_list_error)?; |
| Ok((turns_page.backwards_cursor, items_page.backwards_cursor)) |
| } |
|
|
| async fn thread_items_list_response_inner( |
| &self, |
| params: ThreadItemsListParams, |
| ) -> Result<ThreadItemsListResponse, JSONRPCErrorError> { |
| let ThreadItemsListParams { |
| thread_id, |
| turn_id, |
| cursor, |
| limit, |
| sort_direction, |
| } = params; |
| let thread_id = ThreadId::from_string(&thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
| let page_size = limit |
| .map(|value| value as usize) |
| .unwrap_or(THREAD_ITEMS_DEFAULT_LIMIT) |
| .clamp(1, THREAD_ITEMS_MAX_LIMIT); |
| let page = self |
| .thread_store |
| .list_items(StoreListItemsParams { |
| thread_id, |
| turn_id, |
| include_archived: true, |
| cursor, |
| page_size, |
| sort_direction: match sort_direction.unwrap_or(SortDirection::Asc) { |
| SortDirection::Asc => StoreSortDirection::Asc, |
| SortDirection::Desc => StoreSortDirection::Desc, |
| }, |
| sort_key: StoreItemSortKey::CreatedAtOrdinal, |
| after_updated_at_ordinal: None, |
| }) |
| .await |
| .map_err(|err| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::Unsupported { .. } => { |
| method_not_found("thread/items/list is not supported yet") |
| } |
| ThreadStoreError::ThreadNotFound { thread_id } => { |
| invalid_request(format!("no rollout found for thread id {thread_id}")) |
| } |
| err => internal_error(format!("failed to list thread items: {err}")), |
| })?; |
| let data = page |
| .items |
| .into_iter() |
| .map(|stored_item| { |
| let turn_id = stored_item.turn_id.clone(); |
| let item = deserialize_stored_thread_item(stored_item)?; |
| Ok(ThreadItemEntry { turn_id, item }) |
| }) |
| .collect::<Result<Vec<_>, _>>()?; |
|
|
| Ok(ThreadItemsListResponse { |
| data, |
| next_cursor: page.next_cursor, |
| backwards_cursor: page.backwards_cursor, |
| }) |
| } |
|
|
| async fn load_thread_turns_list_history( |
| &self, |
| thread_id: ThreadId, |
| ) -> Result<Vec<RolloutItem>, ThreadReadViewError> { |
| match self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id, |
| include_archived: true, |
| include_history: true, |
| }) |
| .await |
| { |
| Ok(stored_thread) => { |
| let history = stored_thread.history.ok_or_else(|| { |
| ThreadReadViewError::Internal(format!( |
| "thread store did not return history for thread {thread_id}" |
| )) |
| })?; |
| return Ok(history.items); |
| } |
| Err(ThreadStoreError::InvalidRequest { message }) |
| if message == format!("no rollout found for thread id {thread_id}") => {} |
| Err(ThreadStoreError::ThreadNotFound { |
| thread_id: missing_thread_id, |
| }) if missing_thread_id == thread_id => {} |
| Err(ThreadStoreError::InvalidRequest { message }) => { |
| return Err(ThreadReadViewError::InvalidRequest(message)); |
| } |
| Err(ThreadStoreError::Unsupported { operation }) => { |
| return Err(ThreadReadViewError::Unsupported(operation)); |
| } |
| Err(err) => { |
| return Err(ThreadReadViewError::Internal(format!( |
| "failed to read thread: {err}" |
| ))); |
| } |
| } |
|
|
| let thread = self |
| .thread_manager |
| .get_thread(thread_id) |
| .await |
| .map_err(|_| { |
| ThreadReadViewError::InvalidRequest(format!("thread not loaded: {thread_id}")) |
| })?; |
| let config_snapshot = thread.config_snapshot().await; |
| if config_snapshot.ephemeral { |
| return Err(ThreadReadViewError::InvalidRequest( |
| "ephemeral threads do not support thread/turns/list".to_string(), |
| )); |
| } |
|
|
| thread |
| .load_history( true) |
| .await |
| .map(|history| history.items) |
| .map_err(|err| thread_turns_list_history_load_error(thread_id, err)) |
| } |
|
|
| pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver<ThreadId> { |
| self.thread_manager.subscribe_thread_created() |
| } |
|
|
| pub(crate) async fn connection_initialized( |
| &self, |
| connection_id: ConnectionId, |
| capabilities: ConnectionCapabilities, |
| ) { |
| self.thread_state_manager |
| .connection_initialized(connection_id, capabilities) |
| .await; |
| } |
|
|
| pub(crate) async fn connection_closed(&self, connection_id: ConnectionId) { |
| let thread_ids = self |
| .thread_state_manager |
| .remove_connection(connection_id) |
| .await; |
|
|
| for thread_id in thread_ids { |
| if self.thread_manager.get_thread(thread_id).await.is_err() { |
| |
| |
| self.finalize_thread_teardown(thread_id).await; |
| } |
| } |
| } |
|
|
| pub(crate) fn subscribe_running_assistant_turn_count(&self) -> watch::Receiver<usize> { |
| self.thread_watch_manager.subscribe_running_turn_count() |
| } |
|
|
| |
| pub(crate) async fn try_attach_thread_listener( |
| &self, |
| thread_id: ThreadId, |
| connection_ids: Vec<ConnectionId>, |
| ) { |
| let mut raw_events_enabled = false; |
| if let Ok(thread) = self.thread_manager.get_thread(thread_id).await { |
| let config_snapshot = thread.config_snapshot().await; |
| self.thread_watch_manager |
| .upsert_thread(&thread_id.to_string()) |
| .await; |
| if let Some(parent_thread_id) = config_snapshot.parent_thread_id { |
| raw_events_enabled = self |
| .thread_state_manager |
| .thread_state(parent_thread_id) |
| .await |
| .lock() |
| .await |
| .experimental_raw_events; |
| } |
| } |
|
|
| for connection_id in connection_ids { |
| log_listener_attach_result( |
| self.ensure_conversation_listener(thread_id, connection_id, raw_events_enabled) |
| .await, |
| thread_id, |
| connection_id, |
| "thread", |
| ); |
| } |
| } |
|
|
| async fn thread_resume_inner( |
| &self, |
| target: &ThreadResumeTarget, |
| params: &ThreadResumeParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| client_mcp_extensions: ClientMcpExtensions, |
| prepared_config: &mut Option<PreparedResumeConfig>, |
| ) -> Result<ControlFlow<()>, JSONRPCErrorError> { |
| if let Ok(thread_id) = ThreadId::from_string(¶ms.thread_id) |
| && self |
| .pending_thread_unloads |
| .lock() |
| .await |
| .contains(&thread_id) |
| { |
| return Err(invalid_request(format!( |
| "thread {thread_id} is closing; retry thread/resume after the thread is closed" |
| ))); |
| } |
|
|
| if params.sandbox.is_some() && params.permissions.is_some() { |
| return Err(invalid_request( |
| "`permissions` cannot be combined with `sandbox`", |
| )); |
| } |
| let redact_resume_payloads = |
| should_redact_thread_resume_payloads(app_server_client_name.as_deref()); |
|
|
| let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; |
| let stored_thread_from_running_probe = match target { |
| ThreadResumeTarget::Client(request_id) => match self |
| .resume_running_thread( |
| request_id, |
| params, |
| app_server_client_name.clone(), |
| app_server_client_version.clone(), |
| None, |
| ) |
| .await? |
| { |
| RunningThreadResumeResult::Handled(completion) => { |
| |
| drop(_thread_list_state_permit); |
| let _ = completion.await; |
| return Ok(ControlFlow::Break(())); |
| } |
| RunningThreadResumeResult::NotRunning(stored_thread) => stored_thread, |
| }, |
| ThreadResumeTarget::DaemonRecovery(saved) => { |
| let thread_id = ThreadId::from_string(¶ms.thread_id) |
| .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; |
| |
| if self.thread_manager.get_thread(thread_id).await.is_ok() { |
| if let Some(saved) = saved { |
| self.continue_daemon_turn(¶ms.thread_id, saved.clone()) |
| .await; |
| } |
| return Ok(ControlFlow::Break(())); |
| } |
| None |
| } |
| }; |
|
|
| let ThreadResumeParams { |
| thread_id, |
| history, |
| path, |
| model, |
| model_provider, |
| service_tier, |
| cwd, |
| runtime_workspace_roots, |
| approval_policy, |
| approvals_reviewer, |
| sandbox, |
| permissions, |
| config: mut request_overrides, |
| base_instructions, |
| developer_instructions, |
| personality, |
| exclude_turns, |
| initial_turns_page, |
| } = params.clone(); |
| let include_turns = !exclude_turns; |
|
|
| let resume_result = if let Some(history) = history { |
| self.resume_thread_from_history(history.as_slice()) |
| .await |
| .map(|thread_history| (thread_history, None)) |
| } else if let Some(stored_thread) = stored_thread_from_running_probe { |
| self.load_resume_initial_history_from_stored_thread(*stored_thread) |
| .await |
| .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))) |
| } else { |
| match self |
| .read_stored_thread_for_resume( |
| &thread_id, |
| path.as_ref(), |
| false, |
| ) |
| .await |
| { |
| Ok(stored_thread) => self |
| .load_resume_initial_history_from_stored_thread(stored_thread) |
| .await |
| .map(|(thread_history, stored_thread)| (thread_history, Some(stored_thread))), |
| Err(error) => Err(error), |
| } |
| }; |
| let (thread_history, resume_source_thread) = resume_result?; |
| 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 |
| && prepared_config.is_none() |
| && let ThreadResumeTarget::Client(request_id) = target |
| { |
| self.send_deprecation_notice( |
| request_id.connection_id, |
| PAGINATED_FULL_HISTORY_DEPRECATION_SUMMARY, |
| ) |
| .await; |
| } |
|
|
| |
| if let InitialHistory::Resumed(resumed_history) = &thread_history |
| && let Some((source, _)) = thread_history.get_resumed_session_sources() |
| && !can_accept_direct_input(thread_history.get_multi_agent_version(), &source) |
| { |
| let ThreadResumeTarget::Client(request_id) = target else { |
| return Ok(ControlFlow::Break(())); |
| }; |
| let child_thread_id = resumed_history.conversation_id; |
| self.thread_manager |
| .ensure_multi_agent_v2_child_loaded(child_thread_id) |
| .await |
| .map_err(|err| { |
| tracing::warn!( |
| thread_id = %child_thread_id, |
| error = %err, |
| "failed to resume a multi-agent v2 child through its parent" |
| ); |
| 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", |
| ) |
| })?; |
|
|
| let cold_resume_history = paginated_resume.then(|| thread_history.get_rollout_items()); |
| |
| let attach_params = ThreadResumeParams { |
| thread_id: child_thread_id.to_string(), |
| exclude_turns, |
| initial_turns_page, |
| ..Default::default() |
| }; |
| return match self |
| .resume_running_thread( |
| request_id, |
| &attach_params, |
| app_server_client_name, |
| app_server_client_version, |
| cold_resume_history, |
| ) |
| .await? |
| { |
| RunningThreadResumeResult::Handled(completion) => { |
| drop(_thread_list_state_permit); |
| let _ = completion.await; |
| 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", |
| )), |
| }; |
| } |
|
|
| |
| |
| let history_settings = if let InitialHistory::Resumed(resumed) = &thread_history { |
| resumed.history.iter().rev().find_map(|item| match item { |
| RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(event)) |
| if event.thread_id == Some(resumed.conversation_id) => |
| { |
| Some(&event.thread_settings) |
| } |
| _ => None, |
| }) |
| } else { |
| None |
| }; |
| let history_cwd = history_settings |
| .map(|settings| settings.cwd.to_path_buf()) |
| .or_else(|| thread_history.session_cwd()); |
| let mut runtime_workspace_roots = |
| runtime_workspace_roots.map(resolve_runtime_workspace_roots); |
| if runtime_workspace_roots.is_none() { |
| |
| |
| let saved_workspaces: Option<(&Path, Vec<&Path>)> = match history_settings { |
| Some(settings) => settings.runtime_workspace_roots.as_ref().map(|roots| { |
| ( |
| settings.cwd.as_path(), |
| roots.iter().map(AbsolutePathBuf::as_path).collect(), |
| ) |
| }), |
| None => { |
| if let InitialHistory::Resumed(resumed) = &thread_history |
| && let Some(RolloutItem::SessionMeta(meta)) = resumed.history.first() |
| && meta.meta.id == resumed.conversation_id |
| && let Some(roots) = &meta.meta.runtime_workspace_roots |
| { |
| Some(( |
| meta.meta.cwd.as_path(), |
| roots.iter().map(PathBuf::as_path).collect(), |
| )) |
| } else { |
| None |
| } |
| } |
| }; |
| if let Some((saved_cwd, saved_roots)) = saved_workspaces { |
| let resume_cwd = resolve_request_cwd(cwd.as_ref().map(PathBuf::from))?; |
| |
| |
| let restored_roots = path_utils::replace_path_and_deduplicate( |
| saved_roots, |
| saved_cwd, |
| resume_cwd.as_ref().map_or(saved_cwd, AbsolutePathBuf::as_path), |
| ) |
| .into_iter() |
| .map(|root| { |
| AbsolutePathBuf::from_absolute_path_checked(root).map_err(|err| { |
| invalid_params(format!( |
| "cannot restore workspace root `{}` on this host: {err}. Pass `runtimeWorkspaceRoots` with valid local paths or an empty list.", |
| root.display() |
| )) |
| }) |
| }) |
| .collect::<Result<Vec<_>, _>>()?; |
| |
| runtime_workspace_roots = Some(resolve_runtime_workspace_roots(restored_roots)); |
| } |
| } |
| let mut typesafe_overrides = self.build_thread_config_overrides( |
| model, |
| model_provider, |
| service_tier, |
| cwd, |
| runtime_workspace_roots, |
| approval_policy, |
| approvals_reviewer, |
| sandbox, |
| permissions, |
| base_instructions, |
| developer_instructions, |
| personality, |
| ); |
| if typesafe_overrides.approval_policy.is_none() |
| && let Some(value) = request_overrides |
| .as_mut() |
| .and_then(|overrides| overrides.remove("approval_policy")) |
| { |
| let approval_policy = serde_json::from_value(value).map_err(|err| { |
| invalid_params(format!("invalid `approval_policy` config override: {err}")) |
| })?; |
| typesafe_overrides.approval_policy = Some(approval_policy); |
| } |
| let has_explicit_model_resume_override = |
| has_model_resume_override(request_overrides.as_ref(), &typesafe_overrides); |
| let persisted_metadata = self |
| .load_and_apply_persisted_resume_metadata( |
| &thread_history, |
| &mut request_overrides, |
| &mut typesafe_overrides, |
| ) |
| .await; |
|
|
| let clear_reasoning_effort = !has_explicit_model_resume_override |
| && persisted_metadata |
| .as_ref() |
| .is_some_and(|metadata| metadata.reasoning_effort.is_none()); |
| let config_state = ResumeConfigState { |
| history_cwd: history_cwd.clone(), |
| workspace_roots: typesafe_overrides.workspace_roots.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, |
| _ => { |
| |
| drop(_thread_list_state_permit); |
| let config = self |
| .config_manager |
| .load_for_cwd(request_overrides, typesafe_overrides, history_cwd) |
| .await |
| .map_err(|err| config_load_error(&err))?; |
| *prepared_config = Some(PreparedResumeConfig { |
| state: config_state, |
| config, |
| }); |
| |
| return Ok(ControlFlow::Continue(())); |
| } |
| }; |
| if clear_reasoning_effort { |
| config.model_reasoning_effort = None; |
| } |
|
|
| let response_history = thread_history.clone(); |
|
|
| match self |
| .thread_manager |
| .resume_thread_with_history( |
| config, |
| thread_history, |
| self.auth_manager.clone(), |
| match target { |
| ThreadResumeTarget::Client(request_id) => { |
| self.request_trace_context(request_id).await |
| } |
| ThreadResumeTarget::DaemonRecovery(_) => None, |
| }, |
| client_mcp_extensions, |
| ) |
| .await |
| { |
| Ok(NewThread { |
| thread_id, |
| thread: codex_thread, |
| session_configured, |
| .. |
| }) => { |
| let ThreadResumeTarget::Client(request_id) = target else { |
| |
| self.thread_watch_manager |
| .upsert_thread(&thread_id.to_string()) |
| .await; |
| |
| codex_thread |
| .emit_thread_idle_lifecycle_if_idle(ThreadIdleCause::Completed) |
| .await; |
| if let ThreadResumeTarget::DaemonRecovery(Some(saved)) = target { |
| self.continue_daemon_turn(&thread_id.to_string(), saved.clone()) |
| .await; |
| } |
| let state = self.thread_state_manager.thread_state(thread_id).await; |
| self.ensure_listener_task_running(thread_id, Arc::clone(&codex_thread), state) |
| .await?; |
| return Ok(ControlFlow::Break(())); |
| }; |
| let request_id = request_id.clone(); |
| if let Err(err) = Self::set_app_server_client_info( |
| codex_thread.as_ref(), |
| app_server_client_name, |
| app_server_client_version, |
| ) |
| .await |
| { |
| self.outgoing.send_error(request_id, err).await; |
| return Ok(ControlFlow::Break(())); |
| } |
| let instruction_sources = codex_thread.legacy_instruction_sources().await; |
| let SessionConfiguredEvent { rollout_path, .. } = session_configured; |
| let Some(rollout_path) = rollout_path else { |
| let error = |
| internal_error(format!("rollout path missing for thread {thread_id}")); |
| self.outgoing.send_error(request_id, error).await; |
| return Ok(ControlFlow::Break(())); |
| }; |
| |
| |
| |
| if paginated_resume |
| && let Err(error) = self |
| .thread_store |
| .persist_thread(thread_id, PersistContext::Standard) |
| .await |
| .map_err(thread_store_resume_read_error) |
| { |
| self.outgoing.send_error(request_id, error).await; |
| 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(ControlFlow::Break(())); |
| } |
| } |
| } else { |
| None |
| }; |
| |
| log_listener_attach_result( |
| self.ensure_conversation_listener( |
| thread_id, |
| request_id.connection_id, |
| false, |
| ) |
| .await, |
| thread_id, |
| request_id.connection_id, |
| "thread", |
| ); |
|
|
| let mut thread = match self |
| .load_thread_from_resume_source_or_send_internal( |
| thread_id, |
| codex_thread.as_ref(), |
| &response_history, |
| rollout_path.as_path(), |
| resume_source_thread, |
| include_turns && !paginated_resume, |
| ) |
| .await |
| { |
| Ok(thread) => thread, |
| Err(message) => { |
| self.outgoing |
| .send_error(request_id, internal_error(message)) |
| .await; |
| return Ok(ControlFlow::Break(())); |
| } |
| }; |
| thread.thread_source = codex_thread |
| .config_snapshot() |
| .await |
| .thread_source |
| .map(Into::into); |
| if let Some(materialized_turns) = materialized_turns { |
| thread.turns = materialized_turns; |
| } |
|
|
| self.thread_watch_manager.upsert_thread(&thread.id).await; |
|
|
| let thread_status = self |
| .thread_watch_manager |
| .loaded_status_for_thread(&thread.id) |
| .await; |
|
|
| set_thread_status_and_interrupt_stale_turns( |
| &mut thread, |
| thread_status, |
| false, |
| ); |
| let config_snapshot = codex_thread.config_snapshot().await; |
| let (turns_backwards_cursor, items_backwards_cursor) = |
| if matches!(config_snapshot.history_mode, ThreadHistoryMode::Paginated) { |
| match Self::paginated_resume_backwards_cursors( |
| self.thread_store.as_ref(), |
| thread_id, |
| ) |
| .await |
| { |
| Ok(cursors) => cursors, |
| Err(error) => { |
| self.outgoing.send_error(request_id, error).await; |
| return Ok(ControlFlow::Break(())); |
| } |
| } |
| } else { |
| (None, None) |
| }; |
| let sandbox = config_snapshot.sandbox_policy().into(); |
| let active_permission_profile = thread_response_active_permission_profile( |
| config_snapshot.active_permission_profile, |
| ); |
| let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() { |
| let initial_turns_page_result = if paginated_resume { |
| self.paginated_resume_initial_turns_page(thread_id, params) |
| .await |
| } else { |
| build_thread_resume_initial_turns_page( |
| response_history.get_rollout_items(), |
| thread.status.clone(), |
| false, |
| None, |
| params, |
| ) |
| }; |
| match initial_turns_page_result { |
| Ok(page) => Some(page), |
| Err(error) => { |
| self.outgoing.send_error(request_id, error).await; |
| return Ok(ControlFlow::Break(())); |
| } |
| } |
| } else { |
| None |
| }; |
| let token_usage_turn_id = (include_turns || paginated_resume) |
| .then(|| { |
| let turns = if thread.turns.is_empty() { |
| initial_turns_page |
| .as_ref() |
| .map_or(&[][..], |page| page.data.as_slice()) |
| } else { |
| thread.turns.as_slice() |
| }; |
| restored_token_usage_turn_id(response_history.get_rollout_items(), turns) |
| }) |
| .filter(|turn_id| !turn_id.is_empty()); |
| if redact_resume_payloads { |
| redact_thread_resume_payloads(&mut thread.turns); |
| if let Some(initial_turns_page) = initial_turns_page.as_mut() { |
| redact_thread_resume_payloads(&mut initial_turns_page.data); |
| } |
| } |
|
|
| let thread_originator = config_snapshot.originator.clone(); |
| let response = ThreadResumeResponse { |
| thread, |
| disabled_plugin_ids: config_snapshot.disabled_plugin_ids, |
| model: session_configured.model, |
| model_provider: session_configured.model_provider_id, |
| service_tier: session_configured.service_tier, |
| cwd: session_configured.cwd, |
| runtime_workspace_roots: config_snapshot.workspace_roots, |
| instruction_sources, |
| approval_policy: session_configured.approval_policy.into(), |
| approvals_reviewer: session_configured.approvals_reviewer.into(), |
| sandbox, |
| active_permission_profile, |
| reasoning_effort: session_configured.reasoning_effort, |
| collaboration_mode: Some(config_snapshot.collaboration_mode), |
| multi_agent_mode: MultiAgentMode::ExplicitRequestOnly, |
| initial_turns_page, |
| turns_backwards_cursor, |
| items_backwards_cursor, |
| }; |
|
|
| let connection_id = request_id.connection_id; |
| self.outgoing |
| .send_response_with_thread_originator(request_id, response, thread_originator) |
| .await; |
| |
| |
| if let Some(token_usage_turn_id) = token_usage_turn_id { |
| |
| |
| |
| send_thread_token_usage_update_to_connection( |
| &self.outgoing, |
| connection_id, |
| thread_id, |
| codex_thread.as_ref(), |
| token_usage_turn_id, |
| ) |
| .await; |
| } |
| self.thread_goal_processor |
| .emit_resume_goal_snapshot(thread_id) |
| .await; |
| codex_thread |
| .emit_thread_idle_lifecycle_if_idle(ThreadIdleCause::Completed) |
| .await; |
| } |
| Err(err) => { |
| let error = match err.details() { |
| CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()), |
| _ => internal_error(format!("error resuming thread: {err}")), |
| }; |
| return Err(error); |
| } |
| } |
| Ok(ControlFlow::Break(())) |
| } |
|
|
| async fn load_and_apply_persisted_resume_metadata( |
| &self, |
| thread_history: &InitialHistory, |
| request_overrides: &mut Option<HashMap<String, serde_json::Value>>, |
| typesafe_overrides: &mut ConfigOverrides, |
| ) -> Option<ThreadMetadata> { |
| let InitialHistory::Resumed(resumed_history) = thread_history else { |
| return None; |
| }; |
| if let Some(persisted_settings) = latest_persisted_resume_settings(&resumed_history.history) |
| { |
| if typesafe_overrides.approval_policy.is_none() { |
| typesafe_overrides.approval_policy = Some(persisted_settings.approval_policy); |
| } |
| if typesafe_overrides.approvals_reviewer.is_none() |
| && !request_overrides |
| .as_ref() |
| .is_some_and(|overrides| overrides.contains_key("approvals_reviewer")) |
| { |
| typesafe_overrides.approvals_reviewer = persisted_settings.approvals_reviewer; |
| } |
| if !has_permission_override(request_overrides.as_ref(), typesafe_overrides) { |
| typesafe_overrides.persisted_permission_profile_id = persisted_settings |
| .active_permission_profile |
| .map(|profile| profile.id); |
| } |
| } |
| let state_db_ctx = self.state_db.clone()?; |
| let persisted_metadata = state_db_ctx |
| .get_thread(resumed_history.conversation_id) |
| .await |
| .ok() |
| .flatten()?; |
| merge_persisted_resume_metadata(request_overrides, typesafe_overrides, &persisted_metadata); |
| Some(persisted_metadata) |
| } |
|
|
| #[tracing::instrument(level = "trace", skip_all)] |
| async fn resume_running_thread( |
| &self, |
| request_id: &ConnectionRequestId, |
| params: &ThreadResumeParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| cold_resume_history: Option<&[RolloutItem]>, |
| ) -> Result<RunningThreadResumeResult, JSONRPCErrorError> { |
| let running_thread = if params.history.is_some() { |
| if let Ok(existing_thread_id) = ThreadId::from_string(¶ms.thread_id) |
| && self |
| .thread_manager |
| .get_thread(existing_thread_id) |
| .await |
| .is_ok() |
| { |
| return Err(invalid_request(format!( |
| "cannot resume thread {existing_thread_id} with history while it is already running" |
| ))); |
| } |
| None |
| } else if let Ok(existing_thread_id) = ThreadId::from_string(¶ms.thread_id) |
| && let Ok(existing_thread) = self.thread_manager.get_thread(existing_thread_id).await |
| { |
| let source_thread = self |
| .read_stored_thread_for_resume( |
| ¶ms.thread_id, |
| None, |
| false, |
| ) |
| .await?; |
| Some((existing_thread_id, existing_thread, source_thread)) |
| } else { |
| let source_thread = self |
| .read_stored_thread_for_resume( |
| ¶ms.thread_id, |
| params.path.as_ref(), |
| false, |
| ) |
| .await?; |
| let existing_thread_id = source_thread.thread_id; |
| match self.thread_manager.get_thread(existing_thread_id).await { |
| Ok(existing_thread) => Some((existing_thread_id, existing_thread, source_thread)), |
| Err(_) => { |
| return Ok(RunningThreadResumeResult::NotRunning(Some(Box::new( |
| source_thread, |
| )))); |
| } |
| } |
| }; |
|
|
| if let Some((existing_thread_id, existing_thread, mut source_thread)) = running_thread { |
| let paginated_resume = |
| matches!(source_thread.history_mode, ThreadHistoryMode::Paginated); |
| let existing_thread_rollout_path = existing_thread.rollout_path(); |
| let active_path = existing_thread_rollout_path |
| .as_ref() |
| .or(source_thread.rollout_path.as_ref()); |
| if let (Some(requested_path), Some(active_path)) = (params.path.as_ref(), active_path) |
| && !path_utils::paths_match_after_normalization(requested_path, active_path) |
| { |
| return Err(invalid_request(format!( |
| "cannot resume running thread {existing_thread_id} with stale path: requested `{}`, active `{}`", |
| requested_path.display(), |
| active_path.display() |
| ))); |
| } |
| let config_snapshot = existing_thread.config_snapshot().await; |
| let mismatch_details = collect_resume_override_mismatches(params, &config_snapshot); |
| if !mismatch_details.is_empty() { |
| let has_subscribers = !self |
| .thread_state_manager |
| .subscribed_connection_ids(existing_thread_id) |
| .await |
| .is_empty(); |
| let loaded_status = self |
| .thread_watch_manager |
| .loaded_status_for_thread(&existing_thread_id.to_string()) |
| .await; |
| let is_running = |
| matches!(existing_thread.agent_status().await, AgentStatus::Running); |
|
|
| |
| if can_accept_direct_input( |
| existing_thread.multi_agent_version(), |
| &config_snapshot.session_source, |
| ) && !has_subscribers |
| && matches!(loaded_status, ThreadStatus::Idle) |
| && !is_running |
| { |
| |
| |
| |
| match wait_for_thread_shutdown(&existing_thread).await { |
| ThreadShutdownResult::Complete => { |
| self.thread_manager.remove_thread(&existing_thread_id).await; |
| self.finalize_thread_teardown(existing_thread_id).await; |
| |
| |
| return Ok(RunningThreadResumeResult::NotRunning(None)); |
| } |
| ThreadShutdownResult::SubmitFailed => { |
| warn!("failed to submit Shutdown to thread {existing_thread_id}"); |
| } |
| ThreadShutdownResult::TimedOut => { |
| warn!("thread {existing_thread_id} shutdown timed out"); |
| } |
| } |
| } |
|
|
| |
| |
| tracing::warn!( |
| "thread/resume overrides ignored for loaded thread {}: {}", |
| existing_thread_id, |
| mismatch_details.join("; ") |
| ); |
| } |
| let redact_resume_payloads = |
| should_redact_thread_resume_payloads(app_server_client_name.as_deref()); |
| let include_turns = !params.exclude_turns; |
| if paginated_resume && include_turns { |
| self.send_deprecation_notice( |
| request_id.connection_id, |
| PAGINATED_FULL_HISTORY_DEPRECATION_SUMMARY, |
| ) |
| .await; |
| } |
| let needs_history = |
| !paginated_resume && (include_turns || params.initial_turns_page.is_some()); |
| if needs_history { |
| let source_thread_id = source_thread.thread_id.to_string(); |
| let source_rollout_path = source_thread.rollout_path.clone(); |
| source_thread = self |
| .read_stored_thread_for_resume( |
| &source_thread_id, |
| source_rollout_path.as_ref(), |
| true, |
| ) |
| .await?; |
| } |
| if paginated_resume && (include_turns || params.initial_turns_page.is_some()) { |
| self.thread_store |
| .persist_thread(existing_thread_id, PersistContext::Standard) |
| .await |
| .map_err(thread_store_resume_read_error)?; |
| } |
| let history_items = if needs_history { |
| source_thread |
| .history |
| .take() |
| .map(|history| history.items) |
| .ok_or_else(|| { |
| internal_error(format!( |
| "thread {existing_thread_id} did not include persisted history" |
| )) |
| })? |
| } else { |
| Vec::new() |
| }; |
|
|
| let thread_state = self |
| .thread_state_manager |
| .thread_state(existing_thread_id) |
| .await; |
| self.ensure_listener_task_running( |
| existing_thread_id, |
| existing_thread.clone(), |
| thread_state.clone(), |
| ) |
| .await?; |
| Self::set_app_server_client_info( |
| existing_thread.as_ref(), |
| app_server_client_name, |
| app_server_client_version, |
| ) |
| .await?; |
|
|
| let mut thread_summary = self.stored_thread_to_api_thread( |
| source_thread, |
| config_snapshot.model_provider_id.as_str(), |
| false, |
| ); |
| thread_summary.session_id = existing_thread.session_configured().session_id.to_string(); |
| thread_summary.thread_source = config_snapshot.thread_source.clone().map(Into::into); |
| apply_live_thread_settings(&mut thread_summary, &config_snapshot); |
| thread_summary.can_accept_direct_input = Some(can_accept_direct_input( |
| existing_thread.multi_agent_version(), |
| &config_snapshot.session_source, |
| )); |
| let instruction_sources = existing_thread.legacy_instruction_sources().await; |
|
|
| let listener_command_tx = { |
| let thread_state = thread_state.lock().await; |
| thread_state.listener_command_tx() |
| }; |
| let Some(listener_command_tx) = listener_command_tx else { |
| return Err(internal_error(format!( |
| "failed to enqueue running thread resume for thread {existing_thread_id}: thread listener is not running" |
| ))); |
| }; |
|
|
| let (emit_thread_goal_update, thread_goal_state_db) = self |
| .thread_goal_processor |
| .pending_resume_goal_state(existing_thread.as_ref()) |
| .await; |
| let paginated_turns = if paginated_resume && include_turns { |
| Some(self.paginated_thread_full_turns(existing_thread_id).await?) |
| } else { |
| None |
| }; |
| let paginated_initial_turns_page = if paginated_resume { |
| match params.initial_turns_page.as_ref() { |
| Some(params) => Some( |
| self.paginated_resume_initial_turns_page(existing_thread_id, params) |
| .await?, |
| ), |
| None => None, |
| } |
| } else { |
| None |
| }; |
| let cold_resume_token_usage_turn_id = cold_resume_history |
| .map(|history| { |
| let turns = paginated_turns |
| .as_deref() |
| .filter(|turns| !turns.is_empty()) |
| .unwrap_or_else(|| { |
| paginated_initial_turns_page |
| .as_ref() |
| .map_or(&[][..], |page| page.data.as_slice()) |
| }); |
| restored_token_usage_turn_id(history, turns) |
| }) |
| .filter(|turn_id| !turn_id.is_empty()); |
| let paginated_initial_turns_page_with_active_slot = if paginated_resume { |
| match params.initial_turns_page.as_ref() { |
| Some(params) |
| if matches!( |
| params.sort_direction.unwrap_or(SortDirection::Desc), |
| SortDirection::Desc |
| ) => |
| { |
| Some( |
| self.paginated_resume_initial_turns_page_with_active_slot( |
| existing_thread_id, |
| params, |
| ) |
| .await?, |
| ) |
| } |
| Some(_) | None => None, |
| } |
| } else { |
| None |
| }; |
| let resume_cursor_store = paginated_resume.then(|| Arc::clone(&self.thread_store)); |
|
|
| let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); |
| let command = crate::thread_state::ThreadListenerCommand::SendThreadResumeResponse { |
| request: Box::new(crate::thread_state::PendingThreadResumeRequest { |
| request_id: request_id.clone(), |
| history_items, |
| cold_resume_token_usage_turn_id, |
| config_snapshot, |
| instruction_sources, |
| thread_summary, |
| emit_thread_goal_update, |
| thread_goal_state_db, |
| include_turns, |
| initial_turns_page: params.initial_turns_page.clone(), |
| paginated_turns, |
| paginated_initial_turns_page, |
| paginated_initial_turns_page_with_active_slot, |
| resume_cursor_store, |
| redact_resume_payloads, |
| }), |
| completion_tx, |
| }; |
| if listener_command_tx.send(command).is_err() { |
| return Err(internal_error(format!( |
| "failed to enqueue running thread resume for thread {existing_thread_id}: thread listener command channel is closed" |
| ))); |
| } |
| return Ok(RunningThreadResumeResult::Handled(completion_rx)); |
| } |
| Ok(RunningThreadResumeResult::NotRunning(None)) |
| } |
|
|
| #[tracing::instrument(level = "trace", skip_all)] |
| async fn resume_thread_from_history( |
| &self, |
| history: &[ResponseItem], |
| ) -> Result<InitialHistory, JSONRPCErrorError> { |
| if history.is_empty() { |
| return Err(invalid_request("history must not be empty")); |
| } |
| Ok(InitialHistory::Forked( |
| history |
| .iter() |
| .cloned() |
| .map(|item| RolloutItem::ResponseItem(item.into())) |
| .collect(), |
| )) |
| } |
|
|
| async fn load_resume_initial_history_from_stored_thread( |
| &self, |
| stored_thread: StoredThread, |
| ) -> Result<(InitialHistory, StoredThread), JSONRPCErrorError> { |
| if matches!(stored_thread.history_mode, ThreadHistoryMode::Paginated) { |
| let model_context = self |
| .thread_store |
| .load_latest_model_context(StoreLoadThreadHistoryParams { |
| thread_id: stored_thread.thread_id, |
| include_archived: true, |
| }) |
| .await |
| .map_err(thread_store_resume_read_error)?; |
| let history = InitialHistory::Resumed(ResumedHistory { |
| conversation_id: model_context.thread_id, |
| history: Arc::new(model_context.items), |
| rollout_path: stored_thread.rollout_path.clone(), |
| }); |
| return Ok((history, stored_thread)); |
| } |
|
|
| let thread_id = stored_thread.thread_id.to_string(); |
| let rollout_path = stored_thread.rollout_path.clone(); |
| let mut stored_thread = self |
| .read_stored_thread_for_resume( |
| &thread_id, |
| rollout_path.as_ref(), |
| true, |
| ) |
| .await?; |
| let history = self |
| .stored_thread_to_initial_history(&mut stored_thread) |
| .await?; |
| Ok((history, stored_thread)) |
| } |
|
|
| async fn read_stored_thread_for_resume( |
| &self, |
| thread_id: &str, |
| path: Option<&PathBuf>, |
| include_history: bool, |
| ) -> Result<StoredThread, JSONRPCErrorError> { |
| let result = if let Some(path) = path { |
| self.thread_store |
| .read_thread_by_rollout_path(StoreReadThreadByRolloutPathParams { |
| rollout_path: path.clone(), |
| include_archived: true, |
| include_history, |
| }) |
| .await |
| } else { |
| let existing_thread_id = match ThreadId::from_string(thread_id) { |
| Ok(id) => id, |
| Err(err) => { |
| return Err(invalid_request(format!("invalid session id: {err}"))); |
| } |
| }; |
| let params = StoreReadThreadParams { |
| thread_id: existing_thread_id, |
| include_archived: true, |
| include_history, |
| }; |
| self.thread_store.read_thread(params).await |
| }; |
|
|
| let stored_thread = result.map_err(thread_store_resume_read_error)?; |
| if let Some(requested_path) = path |
| && matches!(stored_thread.history_mode, ThreadHistoryMode::Paginated) |
| { |
| let current_thread = self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id: stored_thread.thread_id, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| .map_err(thread_store_resume_read_error)?; |
| if let Some(current_path) = current_thread.rollout_path.as_ref() |
| && !path_utils::paths_match_after_normalization( |
| codex_rollout::plain_rollout_path(requested_path).as_path(), |
| codex_rollout::plain_rollout_path(current_path).as_path(), |
| ) |
| { |
| return Err(invalid_request(format!( |
| "cannot resume paginated thread {} with stale path: requested {}, current {}; omit path and resume by thread id", |
| stored_thread.thread_id, |
| requested_path.display(), |
| current_path.display() |
| ))); |
| } |
| } |
| if stored_thread.archived_at.is_some() { |
| let thread_id = stored_thread.thread_id; |
| return Err(invalid_request(format!( |
| "session {thread_id} is archived. Run `codex unarchive {thread_id}` to unarchive it first." |
| ))); |
| } |
|
|
| Ok(stored_thread) |
| } |
|
|
| #[tracing::instrument(level = "trace", skip_all)] |
| async fn stored_thread_to_initial_history( |
| &self, |
| stored_thread: &mut StoredThread, |
| ) -> Result<InitialHistory, JSONRPCErrorError> { |
| let thread_id = stored_thread.thread_id; |
| let history = stored_thread |
| .history |
| .take() |
| .map(|history| history.items) |
| .ok_or_else(|| { |
| internal_error(format!( |
| "thread {thread_id} did not include persisted history" |
| )) |
| })?; |
| Ok(InitialHistory::Resumed(ResumedHistory { |
| conversation_id: thread_id, |
| history: Arc::new(history), |
| rollout_path: stored_thread.rollout_path.clone(), |
| })) |
| } |
|
|
| fn stored_thread_to_api_thread( |
| &self, |
| stored_thread: StoredThread, |
| fallback_provider: &str, |
| include_turns: bool, |
| ) -> Thread { |
| let (mut thread, history) = |
| thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); |
| if include_turns && let Some(history) = history { |
| populate_thread_turns_from_history( |
| &mut thread, |
| &history.items, |
| None, |
| ); |
| } |
| thread |
| } |
|
|
| async fn read_stored_thread_for_new_fork( |
| &self, |
| thread_id: ThreadId, |
| include_history: bool, |
| ) -> Result<StoredThread, JSONRPCErrorError> { |
| self.thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id, |
| include_archived: true, |
| include_history, |
| }) |
| .await |
| .map_err(thread_store_resume_read_error) |
| } |
|
|
| async fn load_thread_from_resume_source_or_send_internal( |
| &self, |
| thread_id: ThreadId, |
| thread: &CodexThread, |
| thread_history: &InitialHistory, |
| rollout_path: &Path, |
| resume_source_thread: Option<StoredThread>, |
| include_turns: bool, |
| ) -> std::result::Result<Thread, String> { |
| let config_snapshot = thread.config_snapshot().await; |
| let session_id = thread.session_configured().session_id.to_string(); |
| let can_accept_direct_input = can_accept_direct_input( |
| thread.multi_agent_version(), |
| &config_snapshot.session_source, |
| ); |
| let thread = match thread_history { |
| InitialHistory::Resumed(resumed) => { |
| let fallback_provider = config_snapshot.model_provider_id.as_str(); |
| if let Some(stored_thread) = resume_source_thread { |
| let stored_thread = |
| if let Some(rollout_path) = stored_thread.rollout_path.clone() { |
| self.thread_store |
| .read_thread_by_rollout_path(StoreReadThreadByRolloutPathParams { |
| rollout_path, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| .unwrap_or(StoredThread { |
| history: None, |
| ..stored_thread |
| }) |
| } else { |
| self.thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id: stored_thread.thread_id, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| .unwrap_or(StoredThread { |
| history: None, |
| ..stored_thread |
| }) |
| }; |
| Ok(thread_from_stored_thread( |
| stored_thread, |
| fallback_provider, |
| &self.config.cwd, |
| ) |
| .0) |
| } else { |
| match self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id: resumed.conversation_id, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| { |
| Ok(stored_thread) => Ok(thread_from_stored_thread( |
| stored_thread, |
| fallback_provider, |
| &self.config.cwd, |
| ) |
| .0), |
| Err(read_err) => { |
| Err(format!("failed to read thread from store: {read_err}")) |
| } |
| } |
| } |
| } |
| InitialHistory::Forked(items) => { |
| let mut thread = build_thread_from_snapshot( |
| thread_id, |
| session_id.clone(), |
| thread.multi_agent_version(), |
| &config_snapshot, |
| Some(rollout_path.into()), |
| ); |
| thread.preview = preview_from_rollout_items(items); |
| Ok(thread) |
| } |
| InitialHistory::New | InitialHistory::Cleared => Err(format!( |
| "failed to build resume response for thread {thread_id}: initial history missing" |
| )), |
| }; |
| let mut thread = thread?; |
| apply_live_thread_settings(&mut thread, &config_snapshot); |
| thread.can_accept_direct_input = Some(can_accept_direct_input); |
| thread.id = thread_id.to_string(); |
| thread.session_id = session_id; |
| thread.path = Some(rollout_path.to_path_buf()); |
| if include_turns { |
| let history_items = thread_history.get_rollout_items(); |
| populate_thread_turns_from_history( |
| &mut thread, |
| history_items, |
| None, |
| ); |
| } |
| self.attach_thread_name(thread_id, &mut thread).await; |
| Ok(thread) |
| } |
|
|
| async fn attach_thread_name(&self, thread_id: ThreadId, thread: &mut Thread) { |
| if let Ok(stored_thread) = self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| && let Some(title) = stored_thread.name.as_deref().map(str::trim) |
| && !title.is_empty() |
| { |
| if stored_thread.history_mode == ThreadHistoryMode::Paginated { |
| thread.name = Some(title.to_string()); |
| } else { |
| set_thread_name_from_title(thread, title.to_string()); |
| } |
| } |
| } |
|
|
| async fn thread_fork_inner( |
| &self, |
| request_id: ConnectionRequestId, |
| params: ThreadForkParams, |
| app_server_client_name: Option<String>, |
| app_server_client_version: Option<String>, |
| client_mcp_extensions: ClientMcpExtensions, |
| ) -> Result<(), JSONRPCErrorError> { |
| let ThreadForkParams { |
| thread_id, |
| last_turn_id, |
| before_turn_id, |
| path, |
| model, |
| model_provider, |
| service_tier, |
| cwd, |
| runtime_workspace_roots, |
| approval_policy, |
| approvals_reviewer, |
| sandbox, |
| permissions, |
| config: cli_overrides, |
| base_instructions, |
| developer_instructions, |
| ephemeral, |
| thread_source, |
| exclude_turns, |
| defer_goal_continuation, |
| } = params; |
| let include_turns = !exclude_turns; |
| if sandbox.is_some() && permissions.is_some() { |
| return Err(invalid_request( |
| "`permissions` cannot be combined with `sandbox`", |
| )); |
| } |
| let source_thread = self |
| .read_stored_thread_for_resume( |
| &thread_id, |
| path.as_ref(), |
| false, |
| ) |
| .await?; |
| let paginated_source = matches!(source_thread.history_mode, ThreadHistoryMode::Paginated); |
| if last_turn_id.is_some() && before_turn_id.is_some() { |
| return Err(invalid_request( |
| "`beforeTurnId` cannot be combined with `lastTurnId`", |
| )); |
| } |
| if ephemeral && defer_goal_continuation { |
| return Err(invalid_request( |
| "`deferGoalContinuation` cannot be combined with `ephemeral`", |
| )); |
| } |
| if paginated_source && ephemeral && include_turns { |
| return Err(invalid_request( |
| "ephemeral paginated thread/fork requires `excludeTurns: true`", |
| )); |
| } |
| if paginated_source && include_turns { |
| self.send_deprecation_notice( |
| request_id.connection_id, |
| PAGINATED_FULL_HISTORY_DEPRECATION_SUMMARY, |
| ) |
| .await; |
| } |
| let source_thread_id = source_thread.thread_id; |
| let source_thread_name = source_thread |
| .name |
| .as_deref() |
| .and_then(codex_core::util::normalize_thread_name); |
| let mut prepared_fork = if paginated_source { |
| let boundary = match (last_turn_id.as_deref(), before_turn_id.as_deref()) { |
| (Some(turn_id), None) => { |
| codex_thread_store::ForkBoundary::ThroughTurn(turn_id.to_string()) |
| } |
| (None, Some(turn_id)) => { |
| codex_thread_store::ForkBoundary::BeforeTurn(turn_id.to_string()) |
| } |
| (None, None) => codex_thread_store::ForkBoundary::Latest, |
| (Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"), |
| }; |
| Some( |
| self.thread_store |
| .prepare_fork(codex_thread_store::PrepareForkParams { |
| thread_id: source_thread_id, |
| boundary, |
| }) |
| .await |
| .map_err(|err| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::ThreadNotFound { thread_id } => { |
| invalid_request(format!("no rollout found for thread id {thread_id}")) |
| } |
| ThreadStoreError::Unsupported { .. } => { |
| method_not_found("paginated_threads is not supported yet") |
| } |
| err => internal_error(format!("failed to prepare paginated fork: {err}")), |
| })?, |
| ) |
| } else { |
| None |
| }; |
| let source_history_items = if let Some(prepared_fork) = prepared_fork.as_ref() { |
| Arc::clone(&prepared_fork.model_context) |
| } else { |
| let mut source_thread = self |
| .read_stored_thread_for_resume( |
| &thread_id, |
| path.as_ref(), |
| true, |
| ) |
| .await?; |
| Arc::new( |
| source_thread |
| .history |
| .take() |
| .map(|history| history.items) |
| .ok_or_else(|| { |
| internal_error(format!( |
| "thread {source_thread_id} did not include persisted history" |
| )) |
| })?, |
| ) |
| }; |
| let history_cwd = Some(source_thread.cwd.clone()); |
|
|
| |
| let mut cli_overrides = cli_overrides.unwrap_or_default(); |
| if cfg!(windows) { |
| match WindowsSandboxLevel::from_config(&self.config) { |
| WindowsSandboxLevel::Elevated => { |
| cli_overrides |
| .insert("windows.sandbox".to_string(), serde_json::json!("elevated")); |
| } |
| WindowsSandboxLevel::RestrictedToken => { |
| cli_overrides.insert( |
| "windows.sandbox".to_string(), |
| serde_json::json!("unelevated"), |
| ); |
| } |
| WindowsSandboxLevel::Disabled => {} |
| } |
| } |
| let request_overrides = if cli_overrides.is_empty() { |
| None |
| } else { |
| Some(cli_overrides) |
| }; |
| let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots); |
| let mut typesafe_overrides = self.build_thread_config_overrides( |
| model, |
| model_provider, |
| service_tier, |
| cwd, |
| runtime_workspace_roots, |
| approval_policy, |
| approvals_reviewer, |
| sandbox, |
| permissions, |
| base_instructions, |
| developer_instructions, |
| None, |
| ); |
| typesafe_overrides.ephemeral = ephemeral.then_some(true); |
| let restore_approval_policy = typesafe_overrides.approval_policy.is_none(); |
| let restore_approvals_reviewer = typesafe_overrides.approvals_reviewer.is_none() |
| && !request_overrides |
| .as_ref() |
| .is_some_and(|overrides| overrides.contains_key("approvals_reviewer")); |
| let restore_permission_profile = |
| !has_permission_override(request_overrides.as_ref(), &typesafe_overrides); |
| let needs_latest_settings = |
| restore_approval_policy || restore_approvals_reviewer || restore_permission_profile; |
| let loaded_parent = self.thread_manager.get_thread(source_thread_id).await.ok(); |
| let loaded_parent_settings = if paginated_source && needs_latest_settings { |
| if let Some(parent) = loaded_parent.as_ref() { |
| let snapshot = parent.thread_settings_snapshot().await; |
| Some(PersistedResumeSettings { |
| approval_policy: snapshot.approval_policy, |
| approvals_reviewer: Some(snapshot.approvals_reviewer), |
| active_permission_profile: snapshot.active_permission_profile, |
| }) |
| } else { |
| None |
| } |
| } else { |
| None |
| }; |
| let latest_context = if paginated_source |
| && needs_latest_settings |
| && loaded_parent.is_none() |
| && (last_turn_id.is_some() || before_turn_id.is_some()) |
| { |
| Some(Arc::new( |
| self.thread_store |
| .load_latest_model_context(StoreLoadThreadHistoryParams { |
| thread_id: source_thread_id, |
| include_archived: true, |
| }) |
| .await |
| .map_err(thread_store_resume_read_error)? |
| .items, |
| )) |
| } else { |
| None |
| }; |
| |
| |
| let source_multi_agent_version = InitialHistory::Resumed(ResumedHistory { |
| conversation_id: source_thread_id, |
| history: Arc::clone(latest_context.as_ref().unwrap_or(&source_history_items)), |
| rollout_path: source_thread.rollout_path.clone(), |
| }) |
| .get_multi_agent_version() |
| .or_else(|| { |
| loaded_parent |
| .as_ref() |
| .and_then(|parent| parent.multi_agent_version()) |
| }); |
| let persisted_settings = loaded_parent_settings.or_else(|| { |
| latest_persisted_resume_settings( |
| latest_context |
| .as_deref() |
| .unwrap_or_else(|| source_history_items.as_ref()), |
| ) |
| }); |
| if let Some(persisted_settings) = persisted_settings { |
| if restore_approval_policy { |
| typesafe_overrides.approval_policy = Some(persisted_settings.approval_policy); |
| } |
| if restore_approvals_reviewer { |
| typesafe_overrides.approvals_reviewer = persisted_settings.approvals_reviewer; |
| } |
| if restore_permission_profile { |
| typesafe_overrides.persisted_permission_profile_id = persisted_settings |
| .active_permission_profile |
| .map(|profile| profile.id); |
| } |
| } |
| |
| let config = self |
| .config_manager |
| .load_for_cwd(request_overrides, typesafe_overrides, history_cwd) |
| .await |
| .map_err(|err| config_load_error(&err))?; |
| let goals_enabled = config.features.enabled(Feature::Goals); |
|
|
| let fallback_model_provider = config.model_provider_id.clone(); |
| let parent_trace = self.request_trace_context(&request_id).await; |
| let thread_source = thread_source.map(Into::into); |
|
|
| let mut history_items = if prepared_fork.is_some() { |
| source_history_items |
| } else { |
| let source_history_items = Arc::unwrap_or_clone(source_history_items); |
| let history_items = match (last_turn_id.as_deref(), before_turn_id.as_deref()) { |
| (Some(last_turn_id), None) => { |
| truncate_rollout_after_turn_id(source_history_items, last_turn_id) |
| .map_err(|err| core_thread_write_error("truncate thread for fork", err))? |
| } |
| (None, Some(before_turn_id)) => { |
| truncate_rollout_before_turn_id(source_history_items, before_turn_id) |
| .map_err(|err| core_thread_write_error("truncate thread for fork", err))? |
| } |
| (None, None) => source_history_items, |
| (Some(_), Some(_)) => unreachable!("fork boundaries are mutually exclusive"), |
| }; |
| Arc::new(history_items) |
| }; |
| if let Some(multi_agent_version) = source_multi_agent_version |
| && (last_turn_id.is_some() || before_turn_id.is_some()) |
| { |
| |
| for item in Arc::make_mut(&mut history_items) { |
| if let RolloutItem::SessionMeta(meta) = item |
| && meta.meta.id == source_thread_id |
| { |
| meta.meta.multi_agent_version = Some(multi_agent_version); |
| } |
| } |
| if let Some(prepared_fork) = prepared_fork.as_mut() { |
| prepared_fork.model_context = Arc::clone(&history_items); |
| } |
| } |
|
|
| let ephemeral_preview = if ephemeral { |
| if paginated_source && last_turn_id.is_none() && before_turn_id.is_none() { |
| source_thread.preview.clone() |
| } else { |
| preview_from_rollout_items(&history_items) |
| } |
| } else { |
| String::new() |
| }; |
| let ephemeral_turns = if ephemeral && include_turns { |
| build_legacy_api_turns_from_rollout_items(&history_items) |
| } else { |
| Vec::new() |
| }; |
| let ephemeral_token_usage_turn_id = (ephemeral && include_turns) |
| .then(|| restored_token_usage_turn_id(&history_items, ephemeral_turns.as_slice())); |
| let token_usage_history_items = paginated_source.then(|| Arc::clone(&history_items)); |
| let inherited_project_id = source_thread.project_id.clone(); |
| let reserved_thread_id = if config.ephemeral { |
| None |
| } else { |
| stage_pending_thread_metadata( |
| self.thread_manager.as_ref(), |
| self.thread_store.as_ref(), |
| StoreThreadMetadataPatch { |
| project_id: inherited_project_id.clone().map(Some), |
| daybreak_enabled: source_thread.daybreak_enabled, |
| ..Default::default() |
| }, |
| "thread/fork", |
| ) |
| .await? |
| }; |
|
|
| let fork_options = StartThreadOptions { |
| thread_source, |
| parent_trace, |
| client_mcp_extensions, |
| reserved_thread_id, |
| ..StartThreadOptions::new(config) |
| }; |
| let new_thread = if let Some(prepared_fork) = prepared_fork { |
| self.thread_manager |
| .fork_prepared_thread(fork_options, prepared_fork) |
| .await |
| } else { |
| self.thread_manager |
| .fork_thread_from_history( |
| ForkSnapshot::Interrupted, |
| fork_options, |
| InitialHistory::Resumed(ResumedHistory { |
| conversation_id: source_thread_id, |
| history: history_items, |
| rollout_path: source_thread.rollout_path.clone(), |
| }), |
| ) |
| .await |
| }; |
| let NewThread { |
| thread_id, |
| thread: forked_thread, |
| session_configured, |
| .. |
| } = match new_thread { |
| Ok(new_thread) => new_thread, |
| Err(err) => { |
| remove_pending_thread_metadata(self.thread_store.as_ref(), reserved_thread_id) |
| .await; |
| return Err(match err.details() { |
| CodexErrorDetails::Io(_) | CodexErrorDetails::Json(_) => { |
| invalid_request(format!("failed to load thread {source_thread_id}: {err}")) |
| } |
| CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()), |
| _ => internal_error(format!("error forking thread: {err}")), |
| }); |
| } |
| }; |
|
|
| Self::set_app_server_client_info( |
| forked_thread.as_ref(), |
| app_server_client_name, |
| app_server_client_version, |
| ) |
| .await?; |
| if session_configured.rollout_path.is_some() |
| && let Some(name) = source_thread_name.clone() |
| { |
| self.thread_manager |
| .update_thread_metadata( |
| thread_id, |
| StoreThreadMetadataPatch { |
| name: Some(Some(name)), |
| ..Default::default() |
| }, |
| true, |
| ) |
| .await |
| .map_err(|err| core_thread_write_error("inherit source thread name", err))?; |
| } |
| let inherited_goal = if defer_goal_continuation |
| && session_configured.rollout_path.is_some() |
| && goals_enabled |
| { |
| if let Some(state_db) = forked_thread.state_db().or_else(|| self.state_db.clone()) { |
| self.thread_goal_processor |
| .flush_goal_progress_for_fork(source_thread_id) |
| .await |
| .map_err(|err| { |
| internal_error(format!("failed to flush source thread goal: {err}")) |
| })?; |
| inherit_thread_goal_snapshot(&state_db, source_thread_id, thread_id) |
| .await |
| .map_err(|err| { |
| internal_error(format!("failed to inherit source thread goal: {err}")) |
| })? |
| } else { |
| false |
| } |
| } else { |
| false |
| }; |
| if inherited_goal { |
| self.thread_goal_processor |
| .restore_inherited_goal_runtime(thread_id) |
| .await; |
| } |
|
|
| let instruction_sources = forked_thread.legacy_instruction_sources().await; |
|
|
| |
| log_listener_attach_result( |
| self.ensure_conversation_listener( |
| thread_id, |
| request_id.connection_id, |
| false, |
| ) |
| .await, |
| thread_id, |
| request_id.connection_id, |
| "thread", |
| ); |
|
|
| let config_snapshot = forked_thread.config_snapshot().await; |
|
|
| |
| |
| let (mut thread, mut token_usage_turn_id) = if session_configured.rollout_path.is_some() { |
| let stored_thread = self |
| .read_stored_thread_for_new_fork(thread_id, include_turns && !paginated_source) |
| .await?; |
| let (mut thread, history) = thread_from_stored_thread( |
| stored_thread, |
| fallback_model_provider.as_str(), |
| &self.config.cwd, |
| ); |
| if include_turns && let Some(history) = history.as_ref() { |
| populate_thread_turns_from_history( |
| &mut thread, |
| &history.items, |
| None, |
| ); |
| } |
| let token_usage_turn_id = include_turns.then(|| { |
| restored_token_usage_turn_id( |
| history |
| .as_ref() |
| .map(|history| history.items.as_slice()) |
| .or_else(|| token_usage_history_items.as_deref().map(Vec::as_slice)) |
| .unwrap_or(&[]), |
| thread.turns.as_slice(), |
| ) |
| }); |
| (thread, token_usage_turn_id) |
| } else { |
| let mut thread = build_thread_from_snapshot( |
| thread_id, |
| session_configured.session_id.to_string(), |
| forked_thread.multi_agent_version(), |
| &config_snapshot, |
| None, |
| ); |
| thread.preview = ephemeral_preview; |
| thread.forked_from_id = Some(source_thread_id.to_string()); |
| thread.turns = ephemeral_turns; |
| (thread, ephemeral_token_usage_turn_id) |
| }; |
| if paginated_source && include_turns { |
| thread.turns = self.paginated_thread_full_turns(thread_id).await?; |
| token_usage_turn_id = Some(restored_token_usage_turn_id( |
| token_usage_history_items |
| .as_deref() |
| .map_or(&[], Vec::as_slice), |
| thread.turns.as_slice(), |
| )); |
| } |
| if let Some(name) = source_thread_name { |
| set_thread_name_from_title(&mut thread, name); |
| } |
| thread.can_accept_direct_input = Some(can_accept_direct_input( |
| forked_thread.multi_agent_version(), |
| &config_snapshot.session_source, |
| )); |
| thread.session_id = session_configured.session_id.to_string(); |
| thread.thread_source = config_snapshot.thread_source.clone().map(Into::into); |
| apply_live_thread_settings(&mut thread, &config_snapshot); |
| if thread.path.is_none() { |
| thread.project_id = inherited_project_id.clone(); |
| } |
|
|
| self.thread_watch_manager |
| .upsert_thread_silently(&thread.id) |
| .await; |
|
|
| thread.status = resolve_thread_status( |
| self.thread_watch_manager |
| .loaded_status_for_thread(&thread.id) |
| .await, |
| false, |
| ); |
| let sandbox = config_snapshot.sandbox_policy().into(); |
| let active_permission_profile = |
| thread_response_active_permission_profile(config_snapshot.active_permission_profile); |
| let thread_originator = config_snapshot.originator.clone(); |
| let response = ThreadForkResponse { |
| thread: thread.clone(), |
| disabled_plugin_ids: config_snapshot.disabled_plugin_ids, |
| model: session_configured.model, |
| model_provider: session_configured.model_provider_id, |
| service_tier: session_configured.service_tier, |
| cwd: session_configured.cwd, |
| runtime_workspace_roots: config_snapshot.workspace_roots, |
| instruction_sources, |
| approval_policy: session_configured.approval_policy.into(), |
| approvals_reviewer: session_configured.approvals_reviewer.into(), |
| sandbox, |
| active_permission_profile, |
| reasoning_effort: session_configured.reasoning_effort, |
| multi_agent_mode: MultiAgentMode::ExplicitRequestOnly, |
| }; |
|
|
| let notif = thread_started_notification(thread); |
| let connection_id = request_id.connection_id; |
| self.outgoing |
| .send_response_with_thread_originator(request_id, response, thread_originator) |
| .await; |
| |
| |
| if let Some(token_usage_turn_id) = token_usage_turn_id { |
| |
| |
| send_thread_token_usage_update_to_connection( |
| &self.outgoing, |
| connection_id, |
| thread_id, |
| forked_thread.as_ref(), |
| token_usage_turn_id, |
| ) |
| .await; |
| } |
|
|
| self.outgoing |
| .send_server_notification(ServerNotification::ThreadStarted(notif)) |
| .await; |
| if inherited_goal { |
| self.thread_goal_processor |
| .emit_thread_goal_snapshot(thread_id) |
| .await; |
| } |
| Ok(()) |
| } |
|
|
| async fn get_thread_summary_response_inner( |
| &self, |
| params: GetConversationSummaryParams, |
| ) -> Result<GetConversationSummaryResponse, JSONRPCErrorError> { |
| let fallback_provider = self.config.model_provider_id.as_str(); |
| let read_result = match params { |
| GetConversationSummaryParams::ThreadId { conversation_id } => self |
| .thread_store |
| .read_thread(StoreReadThreadParams { |
| thread_id: conversation_id, |
| include_archived: true, |
| include_history: false, |
| }) |
| .await |
| .map_err(|err| conversation_summary_thread_id_read_error(conversation_id, err)), |
| GetConversationSummaryParams::RolloutPath { rollout_path } => { |
| let Some(local_thread_store) = self |
| .thread_store |
| .as_any() |
| .downcast_ref::<LocalThreadStore>() |
| else { |
| return Err(invalid_request( |
| "rollout path queries are only supported with the local thread store", |
| )); |
| }; |
|
|
| local_thread_store |
| .read_thread_by_rollout_path( |
| rollout_path.clone(), |
| true, |
| false, |
| ) |
| .await |
| .map_err(|err| conversation_summary_rollout_path_read_error(&rollout_path, err)) |
| } |
| }; |
|
|
| let stored_thread = read_result?; |
| let summary = summary_from_stored_thread(stored_thread, fallback_provider); |
| Ok(GetConversationSummaryResponse { summary }) |
| } |
|
|
| async fn list_threads_common( |
| &self, |
| requested_page_size: usize, |
| cursor: Option<String>, |
| sort_key: StoreThreadSortKey, |
| sort_direction: SortDirection, |
| filters: ThreadListFilters, |
| ) -> Result<(Vec<StoredThread>, Option<String>), JSONRPCErrorError> { |
| let ThreadListFilters { |
| model_providers, |
| source_kinds, |
| archived, |
| section_id, |
| project_id, |
| cwd_filters, |
| search_term, |
| use_state_db_only, |
| relation_filter, |
| } = filters; |
| let mut cursor_obj = cursor; |
| let mut last_cursor = cursor_obj.clone(); |
| let mut remaining = requested_page_size; |
| let mut items = Vec::with_capacity(requested_page_size); |
| let mut next_cursor: Option<String> = None; |
|
|
| let model_provider_filter = match model_providers { |
| Some(providers) => { |
| if providers.is_empty() { |
| None |
| } else { |
| Some(providers) |
| } |
| } |
| None if relation_filter.is_some() => None, |
| None => Some(vec![self.config.model_provider_id.clone()]), |
| }; |
| let (allowed_sources_vec, source_kind_filter) = |
| if relation_filter.is_some() && source_kinds.is_none() { |
| (Vec::new(), None) |
| } else { |
| compute_source_filters(source_kinds) |
| }; |
| let allowed_sources = allowed_sources_vec.as_slice(); |
| let store_sort_direction = match sort_direction { |
| SortDirection::Asc => StoreSortDirection::Asc, |
| SortDirection::Desc => StoreSortDirection::Desc, |
| }; |
|
|
| while remaining > 0 { |
| let page_size = remaining.min(THREAD_LIST_MAX_LIMIT); |
| let page = self |
| .thread_store |
| .list_threads(StoreListThreadsParams { |
| page_size, |
| cursor: cursor_obj.clone(), |
| sort_key, |
| sort_direction: store_sort_direction, |
| allowed_sources: allowed_sources.to_vec(), |
| model_providers: model_provider_filter.clone(), |
| cwd_filters: cwd_filters.clone(), |
| archived, |
| section: section_id.clone(), |
| project_id: project_id.clone(), |
| search_term: search_term.clone(), |
| use_state_db_only, |
| relation_filter, |
| }) |
| .await |
| .map_err(thread_store_list_error)?; |
|
|
| let mut filtered = Vec::with_capacity(page.items.len()); |
| for it in page.items { |
| let source = with_thread_spawn_agent_metadata( |
| it.source.clone(), |
| it.agent_nickname.clone(), |
| it.agent_role.clone(), |
| ); |
| if source_kind_filter |
| .as_ref() |
| .is_none_or(|filter| source_kind_matches(&source, filter)) |
| && cwd_filters.as_ref().is_none_or(|expected_cwds| { |
| expected_cwds.iter().any(|expected_cwd| { |
| path_utils::paths_match_after_normalization(&it.cwd, expected_cwd) |
| }) |
| }) |
| { |
| filtered.push(it); |
| if filtered.len() >= remaining { |
| break; |
| } |
| } |
| } |
| items.extend(filtered); |
| remaining = requested_page_size.saturating_sub(items.len()); |
|
|
| next_cursor = page.next_cursor; |
| if remaining == 0 { |
| break; |
| } |
|
|
| let Some(cursor_val) = next_cursor.clone() else { |
| break; |
| }; |
| |
| |
| if last_cursor.as_ref() == Some(&cursor_val) { |
| next_cursor = None; |
| break; |
| } |
| last_cursor = Some(cursor_val.clone()); |
| cursor_obj = Some(cursor_val); |
| } |
|
|
| Ok((items, next_cursor)) |
| } |
| } |
|
|
| fn xcode_26_4_mcp_elicitations_auto_deny( |
| client_name: Option<&str>, |
| client_version: Option<&str>, |
| ) -> bool { |
| |
| |
| |
| client_name == Some("Xcode") |
| && client_version.is_some_and(|version| version.starts_with("26.4")) |
| } |
|
|
| const THREAD_TURNS_DEFAULT_LIMIT: usize = 25; |
| const THREAD_TURNS_MAX_LIMIT: usize = 100; |
| const THREAD_ITEMS_DEFAULT_LIMIT: usize = 25; |
| const THREAD_ITEMS_MAX_LIMIT: usize = 100; |
| const THREAD_SEARCH_OCCURRENCES_DEFAULT_LIMIT: usize = 50; |
| const THREAD_SEARCH_OCCURRENCES_MAX_LIMIT: usize = 250; |
|
|
| pub(super) fn thread_turns_page_size(limit: Option<u32>) -> usize { |
| limit |
| .map(|value| value as usize) |
| .unwrap_or(THREAD_TURNS_DEFAULT_LIMIT) |
| .clamp(1, THREAD_TURNS_MAX_LIMIT) |
| } |
|
|
| fn thread_backwards_cursor_for_sort_key( |
| thread: &StoredThread, |
| sort_key: StoreThreadSortKey, |
| sort_direction: SortDirection, |
| ) -> Option<String> { |
| if sort_key == StoreThreadSortKey::SectionPosition { |
| let position = match sort_direction { |
| SortDirection::Asc => thread.section_position?.checked_add(1)?, |
| SortDirection::Desc => thread.section_position?.checked_sub(1)?, |
| }; |
| return Some(format!("{position}|{}", thread.thread_id)); |
| } |
|
|
| let timestamp = match sort_key { |
| StoreThreadSortKey::CreatedAt => thread.created_at, |
| StoreThreadSortKey::UpdatedAt => thread.updated_at, |
| StoreThreadSortKey::RecencyAt => thread.recency_at, |
| StoreThreadSortKey::SectionPosition => unreachable!("section positions use rank cursors"), |
| }; |
| |
| |
| let timestamp = match sort_direction { |
| SortDirection::Asc => timestamp.checked_add_signed(ChronoDuration::milliseconds(1))?, |
| SortDirection::Desc => timestamp.checked_sub_signed(ChronoDuration::milliseconds(1))?, |
| }; |
| Some(timestamp.to_rfc3339_opts(SecondsFormat::Millis, true)) |
| } |
|
|
| struct ThreadTurnsPage { |
| pub(super) turns: Vec<Turn>, |
| pub(super) next_cursor: Option<String>, |
| pub(super) backwards_cursor: Option<String>, |
| } |
|
|
| #[derive(serde::Serialize, serde::Deserialize)] |
| #[serde(rename_all = "camelCase")] |
| struct ThreadTurnsCursor { |
| turn_id: String, |
| include_anchor: bool, |
| } |
|
|
| fn paginate_thread_turns( |
| turns: Vec<Turn>, |
| cursor: Option<&str>, |
| limit: Option<u32>, |
| sort_direction: SortDirection, |
| ) -> Result<ThreadTurnsPage, JSONRPCErrorError> { |
| if turns.is_empty() { |
| return Ok(ThreadTurnsPage { |
| turns: Vec::new(), |
| next_cursor: None, |
| backwards_cursor: None, |
| }); |
| } |
|
|
| let anchor = cursor.map(parse_thread_turns_cursor).transpose()?; |
| let page_size = limit |
| .map(|value| value as usize) |
| .unwrap_or(THREAD_TURNS_DEFAULT_LIMIT) |
| .clamp(1, THREAD_TURNS_MAX_LIMIT); |
|
|
| let anchor_index = anchor |
| .as_ref() |
| .and_then(|anchor| turns.iter().position(|turn| turn.id == anchor.turn_id)); |
| if anchor.is_some() && anchor_index.is_none() { |
| return Err(invalid_request( |
| "invalid cursor: anchor turn is no longer present", |
| )); |
| } |
|
|
| let mut keyed_turns: Vec<_> = turns.into_iter().enumerate().collect(); |
| match sort_direction { |
| SortDirection::Asc => { |
| if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { |
| keyed_turns.retain(|(index, _)| { |
| if anchor.include_anchor { |
| *index >= anchor_index |
| } else { |
| *index > anchor_index |
| } |
| }); |
| } |
| } |
| SortDirection::Desc => { |
| keyed_turns.reverse(); |
| if let (Some(anchor), Some(anchor_index)) = (anchor.as_ref(), anchor_index) { |
| keyed_turns.retain(|(index, _)| { |
| if anchor.include_anchor { |
| *index <= anchor_index |
| } else { |
| *index < anchor_index |
| } |
| }); |
| } |
| } |
| } |
|
|
| let more_turns_available = keyed_turns.len() > page_size; |
| keyed_turns.truncate(page_size); |
| let backwards_cursor = keyed_turns |
| .first() |
| .map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, true)) |
| .transpose()?; |
| let next_cursor = if more_turns_available { |
| keyed_turns |
| .last() |
| .map(|(_, turn)| serialize_thread_turns_cursor(&turn.id, false)) |
| .transpose()? |
| } else { |
| None |
| }; |
| let turns = keyed_turns.into_iter().map(|(_, turn)| turn).collect(); |
|
|
| Ok(ThreadTurnsPage { |
| turns, |
| next_cursor, |
| backwards_cursor, |
| }) |
| } |
|
|
| fn serialize_thread_turns_cursor( |
| turn_id: &str, |
| include_anchor: bool, |
| ) -> Result<String, JSONRPCErrorError> { |
| serde_json::to_string(&ThreadTurnsCursor { |
| turn_id: turn_id.to_string(), |
| include_anchor, |
| }) |
| .map_err(|err| internal_error(format!("failed to serialize cursor: {err}"))) |
| } |
|
|
| fn parse_thread_turns_cursor(cursor: &str) -> Result<ThreadTurnsCursor, JSONRPCErrorError> { |
| serde_json::from_str(cursor).map_err(|_| invalid_request(format!("invalid cursor: {cursor}"))) |
| } |
|
|
| struct ThreadTurnsPageOptions<'a> { |
| cursor: Option<&'a str>, |
| limit: Option<u32>, |
| sort_direction: SortDirection, |
| items_view: TurnItemsView, |
| } |
|
|
| fn build_thread_turns_page_response( |
| items: &[RolloutItem], |
| loaded_status: ThreadStatus, |
| has_live_running_thread: bool, |
| active_turn: Option<Turn>, |
| options: ThreadTurnsPageOptions<'_>, |
| ) -> Result<ThreadTurnsListResponse, JSONRPCErrorError> { |
| let mut turns = reconstruct_thread_turns_for_turns_list( |
| items, |
| loaded_status, |
| has_live_running_thread, |
| active_turn, |
| ); |
| apply_thread_turns_items_view(&mut turns, options.items_view); |
| let page = paginate_thread_turns(turns, options.cursor, options.limit, options.sort_direction)?; |
| Ok(ThreadTurnsListResponse { |
| data: page.turns, |
| next_cursor: page.next_cursor, |
| backwards_cursor: page.backwards_cursor, |
| }) |
| } |
|
|
| pub(super) fn build_thread_resume_initial_turns_page( |
| items: &[RolloutItem], |
| loaded_status: ThreadStatus, |
| has_live_running_thread: bool, |
| active_turn: Option<Turn>, |
| params: &ThreadResumeInitialTurnsPageParams, |
| ) -> Result<codex_app_server_protocol::TurnsPage, JSONRPCErrorError> { |
| build_thread_turns_page_response( |
| items, |
| loaded_status, |
| has_live_running_thread, |
| active_turn, |
| ThreadTurnsPageOptions { |
| cursor: None, |
| limit: params.limit, |
| sort_direction: params.sort_direction.unwrap_or(SortDirection::Desc), |
| items_view: params.items_view.unwrap_or(TurnItemsView::Summary), |
| }, |
| ) |
| .map(Into::into) |
| } |
|
|
| pub(super) fn apply_thread_turns_items_view(turns: &mut [Turn], items_view: TurnItemsView) { |
| for turn in turns { |
| match items_view { |
| TurnItemsView::NotLoaded => { |
| turn.items.clear(); |
| turn.items_view = TurnItemsView::NotLoaded; |
| } |
| TurnItemsView::Summary => { |
| let first_user_message = turn |
| .items |
| .iter() |
| .find(|item| matches!(item, ThreadItem::UserMessage { .. })) |
| .cloned(); |
| let final_agent_message = turn |
| .items |
| .iter() |
| .rev() |
| .find(|item| matches!(item, ThreadItem::AgentMessage { .. })) |
| .cloned(); |
| turn.items = match (first_user_message, final_agent_message) { |
| (Some(user_message), Some(agent_message)) |
| if user_message.id() != agent_message.id() => |
| { |
| vec![user_message, agent_message] |
| } |
| (Some(user_message), _) => vec![user_message], |
| (None, Some(agent_message)) => vec![agent_message], |
| (None, None) => Vec::new(), |
| }; |
| turn.items_view = TurnItemsView::Summary; |
| } |
| TurnItemsView::Full => { |
| turn.items_view = TurnItemsView::Full; |
| } |
| } |
| } |
| } |
|
|
| fn reconstruct_thread_turns_for_turns_list( |
| items: &[RolloutItem], |
| loaded_status: ThreadStatus, |
| has_live_running_thread: bool, |
| active_turn: Option<Turn>, |
| ) -> Vec<Turn> { |
| let has_live_in_progress_turn = has_live_running_thread |
| || active_turn |
| .as_ref() |
| .is_some_and(|turn| matches!(turn.status, TurnStatus::InProgress)); |
| let mut turns = build_legacy_api_turns_from_rollout_items(items); |
| normalize_thread_turns_status(&mut turns, loaded_status, has_live_in_progress_turn); |
| if let Some(active_turn) = active_turn { |
| merge_turn_history_with_active_turn(&mut turns, active_turn); |
| } |
| turns |
| } |
|
|
| pub(super) fn normalize_thread_turns_status( |
| turns: &mut [Turn], |
| 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 { .. }) { |
| return; |
| } |
| for turn in turns { |
| if matches!(turn.status, TurnStatus::InProgress) { |
| turn.status = TurnStatus::Interrupted; |
| } |
| } |
| } |
|
|
| enum ThreadReadViewError { |
| InvalidRequest(String), |
| Unsupported(&'static str), |
| Internal(String), |
| JsonRpc(JSONRPCErrorError), |
| } |
|
|
| fn thread_read_view_error(err: ThreadReadViewError) -> JSONRPCErrorError { |
| match err { |
| ThreadReadViewError::InvalidRequest(message) => invalid_request(message), |
| ThreadReadViewError::Unsupported(operation) => { |
| unsupported_thread_store_operation(operation) |
| } |
| ThreadReadViewError::Internal(message) => internal_error(message), |
| ThreadReadViewError::JsonRpc(error) => error, |
| } |
| } |
|
|
| fn paginated_history_list_error(err: ThreadStoreError) -> JSONRPCErrorError { |
| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| ThreadStoreError::ThreadNotFound { thread_id } => { |
| invalid_request(format!("no rollout found for thread id {thread_id}")) |
| } |
| err => internal_error(format!("failed to list thread history: {err}")), |
| } |
| } |
|
|
| fn deserialize_stored_thread_item( |
| item: codex_thread_store::StoredThreadItem, |
| ) -> Result<ThreadItem, JSONRPCErrorError> { |
| serde_json::from_slice::<ThreadItem>(&item.item_json).map_err(|err| { |
| internal_error(format!( |
| "failed to deserialize stored thread item {}: {err}", |
| item.item_id |
| )) |
| }) |
| } |
|
|
| fn stored_turn_to_api_turn( |
| turn: StoredTurn, |
| items_view: TurnItemsView, |
| ) -> Result<Turn, JSONRPCErrorError> { |
| let status = match turn.status { |
| StoredTurnStatus::Completed => TurnStatus::Completed, |
| StoredTurnStatus::Interrupted => TurnStatus::Interrupted, |
| StoredTurnStatus::Failed => TurnStatus::Failed, |
| StoredTurnStatus::InProgress => TurnStatus::InProgress, |
| }; |
| let error = turn.error.map(|error| TurnError { |
| misalignment: None, |
| message: error.message, |
| codex_error_info: error.codex_error_info, |
| additional_details: error.additional_details, |
| }); |
| let items = turn |
| .items |
| .into_iter() |
| .map(deserialize_stored_thread_item) |
| .collect::<Result<Vec<_>, _>>()?; |
| Ok(Turn { |
| id: turn.turn_id, |
| items, |
| items_view, |
| status, |
| error, |
| started_at: turn.started_at, |
| completed_at: turn.completed_at, |
| duration_ms: turn.duration_ms, |
| }) |
| } |
|
|
| pub(super) fn unsupported_thread_store_operation(operation: &'static str) -> JSONRPCErrorError { |
| method_not_found(format!("{operation} is not supported yet")) |
| } |
|
|
| fn thread_store_list_error(err: ThreadStoreError) -> JSONRPCErrorError { |
| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| err => internal_error(format!("failed to list threads: {err}")), |
| } |
| } |
|
|
| fn thread_store_resume_read_error(err: ThreadStoreError) -> JSONRPCErrorError { |
| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| ThreadStoreError::ThreadNotFound { thread_id } => { |
| invalid_request(format!("no rollout found for thread id {thread_id}")) |
| } |
| err => internal_error(format!("failed to read thread: {err}")), |
| } |
| } |
|
|
| fn thread_turns_list_history_load_error( |
| thread_id: ThreadId, |
| err: ThreadStoreError, |
| ) -> ThreadReadViewError { |
| match err { |
| ThreadStoreError::InvalidRequest { message } |
| if message.starts_with("failed to resolve rollout path `") => |
| { |
| ThreadReadViewError::InvalidRequest(format!( |
| "thread {thread_id} is not materialized yet; thread/turns/list is unavailable before first user message" |
| )) |
| } |
| ThreadStoreError::InvalidRequest { message } => { |
| ThreadReadViewError::InvalidRequest(message) |
| } |
| ThreadStoreError::Unsupported { operation } => ThreadReadViewError::Unsupported(operation), |
| err => ThreadReadViewError::Internal(format!( |
| "failed to load thread history for thread {thread_id}: {err}" |
| )), |
| } |
| } |
|
|
| fn thread_read_history_load_error( |
| thread_id: ThreadId, |
| err: ThreadStoreError, |
| ) -> ThreadReadViewError { |
| match err { |
| ThreadStoreError::InvalidRequest { message } |
| if message.starts_with("failed to resolve rollout path `") => |
| { |
| ThreadReadViewError::InvalidRequest(format!( |
| "thread {thread_id} is not materialized yet; includeTurns is unavailable before first user message" |
| )) |
| } |
| ThreadStoreError::ThreadNotFound { |
| thread_id: missing_thread_id, |
| } if missing_thread_id == thread_id => ThreadReadViewError::InvalidRequest(format!( |
| "thread {thread_id} is not materialized yet; includeTurns is unavailable before first user message" |
| )), |
| ThreadStoreError::InvalidRequest { message } => { |
| ThreadReadViewError::InvalidRequest(message) |
| } |
| ThreadStoreError::Unsupported { operation } => ThreadReadViewError::Unsupported(operation), |
| err => ThreadReadViewError::Internal(format!( |
| "failed to load thread history for thread {thread_id}: {err}" |
| )), |
| } |
| } |
|
|
| fn conversation_summary_thread_id_read_error( |
| conversation_id: ThreadId, |
| err: ThreadStoreError, |
| ) -> JSONRPCErrorError { |
| let no_rollout_message = format!("no rollout found for thread id {conversation_id}"); |
| match err { |
| ThreadStoreError::InvalidRequest { message } if message == no_rollout_message => { |
| conversation_summary_not_found_error(conversation_id) |
| } |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| ThreadStoreError::ThreadNotFound { thread_id } if thread_id == conversation_id => { |
| conversation_summary_not_found_error(conversation_id) |
| } |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| err => internal_error(format!( |
| "failed to load conversation summary for {conversation_id}: {err}" |
| )), |
| } |
| } |
|
|
| fn conversation_summary_not_found_error(conversation_id: ThreadId) -> JSONRPCErrorError { |
| invalid_request(format!( |
| "no rollout found for conversation id {conversation_id}" |
| )) |
| } |
|
|
| fn conversation_summary_rollout_path_read_error( |
| path: &Path, |
| err: ThreadStoreError, |
| ) -> JSONRPCErrorError { |
| match err { |
| ThreadStoreError::InvalidRequest { message } => invalid_request(message), |
| ThreadStoreError::Unsupported { operation } => { |
| unsupported_thread_store_operation(operation) |
| } |
| err => internal_error(format!( |
| "failed to load conversation summary from {}: {}", |
| path.display(), |
| err |
| )), |
| } |
| } |
|
|
| pub(super) fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPCErrorError { |
| match err.details() { |
| CodexErrorDetails::ThreadNotFound(thread_id) => { |
| invalid_request(format!("thread not found: {thread_id}")) |
| } |
| CodexErrorDetails::InvalidRequest(message) => invalid_request(message.clone()), |
| CodexErrorDetails::UnsupportedOperation(message) => method_not_found(message.clone()), |
| _ => internal_error(format!("failed to {operation}: {err}")), |
| } |
| } |
|
|
| fn thread_store_mutation_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { |
| match err { |
| ThreadStoreError::InvalidRequest { message } | ThreadStoreError::Conflict { message } => { |
| invalid_request(message) |
| } |
| ThreadStoreError::Unsupported { |
| operation: unsupported_operation, |
| } => unsupported_thread_store_operation(unsupported_operation), |
| err => internal_error(format!("failed to {operation} session: {err}")), |
| } |
| } |
|
|
| fn set_thread_name_from_title(thread: &mut Thread, title: String) { |
| if title.trim().is_empty() || thread.preview.trim() == title.trim() { |
| return; |
| } |
| thread.name = Some(title); |
| } |
|
|
| pub(crate) fn thread_from_stored_thread( |
| thread: StoredThread, |
| fallback_provider: &str, |
| fallback_cwd: &AbsolutePathBuf, |
| ) -> (Thread, Option<codex_thread_store::StoredThreadHistory>) { |
| let path = thread.rollout_path; |
| let git_info = thread.git_info.map(|info| ApiGitInfo { |
| sha: info.commit_hash.map(|sha| sha.0), |
| branch: info.branch, |
| origin_url: info.repository_url.map(String::from), |
| }); |
| let cwd = AbsolutePathBuf::relative_to_current_dir(path_utils::normalize_for_native_workdir( |
| thread.cwd, |
| )) |
| .unwrap_or_else(|err| { |
| warn!("failed to normalize thread cwd while reading stored thread: {err}"); |
| fallback_cwd.clone() |
| }); |
| let source = with_thread_spawn_agent_metadata( |
| thread.source, |
| thread.agent_nickname.clone(), |
| thread.agent_role.clone(), |
| ); |
| let history = thread.history; |
| let thread_id = thread.thread_id.to_string(); |
| let thread = Thread { |
| id: thread_id.clone(), |
| environments: None, |
| extra: None, |
| session_id: thread_id, |
| forked_from_id: thread.forked_from_id.map(|id| id.to_string()), |
| parent_thread_id: thread.parent_thread_id.map(|id| id.to_string()), |
| preview: thread.preview, |
| ephemeral: false, |
| section: thread.section.map(|section| ThreadSection { |
| id: section.id, |
| name: section.name, |
| appearance: section |
| .appearance |
| .map(|appearance| ThreadSectionAppearance { |
| icon: appearance.icon, |
| color: appearance.color, |
| }), |
| }), |
| section_entered_at: thread |
| .section_entered_at |
| .map(|entered_at| entered_at.timestamp()), |
| project_id: thread.project_id, |
| history_mode: thread.history_mode.into(), |
| model_provider: if thread.model_provider.is_empty() { |
| fallback_provider.to_string() |
| } else { |
| thread.model_provider |
| }, |
| model: thread.model, |
| reasoning_effort: thread.reasoning_effort, |
| created_at: thread.created_at.timestamp(), |
| updated_at: thread.updated_at.timestamp(), |
| recency_at: Some(thread.recency_at.timestamp()), |
| status: ThreadStatus::NotLoaded, |
| path, |
| cwd, |
| cli_version: thread.cli_version, |
| originator: thread.originator, |
| agent_nickname: source.get_nickname(), |
| agent_role: source.get_agent_role(), |
| source: source.into(), |
| can_accept_direct_input: None, |
| thread_source: thread.thread_source.map(Into::into), |
| git_info, |
| name: thread.name, |
| daybreak_enabled: thread.daybreak_enabled, |
| turns: Vec::new(), |
| }; |
| (thread, history) |
| } |
|
|
| fn summary_from_stored_thread( |
| thread: StoredThread, |
| fallback_provider: &str, |
| ) -> ConversationSummary { |
| let path = thread.rollout_path.unwrap_or_default(); |
| let source = with_thread_spawn_agent_metadata( |
| thread.source, |
| thread.agent_nickname.clone(), |
| thread.agent_role.clone(), |
| ); |
| let git_info = thread.git_info.map(|git| ConversationGitInfo { |
| sha: git.commit_hash.map(|sha| sha.0), |
| branch: git.branch, |
| origin_url: git.repository_url.map(String::from), |
| }); |
| ConversationSummary { |
| conversation_id: thread.thread_id, |
| path, |
| preview: thread.preview, |
| |
| |
| timestamp: Some( |
| thread |
| .created_at |
| .to_rfc3339_opts(SecondsFormat::Millis, true), |
| ), |
| updated_at: Some( |
| thread |
| .updated_at |
| .to_rfc3339_opts(SecondsFormat::Millis, true), |
| ), |
| model_provider: if thread.model_provider.is_empty() { |
| fallback_provider.to_string() |
| } else { |
| thread.model_provider |
| }, |
| cwd: thread.cwd, |
| cli_version: thread.cli_version, |
| source, |
| git_info, |
| } |
| } |
|
|
| #[allow(clippy::too_many_arguments)] |
| #[cfg(test)] |
| fn summary_from_state_db_metadata( |
| conversation_id: ThreadId, |
| path: PathBuf, |
| first_user_message: Option<String>, |
| preview: Option<String>, |
| timestamp: String, |
| updated_at: String, |
| model_provider: String, |
| cwd: PathBuf, |
| cli_version: String, |
| source: String, |
| _thread_source: Option<codex_protocol::protocol::ThreadSource>, |
| agent_nickname: Option<String>, |
| agent_role: Option<String>, |
| git_sha: Option<String>, |
| git_branch: Option<String>, |
| git_origin_url: Option<String>, |
| ) -> ConversationSummary { |
| let preview = preview.or(first_user_message).unwrap_or_default(); |
| let source = serde_json::from_str(&source) |
| .or_else(|_| serde_json::from_value(serde_json::Value::String(source.clone()))) |
| .unwrap_or(codex_protocol::protocol::SessionSource::Unknown); |
| let source = with_thread_spawn_agent_metadata(source, agent_nickname, agent_role); |
| let git_info = if git_sha.is_none() && git_branch.is_none() && git_origin_url.is_none() { |
| None |
| } else { |
| Some(ConversationGitInfo { |
| sha: git_sha, |
| branch: git_branch, |
| origin_url: git_origin_url, |
| }) |
| }; |
| ConversationSummary { |
| conversation_id, |
| path, |
| preview, |
| timestamp: Some(timestamp), |
| updated_at: Some(updated_at), |
| model_provider, |
| cwd, |
| cli_version, |
| source, |
| git_info, |
| } |
| } |
|
|
| #[cfg(test)] |
| fn summary_from_thread_metadata(metadata: &ThreadMetadata) -> ConversationSummary { |
| summary_from_state_db_metadata( |
| metadata.id, |
| metadata.rollout_path.clone(), |
| metadata.first_user_message.clone(), |
| metadata.preview.clone(), |
| metadata |
| .created_at |
| .to_rfc3339_opts(SecondsFormat::Secs, true), |
| metadata |
| .updated_at |
| .to_rfc3339_opts(SecondsFormat::Secs, true), |
| metadata.model_provider.clone(), |
| metadata.cwd.clone(), |
| metadata.cli_version.clone(), |
| metadata.source.clone(), |
| metadata.thread_source.clone(), |
| metadata.agent_nickname.clone(), |
| metadata.agent_role.clone(), |
| metadata.git_sha.clone(), |
| metadata.git_branch.clone(), |
| metadata.git_origin_url.clone().map(String::from), |
| ) |
| } |
|
|
| fn preview_from_rollout_items(items: &[RolloutItem]) -> String { |
| items |
| .iter() |
| .find_map(|item| match item { |
| RolloutItem::ResponseItem(item) => match codex_core::parse_turn_item(&item.item) { |
| Some(codex_protocol::items::TurnItem::UserMessage(user)) => Some(user.message()), |
| _ => None, |
| }, |
| _ => None, |
| }) |
| .map(|preview| strip_user_message_prefix(preview.as_str()).to_string()) |
| .unwrap_or_default() |
| } |
|
|
| fn build_thread_from_snapshot( |
| thread_id: ThreadId, |
| session_id: String, |
| multi_agent_version: Option<codex_protocol::protocol::MultiAgentVersion>, |
| config_snapshot: &ThreadConfigSnapshot, |
| path: Option<PathBuf>, |
| ) -> Thread { |
| let now = time::OffsetDateTime::now_utc().unix_timestamp(); |
| Thread { |
| id: thread_id.to_string(), |
| environments: Some( |
| config_snapshot |
| .environment_selections() |
| .iter() |
| .map(Into::into) |
| .collect(), |
| ), |
| extra: None, |
| session_id, |
| forked_from_id: None, |
| parent_thread_id: config_snapshot.parent_thread_id.map(|id| id.to_string()), |
| preview: String::new(), |
| ephemeral: config_snapshot.ephemeral, |
| section: None, |
| section_entered_at: None, |
| project_id: None, |
| history_mode: config_snapshot.history_mode.into(), |
| model_provider: config_snapshot.model_provider_id.clone(), |
| model: Some(config_snapshot.model.clone()), |
| reasoning_effort: config_snapshot.reasoning_effort.clone(), |
| created_at: now, |
| updated_at: now, |
| recency_at: Some(now), |
| status: ThreadStatus::NotLoaded, |
| path, |
| cwd: config_snapshot.cwd().clone(), |
| cli_version: env!("CARGO_PKG_VERSION").to_string(), |
| originator: (!config_snapshot.originator.is_empty()) |
| .then(|| config_snapshot.originator.clone()), |
| agent_nickname: config_snapshot.session_source.get_nickname(), |
| agent_role: config_snapshot.session_source.get_agent_role(), |
| source: config_snapshot.session_source.clone().into(), |
| can_accept_direct_input: Some(can_accept_direct_input( |
| multi_agent_version, |
| &config_snapshot.session_source, |
| )), |
| thread_source: config_snapshot.thread_source.clone().map(Into::into), |
| git_info: None, |
| name: None, |
| daybreak_enabled: None, |
| turns: Vec::new(), |
| } |
| } |
|
|
| fn paginate_background_terminals( |
| terminals: &[ThreadBackgroundTerminal], |
| cursor: Option<String>, |
| limit: Option<u32>, |
| ) -> Result<(Vec<ThreadBackgroundTerminal>, Option<String>), JSONRPCErrorError> { |
| let start = match cursor { |
| Some(cursor) => { |
| let cursor = cursor |
| .parse::<i32>() |
| .map_err(|err| invalid_request(format!("invalid cursor: {err}")))?; |
| terminals |
| .iter() |
| .position(|terminal| { |
| terminal |
| .process_id |
| .parse::<i32>() |
| .is_ok_and(|process_id| process_id > cursor) |
| }) |
| .unwrap_or(terminals.len()) |
| } |
| None => 0, |
| }; |
| let effective_limit = limit.unwrap_or(terminals.len() as u32).max(1) as usize; |
| let end = start.saturating_add(effective_limit).min(terminals.len()); |
| let next_cursor = (end < terminals.len()).then(|| terminals[end - 1].process_id.clone()); |
| Ok((terminals[start..end].to_vec(), next_cursor)) |
| } |
|
|
| fn build_thread_from_loaded_snapshot( |
| thread_id: ThreadId, |
| config_snapshot: &ThreadConfigSnapshot, |
| loaded_thread: &CodexThread, |
| ) -> Thread { |
| build_thread_from_snapshot( |
| thread_id, |
| loaded_thread.session_configured().session_id.to_string(), |
| loaded_thread.multi_agent_version(), |
| config_snapshot, |
| loaded_thread.rollout_path(), |
| ) |
| } |
|
|
| #[cfg(test)] |
| #[path = "thread_processor_tests.rs"] |
| mod thread_processor_tests; |
|
|
