Wire app-server extension event sink (#24586)

## Why

The goal extension already emits `ThreadGoalUpdated` events, but
production app-server thread extensions were built with the default
no-op extension event sink. That meant extension-driven goal updates
could be produced without ever reaching app-server clients.

## What changed

- Build app-server thread extensions with a host-provided
`ExtensionEventSink`.
- Add an app-server sink that converts extension `ThreadGoalUpdated`
events into `ServerNotification::ThreadGoalUpdated` broadcasts.
- Use the existing bounded outgoing message channel via `try_send` so
event forwarding cannot create an unbounded queue.
- Pass `NoopExtensionEventSink` in app-server tests that construct a
`ThreadManager` without an app-server host.
- Refresh `Cargo.lock` for the existing `codex-memories-extension`
`codex-otel` dependency.

## Verification

- `just test -p codex-app-server
extensions::tests::app_server_event_sink_forwards_thread_goal_updates`

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

@@ -1,27 +1,67 @@
 use std::sync::Arc;
 use std::sync::Weak;
 
+use codex_app_server_protocol::ServerNotification;
+use codex_app_server_protocol::ThreadGoalUpdatedNotification;
 use codex_core::NewThread;
 use codex_core::StartThreadOptions;
 use codex_core::ThreadManager;
 use codex_core::config::Config;
 use codex_extension_api::AgentSpawnFuture;
 use codex_extension_api::AgentSpawner;
+use codex_extension_api::ExtensionEventSink;
 use codex_extension_api::ExtensionRegistry;
 use codex_extension_api::ExtensionRegistryBuilder;
 use codex_protocol::ThreadId;
 use codex_protocol::error::CodexErr;
+use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
 
-pub(crate) fn thread_extensions<S>(guardian_agent_spawner: S) -> Arc<ExtensionRegistry<Config>>
+use crate::outgoing_message::OutgoingMessageSender;
+
+pub(crate) fn thread_extensions<S>(
+    guardian_agent_spawner: S,
+    event_sink: Arc<dyn ExtensionEventSink>,
+) -> Arc<ExtensionRegistry<Config>>
 where
     S: AgentSpawner<StartThreadOptions, Spawned = NewThread, Error = CodexErr> + 'static,
 {
-    let mut builder = ExtensionRegistryBuilder::<Config>::new();
+    let mut builder = ExtensionRegistryBuilder::<Config>::with_event_sink(event_sink);
     codex_guardian::install(&mut builder, guardian_agent_spawner);
     codex_memories_extension::install(&mut builder, codex_otel::global());
     Arc::new(builder.build())
 }
 
+pub(crate) fn app_server_extension_event_sink(
+    outgoing: Arc<OutgoingMessageSender>,
+) -> Arc<dyn ExtensionEventSink> {
+    Arc::new(AppServerExtensionEventSink { outgoing })
+}
+
+struct AppServerExtensionEventSink {
+    outgoing: Arc<OutgoingMessageSender>,
+}
+
+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(),
+                        },
+                    ));
+            }
+            msg => {
+                tracing::debug!(event_id = %event.id, ?msg, "dropping unsupported extension event");
+            }
+        }
+    }
+}
+
 pub(crate) fn guardian_agent_spawner(
     thread_manager: Weak<ThreadManager>,
 ) -> impl AgentSpawner<StartThreadOptions, Spawned = NewThread, Error = CodexErr> {
@@ -39,3 +79,84 @@ pub(crate) fn guardian_agent_spawner(
         })
     }
 }
+
+#[cfg(test)]
+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::ThreadGoalStatus;
+    use codex_protocol::protocol::ThreadGoalUpdatedEvent;
+    use pretty_assertions::assert_eq;
+    use tokio::sync::mpsc;
+    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);
+        let outgoing = Arc::new(OutgoingMessageSender::new(
+            outgoing_tx,
+            AnalyticsEventsClient::disabled(),
+        ));
+        let sink = app_server_extension_event_sink(outgoing);
+        let thread_id = ThreadId::default();
+
+        sink.emit(Event {
+            id: "call-1".to_string(),
+            msg: EventMsg::ThreadGoalUpdated(ThreadGoalUpdatedEvent {
+                thread_id,
+                turn_id: Some("turn-1".to_string()),
+                goal: ThreadGoal {
+                    thread_id,
+                    objective: "wire extension events".to_string(),
+                    status: ThreadGoalStatus::Active,
+                    token_budget: Some(123),
+                    tokens_used: 45,
+                    time_used_seconds: 6,
+                    created_at: 7,
+                    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
+        );
+    }
+}