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

Code before:
                match data.binary_search(&cursor) {
                    Ok(idx) => idx + 1,
                    Err(idx) => idx,
                }
            }
            None => 0,
        };

        let effective_limit = limit.unwrap_or(total as u32).max(1) as usize;
        let end = start.saturating_add(effective_limit).min(total);
        let page = data[start..end].to_vec();
        let next_cursor = page.last().filter(|_| end < total).cloned();

        let response = ThreadLoadedListResponse {
            data: page,
            next_cursor,
        };
        self.outgoing.send_response(request_id, response).await;
    }

    pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver<ThreadId> {
        self.thread_manager.subscribe_thread_created()
    }

    /// Ensure we have a listener for thread_id
    pub(crate) async fn ensure_thread_listener(&mut self, thread_id: ThreadId) {
        if self
            .listener_thread_ids_by_subscription
            .values()
            .any(|entry| *entry == thread_id)
        {
            return;
        }

        if let Err(err) = self
            .attach_conversation_listener(thread_id, false, ApiVersion::V2)
            .await
        {
            warn!(
                "failed to attach listener for thread {thread_id}: {message}",
                message = err.message
            );
        }
    }

    async fn thread_resume(&mut self, request_id: RequestId, params: ThreadResumeParams) {
        let ThreadResumeParams {
            thread_id,
            history,
            path,
            model,

Review comment (bug):
It's a bit of a smell that this is called "ensure" if all it does is warn on error. Perhaps we could better comment under what circumstances `attach_conversation_listener` fails? 

Code after:
                match data.binary_search(&cursor) {
                    Ok(idx) => idx + 1,
                    Err(idx) => idx,
                }
            }
            None => 0,
        };

        let effective_limit = limit.unwrap_or(total as u32).max(1) as usize;
        let end = start.saturating_add(effective_limit).min(total);
        let page = data[start..end].to_vec();
        let next_cursor = page.last().filter(|_| end < total).cloned();

        let response = ThreadLoadedListResponse {
            data: page,
            next_cursor,
        };
        self.outgoing.send_response(request_id, response).await;
    }

    pub(crate) fn thread_created_receiver(&self) -> broadcast::Receiver<ThreadId> {
        self.thread_manager.subscribe_thread_created()
    }

    /// Best-effort: attach a listener for thread_id if missing.
    pub(crate) async fn try_attach_thread_listener(&mut self, thread_id: ThreadId) {
        if self
            .listener_thread_ids_by_subscription
            .values()
            .any(|entry| *entry == thread_id)
        {
            return;
        }

        if let Err(err) = self
            .attach_conversation_listener(thread_id, false, ApiVersion::V2)
            .await
        {
            warn!(
                "failed to attach listener for thread {thread_id}: {message}",
                message = err.message
            );
        }
    }

    async fn thread_resume(&mut self, request_id: RequestId, params: ThreadResumeParams) {
        let ThreadResumeParams {
            thread_id,
            history,
            path,
            model,