[2 of 2] Finish moving goal runtime to extension (#26548)

## Stack

1. [#26547](https://github.com/openai/codex/pull/26547) - [1 of 2] Align
goal extension with core behavior
2. [#26548](https://github.com/openai/codex/pull/26548) - [2 of 2] Move
goal runtime to extension

## Why

This PR completes the switch of the goal behavior to the
extension-backed runtime and removes the old core goal implementation.

## What Changed

- Installs the goal extension for app-server `ThreadManager` sessions.
- Routes app-server thread goal `get`, `set`, and `clear` through
`GoalService`.
- Uses thread-idle lifecycle emission after goal resume and snapshot
ordering so the extension can decide whether to continue the goal.
- Forwards extension goal updates through a FIFO async app-server
notification path so backpressure does not drop them or reorder updates.
- Keeps review turns from enabling goal runtime behavior.
- Plans extension tools before dynamic tools so built-in goal tool names
keep their old precedence when goals are enabled.
- Removes the old core goal runtime, core goal tool handlers, and core
goal tool specs.
- Updates tests that were coupled to the core-owned goal runtime while
leaving the legacy `<goal_context>` compatibility path in core for old
threads.
- Removes the stale cargo-shear ignore now that `codex-goal-extension`
is used by the workspace.
- Keeps realtime event matching exhaustive after removing the old
goal-specific realtime text path.


## Validation

- Ran manual `/goal` runs in TUI. Validated time accounting matched
wall-clock time and goal lifecycle state transitions.

codex-rs/app-server/src/extensions.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::sync::Weak;
 
 use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::ThreadGoal;
 use codex_app_server_protocol::ThreadGoalUpdatedNotification;
 use codex_core::NewThread;
 use codex_core::StartThreadOptions;
@@ -12,23 +13,40 @@ 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;
+use crate::thread_state::ThreadListenerCommand;
+use crate::thread_state::ThreadStateManager;
 
 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());
@@ -38,26 +56,53 @@ where
 
 pub(crate) fn app_server_extension_event_sink(
     outgoing: Arc<OutgoingMessageSender>,
+    thread_state_manager: ThreadStateManager,
 ) -> Arc<dyn ExtensionEventSink> {
-    Arc::new(AppServerExtensionEventSink { outgoing })
+    Arc::new(AppServerExtensionEventSink {
+        outgoing,
+        thread_state_manager,
+    })
 }
 
 struct AppServerExtensionEventSink {
     outgoing: Arc<OutgoingMessageSender>,
+    thread_state_manager: ThreadStateManager,
 }
 
 impl ExtensionEventSink for AppServerExtensionEventSink {
     fn emit(&self, event: Event) {
         match event.msg {
             EventMsg::ThreadGoalUpdated(thread_goal_event) => {
-                self.outgoing
-                    .try_send_server_notification(ServerNotification::ThreadGoalUpdated(
-                        ThreadGoalUpdatedNotification {
-                            thread_id: thread_goal_event.thread_id.to_string(),
-                            turn_id: thread_goal_event.turn_id,
-                            goal: thread_goal_event.goal.into(),
-                        },
-                    ));
+                let thread_id = thread_goal_event.thread_id;
+                let turn_id = thread_goal_event.turn_id;
+                let goal: ThreadGoal = thread_goal_event.goal.into();
+                if let Some(listener_command_tx) = self
+                    .thread_state_manager
+                    .current_listener_command_tx(thread_id)
+                {
+                    let command = ThreadListenerCommand::EmitThreadGoalUpdated {
+                        turn_id: turn_id.clone(),
+                        goal: goal.clone(),
+                    };
+                    if listener_command_tx.send(command).is_ok() {
+                        return;
+                    }
+                    tracing::warn!(
+                        "failed to enqueue extension goal update for {thread_id}: listener command channel is closed"
+                    );
+                }
+                let outgoing = Arc::clone(&self.outgoing);
+                tokio::spawn(async move {
+                    outgoing
+                        .send_server_notification(ServerNotification::ThreadGoalUpdated(
+                            ThreadGoalUpdatedNotification {
+                                thread_id: thread_id.to_string(),
+                                turn_id,
+                                goal,
+                            },
+                        ))
+                        .await;
+                });
             }
             msg => {
                 tracing::debug!(event_id = %event.id, ?msg, "dropping unsupported extension event");
@@ -89,10 +134,7 @@ mod tests {
     use std::time::Duration;
 
     use codex_analytics::AnalyticsEventsClient;
-    use codex_app_server_protocol::ServerNotification;
-    use codex_app_server_protocol::ThreadGoal as AppServerThreadGoal;
-    use codex_app_server_protocol::ThreadGoalStatus as AppServerThreadGoalStatus;
-    use codex_protocol::protocol::ThreadGoal;
+    use codex_protocol::protocol::ThreadGoal as CoreThreadGoal;
     use codex_protocol::protocol::ThreadGoalStatus;
     use codex_protocol::protocol::ThreadGoalUpdatedEvent;
     use pretty_assertions::assert_eq;
@@ -100,25 +142,61 @@ mod tests {
     use tokio::time::timeout;
 
     use super::*;
-    use crate::outgoing_message::OutgoingEnvelope;
-    use crate::outgoing_message::OutgoingMessage;
 
     #[tokio::test]
-    async fn app_server_event_sink_forwards_thread_goal_updates() {
-        let (outgoing_tx, mut outgoing_rx) = mpsc::channel(4);
+    async fn app_server_event_sink_uses_listener_fifo_for_goal_updates_and_clears() {
+        let (outgoing_tx, _outgoing_rx) = mpsc::channel(4);
         let outgoing = Arc::new(OutgoingMessageSender::new(
             outgoing_tx,
             AnalyticsEventsClient::disabled(),
         ));
-        let sink = app_server_extension_event_sink(outgoing);
+        let thread_state_manager = ThreadStateManager::new();
         let thread_id = ThreadId::default();
+        let (listener_command_tx, mut listener_command_rx) = mpsc::unbounded_channel();
+        thread_state_manager.register_listener_command_tx(thread_id, listener_command_tx.clone());
+        let sink = app_server_extension_event_sink(outgoing, thread_state_manager);
 
-        sink.emit(Event {
-            id: "call-1".to_string(),
+        for turn_id in ["turn-1", "turn-2"] {
+            sink.emit(thread_goal_updated_event(thread_id, turn_id));
+        }
+        listener_command_tx
+            .send(ThreadListenerCommand::EmitThreadGoalCleared)
+            .expect("listener command channel should be open");
+
+        let mut observed = Vec::new();
+        for _ in 0..3 {
+            let command = timeout(Duration::from_secs(1), listener_command_rx.recv())
+                .await
+                .expect("timed out waiting for listener command")
+                .expect("listener command channel closed unexpectedly");
+            match command {
+                ThreadListenerCommand::EmitThreadGoalUpdated { turn_id, .. } => {
+                    observed.push(turn_id.expect("extension goal updates should include turn ids"));
+                }
+                ThreadListenerCommand::EmitThreadGoalCleared => {
+                    observed.push("cleared".to_string())
+                }
+                _ => panic!("unexpected listener command"),
+            }
+        }
+
+        assert_eq!(
+            vec![
+                "turn-1".to_string(),
+                "turn-2".to_string(),
+                "cleared".to_string()
+            ],
+            observed
+        );
+    }
+
+    fn thread_goal_updated_event(thread_id: ThreadId, turn_id: &str) -> Event {
+        Event {
+            id: turn_id.to_string(),
             msg: EventMsg::ThreadGoalUpdated(ThreadGoalUpdatedEvent {
                 thread_id,
-                turn_id: Some("turn-1".to_string()),
-                goal: ThreadGoal {
+                turn_id: Some(turn_id.to_string()),
+                goal: CoreThreadGoal {
                     thread_id,
                     objective: "wire extension events".to_string(),
                     status: ThreadGoalStatus::Active,
@@ -129,38 +207,6 @@ mod tests {
                     updated_at: 8,
                 },
             }),
-        });
-
-        let envelope = timeout(Duration::from_secs(1), outgoing_rx.recv())
-            .await
-            .expect("timed out waiting for forwarded extension event")
-            .expect("outgoing channel closed unexpectedly");
-        let OutgoingEnvelope::Broadcast { message } = envelope else {
-            panic!("expected broadcast notification");
-        };
-        let OutgoingMessage::AppServerNotification(ServerNotification::ThreadGoalUpdated(
-            notification,
-        )) = message
-        else {
-            panic!("expected thread goal updated notification");
-        };
-
-        assert_eq!(
-            ThreadGoalUpdatedNotification {
-                thread_id: thread_id.to_string(),
-                turn_id: Some("turn-1".to_string()),
-                goal: AppServerThreadGoal {
-                    thread_id: thread_id.to_string(),
-                    objective: "wire extension events".to_string(),
-                    status: AppServerThreadGoalStatus::Active,
-                    token_budget: Some(123),
-                    tokens_used: 45,
-                    time_used_seconds: 6,
-                    created_at: 7,
-                    updated_at: 8,
-                },
-            },
-            notification
-        );
+        }
     }
 }

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),

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(),
@@ -313,8 +315,11 @@ impl MessageProcessor {
                 environment_manager,
                 thread_extensions(
                     guardian_agent_spawner(thread_manager.clone()),
-                    app_server_extension_event_sink(outgoing.clone()),
+                    app_server_extension_event_sink(outgoing.clone(), thread_state_manager.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/outgoing_message.rs

@@ -555,16 +555,6 @@ impl OutgoingMessageSender {
             .await;
     }
 
-    pub(crate) fn try_send_server_notification(&self, notification: ServerNotification) {
-        tracing::trace!("app-server event: {notification}");
-        let outgoing_message = OutgoingMessage::AppServerNotification(notification);
-        if let Err(err) = self.sender.try_send(OutgoingEnvelope::Broadcast {
-            message: outgoing_message,
-        }) {
-            warn!("failed to send server notification to client without waiting: {err:?}");
-        }
-    }
-
     pub(crate) async fn send_server_notification_to_connections(
         &self,
         connection_ids: &[ConnectionId],

codex-rs/app-server/src/request_processors.rs

@@ -279,8 +279,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

@@ -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,
@@ -405,6 +306,7 @@ impl ThreadGoalRequestProcessor {
     ) {
         if let Some(listener_command_tx) = listener_command_tx {
             let command = crate::thread_state::ThreadListenerCommand::EmitThreadGoalUpdated {
+                turn_id: None,
                 goal: goal.clone(),
             };
             if listener_command_tx.send(command).is_ok() {
@@ -449,27 +351,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 +375,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

@@ -244,12 +244,22 @@ pub(super) async fn ensure_listener_task_running(
         if thread_state.listener_matches(&conversation) {
             return Ok(());
         }
-        thread_state.set_listener(
+        let (listener_command_rx, listener_generation) = thread_state.set_listener(
             cancel_tx,
             &conversation,
             watch_registration,
             thread_settings_baseline,
-        )
+        );
+        let Some(listener_command_tx) = thread_state.listener_command_tx() else {
+            tracing::warn!(
+                "thread listener command sender missing immediately after listener registration"
+            );
+            return Ok(());
+        };
+        listener_task_context
+            .thread_state_manager
+            .register_listener_command_tx(conversation_id, listener_command_tx);
+        (listener_command_rx, listener_generation)
     };
     let ListenerTaskContext {
         outgoing,
@@ -378,6 +388,7 @@ pub(super) async fn ensure_listener_task_running(
 
         let mut thread_state = thread_state.lock().await;
         if thread_state.listener_generation == listener_generation {
+            thread_state_manager.unregister_listener_command_tx(conversation_id);
             thread_state.clear_listener();
         }
     });
@@ -471,12 +482,12 @@ pub(super) async fn handle_thread_listener_command(
             )
             .await;
         }
-        ThreadListenerCommand::EmitThreadGoalUpdated { goal } => {
+        ThreadListenerCommand::EmitThreadGoalUpdated { turn_id, goal } => {
             outgoing
                 .send_server_notification(ServerNotification::ThreadGoalUpdated(
                     ThreadGoalUpdatedNotification {
                         thread_id: conversation_id.to_string(),
-                        turn_id: None,
+                        turn_id,
                         goal,
                     },
                 ))
@@ -616,12 +627,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 +696,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/app-server/src/thread_state.rs

@@ -17,6 +17,7 @@ use codex_utils_absolute_path::AbsolutePathBuf;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
 use std::sync::Weak;
 use tokio::sync::Mutex;
 use tokio::sync::mpsc;
@@ -44,8 +45,9 @@ pub(crate) struct PendingThreadResumeRequest {
 pub(crate) enum ThreadListenerCommand {
     // SendThreadResumeResponse is used to resume an already running thread by sending the thread's history to the client and atomically subscribing for new updates.
     SendThreadResumeResponse(Box<PendingThreadResumeRequest>),
-    // EmitThreadGoalUpdated is used to order app-server goal updates with running-thread resume responses.
+    // EmitThreadGoalUpdated is used to order goal updates with running-thread resume responses and goal clears.
     EmitThreadGoalUpdated {
+        turn_id: Option<String>,
         goal: ThreadGoal,
     },
     // EmitThreadGoalCleared is used to order app-server goal clears with running-thread resume responses.
@@ -284,6 +286,10 @@ pub(crate) struct ConnectionCapabilities {
 #[derive(Clone, Default)]
 pub(crate) struct ThreadStateManager {
     state: Arc<Mutex<ThreadStateManagerInner>>,
+    // Extension event sinks are synchronous, so they need an await-free way to
+    // enqueue work on the active per-thread listener.
+    listener_commands:
+        Arc<StdMutex<HashMap<ThreadId, mpsc::UnboundedSender<ThreadListenerCommand>>>>,
 }
 
 impl ThreadStateManager {
@@ -337,6 +343,35 @@ impl ThreadStateManager {
         state.threads.entry(thread_id).or_default().state.clone()
     }
 
+    pub(crate) fn current_listener_command_tx(
+        &self,
+        thread_id: ThreadId,
+    ) -> Option<mpsc::UnboundedSender<ThreadListenerCommand>> {
+        self.listener_commands
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .get(&thread_id)
+            .cloned()
+    }
+
+    pub(crate) fn register_listener_command_tx(
+        &self,
+        thread_id: ThreadId,
+        tx: mpsc::UnboundedSender<ThreadListenerCommand>,
+    ) {
+        self.listener_commands
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(thread_id, tx);
+    }
+
+    pub(crate) fn unregister_listener_command_tx(&self, thread_id: ThreadId) {
+        self.listener_commands
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .remove(&thread_id);
+    }
+
     pub(crate) async fn remove_thread_state(&self, thread_id: ThreadId) {
         let thread_state = {
             let mut state = self.state.lock().await;
@@ -350,6 +385,7 @@ impl ThreadStateManager {
             });
             thread_state
         };
+        self.unregister_listener_command_tx(thread_id);
 
         if let Some(thread_state) = thread_state {
             let mut thread_state = thread_state.lock().await;
@@ -375,6 +411,7 @@ impl ThreadStateManager {
         };
 
         for (thread_id, thread_state) in thread_states {
+            self.unregister_listener_command_tx(thread_id);
             let mut thread_state = thread_state.lock().await;
             tracing::debug!(
                 thread_id = %thread_id,