extension-api: add TurnItemEmitter to tool calls (#24813)

## Why
Extension-contributed tools need to emit visible turn items through
Codex's normal event and persistence pipeline.

## What
- Add `TurnItemEmitter` to extension `ToolCall`s and route the core
implementation through `Session::emit_turn_item_*`.
- Hold weak session and turn references so retained tool calls cannot
keep host state alive.
- Provide a no-op emitter for extension test callers.

## Test Plan
- `just test -p codex-core -E
'test(passes_turn_fields_and_scoped_turn_item_emitter_to_extension_call)'`

---------

Co-authored-by: jif-oai <jif@openai.com>

codex-rs/core/src/tools/handlers/extension_tools.rs

@@ -1,11 +1,18 @@
 use std::sync::Arc;
+use std::sync::Weak;
 
+use codex_protocol::items::TurnItem;
 use codex_tools::ConversationHistory;
+use codex_tools::ExtensionTurnItem;
 use codex_tools::ToolCall as ExtensionToolCall;
 use codex_tools::ToolName;
 use codex_tools::ToolSpec;
+use codex_tools::TurnItemEmissionFuture;
+use codex_tools::TurnItemEmitter;
 
 use crate::function_tool::FunctionCallError;
+use crate::session::session::Session;
+use crate::session::turn_context::TurnContext;
 use crate::tools::context::ToolInvocation;
 use crate::tools::context::ToolOutput;
 use crate::tools::context::ToolPayload;
@@ -52,6 +59,39 @@ impl CoreToolRuntime for ExtensionToolAdapter {
     }
 }
 
+struct CoreTurnItemEmitter {
+    session: Weak<Session>,
+    turn: Weak<TurnContext>,
+}
+
+fn extension_turn_item(item: ExtensionTurnItem) -> TurnItem {
+    match item {
+        ExtensionTurnItem::WebSearch(item) => TurnItem::WebSearch(item),
+    }
+}
+
+impl TurnItemEmitter for CoreTurnItemEmitter {
+    fn emit_started<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> {
+        Box::pin(async move {
+            let (Some(session), Some(turn)) = (self.session.upgrade(), self.turn.upgrade()) else {
+                return;
+            };
+            let item = extension_turn_item(item);
+            session.emit_turn_item_started(turn.as_ref(), &item).await;
+        })
+    }
+
+    fn emit_completed<'a>(&'a self, item: ExtensionTurnItem) -> TurnItemEmissionFuture<'a> {
+        Box::pin(async move {
+            let (Some(session), Some(turn)) = (self.session.upgrade(), self.turn.upgrade()) else {
+                return;
+            };
+            let item = extension_turn_item(item);
+            session.emit_turn_item_completed(turn.as_ref(), item).await;
+        })
+    }
+}
+
 async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall {
     let conversation_history =
         ConversationHistory::new(invocation.session.clone_history().await.into_raw_items());
@@ -61,6 +101,10 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall {
         tool_name: invocation.tool_name.clone(),
         truncation_policy: invocation.turn.truncation_policy,
         conversation_history,
+        turn_item_emitter: Arc::new(CoreTurnItemEmitter {
+            session: Arc::downgrade(&invocation.session),
+            turn: Arc::downgrade(&invocation.turn),
+        }),
         payload: invocation.payload.clone(),
     }
 }
@@ -69,8 +113,13 @@ async fn to_extension_call(invocation: &ToolInvocation) -> ExtensionToolCall {
 mod tests {
     use std::sync::Arc;
 
+    use codex_protocol::items::TurnItem;
+    use codex_protocol::items::WebSearchItem;
     use codex_protocol::models::ContentItem;
     use codex_protocol::models::ResponseItem;
+    use codex_protocol::models::WebSearchAction;
+    use codex_protocol::protocol::EventMsg;
+    use codex_tools::ExtensionTurnItem;
     use pretty_assertions::assert_eq;
     use serde_json::json;
     use tokio::sync::Mutex;
@@ -147,6 +196,16 @@ mod tests {
             &self,
             call: codex_tools::ToolCall,
         ) -> Result<Box<dyn codex_tools::ToolOutput>, codex_tools::FunctionCallError> {
+            let item = ExtensionTurnItem::WebSearch(WebSearchItem {
+                id: call.call_id.clone(),
+                query: "rust trait object".to_string(),
+                action: WebSearchAction::Search {
+                    query: Some("rust trait object".to_string()),
+                    queries: None,
+                },
+            });
+            call.turn_item_emitter.emit_started(item.clone()).await;
+            call.turn_item_emitter.emit_completed(item).await;
             *self.captured_call.lock().await = Some(call);
             Ok(Box::new(codex_tools::JsonToolOutput::new(
                 json!({ "ok": true }),
@@ -191,12 +250,14 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn passes_turn_fields_to_extension_call() {
+    async fn passes_turn_fields_and_scoped_turn_item_emitter_to_extension_call() {
         let captured_call = Arc::new(Mutex::new(None));
         let handler = ExtensionToolAdapter::new(Arc::new(CapturingExtensionExecutor {
             captured_call: Arc::clone(&captured_call),
         }));
-        let (session, turn) = crate::session::tests::make_session_and_context().await;
+        let (session, turn, rx) = crate::session::tests::make_session_and_context_with_rx().await;
+        let weak_session = Arc::downgrade(&session);
+        let weak_turn = Arc::downgrade(&turn);
         let turn_id = turn.sub_id.clone();
         let truncation_policy = turn.truncation_policy;
         let history_item = ResponseItem::Message {
@@ -211,8 +272,8 @@ mod tests {
             .record_conversation_items(&turn, std::slice::from_ref(&history_item))
             .await;
         let invocation = ToolInvocation {
-            session: session.into(),
-            turn: turn.into(),
+            session,
+            turn,
             cancellation_token: tokio_util::sync::CancellationToken::new(),
             tracker: Arc::new(tokio::sync::Mutex::new(TurnDiffTracker::new())),
             call_id: "call-extension".to_string(),
@@ -228,6 +289,8 @@ mod tests {
             .expect("extension call should succeed");
 
         let captured_call = captured_call.lock().await.clone().expect("captured call");
+        assert!(weak_session.upgrade().is_none());
+        assert!(weak_turn.upgrade().is_none());
         assert_eq!(captured_call.turn_id, turn_id);
         assert_eq!(captured_call.call_id, "call-extension");
         assert_eq!(
@@ -245,5 +308,43 @@ mod tests {
             }
             payload => panic!("expected function payload, got {payload:?}"),
         }
+
+        let started = rx.recv().await.expect("item started event");
+        let EventMsg::ItemStarted(started) = started.msg else {
+            panic!("expected item started event");
+        };
+        let TurnItem::WebSearch(started_item) = started.item else {
+            panic!("expected web search item");
+        };
+        let begin = rx.recv().await.expect("legacy web search begin event");
+        let EventMsg::WebSearchBegin(begin) = begin.msg else {
+            panic!("expected legacy web search begin event");
+        };
+        let completed = rx.recv().await.expect("item completed event");
+        let EventMsg::ItemCompleted(completed) = completed.msg else {
+            panic!("expected item completed event");
+        };
+        let TurnItem::WebSearch(completed_item) = completed.item else {
+            panic!("expected web search item");
+        };
+        let end = rx.recv().await.expect("legacy web search end event");
+        let EventMsg::WebSearchEnd(end) = end.msg else {
+            panic!("expected legacy web search end event");
+        };
+
+        let expected = WebSearchItem {
+            id: "call-extension".to_string(),
+            query: "rust trait object".to_string(),
+            action: WebSearchAction::Search {
+                query: Some("rust trait object".to_string()),
+                queries: None,
+            },
+        };
+        assert_eq!(started_item, expected);
+        assert_eq!(completed_item, expected);
+        assert_eq!(begin.call_id, expected.id);
+        assert_eq!(end.call_id, expected.id);
+        assert_eq!(end.query, expected.query);
+        assert_eq!(end.action, expected.action);
     }
 }