Repository: openai/codex
File: codex-rs/app-server/src/codex_message_processor.rs (Rust)

Code before:
    pub(crate) turn_summary: TurnSummary,
    pub(crate) listener_cancel_txs: HashMap<Uuid, oneshot::Sender<()>>,
}

impl ThreadState {
    fn set_listener(&mut self, subscription_id: Uuid, cancel_tx: oneshot::Sender<()>) {
        if let Some(previous) = self.listener_cancel_txs.insert(subscription_id, cancel_tx) {
            let _ = previous.send(());
        }
    }

    fn clear_listener(&mut self, subscription_id: Uuid) {
        if let Some(cancel_tx) = self.listener_cancel_txs.remove(&subscription_id) {
            let _ = cancel_tx.send(());
        }
    }

    fn clear_listeners(&mut self) {
        for (_, cancel_tx) in self.listener_cancel_txs.drain() {
            let _ = cancel_tx.send(());
        }
    }
}

#[derive(Default)]
struct ThreadStateManager {
    thread_states: HashMap<ThreadId, Arc<Mutex<ThreadState>>>,
    thread_id_by_subscription: HashMap<Uuid, ThreadId>,
}

impl ThreadStateManager {
    fn new() -> Self {
        Self::default()
    }

    fn has_listener_for_thread(&self, thread_id: ThreadId) -> bool {
        self.thread_id_by_subscription
            .values()
            .any(|existing| *existing == thread_id)
    }

    fn thread_state(&mut self, thread_id: ThreadId) -> Arc<Mutex<ThreadState>> {
        self.thread_states
            .entry(thread_id)
            .or_insert_with(|| Arc::new(Mutex::new(ThreadState::default())))
            .clone()
    }

    async fn remove_listener(&mut self, subscription_id: Uuid) -> Option<ThreadId> {
        let thread_id = self.thread_id_by_subscription.remove(&subscription_id)?;
        if let Some(thread_state) = self.thread_states.get(&thread_id) {

Review comment (suggestion):
app-server has a few monolithic files (my bad), but probably worth breaking this  and `ThreadState` out to a new file

Code after:
    NotFound(Uuid),
}

enum AppListLoadResult {
    Accessible(Result<Vec<AppInfo>, String>),
    Directory(Result<Vec<AppInfo>, String>),
}

impl Drop for ActiveLogin {
    fn drop(&mut self) {
        self.shutdown_handle.shutdown();
    }
}

/// Handles JSON-RPC messages for Codex threads (and legacy conversation APIs).
pub(crate) struct CodexMessageProcessor {
    auth_manager: Arc<AuthManager>,
    thread_manager: Arc<ThreadManager>,
    outgoing: Arc<OutgoingMessageSender>,
    codex_linux_sandbox_exe: Option<PathBuf>,
    config: Arc<Config>,
    cli_overrides: Vec<(String, TomlValue)>,
    cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
    active_login: Arc<Mutex<Option<ActiveLogin>>>,
    thread_state_manager: ThreadStateManager,
    pending_fuzzy_searches: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
    feedback: CodexFeedback,
}

#[derive(Clone, Copy, Debug, Default)]
pub(crate) enum ApiVersion {
    V1,
    #[default]
    V2,
}

pub(crate) struct CodexMessageProcessorArgs {
    pub(crate) auth_manager: Arc<AuthManager>,
    pub(crate) thread_manager: Arc<ThreadManager>,
    pub(crate) outgoing: Arc<OutgoingMessageSender>,
    pub(crate) codex_linux_sandbox_exe: Option<PathBuf>,
    pub(crate) config: Arc<Config>,
    pub(crate) cli_overrides: Vec<(String, TomlValue)>,
    pub(crate) cloud_requirements: Arc<RwLock<CloudRequirementsLoader>>,
    pub(crate) feedback: CodexFeedback,
}

impl CodexMessageProcessor {
    async fn load_thread(
        &self,
        thread_id: &str,