[codex] Add rollout-backed thread content search (#23519)

## Summary
- add experimental `thread/search` for local rollout-backed thread
search using `rg` over JSONL rollouts
- return search-specific result rows with optional previews instead of
storing preview data on `StoredThread` or ordinary `Thread` responses
- keep `thread/list` separate from full-content search and document the
new app-server surface

## Testing
- `cargo test -p codex-app-server-protocol`
- `cargo test -p codex-app-server
thread_search_returns_content_and_title_matches -- --nocapture`

codex-rs/rollout/src/lib.rs

@@ -9,6 +9,7 @@ pub(crate) mod list;
 pub(crate) mod metadata;
 pub(crate) mod policy;
 pub(crate) mod recorder;
+pub(crate) mod search;
 pub(crate) mod session_index;
 mod sqlite_metrics;
 pub mod state_db;
@@ -60,6 +61,8 @@ pub use policy::should_persist_response_item_for_memories;
 pub use recorder::RolloutRecorder;
 pub use recorder::RolloutRecorderParams;
 pub use recorder::append_rollout_item_to_path;
+pub use search::first_rollout_content_match_snippet;
+pub use search::search_rollout_paths;
 pub use session_index::append_thread_name;
 pub use session_index::find_thread_meta_by_name_str;
 pub use session_index::find_thread_name_by_id;

codex-rs/rollout/src/search.rs

@@ -0,0 +1,251 @@
+use std::collections::HashSet;
+use std::io;
+use std::path::Path;
+use std::path::PathBuf;
+
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
+use codex_protocol::protocol::USER_MESSAGE_BEGIN;
+use tokio::io::AsyncBufReadExt;
+use tokio::process::Command;
+
+use super::ARCHIVED_SESSIONS_SUBDIR;
+use super::SESSIONS_SUBDIR;
+
+const MATCH_CONTEXT_BEFORE_CHARS: usize = 48;
+const MATCH_CONTEXT_AFTER_CHARS: usize = 96;
+
+pub async fn search_rollout_paths(
+    rg_command: &Path,
+    codex_home: &Path,
+    archived: bool,
+    search_term: &str,
+) -> io::Result<HashSet<PathBuf>> {
+    let root = codex_home.join(if archived {
+        ARCHIVED_SESSIONS_SUBDIR
+    } else {
+        SESSIONS_SUBDIR
+    });
+    let search_term = json_escaped_search_term(search_term)?;
+    ripgrep_rollout_paths(rg_command, root.as_path(), search_term.as_str()).await
+}
+
+async fn ripgrep_rollout_paths(
+    rg_command: &Path,
+    root: &Path,
+    search_term: &str,
+) -> io::Result<HashSet<PathBuf>> {
+    if !tokio::fs::try_exists(root).await.unwrap_or(false) {
+        return Ok(HashSet::new());
+    }
+
+    let output = match Command::new(rg_command)
+        .arg("-l")
+        .arg("--fixed-strings")
+        .arg("--no-ignore")
+        .arg("--glob")
+        .arg("*.jsonl")
+        .arg("--")
+        .arg(search_term)
+        .arg(root)
+        .output()
+        .await
+    {
+        Ok(output) => output,
+        Err(err) if err.kind() == io::ErrorKind::NotFound => {
+            return scan_rollout_paths(root, search_term).await;
+        }
+        Err(err) => return Err(err),
+    };
+    if !output.status.success() {
+        if output.status.code() == Some(1) && output.stderr.is_empty() {
+            return Ok(HashSet::new());
+        }
+
+        return Err(io::Error::other(format!(
+            "ripgrep rollout search failed under {}",
+            root.display()
+        )));
+    }
+
+    let mut matches = HashSet::new();
+    for line in String::from_utf8_lossy(output.stdout.as_slice()).lines() {
+        let path = PathBuf::from(line);
+        let path = if path.is_absolute() {
+            path
+        } else {
+            root.join(path)
+        };
+        matches.insert(path);
+    }
+
+    Ok(matches)
+}
+
+async fn scan_rollout_paths(root: &Path, search_term: &str) -> io::Result<HashSet<PathBuf>> {
+    let mut matches = HashSet::new();
+    let mut dirs = vec![root.to_path_buf()];
+
+    while let Some(dir) = dirs.pop() {
+        let mut entries = match tokio::fs::read_dir(dir).await {
+            Ok(entries) => entries,
+            Err(err) if err.kind() == io::ErrorKind::NotFound => continue,
+            Err(err) => return Err(err),
+        };
+        while let Some(entry) = entries.next_entry().await? {
+            let path = entry.path();
+            let file_type = entry.file_type().await?;
+            if file_type.is_dir() {
+                dirs.push(path);
+                continue;
+            }
+            if !file_type.is_file()
+                || path.extension().and_then(|extension| extension.to_str()) != Some("jsonl")
+            {
+                continue;
+            }
+            if rollout_contains(path.as_path(), search_term).await? {
+                matches.insert(path);
+            }
+        }
+    }
+
+    Ok(matches)
+}
+
+async fn rollout_contains(path: &Path, search_term: &str) -> io::Result<bool> {
+    let file = tokio::fs::File::open(path).await?;
+    let mut lines = tokio::io::BufReader::new(file).lines();
+    while let Some(line) = lines.next_line().await? {
+        if line.contains(search_term) {
+            return Ok(true);
+        }
+    }
+    Ok(false)
+}
+
+pub async fn first_rollout_content_match_snippet(
+    path: &Path,
+    search_term: &str,
+) -> io::Result<Option<String>> {
+    let file = tokio::fs::File::open(path).await?;
+    let mut lines = tokio::io::BufReader::new(file).lines();
+    let json_search_term = json_escaped_search_term(search_term)?;
+    while let Some(line) = lines.next_line().await? {
+        if line.contains(json_search_term.as_str())
+            && let Some(snippet) = content_match_snippet(line.as_str(), search_term)
+        {
+            return Ok(Some(snippet));
+        }
+    }
+    Ok(None)
+}
+
+fn json_escaped_search_term(search_term: &str) -> io::Result<String> {
+    let serialized = serde_json::to_string(search_term).map_err(io::Error::other)?;
+    Ok(serialized[1..serialized.len() - 1].to_string())
+}
+
+fn content_match_snippet(jsonl_line: &str, search_term: &str) -> Option<String> {
+    let rollout_line = serde_json::from_str::<RolloutLine>(jsonl_line.trim()).ok()?;
+    let text = conversation_text_from_item(&rollout_line.item)?;
+    excerpt_around_match(text.as_str(), search_term)
+}
+
+fn conversation_text_from_item(item: &RolloutItem) -> Option<String> {
+    match item {
+        RolloutItem::EventMsg(EventMsg::UserMessage(user)) => {
+            let text = strip_user_message_prefix(user.message.as_str());
+            if text.is_empty() {
+                None
+            } else {
+                Some(text.to_string())
+            }
+        }
+        RolloutItem::EventMsg(EventMsg::AgentMessage(agent)) => {
+            if agent.message.trim().is_empty() {
+                None
+            } else {
+                Some(agent.message.trim().to_string())
+            }
+        }
+        RolloutItem::ResponseItem(ResponseItem::Message { role, content, .. }) => {
+            let text = content
+                .iter()
+                .filter_map(content_item_text)
+                .collect::<Vec<_>>()
+                .join(" ");
+            if text.trim().is_empty() || (role != "user" && role != "assistant") {
+                None
+            } else {
+                Some(text)
+            }
+        }
+        RolloutItem::SessionMeta(_)
+        | RolloutItem::TurnContext(_)
+        | RolloutItem::EventMsg(_)
+        | RolloutItem::ResponseItem(_)
+        | RolloutItem::Compacted(_) => None,
+    }
+}
+
+fn content_item_text(item: &ContentItem) -> Option<&str> {
+    match item {
+        ContentItem::InputText { text } | ContentItem::OutputText { text } => Some(text.as_str()),
+        ContentItem::InputImage { .. } => None,
+    }
+}
+
+fn strip_user_message_prefix(text: &str) -> &str {
+    match text.find(USER_MESSAGE_BEGIN) {
+        Some(idx) => text[idx + USER_MESSAGE_BEGIN.len()..].trim(),
+        None => text.trim(),
+    }
+}
+
+fn excerpt_around_match(text: &str, search_term: &str) -> Option<String> {
+    let normalized = normalize_preview_text(text);
+    let match_start = normalized.find(search_term)?;
+    let match_end = match_start.saturating_add(search_term.len());
+    let excerpt_start =
+        char_start_before(normalized.as_str(), match_start, MATCH_CONTEXT_BEFORE_CHARS);
+    let excerpt_end = char_end_after(normalized.as_str(), match_end, MATCH_CONTEXT_AFTER_CHARS);
+    let excerpt = normalized[excerpt_start..excerpt_end].trim();
+    if excerpt.is_empty() {
+        return None;
+    }
+
+    let mut snippet = String::new();
+    if excerpt_start > 0 {
+        snippet.push_str("... ");
+    }
+    snippet.push_str(excerpt);
+    if excerpt_end < normalized.len() {
+        snippet.push_str(" ...");
+    }
+    Some(snippet)
+}
+
+fn normalize_preview_text(text: &str) -> String {
+    text.split_whitespace().collect::<Vec<_>>().join(" ")
+}
+
+fn char_start_before(text: &str, byte_index: usize, chars_before: usize) -> usize {
+    text[..byte_index]
+        .char_indices()
+        .rev()
+        .nth(chars_before)
+        .map(|(idx, _)| idx)
+        .unwrap_or(0)
+}
+
+fn char_end_after(text: &str, byte_index: usize, chars_after: usize) -> usize {
+    text[byte_index..]
+        .char_indices()
+        .nth(chars_after)
+        .map(|(offset, _)| byte_index.saturating_add(offset))
+        .unwrap_or(text.len())
+}