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

Wire app-server goals through extension

Browse Source
This commit is contained in:
Eric Traut
2026-06-04 23:19:46 -07:00
parent 919dbbefd1
commit 64e035e530
23 changed files with 130 additions and 1781 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+1
View File
@@ -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",
codex-rs/app-server/Cargo.toml
+1
View File
@@ -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 }
codex-rs/app-server/src/extensions.rs
+15
View File
@@ -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<S>(
guardian_agent_spawner: S,
event_sink: Arc<dyn ExtensionEventSink>,
auth_manager: Arc<AuthManager>,
state_db: Option<StateDbHandle>,
thread_manager: Weak<ThreadManager>,
goal_service: Arc<GoalService>,
) -> Arc<ExtensionRegistry<Config>>
where
S: AgentSpawner<StartThreadOptions, Spawned = NewThread, Error = CodexErr> + 'static,
{
let mut builder = ExtensionRegistryBuilder::<Config>::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());
codex-rs/app-server/src/mcp_refresh.rs
+3
View File
@@ -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),
codex-rs/app-server/src/message_processor.rs
+6
View File
@@ -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(),
codex-rs/app-server/src/request_processors.rs
-2
View File
@@ -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)]
codex-rs/app-server/src/request_processors/thread_goal_processor.rs
+94 -206
View File
@@ -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<Config>,
thread_state_manager: ThreadStateManager,
state_db: Option<StateDbHandle>,
goal_service: Arc<GoalService>,
}
impl ThreadGoalRequestProcessor {
@@ -17,6 +22,7 @@ impl ThreadGoalRequestProcessor {
config: Arc<Config>,
thread_state_manager: ThreadStateManager,
state_db: Option<StateDbHandle>,
goal_service: Arc<GoalService>,
) -> 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<i64>) -> 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),
}
}
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+3 -11
View File
@@ -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;
}
}
codex-rs/core/src/codex_thread.rs
+3 -45
View File
@@ -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)]
codex-rs/core/src/lib.rs
-3
View File
@@ -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;
codex-rs/core/src/session/review.rs
-3
View File
@@ -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(),
codex-rs/core/src/session/session.rs
-3
View File
@@ -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<RealtimeConversationManager>,
pub(crate) active_turn: Mutex<Option<ActiveTurn>>,
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),
codex-rs/core/src/session/tests.rs
+1 -1181
View File
File diff suppressed because it is too large Load Diff
codex-rs/core/src/session/turn.rs
+1 -102
View File
@@ -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<String> {
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,
}
}
codex-rs/core/src/session/turn_context.rs
-10
View File
@@ -91,7 +91,6 @@ pub struct TurnContext {
pub(crate) shell_environment_policy: ShellEnvironmentPolicy,
pub(crate) available_models: Vec<ModelPreset>,
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<Value>,
@@ -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<SkillLoadOutcome>,
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();
codex-rs/core/src/tasks/mod.rs
-42
View File
@@ -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;
}
codex-rs/core/src/thread_manager.rs
-3
View File
@@ -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)
}
codex-rs/core/src/thread_manager_tests.rs
-101
View File
@@ -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(())
}
codex-rs/core/src/tools/handlers/mod.rs
-5
View File
@@ -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;
codex-rs/core/src/tools/registry.rs
+1 -15
View File
@@ -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;
codex-rs/core/src/tools/spec_plan.rs
-16
View File
@@ -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 {
codex-rs/core/src/tools/spec_plan_tests.rs
+1 -30
View File
@@ -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);
})
codex-rs/core/tests/suite/prompt_caching.rs
-3
View File
@@ -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",