From 3be49da40ae2605b24cdef058f8fd3fe23e476fe Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Mon, 20 Oct 2025 17:51:54 -0700 Subject: [PATCH] core: refactor TurnEvents to use Arc instead of Sender - TurnEvents no longer holds a Sender, but an Arc, and events are sent via Session methods. - Update TurnEvents::new signature and all call sites to pass Arc. - Update TurnEvents usage in Session and tests for Arc 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 | 56 ++++++++++--------- codex-rs/core/src/state/mod.rs | 4 +- .../{item_collector.rs => turn_events.rs} | 34 +++++++---- codex-rs/core/src/unified_exec/mod.rs | 2 +- 4 files changed, 55 insertions(+), 41 deletions(-) rename codex-rs/core/src/state/{item_collector.rs => turn_events.rs} (77%) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 5bafd1f0f2..9583f0187d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -357,7 +357,7 @@ impl Session { provider: ModelProviderInfo, session_configuration: &SessionConfiguration, conversation_id: ConversationId, - tx_event: Sender, + session: Arc, ) -> 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, 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, sub_id: String, updates: SessionSettingsUpdate, ) -> Arc { @@ -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, 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) } diff --git a/codex-rs/core/src/state/mod.rs b/codex-rs/core/src/state/mod.rs index 65d80372b1..8a735347ff 100644 --- a/codex-rs/core/src/state/mod.rs +++ b/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; diff --git a/codex-rs/core/src/state/item_collector.rs b/codex-rs/core/src/state/turn_events.rs similarity index 77% rename from codex-rs/core/src/state/item_collector.rs rename to codex-rs/core/src/state/turn_events.rs index 7f6041157a..77661a5dc2 100644 --- a/codex-rs/core/src/state/item_collector.rs +++ b/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, + session: Arc, +} + +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, + session: Arc, 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; } } diff --git a/codex-rs/core/src/unified_exec/mod.rs b/codex-rs/core/src/unified_exec/mod.rs index fb791fe89f..25e85a51aa 100644 --- a/codex-rs/core/src/unified_exec/mod.rs +++ b/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(