From 64e035e53068ba414403f5a318a6ebdc8eebcf06 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Thu, 4 Jun 2026 23:19:46 -0700 Subject: [PATCH] Wire app-server goals through extension --- codex-rs/Cargo.lock | 1 + codex-rs/app-server/Cargo.toml | 1 + codex-rs/app-server/src/extensions.rs | 15 + codex-rs/app-server/src/mcp_refresh.rs | 3 + codex-rs/app-server/src/message_processor.rs | 6 + codex-rs/app-server/src/request_processors.rs | 2 - .../thread_goal_processor.rs | 300 ++--- .../request_processors/thread_lifecycle.rs | 14 +- codex-rs/core/src/codex_thread.rs | 48 +- codex-rs/core/src/lib.rs | 3 - codex-rs/core/src/session/review.rs | 3 - codex-rs/core/src/session/session.rs | 3 - codex-rs/core/src/session/tests.rs | 1182 +---------------- codex-rs/core/src/session/turn.rs | 103 +- codex-rs/core/src/session/turn_context.rs | 10 - codex-rs/core/src/tasks/mod.rs | 42 - codex-rs/core/src/thread_manager.rs | 3 - codex-rs/core/src/thread_manager_tests.rs | 101 -- codex-rs/core/src/tools/handlers/mod.rs | 5 - codex-rs/core/src/tools/registry.rs | 16 +- codex-rs/core/src/tools/spec_plan.rs | 16 - codex-rs/core/src/tools/spec_plan_tests.rs | 31 +- codex-rs/core/tests/suite/prompt_caching.rs | 3 - 23 files changed, 130 insertions(+), 1781 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index b899222384..a61df434f1 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1926,6 +1926,7 @@ dependencies = [ "codex-file-search", "codex-file-watcher", "codex-git-utils", + "codex-goal-extension", "codex-guardian", "codex-hooks", "codex-image-generation-extension", diff --git a/codex-rs/app-server/Cargo.toml b/codex-rs/app-server/Cargo.toml index 37cdf219bf..4bdc430a1a 100644 --- a/codex-rs/app-server/Cargo.toml +++ b/codex-rs/app-server/Cargo.toml @@ -41,6 +41,7 @@ codex-extension-api = { workspace = true } codex-external-agent-migration = { workspace = true } codex-external-agent-sessions = { workspace = true } codex-features = { workspace = true } +codex-goal-extension = { workspace = true } codex-guardian = { workspace = true } codex-git-utils = { workspace = true } codex-file-watcher = { workspace = true } diff --git a/codex-rs/app-server/src/extensions.rs b/codex-rs/app-server/src/extensions.rs index 7b2673ca06..6c2251bab7 100644 --- a/codex-rs/app-server/src/extensions.rs +++ b/codex-rs/app-server/src/extensions.rs @@ -12,11 +12,13 @@ use codex_extension_api::AgentSpawner; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionRegistry; use codex_extension_api::ExtensionRegistryBuilder; +use codex_goal_extension::GoalService; use codex_login::AuthManager; use codex_protocol::ThreadId; use codex_protocol::error::CodexErr; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; +use codex_rollout::state_db::StateDbHandle; use crate::outgoing_message::OutgoingMessageSender; @@ -24,11 +26,24 @@ pub(crate) fn thread_extensions( guardian_agent_spawner: S, event_sink: Arc, auth_manager: Arc, + state_db: Option, + thread_manager: Weak, + goal_service: Arc, ) -> Arc> where S: AgentSpawner + 'static, { let mut builder = ExtensionRegistryBuilder::::with_event_sink(event_sink); + if let Some(state_db) = state_db { + codex_goal_extension::install_with_backend( + &mut builder, + state_db, + codex_otel::global(), + thread_manager, + goal_service, + |config: &Config| config.features.enabled(codex_features::Feature::Goals), + ); + } codex_guardian::install(&mut builder, guardian_agent_spawner); codex_memories_extension::install(&mut builder, codex_otel::global()); codex_web_search_extension::install(&mut builder, auth_manager.clone()); diff --git a/codex-rs/app-server/src/mcp_refresh.rs b/codex-rs/app-server/src/mcp_refresh.rs index 0996c409bc..8c72533ba5 100644 --- a/codex-rs/app-server/src/mcp_refresh.rs +++ b/codex-rs/app-server/src/mcp_refresh.rs @@ -191,6 +191,9 @@ mod tests { guardian_agent_spawner(thread_manager.clone()), Arc::new(NoopExtensionEventSink), auth_manager.clone(), + Some(state_db.clone()), + thread_manager.clone(), + Arc::new(codex_goal_extension::GoalService::new()), ), /*analytics_events_client*/ None, Arc::clone(&thread_store), diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 6c578ce8e5..552e4cd58b 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -70,6 +70,7 @@ use codex_core::ThreadManager; use codex_core::config::Config; use codex_exec_server::EnvironmentManager; use codex_feedback::CodexFeedback; +use codex_goal_extension::GoalService; use codex_login::AuthManager; use codex_login::auth::ExternalAuth; use codex_login::auth::ExternalAuthRefreshContext; @@ -305,6 +306,7 @@ impl MessageProcessor { // resumed, or forked threads to a different persistence backend/root. let thread_store = codex_core::thread_store_from_config(config.as_ref(), state_db.clone()); let environment_manager_for_requests = Arc::clone(&environment_manager); + let goal_service = Arc::new(GoalService::new()); let thread_manager = Arc::new_cyclic(|thread_manager| { ThreadManager::new( config.as_ref(), @@ -315,6 +317,9 @@ impl MessageProcessor { guardian_agent_spawner(thread_manager.clone()), app_server_extension_event_sink(outgoing.clone()), auth_manager.clone(), + state_db.clone(), + thread_manager.clone(), + Arc::clone(&goal_service), ), Some(analytics_events_client.clone()), Arc::clone(&thread_store), @@ -416,6 +421,7 @@ impl MessageProcessor { Arc::clone(&config), thread_state_manager.clone(), state_db.clone(), + Arc::clone(&goal_service), ); let thread_processor = ThreadRequestProcessor::new( auth_manager.clone(), diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 0e94528a7a..13128125e1 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -275,8 +275,6 @@ use codex_config::loader::project_trust_key; use codex_config::types::McpServerTransportConfig; use codex_core::CodexThread; use codex_core::CodexThreadSettingsOverrides; -use codex_core::ExternalGoalPreviousStatus; -use codex_core::ExternalGoalSet; use codex_core::ForkSnapshot; use codex_core::NewThread; #[cfg(test)] diff --git a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs index c18806ea03..a2fd561ec8 100644 --- a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs @@ -1,5 +1,9 @@ use super::*; -use codex_protocol::protocol::validate_thread_goal_objective; +use codex_goal_extension::GoalObjectiveUpdate; +use codex_goal_extension::GoalService; +use codex_goal_extension::GoalServiceError; +use codex_goal_extension::GoalSetRequest; +use codex_goal_extension::GoalTokenBudgetUpdate; #[derive(Clone)] pub(crate) struct ThreadGoalRequestProcessor { @@ -8,6 +12,7 @@ pub(crate) struct ThreadGoalRequestProcessor { config: Arc, thread_state_manager: ThreadStateManager, state_db: Option, + goal_service: Arc, } impl ThreadGoalRequestProcessor { @@ -17,6 +22,7 @@ impl ThreadGoalRequestProcessor { config: Arc, thread_state_manager: ThreadStateManager, state_db: Option, + goal_service: Arc, ) -> Self { Self { thread_manager, @@ -24,6 +30,7 @@ impl ThreadGoalRequestProcessor { config, thread_state_manager, state_db, + goal_service, } } @@ -66,10 +73,8 @@ impl ThreadGoalRequestProcessor { } self.emit_thread_goal_snapshot(thread_id).await; // App-server owns resume response and snapshot ordering, so wait until - // those are sent before letting core start goal continuation. - if let Err(err) = thread.continue_active_goal_if_idle().await { - tracing::warn!("failed to continue active goal after resume: {err}"); - } + // those are sent before letting extensions react to the idle thread. + thread.emit_thread_idle_lifecycle_if_idle().await; } pub(crate) async fn pending_resume_goal_state( @@ -100,140 +105,36 @@ impl ThreadGoalRequestProcessor { let thread_id = parse_thread_id_for_request(params.thread_id.as_str())?; let state_db = self.state_db_for_materialized_thread(thread_id).await?; - let running_thread = self.thread_manager.get_thread(thread_id).await.ok(); - let rollout_path = match running_thread.as_ref() { - Some(thread) => thread.rollout_path().ok_or_else(|| { - invalid_request(format!( - "ephemeral thread does not support goals: {thread_id}" - )) - })?, - None => codex_rollout::find_thread_path_by_id_str( - &self.config.codex_home, - &thread_id.to_string(), - self.state_db.as_deref(), - ) - .await - .map_err(|err| { - internal_error(format!("failed to locate thread id {thread_id}: {err}")) - })? - .ok_or_else(|| invalid_request(format!("thread not found: {thread_id}")))?, - }; - reconcile_rollout( - Some(&state_db), - rollout_path.as_path(), - self.config.model_provider_id.as_str(), - /*builder*/ None, - &[], - /*archived_only*/ None, - /*new_thread_memory_mode*/ None, - ) - .await; + self.reconcile_thread_goal_rollout(thread_id, &state_db) + .await?; let listener_command_tx = { let thread_state = self.thread_state_manager.thread_state(thread_id).await; let thread_state = thread_state.lock().await; thread_state.listener_command_tx() }; - let status = params.status.map(thread_goal_status_to_state); - let objective = params.objective.as_deref().map(str::trim); + let status = params.status.map(ThreadGoalStatus::to_core); + let objective = params.objective.as_deref(); - if let Some(objective) = objective { - validate_thread_goal_objective(objective).map_err(invalid_request)?; - } - if objective.is_some() || params.token_budget.is_some() { - validate_goal_budget(params.token_budget.flatten()).map_err(invalid_request)?; - } - - if let Some(thread) = running_thread.as_ref() { - thread.prepare_external_goal_mutation().await; - } - - let should_set_thread_preview = objective.is_some(); - let (goal, previous_status) = (if let Some(objective) = objective { - let existing_goal = state_db - .thread_goals() - .get_thread_goal(thread_id) - .await - .map_err(|err| invalid_request(err.to_string()))?; - if let Some(goal) = existing_goal.as_ref() { - let previous_status = ExternalGoalPreviousStatus::from(goal); - state_db - .thread_goals() - .update_thread_goal( - thread_id, - codex_state::GoalUpdate { - objective: Some(objective.to_string()), - status, - token_budget: params.token_budget, - expected_goal_id: Some(goal.goal_id.clone()), - }, - ) - .await - .and_then(|goal| { - goal.ok_or_else(|| { - anyhow::anyhow!( - "cannot update goal for thread {thread_id}: no goal exists" - ) - }) - }) - .map(|goal| (goal, previous_status)) - } else { - let previous_status = ExternalGoalPreviousStatus::NewGoal; - state_db - .thread_goals() - .replace_thread_goal( - thread_id, - objective, - status.unwrap_or(codex_state::ThreadGoalStatus::Active), - params.token_budget.flatten(), - ) - .await - .map(|goal| (goal, previous_status)) - } - } else { - let existing_goal = state_db - .thread_goals() - .get_thread_goal(thread_id) - .await - .map_err(|err| invalid_request(err.to_string()))?; - let Some(existing_goal) = existing_goal else { - return Err(invalid_request(format!( - "cannot update goal for thread {thread_id}: no goal exists" - ))); - }; - let previous_status = ExternalGoalPreviousStatus::from(&existing_goal); - state_db - .thread_goals() - .update_thread_goal( + let outcome = self + .goal_service + .set_thread_goal( + &state_db, + GoalSetRequest { thread_id, - codex_state::GoalUpdate { - objective: None, - status, - token_budget: params.token_budget, - expected_goal_id: None, + objective: objective + .map(GoalObjectiveUpdate::Set) + .unwrap_or(GoalObjectiveUpdate::Keep), + status, + token_budget: match params.token_budget { + Some(token_budget) => GoalTokenBudgetUpdate::Set(token_budget), + None => GoalTokenBudgetUpdate::Keep, }, - ) - .await - .and_then(|goal| { - goal.ok_or_else(|| { - anyhow::anyhow!("cannot update goal for thread {thread_id}: no goal exists") - }) - }) - .map(|goal| (goal, previous_status)) - }) - .map_err(|err| invalid_request(err.to_string()))?; - if should_set_thread_preview - && let Err(err) = state_db - .set_thread_preview_if_empty(thread_id, goal.objective.as_str()) - .await - { - warn!("failed to set empty thread preview from goal objective for {thread_id}: {err}"); - } - let external_goal_set = ExternalGoalSet { - goal: goal.clone(), - previous_status, - }; - let goal = api_thread_goal_from_state(goal); + }, + ) + .await + .map_err(goal_service_error)?; + let goal = ThreadGoal::from(outcome.goal.clone()); self.outgoing .send_response( request_id.clone(), @@ -242,9 +143,7 @@ impl ThreadGoalRequestProcessor { .await; self.emit_thread_goal_updated_ordered(thread_id, goal, listener_command_tx) .await; - if let Some(thread) = running_thread.as_ref() { - thread.apply_external_goal_set(external_goal_set).await; - } + outcome.apply_runtime_effects(&self.goal_service).await; Ok(()) } @@ -258,12 +157,12 @@ impl ThreadGoalRequestProcessor { let thread_id = parse_thread_id_for_request(params.thread_id.as_str())?; let state_db = self.state_db_for_materialized_thread(thread_id).await?; - let goal = state_db - .thread_goals() - .get_thread_goal(thread_id) + let goal = self + .goal_service + .get_thread_goal(&state_db, thread_id) .await - .map_err(|err| internal_error(format!("failed to read thread goal: {err}")))? - .map(api_thread_goal_from_state); + .map_err(goal_service_error)? + .map(ThreadGoal::from); Ok(ThreadGoalGetResponse { goal }) } @@ -278,53 +177,19 @@ impl ThreadGoalRequestProcessor { let thread_id = parse_thread_id_for_request(params.thread_id.as_str())?; let state_db = self.state_db_for_materialized_thread(thread_id).await?; - let running_thread = self.thread_manager.get_thread(thread_id).await.ok(); - let rollout_path = match running_thread.as_ref() { - Some(thread) => thread.rollout_path().ok_or_else(|| { - invalid_request(format!( - "ephemeral thread does not support goals: {thread_id}" - )) - })?, - None => codex_rollout::find_thread_path_by_id_str( - &self.config.codex_home, - &thread_id.to_string(), - self.state_db.as_deref(), - ) - .await - .map_err(|err| { - internal_error(format!("failed to locate thread id {thread_id}: {err}")) - })? - .ok_or_else(|| invalid_request(format!("thread not found: {thread_id}")))?, - }; - reconcile_rollout( - Some(&state_db), - rollout_path.as_path(), - self.config.model_provider_id.as_str(), - /*builder*/ None, - &[], - /*archived_only*/ None, - /*new_thread_memory_mode*/ None, - ) - .await; - - if let Some(thread) = running_thread.as_ref() { - thread.prepare_external_goal_mutation().await; - } + self.reconcile_thread_goal_rollout(thread_id, &state_db) + .await?; let listener_command_tx = { let thread_state = self.thread_state_manager.thread_state(thread_id).await; let thread_state = thread_state.lock().await; thread_state.listener_command_tx() }; - let cleared = state_db - .thread_goals() - .delete_thread_goal(thread_id) + let cleared = self + .goal_service + .clear_thread_goal(&state_db, thread_id) .await - .map_err(|err| internal_error(format!("failed to clear thread goal: {err}")))?; - - if cleared && let Some(thread) = running_thread.as_ref() { - thread.apply_external_goal_clear().await; - } + .map_err(goal_service_error)?; self.outgoing .send_response(request_id, ThreadGoalClearResponse { cleared }) @@ -367,6 +232,42 @@ impl ThreadGoalRequestProcessor { .ok_or_else(|| internal_error("sqlite state db unavailable for thread goals")) } + async fn reconcile_thread_goal_rollout( + &self, + thread_id: ThreadId, + state_db: &StateDbHandle, + ) -> Result<(), JSONRPCErrorError> { + let running_thread = self.thread_manager.get_thread(thread_id).await.ok(); + let rollout_path = match running_thread.as_ref() { + Some(thread) => thread.rollout_path().ok_or_else(|| { + invalid_request(format!( + "ephemeral thread does not support goals: {thread_id}" + )) + })?, + None => codex_rollout::find_thread_path_by_id_str( + &self.config.codex_home, + &thread_id.to_string(), + self.state_db.as_deref(), + ) + .await + .map_err(|err| { + internal_error(format!("failed to locate thread id {thread_id}: {err}")) + })? + .ok_or_else(|| invalid_request(format!("thread not found: {thread_id}")))?, + }; + reconcile_rollout( + Some(state_db), + rollout_path.as_path(), + self.config.model_provider_id.as_str(), + /*builder*/ None, + &[], + /*archived_only*/ None, + /*new_thread_memory_mode*/ None, + ) + .await; + Ok(()) + } + async fn emit_thread_goal_snapshot(&self, thread_id: ThreadId) { let state_db = match self.state_db_for_materialized_thread(thread_id).await { Ok(state_db) => state_db, @@ -449,27 +350,20 @@ impl ThreadGoalRequestProcessor { } } -fn validate_goal_budget(value: Option) -> Result<(), String> { - if let Some(value) = value - && value <= 0 - { - return Err("goal budgets must be positive when provided".to_string()); - } - Ok(()) -} - -fn thread_goal_status_to_state(status: ThreadGoalStatus) -> codex_state::ThreadGoalStatus { - match status { - ThreadGoalStatus::Active => codex_state::ThreadGoalStatus::Active, - ThreadGoalStatus::Paused => codex_state::ThreadGoalStatus::Paused, - ThreadGoalStatus::Blocked => codex_state::ThreadGoalStatus::Blocked, - ThreadGoalStatus::UsageLimited => codex_state::ThreadGoalStatus::UsageLimited, - ThreadGoalStatus::BudgetLimited => codex_state::ThreadGoalStatus::BudgetLimited, - ThreadGoalStatus::Complete => codex_state::ThreadGoalStatus::Complete, +pub(super) fn api_thread_goal_from_state(goal: codex_state::ThreadGoal) -> ThreadGoal { + ThreadGoal { + thread_id: goal.thread_id.to_string(), + objective: goal.objective, + status: api_thread_goal_status_from_state(goal.status), + token_budget: goal.token_budget, + tokens_used: goal.tokens_used, + time_used_seconds: goal.time_used_seconds, + created_at: goal.created_at.timestamp(), + updated_at: goal.updated_at.timestamp(), } } -fn thread_goal_status_from_state(status: codex_state::ThreadGoalStatus) -> ThreadGoalStatus { +fn api_thread_goal_status_from_state(status: codex_state::ThreadGoalStatus) -> ThreadGoalStatus { match status { codex_state::ThreadGoalStatus::Active => ThreadGoalStatus::Active, codex_state::ThreadGoalStatus::Paused => ThreadGoalStatus::Paused, @@ -480,16 +374,10 @@ fn thread_goal_status_from_state(status: codex_state::ThreadGoalStatus) -> Threa } } -pub(super) fn api_thread_goal_from_state(goal: codex_state::ThreadGoal) -> ThreadGoal { - ThreadGoal { - thread_id: goal.thread_id.to_string(), - objective: goal.objective, - status: thread_goal_status_from_state(goal.status), - token_budget: goal.token_budget, - tokens_used: goal.tokens_used, - time_used_seconds: goal.time_used_seconds, - created_at: goal.created_at.timestamp(), - updated_at: goal.updated_at.timestamp(), +fn goal_service_error(err: GoalServiceError) -> JSONRPCErrorError { + match err { + GoalServiceError::InvalidRequest(message) => invalid_request(message), + GoalServiceError::Internal(message) => internal_error(message), } } diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index f72cfb5a62..410b01ae15 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -616,12 +616,6 @@ pub(super) async fn handle_pending_thread_resume_request( } } - if pending.emit_thread_goal_update - && let Err(err) = conversation.apply_goal_resume_runtime_effects().await - { - tracing::warn!("failed to apply goal resume runtime effects: {err}"); - } - let ThreadConfigSnapshot { model, model_provider_id, @@ -691,11 +685,9 @@ pub(super) async fn handle_pending_thread_resume_request( .replay_requests_to_connection_for_thread(connection_id, conversation_id) .await; // App-server owns resume response and snapshot ordering, so wait until - // replay completes before letting core start goal continuation. - if pending.emit_thread_goal_update - && let Err(err) = conversation.continue_active_goal_if_idle().await - { - tracing::warn!("failed to continue active goal after running-thread resume: {err}"); + // replay completes before letting extensions react to the idle thread. + if pending.emit_thread_goal_update { + conversation.emit_thread_idle_lifecycle_if_idle().await; } } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index f9f10881b5..4ac0757b5f 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -1,7 +1,5 @@ use crate::agent::AgentStatus; use crate::config::ConstraintResult; -use crate::goals::ExternalGoalSet; -use crate::goals::GoalRuntimeEvent; use crate::session::Codex; use crate::session::SessionSettingsUpdate; use crate::session::SteerInputError; @@ -205,51 +203,11 @@ impl CodexThread { } } - pub async fn apply_goal_resume_runtime_effects(&self) -> anyhow::Result<()> { + pub async fn emit_thread_idle_lifecycle_if_idle(&self) { self.codex .session - .goal_runtime_apply(GoalRuntimeEvent::ThreadResumed) - .await - } - - pub async fn continue_active_goal_if_idle(&self) -> anyhow::Result<()> { - self.codex - .session - .goal_runtime_apply(GoalRuntimeEvent::MaybeContinueIfIdle) - .await - } - - pub async fn prepare_external_goal_mutation(&self) { - if let Err(err) = self - .codex - .session - .goal_runtime_apply(GoalRuntimeEvent::ExternalMutationStarting) - .await - { - tracing::warn!("failed to prepare external goal mutation: {err}"); - } - } - - pub async fn apply_external_goal_set(&self, external_set: ExternalGoalSet) { - if let Err(err) = self - .codex - .session - .goal_runtime_apply(GoalRuntimeEvent::ExternalSet { external_set }) - .await - { - tracing::warn!("failed to apply external goal status runtime effects: {err}"); - } - } - - pub async fn apply_external_goal_clear(&self) { - if let Err(err) = self - .codex - .session - .goal_runtime_apply(GoalRuntimeEvent::ExternalClear) - .await - { - tracing::warn!("failed to apply external goal clear runtime effects: {err}"); - } + .emit_thread_idle_lifecycle_if_idle() + .await; } #[doc(hidden)] diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 4dc19de7e6..400c803680 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -39,9 +39,6 @@ pub mod exec_env; mod exec_policy; #[cfg(test)] mod git_info_tests; -mod goals; -pub use goals::ExternalGoalPreviousStatus; -pub use goals::ExternalGoalSet; mod guardian; mod hook_runtime; mod installation_id; diff --git a/codex-rs/core/src/session/review.rs b/codex-rs/core/src/session/review.rs index ff4c77aab9..06378c1250 100644 --- a/codex-rs/core/src/session/review.rs +++ b/codex-rs/core/src/session/review.rs @@ -24,9 +24,7 @@ pub(super) async fn spawn_review_thread( let mut review_features = sess.features.clone(); let _ = review_features.disable(Feature::WebSearchRequest); let _ = review_features.disable(Feature::WebSearchCached); - let _ = review_features.disable(Feature::Goals); let review_web_search_mode = WebSearchMode::Disabled; - let goal_tools_supported = !config.ephemeral && parent_turn_context.goal_tools_enabled(); let available_models = sess .services .models_manager @@ -126,7 +124,6 @@ pub(super) async fn spawn_review_thread( environments: parent_turn_context.environments.clone(), available_models, unified_exec_shell_mode, - goal_tools_supported, features: review_features, ghost_snapshot: parent_turn_context.ghost_snapshot.clone(), current_date: parent_turn_context.current_date.clone(), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 9314a743a5..74c4b5e4e4 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -2,7 +2,6 @@ use super::input_queue::InputQueue; use super::*; use crate::agents_md::LoadedAgentsMd; use crate::config::ConstraintError; -use crate::goals::GoalRuntimeState; use crate::skills::SkillError; use crate::state::ActiveTurn; use codex_protocol::SessionId; @@ -37,7 +36,6 @@ pub(crate) struct Session { pub(crate) conversation: Arc, pub(crate) active_turn: Mutex>, pub(crate) input_queue: InputQueue, - pub(crate) goal_runtime: GoalRuntimeState, pub(crate) guardian_review_session: GuardianReviewSessionManager, pub(crate) services: SessionServices, pub(super) next_internal_sub_id: AtomicU64, @@ -1071,7 +1069,6 @@ impl Session { conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: InputQueue::new(), - goal_runtime: GoalRuntimeState::new(), guardian_review_session: GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index a2023ea378..f97bbff4fd 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -57,11 +57,6 @@ use codex_protocol::request_permissions::PermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile; use tracing::Span; -use crate::goals::CreateGoalRequest; -use crate::goals::ExternalGoalPreviousStatus; -use crate::goals::ExternalGoalSet; -use crate::goals::GoalRuntimeEvent; -use crate::goals::SetGoalRequest; use crate::rollout::recorder::RolloutRecorder; use crate::state::ActiveTurn; use crate::state::TaskKind; @@ -72,11 +67,9 @@ use crate::tasks::execute_user_shell_command; use crate::tools::ToolRouter; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolPayload; -use crate::tools::handlers::CreateGoalHandler; use crate::tools::handlers::ExecCommandHandler; use crate::tools::handlers::RequestPermissionsHandler; use crate::tools::handlers::ShellCommandHandler; -use crate::tools::handlers::UpdateGoalHandler; use crate::tools::registry::ToolExecutor; use crate::tools::router::ToolCallSource; use crate::turn_diff_tracker::TurnDiffTracker; @@ -128,7 +121,6 @@ use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SkillScope; use codex_protocol::protocol::Submission; -use codex_protocol::protocol::ThreadGoalStatus; use codex_protocol::protocol::ThreadRolledBackEvent; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::TokenCountEvent; @@ -139,28 +131,21 @@ use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::W3cTraceContext; -use codex_protocol::request_user_input::RequestUserInputAnswer; -use codex_protocol::request_user_input::RequestUserInputResponse; use codex_rmcp_client::ElicitationAction; use core_test_support::PathBufExt; use core_test_support::PathExt; use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; use core_test_support::context_snapshot::ContextSnapshotRenderMode; -use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; -use core_test_support::responses::ev_completed_with_tokens; -use core_test_support::responses::ev_function_call; use core_test_support::responses::ev_response_created; use core_test_support::responses::mount_sse_once; -use core_test_support::responses::mount_sse_sequence; use core_test_support::responses::sse; use core_test_support::responses::start_mock_server; use core_test_support::test_codex::test_codex; use core_test_support::test_path_buf; use core_test_support::tracing::install_test_tracing; use core_test_support::wait_for_event; -use core_test_support::wait_for_event_match; use opentelemetry::trace::TraceContextExt; use opentelemetry::trace::TraceId; use opentelemetry_sdk::metrics::InMemoryMetricExporter; @@ -4896,7 +4881,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { session_configuration.cwd.clone(), "turn_id".to_string(), skills_outcome, - /*goal_tools_supported*/ true, ); let session = Session { @@ -4913,7 +4897,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: super::input_queue::InputQueue::new(), - goal_runtime: crate::goals::GoalRuntimeState::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), @@ -6780,18 +6763,7 @@ where let (tx_event, rx_event) = async_channel::unbounded(); let mut config = build_test_config(codex_home).await; configure_config(&mut config); - let state_db = if config.features.enabled(Feature::Goals) { - Some( - codex_state::StateRuntime::init( - config.sqlite_home.clone(), - config.model_provider_id.clone(), - ) - .await - .expect("goal tests should initialize sqlite state db"), - ) - } else { - None - }; + let state_db = None; let config = Arc::new(config); let thread_id = ThreadId::default(); let auth_manager = AuthManager::from_auth_for_testing(auth); @@ -6986,7 +6958,6 @@ where session_configuration.cwd.clone(), "turn_id".to_string(), skills_outcome, - /*goal_tools_supported*/ true, )); let session = Arc::new(Session { @@ -7003,7 +6974,6 @@ where conversation: Arc::new(RealtimeConversationManager::new()), active_turn: Mutex::new(None), input_queue: super::input_queue::InputQueue::new(), - goal_runtime: crate::goals::GoalRuntimeState::new(), guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(), services, next_internal_sub_id: AtomicU64::new(0), @@ -7027,52 +6997,6 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx( .await } -async fn make_goal_session_and_context_with_rx() -> ( - Arc, - Arc, - async_channel::Receiver, - tempfile::TempDir, -) { - let codex_home = tempfile::tempdir().expect("create temp dir"); - let (session, turn_context, rx) = make_session_and_context_with_auth_config_home_and_rx( - CodexAuth::from_api_key("Test API Key"), - Vec::new(), - codex_home.path(), - |config| { - config - .features - .enable(Feature::Goals) - .expect("goal mode should be enableable in tests"); - }, - ) - .await; - upsert_goal_test_thread(session.as_ref()).await; - (session, turn_context, rx, codex_home) -} - -async fn upsert_goal_test_thread(session: &Session) { - let config = session.get_config().await; - let state_db = session - .state_db() - .expect("goal test session should have a state db"); - let mut builder = codex_state::ThreadMetadataBuilder::new( - session.thread_id, - config - .codex_home - .join("goal-test-rollout.jsonl") - .to_path_buf(), - chrono::Utc::now(), - SessionSource::Cli, - ); - builder.cwd = config.cwd.to_path_buf(); - builder.model_provider = Some(config.model_provider_id.clone()); - let metadata = builder.build(config.model_provider_id.as_str()); - state_db - .upsert_thread(&metadata) - .await - .expect("goal test thread should be upserted"); -} - // Like make_session_and_context, but returns Arc and the event receiver // so tests can assert on emitted events. pub(crate) async fn make_session_and_context_with_rx() -> ( @@ -9129,282 +9053,6 @@ async fn abort_empty_active_turn_preserves_pending_input() { ); } -#[tokio::test] -async fn interrupt_accounts_active_goal_without_pausing() -> anyhow::Result<()> { - let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.set_thread_goal( - tc.as_ref(), - SetGoalRequest { - objective: Some("Keep improving the benchmark".to_string()), - status: None, - token_budget: None, - }, - ) - .await?; - - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; - set_total_token_usage(&sess, post_goal_token_usage()).await; - - sess.abort_all_tasks(TurnAbortReason::Interrupted).await; - - let goal = sess - .get_thread_goal() - .await? - .expect("goal should remain persisted after interrupt"); - assert_eq!( - codex_protocol::protocol::ThreadGoalStatus::Active, - goal.status - ); - assert_eq!(70, goal.tokens_used); - - assert!(sess.active_turn.lock().await.is_none()); - - Ok(()) -} - -#[tokio::test] -async fn shutdown_without_active_turn_keeps_active_goal_active() -> anyhow::Result<()> { - let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.set_thread_goal( - tc.as_ref(), - SetGoalRequest { - objective: Some("Keep improving the benchmark".to_string()), - status: None, - token_budget: None, - }, - ) - .await?; - - assert!(sess.active_turn.lock().await.is_none()); - assert!(handlers::shutdown(&sess, "shutdown".to_string()).await); - - let goal = sess - .get_thread_goal() - .await? - .expect("goal should remain persisted after shutdown"); - assert_eq!( - codex_protocol::protocol::ThreadGoalStatus::Active, - goal.status - ); - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn active_goal_continuation_runs_again_after_no_tool_turn() -> anyhow::Result<()> { - let server = start_mock_server().await; - let mut builder = test_codex().with_config(|config| { - config - .features - .enable(Feature::Goals) - .expect("goal mode should be enableable in tests"); - }); - let test = builder.build(&server).await?; - let responses = mount_sse_sequence( - &server, - vec![ - sse(vec![ - ev_response_created("resp-1"), - ev_function_call( - "call-create-goal", - "create_goal", - r#"{"objective":"write a benchmark note"}"#, - ), - ev_completed("resp-1"), - ]), - sse(vec![ - ev_assistant_message("msg-1", "Draft ready."), - ev_completed("resp-2"), - ]), - sse(vec![ - ev_assistant_message("msg-2", "I am still working on the benchmark note."), - ev_completed("resp-3"), - ]), - sse(vec![ - ev_response_created("resp-4"), - ev_function_call( - "call-complete-goal", - "update_goal", - r#"{"status":"complete"}"#, - ), - ev_completed("resp-4"), - ]), - sse(vec![ - ev_assistant_message("msg-3", "Goal complete."), - ev_completed("resp-5"), - ]), - ], - ) - .await; - - test.codex - .submit(Op::UserInput { - environments: None, - items: vec![UserInput::Text { - text: "write a benchmark note".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }) - .await?; - - let mut completed_turns = 0; - tokio::time::timeout(std::time::Duration::from_secs(8), async { - loop { - let event = test.codex.next_event().await?; - if matches!(event.msg, EventMsg::TurnComplete(_)) { - completed_turns += 1; - if completed_turns == 3 { - return anyhow::Ok(()); - } - } - } - }) - .await??; - - let goal_context_text = responses - .requests() - .into_iter() - .flat_map(|request| request.message_input_texts("user")) - .find(|text| text.contains("")) - .expect("goal context message should be present"); - assert!(goal_context_text.contains("Continue working toward the active thread goal.")); - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn pending_request_user_input_does_not_spawn_extra_goal_continuation() -> anyhow::Result<()> { - let server = start_mock_server().await; - let mut builder = test_codex().with_config(|config| { - config - .features - .enable(Feature::Goals) - .expect("goal mode should be enableable in tests"); - config - .features - .enable(Feature::DefaultModeRequestUserInput) - .expect("default-mode request_user_input should be enableable in tests"); - }); - let test = builder.build(&server).await?; - let responses = mount_sse_sequence( - &server, - vec![ - sse(vec![ - ev_response_created("resp-1"), - ev_function_call( - "call-create-goal", - "create_goal", - r#"{"objective":"write a benchmark note"}"#, - ), - ev_completed("resp-1"), - ]), - sse(vec![ - ev_assistant_message("msg-1", "Draft ready."), - ev_completed("resp-2"), - ]), - sse(vec![ - ev_response_created("resp-3"), - ev_function_call( - "call-ask-user", - "request_user_input", - r#"{"questions":[{"header":"Choice","id":"next_step","question":"Pick one","options":[{"label":"Outline","description":"Start with an outline."},{"label":"Draft","description":"Write a full draft."}]}]}"#, - ), - ev_completed("resp-3"), - ]), - sse(vec![ - ev_response_created("resp-4"), - ev_function_call( - "call-complete-goal", - "update_goal", - r#"{"status":"complete"}"#, - ), - ev_completed("resp-4"), - ]), - sse(vec![ - ev_assistant_message("msg-2", "Goal complete."), - ev_completed("resp-5"), - ]), - ], - ) - .await; - - test.codex - .submit(Op::UserInput { - environments: None, - items: vec![UserInput::Text { - text: "write a benchmark note".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }) - .await?; - - let request_user_input_event = wait_for_event_match(&test.codex, |event| match event { - EventMsg::RequestUserInput(event) => Some(event.clone()), - _ => None, - }) - .await; - assert_eq!(3, responses.requests().len()); - assert!( - timeout(Duration::from_millis(200), test.codex.next_event()) - .await - .is_err(), - "waiting for request_user_input should keep the turn open without emitting more events" - ); - assert_eq!( - 3, - responses.requests().len(), - "waiting for request_user_input should not start another continuation request" - ); - - test.codex - .submit(Op::UserInputAnswer { - id: request_user_input_event.turn_id, - response: RequestUserInputResponse { - answers: std::collections::HashMap::from([( - "next_step".to_string(), - RequestUserInputAnswer { - answers: vec!["Outline".to_string()], - }, - )]), - }, - }) - .await?; - - let mut completed_turns = 0; - timeout(Duration::from_secs(8), async { - loop { - let event = test.codex.next_event().await?; - if matches!(event.msg, EventMsg::TurnComplete(_)) { - completed_turns += 1; - if completed_turns == 1 { - return anyhow::Ok(()); - } - } - } - }) - .await??; - - assert_eq!(5, responses.requests().len()); - - Ok(()) -} - async fn set_total_token_usage(sess: &Session, total_token_usage: TokenUsage) { let mut state = sess.state.lock().await; state.set_token_info(Some(TokenUsageInfo { @@ -9414,543 +9062,6 @@ async fn set_total_token_usage(sess: &Session, total_token_usage: TokenUsage) { })); } -fn post_goal_token_usage() -> TokenUsage { - TokenUsage { - input_tokens: 50, - cached_input_tokens: 10, - output_tokens: 30, - reasoning_output_tokens: 5, - total_tokens: 75, - } -} - -async fn goal_test_state_db(sess: &Session) -> anyhow::Result { - if let Some(state_db) = sess.state_db() { - return Ok(state_db); - } - let config = sess.get_config().await; - codex_state::StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()) - .await -} - -#[tokio::test] -async fn create_thread_goal_fills_empty_thread_preview() -> anyhow::Result<()> { - let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - let state_db = goal_test_state_db(sess.as_ref()).await?; - - let page = state_db - .list_threads( - /*page_size*/ 10, - codex_state::ThreadFilterOptions { - archived_only: false, - allowed_sources: &[], - model_providers: None, - cwd_filters: None, - anchor: None, - sort_key: codex_state::SortKey::UpdatedAt, - sort_direction: codex_state::SortDirection::Desc, - search_term: None, - }, - ) - .await?; - assert!(page.items.is_empty()); - - sess.create_thread_goal( - tc.as_ref(), - CreateGoalRequest { - objective: "Keep improving the benchmark".to_string(), - token_budget: None, - }, - ) - .await?; - - let page = state_db - .list_threads( - /*page_size*/ 10, - codex_state::ThreadFilterOptions { - archived_only: false, - allowed_sources: &[], - model_providers: None, - cwd_filters: None, - anchor: None, - sort_key: codex_state::SortKey::UpdatedAt, - sort_direction: codex_state::SortDirection::Desc, - search_term: None, - }, - ) - .await?; - let ids = page - .items - .iter() - .map(|thread| thread.id) - .collect::>(); - assert_eq!(vec![sess.thread_id], ids); - assert_eq!( - Some("Keep improving the benchmark"), - page.items[0].preview.as_deref() - ); - - Ok(()) -} - -#[tokio::test] -async fn budget_limited_accounting_steers_active_turn_without_aborting() -> anyhow::Result<()> { - let (sess, tc, rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.set_thread_goal( - tc.as_ref(), - SetGoalRequest { - objective: Some("Keep improving the benchmark".to_string()), - status: None, - token_budget: Some(Some(10)), - }, - ) - .await?; - sess.goal_runtime_apply(GoalRuntimeEvent::TurnStarted { - turn_context: tc.as_ref(), - token_usage: TokenUsage::default(), - }) - .await?; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; - while rx.try_recv().is_ok() {} - - set_total_token_usage( - &sess, - TokenUsage { - input_tokens: 20, - cached_input_tokens: 0, - output_tokens: 5, - reasoning_output_tokens: 0, - total_tokens: 25, - }, - ) - .await; - - sess.goal_runtime_apply(GoalRuntimeEvent::ToolCompleted { - turn_context: tc.as_ref(), - tool_name: "shell_command", - }) - .await?; - - let pending_input = sess.input_queue.get_pending_input(&sess.active_turn).await; - let [TurnInput::ResponseItem(ResponseItem::Message { role, content, .. })] = - pending_input.as_slice() - else { - panic!("expected one budget-limit steering message, got {pending_input:#?}"); - }; - assert_eq!("user", role); - let [ContentItem::InputText { text }] = content.as_slice() else { - panic!("expected one text span in budget-limit steering message, got {content:#?}"); - }; - assert!(text.starts_with("")); - assert!(text.trim_end().ends_with("")); - assert!(text.contains("budget_limited")); - assert!(text.to_lowercase().contains("wrap up this turn soon")); - assert!(sess.active_turn.lock().await.is_some()); - while let Ok(event) = rx.try_recv() { - assert!( - !matches!(event.msg, EventMsg::TurnAborted(_)), - "budget limit should steer the active turn instead of aborting it" - ); - } - - let state_db = goal_test_state_db(sess.as_ref()).await?; - let goal = state_db - .thread_goals() - .get_thread_goal(sess.thread_id) - .await? - .expect("goal should remain persisted after accounting"); - assert_eq!(codex_state::ThreadGoalStatus::BudgetLimited, goal.status); - assert_eq!(25, goal.tokens_used); - - set_total_token_usage( - &sess, - TokenUsage { - input_tokens: 30, - cached_input_tokens: 0, - output_tokens: 10, - reasoning_output_tokens: 0, - total_tokens: 40, - }, - ) - .await; - sess.goal_runtime_apply(GoalRuntimeEvent::ToolCompletedGoal { - turn_context: tc.as_ref(), - }) - .await?; - - let goal = state_db - .thread_goals() - .get_thread_goal(sess.thread_id) - .await? - .expect("goal should remain persisted after follow-up accounting"); - assert_eq!(codex_state::ThreadGoalStatus::BudgetLimited, goal.status); - assert_eq!(40, goal.tokens_used); - - sess.abort_all_tasks(TurnAbortReason::Interrupted).await; - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn usage_limit_runtime_stops_active_goal_and_prevents_idle_continuation() -> anyhow::Result<()> -{ - let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.set_thread_goal( - tc.as_ref(), - SetGoalRequest { - objective: Some("Keep improving the benchmark".to_string()), - status: None, - token_budget: Some(Some(50)), - }, - ) - .await?; - sess.goal_runtime_apply(GoalRuntimeEvent::TurnStarted { - turn_context: tc.as_ref(), - token_usage: TokenUsage::default(), - }) - .await?; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; - set_total_token_usage(&sess, post_goal_token_usage()).await; - - sess.goal_runtime_apply(GoalRuntimeEvent::UsageLimitReached { - turn_context: tc.as_ref(), - }) - .await?; - - let state_db = goal_test_state_db(sess.as_ref()).await?; - let goal = state_db - .thread_goals() - .get_thread_goal(sess.thread_id) - .await? - .expect("goal should remain persisted after usage limiting"); - assert_eq!(codex_state::ThreadGoalStatus::UsageLimited, goal.status); - assert_eq!(70, goal.tokens_used); - - sess.abort_all_tasks(TurnAbortReason::Replaced).await; - sess.goal_runtime_apply(GoalRuntimeEvent::MaybeContinueIfIdle) - .await?; - assert!(sess.active_turn.lock().await.is_none()); - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn external_goal_mutation_accounts_active_turn_before_status_change() -> anyhow::Result<()> { - let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.set_thread_goal( - tc.as_ref(), - SetGoalRequest { - objective: Some("Keep improving the benchmark".to_string()), - status: None, - token_budget: None, - }, - ) - .await?; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; - set_total_token_usage(&sess, post_goal_token_usage()).await; - - sess.goal_runtime_apply(GoalRuntimeEvent::ExternalMutationStarting) - .await?; - - let state_db = goal_test_state_db(sess.as_ref()).await?; - let goal = state_db - .thread_goals() - .get_thread_goal(sess.thread_id) - .await? - .expect("goal should remain persisted"); - assert_eq!(70, goal.tokens_used); - - let previous_goal = goal.clone(); - let goal_id = goal.goal_id.clone(); - let updated_goal = state_db - .thread_goals() - .update_thread_goal( - sess.thread_id, - codex_state::GoalUpdate { - objective: None, - status: Some(codex_state::ThreadGoalStatus::Complete), - token_budget: None, - expected_goal_id: Some(goal_id), - }, - ) - .await? - .expect("goal status update should succeed"); - sess.goal_runtime_apply(GoalRuntimeEvent::ExternalSet { - external_set: ExternalGoalSet { - goal: updated_goal, - previous_status: ExternalGoalPreviousStatus::from(&previous_goal), - }, - }) - .await?; - - assert!(sess.active_turn.lock().await.is_some()); - let goal = state_db - .thread_goals() - .get_thread_goal(sess.thread_id) - .await? - .expect("goal should remain persisted"); - assert_eq!(codex_state::ThreadGoalStatus::Complete, goal.status); - assert_eq!(70, goal.tokens_used); - - sess.abort_all_tasks(TurnAbortReason::Replaced).await; - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn external_objective_change_steers_active_turn() -> anyhow::Result<()> { - let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; - - let state_db = goal_test_state_db(sess.as_ref()).await?; - let old_goal = state_db - .thread_goals() - .replace_thread_goal( - sess.thread_id, - "Keep improving the benchmark", - codex_state::ThreadGoalStatus::Active, - /*token_budget*/ Some(10_000), - ) - .await?; - let new_goal = state_db - .thread_goals() - .replace_thread_goal( - sess.thread_id, - "Write a concise benchmark summary", - codex_state::ThreadGoalStatus::Active, - /*token_budget*/ Some(10_000), - ) - .await?; - - sess.goal_runtime_apply(GoalRuntimeEvent::ExternalSet { - external_set: ExternalGoalSet { - goal: new_goal, - previous_status: ExternalGoalPreviousStatus::from(&old_goal), - }, - }) - .await?; - - let pending_input = sess.input_queue.get_pending_input(&sess.active_turn).await; - assert!( - pending_input.iter().any(|item| { - matches!( - item, - TurnInput::ResponseItem(ResponseItem::Message { role, content, .. }) - if role == "user" - && content.iter().any(|content| matches!( - content, - ContentItem::InputText { text } - if text.starts_with("") - && text.trim_end().ends_with("") - && text.contains("The active thread goal objective was edited") - && text.contains("Write a concise benchmark summary") - )) - ) - }), - "expected objective-updated steering prompt in pending input: {pending_input:?}" - ); - - sess.abort_all_tasks(TurnAbortReason::Replaced).await; - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn external_active_goal_set_marks_current_turn_for_accounting() -> anyhow::Result<()> { - let (sess, tc, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; - set_total_token_usage(&sess, post_goal_token_usage()).await; - - let state_db = goal_test_state_db(sess.as_ref()).await?; - let goal = state_db - .thread_goals() - .replace_thread_goal( - sess.thread_id, - "Keep improving the benchmark", - codex_state::ThreadGoalStatus::Active, - /*token_budget*/ None, - ) - .await?; - sess.goal_runtime_apply(GoalRuntimeEvent::ExternalSet { - external_set: ExternalGoalSet { - goal, - previous_status: ExternalGoalPreviousStatus::NewGoal, - }, - }) - .await?; - - set_total_token_usage( - &sess, - TokenUsage { - input_tokens: 65, - cached_input_tokens: 10, - output_tokens: 40, - reasoning_output_tokens: 5, - total_tokens: 110, - }, - ) - .await; - sess.goal_runtime_apply(GoalRuntimeEvent::ToolCompleted { - turn_context: tc.as_ref(), - tool_name: "shell_command", - }) - .await?; - - let goal = state_db - .thread_goals() - .get_thread_goal(sess.thread_id) - .await? - .expect("goal should remain persisted"); - assert_eq!(codex_state::ThreadGoalStatus::Active, goal.status); - assert_eq!(25, goal.tokens_used); - - sess.abort_all_tasks(TurnAbortReason::Replaced).await; - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn completed_goal_accounts_current_turn_tokens_before_tool_response() -> anyhow::Result<()> { - let server = start_mock_server().await; - let mut builder = test_codex().with_config(|config| { - config - .features - .enable(Feature::Goals) - .expect("goal mode should be enableable in tests"); - }); - let test = builder.build(&server).await?; - let responses = mount_sse_sequence( - &server, - vec![ - sse(vec![ - ev_response_created("resp-1"), - ev_function_call( - "call-create-goal", - "create_goal", - r#"{"objective":"write a report","token_budget":500}"#, - ), - ev_completed("resp-1"), - ]), - sse(vec![ - ev_response_created("resp-2"), - ev_function_call( - "call-complete-goal", - "update_goal", - r#"{"status":"complete"}"#, - ), - ev_completed_with_tokens("resp-2", /*total_tokens*/ 580), - ]), - sse(vec![ - ev_assistant_message("msg-1", "Goal complete."), - ev_completed("resp-3"), - ]), - ], - ) - .await; - - test.codex - .submit(Op::UserInput { - environments: None, - items: vec![UserInput::Text { - text: "write a report".into(), - text_elements: Vec::new(), - }], - final_output_json_schema: None, - responsesapi_client_metadata: None, - additional_context: Default::default(), - thread_settings: Default::default(), - }) - .await?; - - tokio::time::timeout(std::time::Duration::from_secs(8), async { - loop { - let event = test.codex.next_event().await?; - if matches!(event.msg, EventMsg::TurnComplete(_)) { - return anyhow::Ok(()); - } - } - }) - .await??; - - let complete_output = responses - .function_call_output_text("call-complete-goal") - .expect("complete tool output should be sent to the model"); - let complete_output: serde_json::Value = serde_json::from_str(&complete_output)?; - assert_eq!(complete_output["goal"]["tokensUsed"], 580); - assert_eq!(complete_output["goal"]["status"], "complete"); - assert_eq!(complete_output["remainingTokens"], 0); - assert_eq!( - complete_output["completionBudgetReport"], - "Goal achieved. Report final usage from this tool result's structured goal fields. If `goal.tokenBudget` is present, include token usage from `goal.tokensUsed` and `goal.tokenBudget`. If `goal.timeUsedSeconds` is greater than 0, summarize elapsed time in a concise, human-friendly form appropriate to the response language." - ); - let requests = responses.requests(); - let completion_followup_request = requests - .last() - .expect("completion tool output should be sent in a follow-up request"); - assert!( - !completion_followup_request.body_contains_text("budget_limited"), - "completion follow-up should not include budget-limit steering" - ); - - let state_db = codex_state::StateRuntime::init( - test.config.sqlite_home.clone(), - test.config.model_provider_id.clone(), - ) - .await?; - let persisted_goal = state_db - .thread_goals() - .get_thread_goal(test.session_configured.thread_id) - .await? - .expect("goal should be persisted"); - assert_eq!( - codex_state::ThreadGoalStatus::Complete, - persisted_goal.status - ); - assert_eq!(580, persisted_goal.tokens_used); - - Ok(()) -} - #[tokio::test] async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { let (sess, tc, _rx) = make_session_and_context_with_rx().await; @@ -10502,297 +9613,6 @@ async fn sample_rollout( ) } -#[tokio::test] -async fn create_goal_tool_rejects_existing_goal() { - let (session, turn_context, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); - let handler = CreateGoalHandler; - - handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker: Arc::clone(&tracker), - call_id: "create-goal-1".to_string(), - tool_name: codex_tools::ToolName::plain("create_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "objective": "Keep the watcher alive", - "token_budget": 123, - }) - .to_string(), - }, - }) - .await - .expect("initial create_goal should succeed"); - - let response = handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker, - call_id: "create-goal-2".to_string(), - tool_name: codex_tools::ToolName::plain("create_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "objective": "Replace the watcher", - "token_budget": 456, - }) - .to_string(), - }, - }) - .await; - - let Err(FunctionCallError::RespondToModel(output)) = response else { - panic!("expected create_goal to reject an existing goal"); - }; - assert_eq!( - output, - "cannot create a new goal because this thread already has a goal; use update_goal only when the existing goal is complete" - ); - - let goal = session - .get_thread_goal() - .await - .expect("read thread goal") - .expect("goal should still exist"); - assert_eq!(goal.objective, "Keep the watcher alive"); - assert_eq!(goal.token_budget, Some(123)); -} - -#[tokio::test] -async fn update_goal_tool_rejects_pausing_goal() { - let (session, turn_context, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); - let create_handler = CreateGoalHandler; - let update_handler = UpdateGoalHandler; - - create_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker: Arc::clone(&tracker), - call_id: "create-goal".to_string(), - tool_name: codex_tools::ToolName::plain("create_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "objective": "Keep the watcher alive", - "token_budget": 123, - }) - .to_string(), - }, - }) - .await - .expect("initial create_goal should succeed"); - - let response = update_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker, - call_id: "pause-goal".to_string(), - tool_name: codex_tools::ToolName::plain("update_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "status": "paused", - }) - .to_string(), - }, - }) - .await; - - let Err(FunctionCallError::RespondToModel(output)) = response else { - panic!("expected update_goal to reject pausing a goal"); - }; - assert_eq!( - output, - "update_goal can only mark the existing goal complete or blocked; pause, resume, budget-limited, and usage-limited status changes are controlled by the user or system" - ); - - let goal = session - .get_thread_goal() - .await - .expect("read thread goal") - .expect("goal should still exist"); - assert_eq!(goal.status, ThreadGoalStatus::Active); -} - -#[tokio::test] -async fn update_goal_tool_marks_goal_blocked() { - let (session, turn_context, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); - let create_handler = CreateGoalHandler; - let update_handler = UpdateGoalHandler; - - create_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker: Arc::clone(&tracker), - call_id: "create-goal".to_string(), - tool_name: codex_tools::ToolName::plain("create_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "objective": "Keep the watcher alive", - "token_budget": 123, - }) - .to_string(), - }, - }) - .await - .expect("initial create_goal should succeed"); - - update_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker, - call_id: "block-goal".to_string(), - tool_name: codex_tools::ToolName::plain("update_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "status": "blocked", - }) - .to_string(), - }, - }) - .await - .expect("update_goal should mark the goal blocked"); - - let goal = session - .get_thread_goal() - .await - .expect("read thread goal") - .expect("goal should still exist"); - assert_eq!(goal.status, ThreadGoalStatus::Blocked); -} - -#[tokio::test] -async fn update_goal_tool_rejects_usage_limited_goal() { - let (session, turn_context, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); - let create_handler = CreateGoalHandler; - let update_handler = UpdateGoalHandler; - - create_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker: Arc::clone(&tracker), - call_id: "create-goal".to_string(), - tool_name: codex_tools::ToolName::plain("create_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "objective": "Keep the watcher alive", - }) - .to_string(), - }, - }) - .await - .expect("initial create_goal should succeed"); - - let response = update_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker, - call_id: "usage-limit-goal".to_string(), - tool_name: codex_tools::ToolName::plain("update_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "status": "usageLimited", - }) - .to_string(), - }, - }) - .await; - - let Err(FunctionCallError::RespondToModel(output)) = response else { - panic!("expected update_goal to reject usage-limiting a goal"); - }; - assert_eq!( - output, - "update_goal can only mark the existing goal complete or blocked; pause, resume, budget-limited, and usage-limited status changes are controlled by the user or system" - ); - - let goal = session - .get_thread_goal() - .await - .expect("read thread goal") - .expect("goal should still exist"); - assert_eq!(goal.status, ThreadGoalStatus::Active); -} - -#[tokio::test] -async fn update_goal_tool_marks_goal_complete() { - let (session, turn_context, _rx, _codex_home) = make_goal_session_and_context_with_rx().await; - let tracker = Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())); - let create_handler = CreateGoalHandler; - let update_handler = UpdateGoalHandler; - - create_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker: Arc::clone(&tracker), - call_id: "create-goal".to_string(), - tool_name: codex_tools::ToolName::plain("create_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "objective": "Keep the watcher alive", - "token_budget": 123, - }) - .to_string(), - }, - }) - .await - .expect("initial create_goal should succeed"); - - update_handler - .handle(ToolInvocation { - session: Arc::clone(&session), - turn: Arc::clone(&turn_context), - cancellation_token: CancellationToken::new(), - tracker, - call_id: "complete-goal".to_string(), - tool_name: codex_tools::ToolName::plain("update_goal"), - source: ToolCallSource::Direct, - payload: ToolPayload::Function { - arguments: serde_json::json!({ - "status": "complete", - }) - .to_string(), - }, - }) - .await - .expect("update_goal should mark the goal complete"); - - let goal = session - .get_thread_goal() - .await - .expect("read thread goal") - .expect("goal should still exist"); - assert_eq!(goal.status, ThreadGoalStatus::Complete); -} - #[tokio::test] async fn rejects_escalated_permissions_when_policy_not_on_request() { use crate::exec_policy::ExecApprovalRequest; diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 4a9316496c..451dfdd61f 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -17,7 +17,6 @@ use crate::compact_remote_v2::run_inline_remote_auto_compact_task as run_inline_ use crate::connectors; use crate::context::ContextualUserFragment; use crate::feedback_tags; -use crate::goals::GoalRuntimeEvent; use crate::hook_runtime::inspect_pending_input; use crate::hook_runtime::record_additional_contexts; use crate::hook_runtime::record_pending_input; @@ -151,15 +150,6 @@ pub(crate) async fn run_turn( let error = err.to_codex_protocol_error(); sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone()) .await; - if error == CodexErrorInfo::UsageLimitExceeded - && let Err(err) = sess - .goal_runtime_apply(GoalRuntimeEvent::UsageLimitReached { - turn_context: turn_context.as_ref(), - }) - .await - { - warn!("failed to usage-limit active goal after usage-limit error: {err}"); - } error!("Failed to run pre-sampling compact"); return None; } @@ -304,17 +294,6 @@ pub(crate) async fn run_turn( let error = err.to_codex_protocol_error(); sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone()) .await; - if error == CodexErrorInfo::UsageLimitExceeded - && let Err(err) = sess - .goal_runtime_apply(GoalRuntimeEvent::UsageLimitReached { - turn_context: turn_context.as_ref(), - }) - .await - { - warn!( - "failed to usage-limit active goal after usage-limit error: {err}" - ); - } return None; } can_drain_pending_input = !model_needs_follow_up; @@ -400,15 +379,6 @@ pub(crate) async fn run_turn( let error = e.to_codex_protocol_error(); sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone()) .await; - if error == CodexErrorInfo::UsageLimitExceeded - && let Err(err) = sess - .goal_runtime_apply(GoalRuntimeEvent::UsageLimitReached { - turn_context: turn_context.as_ref(), - }) - .await - { - warn!("failed to usage-limit active goal after usage-limit error: {err}"); - } sess.track_turn_codex_error(turn_context.as_ref(), &e); let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None)); sess.send_event(&turn_context, event).await; @@ -1383,78 +1353,7 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option { TurnItem::AgentMessage(item) => Some(agent_message_text(item)), _ => None, }, - EventMsg::Error(_) - | EventMsg::Warning(_) - | EventMsg::GuardianWarning(_) - | EventMsg::RealtimeConversationStarted(_) - | EventMsg::RealtimeConversationSdp(_) - | EventMsg::RealtimeConversationRealtime(_) - | EventMsg::RealtimeConversationClosed(_) - | EventMsg::ModelReroute(_) - | EventMsg::ModelVerification(_) - | EventMsg::ContextCompacted(_) - | EventMsg::ThreadRolledBack(_) - | EventMsg::TurnStarted(_) - | EventMsg::ThreadSettingsApplied(_) - | EventMsg::TurnComplete(_) - | EventMsg::TokenCount(_) - | EventMsg::UserMessage(_) - | EventMsg::AgentReasoning(_) - | EventMsg::AgentReasoningRawContent(_) - | EventMsg::AgentReasoningSectionBreak(_) - | EventMsg::SessionConfigured(_) - | EventMsg::ThreadGoalUpdated(_) - | EventMsg::McpStartupUpdate(_) - | EventMsg::McpStartupComplete(_) - | EventMsg::McpToolCallBegin(_) - | EventMsg::McpToolCallEnd(_) - | EventMsg::WebSearchBegin(_) - | EventMsg::WebSearchEnd(_) - | EventMsg::ExecCommandBegin(_) - | EventMsg::ExecCommandOutputDelta(_) - | EventMsg::TerminalInteraction(_) - | EventMsg::ExecCommandEnd(_) - | EventMsg::PatchApplyBegin(_) - | EventMsg::PatchApplyUpdated(_) - | EventMsg::PatchApplyEnd(_) - | EventMsg::ImageGenerationBegin(_) - | EventMsg::ImageGenerationEnd(_) - | EventMsg::ViewImageToolCall(_) - | EventMsg::ExecApprovalRequest(_) - | EventMsg::RequestPermissions(_) - | EventMsg::RequestUserInput(_) - | EventMsg::DynamicToolCallRequest(_) - | EventMsg::DynamicToolCallResponse(_) - | EventMsg::GuardianAssessment(_) - | EventMsg::ElicitationRequest(_) - | EventMsg::ApplyPatchApprovalRequest(_) - | EventMsg::DeprecationNotice(_) - | EventMsg::StreamError(_) - | EventMsg::TurnDiff(_) - | EventMsg::RealtimeConversationListVoicesResponse(_) - | EventMsg::PlanUpdate(_) - | EventMsg::TurnAborted(_) - | EventMsg::ShutdownComplete - | EventMsg::EnteredReviewMode(_) - | EventMsg::ExitedReviewMode(_) - | EventMsg::RawResponseItem(_) - | EventMsg::ItemStarted(_) - | EventMsg::HookStarted(_) - | EventMsg::HookCompleted(_) - | EventMsg::AgentMessageContentDelta(_) - | EventMsg::PlanDelta(_) - | EventMsg::ReasoningContentDelta(_) - | EventMsg::ReasoningRawContentDelta(_) - | EventMsg::CollabAgentSpawnBegin(_) - | EventMsg::CollabAgentSpawnEnd(_) - | EventMsg::CollabAgentInteractionBegin(_) - | EventMsg::CollabAgentInteractionEnd(_) - | EventMsg::CollabWaitingBegin(_) - | EventMsg::CollabWaitingEnd(_) - | EventMsg::CollabCloseBegin(_) - | EventMsg::CollabCloseEnd(_) - | EventMsg::CollabResumeBegin(_) - | EventMsg::CollabResumeEnd(_) => None, + _ => None, } } diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 66a32f2aa6..5f52912d59 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -91,7 +91,6 @@ pub struct TurnContext { pub(crate) shell_environment_policy: ShellEnvironmentPolicy, pub(crate) available_models: Vec, pub(crate) unified_exec_shell_mode: UnifiedExecShellMode, - pub(crate) goal_tools_supported: bool, pub features: ManagedFeatures, pub(crate) ghost_snapshot: GhostSnapshotConfig, pub(crate) final_output_json_schema: Option, @@ -170,10 +169,6 @@ impl TurnContext { ToolEnvironmentMode::from_count(self.environments.turn_environments.len()) } - pub(crate) fn goal_tools_enabled(&self) -> bool { - self.goal_tools_supported && self.features.get().enabled(Feature::Goals) - } - pub(crate) async fn with_model( &self, model: String, @@ -264,7 +259,6 @@ impl TurnContext { shell_environment_policy: self.shell_environment_policy.clone(), available_models, unified_exec_shell_mode: self.unified_exec_shell_mode.clone(), - goal_tools_supported: self.goal_tools_supported, features, ghost_snapshot: self.ghost_snapshot.clone(), final_output_json_schema: self.final_output_json_schema.clone(), @@ -477,7 +471,6 @@ impl Session { cwd: AbsolutePathBuf, sub_id: String, skills_outcome: Arc, - goal_tools_supported: bool, ) -> TurnContext { let reasoning_effort = session_configuration.collaboration_mode.reasoning_effort(); let reasoning_summary = session_configuration @@ -568,7 +561,6 @@ impl Session { shell_environment_policy: per_turn_config.permissions.shell_environment_policy.clone(), available_models, unified_exec_shell_mode, - goal_tools_supported, features: per_turn_config.features.clone(), ghost_snapshot: per_turn_config.ghost_snapshot.clone(), final_output_json_schema: None, @@ -781,7 +773,6 @@ impl Session { .skills_for_config(&skills_input, fs) .await, ); - let goal_tools_supported = !per_turn_config.ephemeral && self.state_db().is_some(); let mut turn_context: TurnContext = Self::make_turn_context( self.thread_id(), self.session_id(), @@ -810,7 +801,6 @@ impl Session { cwd, sub_id, skills_outcome, - goal_tools_supported, ); turn_context.realtime_active = self.conversation.running_state().await.is_some(); diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index ef2ba0f16d..3bfcc1fbfc 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -23,7 +23,6 @@ use tracing::warn; use crate::config::Config; use crate::context::ContextualUserFragment; -use crate::goals::GoalRuntimeEvent; use crate::hook_runtime::inspect_pending_input; use crate::hook_runtime::record_additional_contexts; use crate::hook_runtime::record_pending_input; @@ -341,15 +340,6 @@ impl Session { .await .clear_turn(&turn_context.sub_id); - if let Err(err) = self - .goal_runtime_apply(GoalRuntimeEvent::TurnStarted { - turn_context: turn_context.as_ref(), - token_usage: token_usage_at_turn_start.clone(), - }) - .await - { - warn!("failed to apply goal runtime turn-start event: {err}"); - } let pending_items = self.input_queue.get_pending_input(&self.active_turn).await; let turn_state = { let mut active = self.active_turn.lock().await; @@ -503,15 +493,6 @@ impl Session { self.emit_turn_abort_lifecycle(reason.clone(), turn_context.extension_data.as_ref()) .await; } - if (aborted_turn || reason == TurnAbortReason::Interrupted) - && let Err(err) = self - .goal_runtime_apply(GoalRuntimeEvent::TaskAborted { - turn_context: turn_context.as_deref(), - }) - .await - { - warn!("failed to apply goal runtime abort event: {err}"); - } if let Some(active_turn) = active_turn_to_clear { // Let interrupted tasks observe cancellation before dropping pending approvals, or an // in-flight approval wait can surface as a model-visible rejection before TurnAborted. @@ -552,14 +533,6 @@ impl Session { self.emit_turn_abort_lifecycle(reason.clone(), turn_context.extension_data.as_ref()) .await; } - if let Err(err) = self - .goal_runtime_apply(GoalRuntimeEvent::TaskAborted { - turn_context: turn_context.as_deref(), - }) - .await - { - warn!("failed to apply goal runtime abort event: {err}"); - } // Let interrupted tasks observe cancellation before dropping pending approvals, or an // in-flight approval wait can surface as a model-visible rejection before TurnAborted. self.input_queue.clear_pending(&active_turn).await; @@ -753,15 +726,6 @@ impl Session { .await; self.emit_turn_stop_lifecycle(turn_context.extension_data.as_ref()) .await; - if let Err(err) = self - .goal_runtime_apply(GoalRuntimeEvent::TurnFinished { - turn_context: turn_context.as_ref(), - turn_completed: true, - }) - .await - { - warn!("failed to apply goal runtime turn-finished event: {err}"); - } let event = EventMsg::TurnComplete(TurnCompleteEvent { turn_id: turn_context.sub_id.clone(), last_agent_message, @@ -791,12 +755,6 @@ impl Session { if !cleared_active_turn { return; } - if let Err(err) = self - .goal_runtime_apply(GoalRuntimeEvent::MaybeContinueIfIdle) - .await - { - warn!("failed to apply goal runtime maybe-continue event: {err}"); - } self.emit_thread_idle_lifecycle_if_idle().await; } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index d130a35815..1cde8c9320 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -1343,9 +1343,6 @@ impl ThreadManagerState { .await?; if is_resumed_thread { new_thread.thread.emit_thread_resume_lifecycle().await; - if let Err(err) = new_thread.thread.apply_goal_resume_runtime_effects().await { - warn!("failed to apply goal resume runtime effects: {err}"); - } } Ok(new_thread) } diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 52a0f8eb3b..34ec682110 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -8,7 +8,6 @@ use crate::session::tests::make_session_and_context; use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; use codex_extension_api::empty_extension_registry; -use codex_features::Feature; use codex_models_manager::manager::RefreshStrategy; use codex_protocol::models::ContentItem; use codex_protocol::models::ReasoningItemReasoningSummary; @@ -1501,103 +1500,3 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ 1, ); } - -#[tokio::test] -async fn resumed_thread_keeps_paused_goal_paused() -> anyhow::Result<()> { - let temp_dir = tempdir().expect("tempdir"); - let mut config = test_config().await; - config.codex_home = temp_dir.path().join("codex-home").abs(); - config.cwd = config.codex_home.abs(); - config - .features - .enable(Feature::Goals) - .expect("goals should be enableable in tests"); - std::fs::create_dir_all(&config.codex_home).expect("create codex home"); - - let auth_manager = - AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing()); - let state_db = init_state_db(&config).await; - let manager = ThreadManager::new( - &config, - auth_manager.clone(), - SessionSource::Exec, - Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), - empty_extension_registry(), - /*analytics_events_client*/ None, - thread_store_from_config(&config, state_db.clone()), - state_db.clone(), - TEST_INSTALLATION_ID.to_string(), - /*attestation_provider*/ None, - ); - - let source = manager - .resume_thread_with_history( - config.clone(), - InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("keep working"))]), - auth_manager.clone(), - /*parent_trace*/ None, - ) - .await - .expect("create source thread"); - let source_path = source - .thread - .rollout_path() - .expect("source rollout path should exist"); - source.thread.flush_rollout().await?; - let state_db = source - .thread - .state_db() - .expect("source thread should have a state db"); - state_db - .thread_goals() - .replace_thread_goal( - source.thread_id, - "Keep working until the task is done", - codex_state::ThreadGoalStatus::Paused, - /*token_budget*/ None, - ) - .await?; - source.thread.shutdown_and_wait().await?; - manager.remove_thread(&source.thread_id).await; - - let resumed = manager - .resume_thread_from_rollout( - config.clone(), - source_path, - auth_manager, - /*parent_trace*/ None, - ) - .await - .expect("resume source thread"); - let goal = state_db - .thread_goals() - .get_thread_goal(resumed.thread_id) - .await? - .expect("goal should still exist after resume"); - assert_eq!(codex_state::ThreadGoalStatus::Paused, goal.status); - assert!( - resumed - .thread - .codex - .session - .active_turn - .lock() - .await - .is_none() - ); - - resumed.thread.continue_active_goal_if_idle().await?; - assert!( - resumed - .thread - .codex - .session - .active_turn - .lock() - .await - .is_none() - ); - - resumed.thread.shutdown_and_wait().await?; - Ok(()) -} diff --git a/codex-rs/core/src/tools/handlers/mod.rs b/codex-rs/core/src/tools/handlers/mod.rs index d6963f0761..fdfb809965 100644 --- a/codex-rs/core/src/tools/handlers/mod.rs +++ b/codex-rs/core/src/tools/handlers/mod.rs @@ -4,8 +4,6 @@ pub(crate) mod apply_patch; pub(crate) mod apply_patch_spec; mod dynamic; pub(crate) mod extension_tools; -mod goal; -pub(crate) mod goal_spec; mod list_available_plugins_to_install; pub(crate) mod list_available_plugins_to_install_spec; mod mcp; @@ -53,9 +51,6 @@ pub use apply_patch::ApplyPatchHandler; use codex_protocol::models::AdditionalPermissionProfile; use codex_protocol::protocol::AskForApproval; pub use dynamic::DynamicToolHandler; -pub use goal::CreateGoalHandler; -pub use goal::GetGoalHandler; -pub use goal::UpdateGoalHandler; pub use list_available_plugins_to_install::ListAvailablePluginsToInstallHandler; pub use mcp::McpHandler; pub use mcp_resource::ListMcpResourceTemplatesHandler; diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index b2a0b7704a..1cc5e67eb5 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -5,7 +5,6 @@ use std::sync::atomic::Ordering; use std::time::Duration; use crate::function_tool::FunctionCallError; -use crate::goals::GoalRuntimeEvent; use crate::hook_runtime::PreToolUseHookResult; use crate::hook_runtime::record_additional_contexts; use crate::hook_runtime::run_post_tool_use_hooks; @@ -34,7 +33,6 @@ use codex_tools::ToolSearchInfo; use codex_tools::ToolSpec; use futures::future::BoxFuture; use serde_json::Value; -use tracing::warn; pub(crate) type ToolTelemetryTags = Vec<(&'static str, String)>; @@ -649,25 +647,13 @@ impl ToolRegistry { handler_executed: true, }, }; - let finished = notify_tool_finish_if_unclaimed( + notify_tool_finish_if_unclaimed( &invocation, terminal_outcome_reached.as_deref(), lifecycle_outcome, ) .await; - if finished - && let Err(err) = invocation - .session - .goal_runtime_apply(GoalRuntimeEvent::ToolCompleted { - turn_context: invocation.turn.as_ref(), - tool_name: tool_name.name.as_str(), - }) - .await - { - warn!("failed to account thread goal progress after tool call: {err}"); - } - match result { Ok(_) => { let mut guard = response_cell.lock().await; diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index eae0644b83..fd65119cb9 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/codex-rs/core/src/tools/spec_plan.rs @@ -6,11 +6,9 @@ use crate::tools::context::ToolInvocation; use crate::tools::handlers::ApplyPatchHandler; use crate::tools::handlers::CodeModeExecuteHandler; use crate::tools::handlers::CodeModeWaitHandler; -use crate::tools::handlers::CreateGoalHandler; use crate::tools::handlers::DynamicToolHandler; use crate::tools::handlers::ExecCommandHandler; use crate::tools::handlers::ExecCommandHandlerOptions; -use crate::tools::handlers::GetGoalHandler; use crate::tools::handlers::ListAvailablePluginsToInstallHandler; use crate::tools::handlers::ListMcpResourceTemplatesHandler; use crate::tools::handlers::ListMcpResourcesHandler; @@ -24,7 +22,6 @@ use crate::tools::handlers::ShellCommandHandler; use crate::tools::handlers::ShellCommandHandlerOptions; use crate::tools::handlers::TestSyncHandler; use crate::tools::handlers::ToolSearchHandler; -use crate::tools::handlers::UpdateGoalHandler; use crate::tools::handlers::ViewImageHandler; use crate::tools::handlers::WriteStdinHandler; use crate::tools::handlers::agent_jobs::ReportAgentJobResultHandler; @@ -305,14 +302,6 @@ fn collab_tools_enabled(turn_context: &TurnContext) -> bool { } } -fn goal_tools_enabled(turn_context: &TurnContext) -> bool { - turn_context.goal_tools_enabled() - && !matches!( - turn_context.session_source, - SessionSource::SubAgent(SubAgentSource::Review) - ) -} - fn agent_jobs_tools_enabled(turn_context: &TurnContext) -> bool { turn_context.features.get().enabled(Feature::SpawnCsv) && collab_tools_enabled(turn_context) } @@ -639,11 +628,6 @@ fn add_core_utility_tools(context: &CoreToolPlanContext<'_>, planned_tools: &mut let environment_mode = turn_context.tool_environment_mode(); planned_tools.add(PlanHandler); - if goal_tools_enabled(turn_context) { - planned_tools.add(GetGoalHandler); - planned_tools.add(CreateGoalHandler); - planned_tools.add(UpdateGoalHandler); - } if turn_context.config.experimental_request_user_input_enabled { planned_tools.add(RequestUserInputHandler { diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index bf5b1c01bb..0fa458d2e9 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -619,36 +619,7 @@ async fn environment_count_controls_environment_backed_tools() { } #[tokio::test] -async fn host_context_gates_goal_and_agent_job_tools() { - let feature_disabled = probe(|turn| { - set_feature(turn, Feature::Goals, /*enabled*/ false); - turn.goal_tools_supported = true; - }) - .await; - feature_disabled.assert_visible_lacks(&["get_goal", "create_goal", "update_goal"]); - - let host_disabled = probe(|turn| { - set_feature(turn, Feature::Goals, /*enabled*/ true); - turn.goal_tools_supported = false; - }) - .await; - host_disabled.assert_visible_lacks(&["get_goal", "create_goal", "update_goal"]); - - let enabled = probe(|turn| { - set_feature(turn, Feature::Goals, /*enabled*/ true); - turn.goal_tools_supported = true; - }) - .await; - enabled.assert_visible_contains(&["get_goal", "create_goal", "update_goal"]); - - let review_thread = probe(|turn| { - set_feature(turn, Feature::Goals, /*enabled*/ true); - turn.goal_tools_supported = true; - turn.session_source = SessionSource::SubAgent(SubAgentSource::Review); - }) - .await; - review_thread.assert_visible_lacks(&["get_goal", "create_goal", "update_goal"]); - +async fn host_context_gates_agent_job_tools() { let normal_agent_job = probe(|turn| { set_feature(turn, Feature::SpawnCsv, /*enabled*/ true); }) diff --git a/codex-rs/core/tests/suite/prompt_caching.rs b/codex-rs/core/tests/suite/prompt_caching.rs index ea646a4dbc..b931093883 100644 --- a/codex-rs/core/tests/suite/prompt_caching.rs +++ b/codex-rs/core/tests/suite/prompt_caching.rs @@ -188,9 +188,6 @@ async fn prompt_tools_are_consistent_across_requests() -> anyhow::Result<()> { }; expected_tools_names.extend([ "update_plan", - "get_goal", - "create_goal", - "update_goal", "request_user_input", "apply_patch", "view_image",