core: refactor TurnEvents to use Arc<Session> instead of Sender<Event>

- TurnEvents no longer holds a Sender<Event>, but an Arc<Session>, and events are sent via Session methods.
- Update TurnEvents::new signature and all call sites to pass Arc<Session>.
- Update TurnEvents usage in Session and tests for Arc<Session> changes.
- Rename item_collector.rs to turn_events.rs and update module visibility.
- Remove unused item_collector mod.
- Update debug

codex-rs/core/src/codex.rs

@@ -357,7 +357,7 @@ impl Session {
         provider: ModelProviderInfo,
         session_configuration: &SessionConfiguration,
         conversation_id: ConversationId,
-        tx_event: Sender<Event>,
+        session: Arc<Session>,
     ) -> TurnContext {
         let config = session_configuration.original_config_do_not_use.clone();
         let model_family = find_family_for_model(&session_configuration.model)
@@ -403,7 +403,7 @@ impl Session {
             is_review_mode: false,
             final_output_json_schema: None,
             codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(),
-            turn_events: TurnEvents::new(tx_event, conversation_id, sub_id.clone(), sub_id),
+            turn_events: TurnEvents::new(session, conversation_id, sub_id.clone(), sub_id),
         }
     }
 
@@ -616,7 +616,7 @@ impl Session {
         format!("auto-compact-{id}")
     }
 
-    async fn record_initial_history(&self, conversation_history: InitialHistory) {
+    async fn record_initial_history(self: &Arc<Self>, conversation_history: InitialHistory) {
         // TODO(pakrym): Ideally we shouldn't need to create a fake turn context here.
         let turn_context = self
             .new_turn("init".to_string(), SessionSettingsUpdate::default())
@@ -653,7 +653,7 @@ impl Session {
     }
 
     pub(crate) async fn new_turn(
-        &self,
+        self: &Arc<Self>,
         sub_id: String,
         updates: SessionSettingsUpdate,
     ) -> Arc<TurnContext> {
@@ -671,7 +671,7 @@ impl Session {
             session_configuration.provider.clone(),
             &session_configuration,
             self.conversation_id,
-            self.get_tx_event(),
+            self.clone(),
         );
         if let Some(final_schema) = updates.final_output_json_schema {
             turn_context.final_output_json_schema = final_schema;
@@ -1448,7 +1448,7 @@ async fn spawn_review_thread(
         final_output_json_schema: None,
         codex_linux_sandbox_exe: parent_turn_context.codex_linux_sandbox_exe.clone(),
         turn_events: TurnEvents::new(
-            sess.get_tx_event(),
+            sess.clone(),
             sess.conversation_id,
             sub_id.to_string(),
             sub_id.clone(),
@@ -2614,7 +2614,7 @@ mod tests {
         )
     }
 
-    pub(crate) fn make_session_and_context() -> (Session, TurnContext) {
+    pub(crate) fn make_session_and_context() -> (Arc<Session>, TurnContext) {
         let (tx_event, _rx_event) = async_channel::unbounded();
         let codex_home = tempfile::tempdir().expect("create temp dir");
         let config = Config::load_from_base_config_with_overrides(
@@ -2655,16 +2655,6 @@ mod tests {
             tool_approvals: Mutex::new(ApprovalStore::default()),
         };
 
-        let turn_context = Session::make_turn_context(
-            "sub_1".to_string(),
-            Some(Arc::clone(&auth_manager)),
-            &otel_event_manager,
-            session_configuration.provider.clone(),
-            &session_configuration,
-            conversation_id,
-            tx_event.clone(),
-        );
-
         let session = Session {
             conversation_id,
             tx_event,
@@ -2674,6 +2664,18 @@ mod tests {
             next_internal_sub_id: AtomicU64::new(0),
         };
 
+        let session = Arc::new(session);
+
+        let turn_context = Session::make_turn_context(
+            "sub_1".to_string(),
+            Some(Arc::clone(&auth_manager)),
+            &otel_event_manager,
+            session_configuration.provider.clone(),
+            &session_configuration,
+            conversation_id,
+            session.clone(),
+        );
+
         (session, turn_context)
     }
 
@@ -2724,16 +2726,6 @@ mod tests {
             tool_approvals: Mutex::new(ApprovalStore::default()),
         };
 
-        let turn_context = Arc::new(Session::make_turn_context(
-            "sub_1".to_string(),
-            Some(Arc::clone(&auth_manager)),
-            &otel_event_manager,
-            session_configuration.provider.clone(),
-            &session_configuration,
-            conversation_id,
-            tx_event.clone(),
-        ));
-
         let session = Arc::new(Session {
             conversation_id,
             tx_event,
@@ -2742,6 +2734,16 @@ mod tests {
             services,
             next_internal_sub_id: AtomicU64::new(0),
         });
+        let session = Arc::clone(&session);
+        let turn_context = Arc::new(Session::make_turn_context(
+            "sub_1".to_string(),
+            Some(Arc::clone(&auth_manager)),
+            &otel_event_manager,
+            session_configuration.provider.clone(),
+            &session_configuration,
+            conversation_id,
+            session.clone(),
+        ));
 
         (session, turn_context, rx_event)
     }

codex-rs/core/src/state/mod.rs

@@ -1,11 +1,11 @@
-mod item_collector;
 mod service;
 mod session;
 mod turn;
+mod turn_events;
 
-pub(crate) use item_collector::TurnEvents;
 pub(crate) use service::SessionServices;
 pub(crate) use session::SessionState;
 pub(crate) use turn::ActiveTurn;
 pub(crate) use turn::RunningTask;
 pub(crate) use turn::TaskKind;
+pub(crate) use turn_events::TurnEvents;

codex-rs/core/src/state/turn_events.rs

@@ -1,4 +1,6 @@
-use async_channel::Sender;
+use core::fmt;
+use std::sync::Arc;
+
 use codex_protocol::ConversationId;
 use codex_protocol::items::TurnItem;
 use codex_protocol::protocol::Event;
@@ -7,17 +9,28 @@ use codex_protocol::protocol::ItemCompletedEvent;
 use codex_protocol::protocol::ItemStartedEvent;
 use tracing::error;
 
-#[derive(Debug)]
+use crate::codex::Session;
+
 pub(crate) struct TurnEvents {
     thread_id: ConversationId,
     sub_id: String,
     turn_id: String,
-    tx_event: Sender<Event>,
+    session: Arc<Session>,
+}
+
+impl fmt::Debug for TurnEvents {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "TurnEvents {{ thread_id: {}, sub_id: {}, turn_id: {} }}",
+            self.thread_id, self.sub_id, self.turn_id
+        )
+    }
 }
 
 impl TurnEvents {
     pub fn new(
-        tx_event: Sender<Event>,
+        session: Arc<Session>,
         thread_id: ConversationId,
         sub_id: String,
         turn_id: String,
@@ -26,13 +39,14 @@ impl TurnEvents {
             thread_id,
             sub_id,
             turn_id,
-            tx_event,
+            session,
         }
     }
 
     pub async fn started(&self, item: TurnItem) {
         let err = self
-            .tx_event
+            .session
+            .get_tx_event()
             .send(Event {
                 id: self.turn_id.clone(),
                 msg: EventMsg::ItemStarted(ItemStartedEvent {
@@ -49,7 +63,8 @@ impl TurnEvents {
 
     pub async fn completed(&self, item: TurnItem) {
         let err = self
-            .tx_event
+            .session
+            .get_tx_event()
             .send(Event {
                 id: self.turn_id.clone(),
                 msg: EventMsg::ItemCompleted(ItemCompletedEvent {
@@ -74,9 +89,6 @@ impl TurnEvents {
             id: self.sub_id.clone(),
             msg,
         };
-        let err = self.tx_event.send(event).await;
-        if let Err(e) = err {
-            error!("failed to send legacy event: {e}");
-        }
+        self.session.send_event(event).await;
     }
 }

codex-rs/core/src/unified_exec/mod.rs

@@ -86,7 +86,7 @@ mod tests {
         let (session, mut turn) = make_session_and_context();
         turn.approval_policy = AskForApproval::Never;
         turn.sandbox_policy = SandboxPolicy::DangerFullAccess;
-        (Arc::new(session), Arc::new(turn))
+        (session, Arc::new(turn))
     }
 
     async fn run_unified_exec_request(