diff --git a/MODULE.bazel.lock b/MODULE.bazel.lock index 118ccac6b9..24d86a4836 100644 --- a/MODULE.bazel.lock +++ b/MODULE.bazel.lock @@ -1353,7 +1353,6 @@ "psl-types_2.0.11": "{\"dependencies\":[],\"features\":{}}", "psl_2.1.184": "{\"dependencies\":[{\"name\":\"psl-types\",\"req\":\"^2.0.11\"},{\"kind\":\"dev\",\"name\":\"rspec\",\"req\":\"^1.0.0\"}],\"features\":{\"default\":[\"helpers\"],\"helpers\":[]}}", "publicsuffix_2.3.0": "{\"dependencies\":[{\"features\":[\"inline-more\"],\"name\":\"hashbrown\",\"optional\":true,\"req\":\"^0.15.1\"},{\"name\":\"idna\",\"optional\":true,\"req\":\"^1.0\"},{\"name\":\"psl-types\",\"req\":\"^2.0.11\"},{\"kind\":\"dev\",\"name\":\"rspec\",\"req\":\"^1.0.0\"},{\"default_features\":false,\"name\":\"unicase\",\"optional\":true,\"req\":\"^2.6.0\"}],\"features\":{\"anycase\":[\"unicase\"],\"default\":[\"punycode\"],\"punycode\":[\"idna\"],\"std\":[]}}", - "pulldown-cmark-escape_0.10.1": "{\"dependencies\":[],\"features\":{\"simd\":[]}}", "pulldown-cmark_0.10.3": "{\"dependencies\":[{\"kind\":\"dev\",\"name\":\"bincode\",\"req\":\"^1.3.1\"},{\"name\":\"bitflags\",\"req\":\"^2\"},{\"kind\":\"dev\",\"name\":\"criterion\",\"req\":\"^0.5\"},{\"name\":\"getopts\",\"optional\":true,\"req\":\"^0.2\"},{\"kind\":\"dev\",\"name\":\"lazy_static\",\"req\":\"^1.4\"},{\"name\":\"memchr\",\"req\":\"^2.5\"},{\"name\":\"pulldown-cmark-escape\",\"optional\":true,\"req\":\"^0.10.0\"},{\"kind\":\"dev\",\"name\":\"regex\",\"req\":\"^1.6\"},{\"features\":[\"derive\"],\"name\":\"serde\",\"optional\":true,\"req\":\"^1.0\"},{\"kind\":\"dev\",\"name\":\"serde_json\",\"req\":\"^1.0.61\"},{\"name\":\"unicase\",\"req\":\"^2.6\"}],\"features\":{\"default\":[\"getopts\",\"html\"],\"gen-tests\":[],\"html\":[\"pulldown-cmark-escape\"],\"simd\":[\"pulldown-cmark-escape?/simd\"]}}", "pxfm_0.1.27": "{\"dependencies\":[{\"name\":\"num-traits\",\"req\":\"^0.2.3\"}],\"features\":{}}", "quick-error_2.0.1": "{\"dependencies\":[],\"features\":{}}", diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 2e045fb2b8..6138bddc8d 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -4052,7 +4052,9 @@ dependencies = [ "codex-rollout", "codex-state", "codex-utils-path", + "futures", "pretty_assertions", + "pulldown-cmark", "serde", "serde_json", "sqlx", @@ -6348,15 +6350,6 @@ dependencies = [ "windows-link", ] -[[package]] -name = "getopts" -version = "0.2.24" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfe4fbac503b8d1f88e6676011885f34b7174f46e59956bba534ba83abded4df" -dependencies = [ - "unicode-width 0.2.1", -] - [[package]] name = "getrandom" version = "0.2.17" @@ -10346,18 +10339,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76979bea66e7875e7509c4ec5300112b316af87fa7a252ca91c448b32dfe3993" dependencies = [ "bitflags 2.10.0", - "getopts", "memchr", - "pulldown-cmark-escape", "unicase", ] -[[package]] -name = "pulldown-cmark-escape" -version = "0.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd348ff538bc9caeda7ee8cad2d1d48236a1f443c1fa3913c6a02fe0043b1dd3" - [[package]] name = "pxfm" version = "0.1.27" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index ec67cd5da1..376bad36a5 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -362,7 +362,7 @@ pathdiff = "0.2" portable-pty = "0.9.0" predicates = "3" pretty_assertions = "1.4.1" -pulldown-cmark = "0.10" +pulldown-cmark = { version = "0.10", default-features = false } quick-xml = "0.41.0" rand = "0.9" ratatui = "0.29.0" diff --git a/codex-rs/app-server-protocol/src/export.rs b/codex-rs/app-server-protocol/src/export.rs index 5c4b83a8a9..012f62a8ec 100644 --- a/codex-rs/app-server-protocol/src/export.rs +++ b/codex-rs/app-server-protocol/src/export.rs @@ -50,6 +50,8 @@ const EXPERIMENTAL_CLIENT_METHOD_DEPENDENCY_TYPES: &[&str] = &[ "RemoteControlClient", "RemoteControlClientsListOrder", "ThreadBackgroundTerminal", + "ThreadSearchOccurrence", + "ThreadSearchTextRange", ]; const SPECIAL_DEFINITIONS: &[&str] = &[ "ClientNotification", diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 4c906d134d..091251a2c5 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -630,6 +630,13 @@ client_request_definitions! { serialization: None, response: v2::ThreadSearchResponse, }, + #[experimental("thread/searchOccurrences")] + ThreadSearchOccurrences => "thread/searchOccurrences" { + params: v2::ThreadSearchOccurrencesParams, + // Explicitly concurrent: this reads persisted paginated history. + serialization: None, + response: v2::ThreadSearchOccurrencesResponse, + }, ThreadLoadedList => "thread/loaded/list" { params: v2::ThreadLoadedListParams, serialization: None, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 5210b6a85c..acebeb546a 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -1249,6 +1249,58 @@ pub struct ThreadSearchResponse { pub backwards_cursor: Option, } +/// Parameters for searching visible message occurrences within one paginated thread. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadSearchOccurrencesParams { + pub thread_id: String, + /// Case-insensitive literal substring to find in visible user messages and final assistant + /// messages. + pub search_term: String, + /// Opaque cursor returned by a previous call for the same thread and search term. + #[ts(optional = nullable)] + pub cursor: Option, + /// Optional occurrence page size. + #[ts(optional = nullable)] + pub limit: Option, +} + +/// UTF-16 code-unit range within `snippet`. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadSearchTextRange { + /// Inclusive UTF-16 code-unit offset. + pub start: u32, + /// Exclusive UTF-16 code-unit offset. + pub end: u32, +} + +/// One visible message occurrence returned by [`ThreadSearchOccurrencesResponse`]. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadSearchOccurrence { + pub turn_id: String, + pub item_id: String, + pub snippet: String, + /// Match range within `snippet`, in UTF-16 code units. + pub snippet_match_range: ThreadSearchTextRange, + /// Opaque inclusive cursor accepted by `thread/turns/list` for this turn. + pub turn_cursor: String, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ThreadSearchOccurrencesResponse { + /// Occurrences in chronological message order. + pub data: Vec, + /// Opaque cursor to continue after the last returned occurrence. + pub next_cursor: Option, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 0b10de1109..afa340e8ec 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -146,6 +146,7 @@ Example with notification opt-out: - `thread/read` — read a stored thread by id without resuming it; optionally include turns via `includeTurns`. The returned `thread` includes `status` (`ThreadStatus`), defaulting to `notLoaded` when the thread is not currently loaded. For loaded threads, experimental clients can use `canAcceptDirectInput` to determine whether `turn/start` and `turn/steer` are accepted; unloaded stored threads report `null` when that capability is unavailable. - `thread/turns/list` — experimental; page through a stored thread’s turn history without resuming it; supports cursor-based pagination with `sortDirection`, `itemsView`, `nextCursor`, and `backwardsCursor`. - `thread/items/list` — experimental; page through persisted thread items without resuming the thread. Pass `turnId` to restrict results to one turn, or omit it to page items across the thread. The active thread store must support item pagination. +- `thread/searchOccurrences` — experimental; find literal, case-insensitive matches in visible user messages and summary-selected final assistant messages within one paginated thread. - `thread/metadata/update` — patch stored thread metadata in sqlite; currently supports updating persisted `gitInfo` fields and returns the refreshed `thread`. - `thread/settings/update` — experimental; queue a partial update to a loaded thread’s next-turn settings without starting a turn or adding transcript items. Omitted fields leave settings unchanged; `serviceTier: null` clears the tier; deprecated `multiAgentMode` is ignored, while Ultra reasoning effort enables proactive multi-agent behavior; `sandboxPolicy` and `permissions` cannot be combined. Returns `{}` when the update is accepted and emits `thread/settings/updated` with the full effective settings only if they actually change. `turn/start` settings overrides emit the same notification when they change the stored settings. - `thread/memoryMode/set` — experimental; set a thread’s persisted memory eligibility to `"enabled"` or `"disabled"` for either a loaded thread or a stored rollout; returns `{}` on success. @@ -548,6 +549,30 @@ cursors can be reused with or without `turnId`; the filter does not change the c Thread stores that do not implement item pagination return JSON-RPC `-32601` with message `thread/items/list is not supported yet`. +`thread/searchOccurrences` searches one paginated thread without replaying its rollout. It returns +occurrences in chronological message order from every visible user message, including steering +messages, and final assistant messages. `snippetMatchRange` uses +UTF-16 offsets within `snippet`, and `turnCursor` can be passed directly to `thread/turns/list` +to load the containing turn. + +```json +{ "method": "thread/searchOccurrences", "id": 26, "params": { + "threadId": "thr_123", + "searchTerm": "needle", + "limit": 50 +} } +{ "id": 26, "result": { + "data": [{ + "turnId": "turn_456", + "itemId": "item_789", + "snippet": "The needle is here.", + "snippetMatchRange": { "start": 4, "end": 10 }, + "turnCursor": "opaque-inclusive-turn-cursor" + }], + "nextCursor": null +} } +``` + ### Example: Update stored thread metadata Use `thread/metadata/update` to patch sqlite-backed metadata for a thread without resuming it. Today this supports persisted `gitInfo`; omitted fields are left unchanged, while explicit `null` clears a stored value. diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index a7e2053270..133c0c84fc 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1147,6 +1147,11 @@ impl MessageProcessor { ClientRequest::ThreadSearch { params, .. } => { self.thread_processor.thread_search(params).await } + ClientRequest::ThreadSearchOccurrences { params, .. } => { + self.thread_processor + .thread_search_occurrences(params) + .await + } ClientRequest::ThreadLoadedList { params, .. } => { self.thread_processor.thread_loaded_list(params).await } diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 0317db876d..d0278b4cfa 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -259,9 +259,13 @@ use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadRollbackParams; +use codex_app_server_protocol::ThreadSearchOccurrence; +use codex_app_server_protocol::ThreadSearchOccurrencesParams; +use codex_app_server_protocol::ThreadSearchOccurrencesResponse; use codex_app_server_protocol::ThreadSearchParams; use codex_app_server_protocol::ThreadSearchResponse; use codex_app_server_protocol::ThreadSearchResult; +use codex_app_server_protocol::ThreadSearchTextRange; use codex_app_server_protocol::ThreadSetNameParams; use codex_app_server_protocol::ThreadSetNameResponse; use codex_app_server_protocol::ThreadSettings; @@ -470,6 +474,7 @@ use codex_thread_store::LoadThreadHistoryParams as StoreLoadThreadHistoryParams; use codex_thread_store::LocalThreadStore; use codex_thread_store::ReadThreadByRolloutPathParams as StoreReadThreadByRolloutPathParams; use codex_thread_store::ReadThreadParams as StoreReadThreadParams; +use codex_thread_store::SearchThreadOccurrencesParams as StoreSearchThreadOccurrencesParams; use codex_thread_store::SearchThreadsParams as StoreSearchThreadsParams; use codex_thread_store::SortDirection as StoreSortDirection; use codex_thread_store::StoredThread; diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index b141657df9..6da21bd82b 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -710,6 +710,15 @@ impl ThreadRequestProcessor { .map(|response| Some(response.into())) } + pub(crate) async fn thread_search_occurrences( + &self, + params: ThreadSearchOccurrencesParams, + ) -> Result, JSONRPCErrorError> { + self.thread_search_occurrences_response_inner(params) + .await + .map(|response| Some(response.into())) + } + pub(crate) async fn thread_loaded_list( &self, params: ThreadLoadedListParams, @@ -2548,6 +2557,65 @@ impl ThreadRequestProcessor { ) } + async fn thread_search_occurrences_response_inner( + &self, + params: ThreadSearchOccurrencesParams, + ) -> Result { + let ThreadSearchOccurrencesParams { + thread_id, + search_term, + cursor, + limit, + } = params; + let thread_id = ThreadId::from_string(&thread_id) + .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?; + if search_term.trim().is_empty() { + return Err(invalid_request( + "thread/searchOccurrences requires a non-empty searchTerm", + )); + } + let page_size = limit + .map(|value| value as usize) + .unwrap_or(THREAD_SEARCH_OCCURRENCES_DEFAULT_LIMIT) + .clamp(1, THREAD_SEARCH_OCCURRENCES_MAX_LIMIT); + let page = self + .thread_store + .search_thread_occurrences(StoreSearchThreadOccurrencesParams { + thread_id, + search_term, + cursor, + page_size, + }) + .await + .map_err(|err| match err { + ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::Unsupported { operation } => { + unsupported_thread_store_operation(operation) + } + ThreadStoreError::ThreadNotFound { thread_id } => { + invalid_request(format!("no rollout found for thread id {thread_id}")) + } + err => internal_error(format!("failed to search thread occurrences: {err}")), + })?; + Ok(ThreadSearchOccurrencesResponse { + data: page + .items + .into_iter() + .map(|item| ThreadSearchOccurrence { + turn_id: item.turn_id, + item_id: item.item_id, + snippet: item.snippet, + snippet_match_range: ThreadSearchTextRange { + start: item.snippet_match_range.start, + end: item.snippet_match_range.end, + }, + turn_cursor: item.turn_cursor, + }) + .collect(), + next_cursor: page.next_cursor, + }) + } + async fn paginated_thread_turns_list_response( &self, thread_id: ThreadId, @@ -4224,6 +4292,8 @@ const THREAD_TURNS_DEFAULT_LIMIT: usize = 25; const THREAD_TURNS_MAX_LIMIT: usize = 100; const THREAD_ITEMS_DEFAULT_LIMIT: usize = 25; const THREAD_ITEMS_MAX_LIMIT: usize = 100; +const THREAD_SEARCH_OCCURRENCES_DEFAULT_LIMIT: usize = 50; +const THREAD_SEARCH_OCCURRENCES_MAX_LIMIT: usize = 250; fn thread_backwards_cursor_for_sort_key( thread: &StoredThread, diff --git a/codex-rs/app-server/tests/common/test_app_server.rs b/codex-rs/app-server/tests/common/test_app_server.rs index 3e2a6903d8..0d2291d5ec 100644 --- a/codex-rs/app-server/tests/common/test_app_server.rs +++ b/codex-rs/app-server/tests/common/test_app_server.rs @@ -103,6 +103,7 @@ use codex_app_server_protocol::ThreadRealtimeStartParams; use codex_app_server_protocol::ThreadRealtimeStopParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadRollbackParams; +use codex_app_server_protocol::ThreadSearchOccurrencesParams; use codex_app_server_protocol::ThreadSearchParams; use codex_app_server_protocol::ThreadSetNameParams; use codex_app_server_protocol::ThreadSettingsUpdateParams; @@ -619,6 +620,15 @@ impl TestAppServer { self.send_request("thread/search", params).await } + /// Send a `thread/searchOccurrences` JSON-RPC request. + pub async fn send_thread_search_occurrences_request( + &mut self, + params: ThreadSearchOccurrencesParams, + ) -> anyhow::Result { + let params = Some(serde_json::to_value(params)?); + self.send_request("thread/searchOccurrences", params).await + } + /// Send a `thread/loaded/list` JSON-RPC request. pub async fn send_thread_loaded_list_request( &mut self, diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 4bc71157dd..482a6ff1dd 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -31,6 +31,8 @@ use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadResumeInitialTurnsPageParams; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; +use codex_app_server_protocol::ThreadSearchOccurrencesParams; +use codex_app_server_protocol::ThreadSearchOccurrencesResponse; use codex_app_server_protocol::ThreadSetNameParams; use codex_app_server_protocol::ThreadSetNameResponse; use codex_app_server_protocol::ThreadStartParams; @@ -56,6 +58,7 @@ use codex_protocol::items::AgentMessageItem; use codex_protocol::items::TurnItem as CoreTurnItem; use codex_protocol::items::UserMessageItem; use codex_protocol::models::BaseInstructions; +use codex_protocol::models::MessagePhase; use codex_protocol::protocol::AgentMessageEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ItemCompletedEvent; @@ -491,6 +494,191 @@ async fn thread_turns_list_supports_requested_items_view() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_search_occurrences_reads_paginated_projection() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + let thread_id = codex_protocol::ThreadId::default(); + let state_db = codex_state::StateRuntime::init( + codex_home.path().to_path_buf(), + "mock_provider".to_string(), + ) + .await?; + let store = LocalThreadStore::new( + LocalThreadStoreConfig { + codex_home: codex_home.path().to_path_buf(), + sqlite_home: codex_home.path().to_path_buf(), + default_model_provider_id: "mock_provider".to_string(), + }, + Some(state_db), + ); + store + .create_thread(CreateThreadParams { + session_id: thread_id.into(), + thread_id, + extra_config: None, + forked_from_id: None, + parent_thread_id: None, + source: ProtocolSessionSource::Cli, + thread_source: None, + originator: "test_originator".to_string(), + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + selected_capability_roots: Vec::new(), + multi_agent_version: None, + history_mode: codex_protocol::protocol::ThreadHistoryMode::Paginated, + subagent_history_start_ordinal: None, + initial_window_id: Uuid::now_v7().to_string(), + metadata: ThreadPersistenceMetadata { + cwd: Some(codex_home.path().to_path_buf()), + model_provider: "mock_provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }) + .await?; + store.persist_thread(thread_id).await?; + store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![ + paginated_turn_started("turn-1"), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::UserMessage(UserMessageItem { + id: "user-1".to_string(), + client_id: None, + content: vec![ + codex_protocol::user_input::UserInput::Text { + text: "Nee".to_string(), + text_elements: Vec::new(), + }, + codex_protocol::user_input::UserInput::Text { + text: "dle needle needle needle".to_string(), + text_elements: Vec::new(), + }, + ], + }), + ), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::UserMessage(UserMessageItem { + id: "steer-1".to_string(), + client_id: None, + content: vec![codex_protocol::user_input::UserInput::Text { + text: "steer toward needle".to_string(), + text_elements: Vec::new(), + }], + }), + ), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::AgentMessage(AgentMessageItem { + id: "commentary-1".to_string(), + content: vec![AgentMessageContent::Text { + text: "commentary needle".to_string(), + }], + phase: Some(MessagePhase::Commentary), + memory_citation: None, + }), + ), + paginated_completed_item( + thread_id, + "turn-1", + CoreTurnItem::AgentMessage(AgentMessageItem { + id: "final-1".to_string(), + content: vec![AgentMessageContent::Text { + text: "😀 **Final** \nneedle".to_string(), + }], + phase: Some(MessagePhase::FinalAnswer), + memory_citation: None, + }), + ), + paginated_turn_completed("turn-1"), + ], + }) + .await?; + store.shutdown_thread(thread_id).await?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + let request_id = mcp + .send_thread_search_occurrences_request(ThreadSearchOccurrencesParams { + thread_id: thread_id.to_string(), + search_term: "needle".to_string(), + cursor: None, + limit: Some(3), + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let ThreadSearchOccurrencesResponse { data, next_cursor } = to_response(response)?; + + assert_eq!( + data.iter() + .map(|occurrence| occurrence.item_id.as_str()) + .collect::>(), + vec!["user-1", "user-1", "user-1"] + ); + assert_eq!( + data.iter() + .map(|occurrence| occurrence.turn_id.as_str()) + .collect::>(), + vec!["turn-1", "turn-1", "turn-1"] + ); + assert_eq!( + data.iter() + .map(|occurrence| occurrence.snippet_match_range.start) + .collect::>(), + vec![0, 7, 14] + ); + let next_cursor = next_cursor.expect("first page should have another occurrence"); + + let request_id = mcp + .send_thread_search_occurrences_request(ThreadSearchOccurrencesParams { + thread_id: thread_id.to_string(), + search_term: "needle".to_string(), + cursor: Some(next_cursor), + limit: Some(3), + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let ThreadSearchOccurrencesResponse { data, next_cursor } = to_response(response)?; + + assert_eq!( + data.iter() + .map(|occurrence| occurrence.item_id.as_str()) + .collect::>(), + vec!["user-1", "steer-1", "final-1"] + ); + assert_eq!( + data.iter() + .map(|occurrence| occurrence.turn_id.as_str()) + .collect::>(), + vec!["turn-1", "turn-1", "turn-1"] + ); + assert_eq!(data[2].snippet, "😀 Final needle"); + assert_eq!(data[2].snippet_match_range.start, 9); + assert_eq!(data[2].snippet_match_range.end, 15); + assert_eq!(next_cursor, None); + + Ok(()) +} + #[tokio::test] async fn thread_turns_list_reads_store_history_without_rollout_path() -> Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/state/thread_history_migrations/0002_thread_items_item_type.sql b/codex-rs/state/thread_history_migrations/0002_thread_items_item_type.sql new file mode 100644 index 0000000000..b29b90bdc3 --- /dev/null +++ b/codex-rs/state/thread_history_migrations/0002_thread_items_item_type.sql @@ -0,0 +1,9 @@ +ALTER TABLE thread_items ADD COLUMN item_type TEXT NOT NULL DEFAULT ''; + +UPDATE thread_items +SET item_type = json_extract(item_json, '$.type') +WHERE item_type = ''; + +CREATE INDEX idx_thread_items_user_messages + ON thread_items(thread_id, rollout_ordinal) + WHERE item_type = 'userMessage'; diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml index 8260c250a2..844d1769f8 100644 --- a/codex-rs/thread-store/Cargo.toml +++ b/codex-rs/thread-store/Cargo.toml @@ -22,6 +22,8 @@ codex-protocol = { workspace = true } codex-rollout = { workspace = true } codex-state = { workspace = true } codex-utils-path = { workspace = true } +futures = { workspace = true } +pulldown-cmark = { workspace = true } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } sqlx = { workspace = true } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 7dcf07c7a2..e218387135 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -37,18 +37,22 @@ pub use types::LoadThreadHistoryParams; pub use types::ReadThreadByRolloutPathParams; pub use types::ReadThreadParams; pub use types::ResumeThreadParams; +pub use types::SearchTextRange; +pub use types::SearchThreadOccurrencesParams; pub use types::SearchThreadsParams; pub use types::SortDirection; pub use types::StoredModelContext; pub use types::StoredThread; pub use types::StoredThreadHistory; pub use types::StoredThreadItem; +pub use types::StoredThreadOccurrence; pub use types::StoredThreadSearchResult; pub use types::StoredTurn; pub use types::StoredTurnError; pub use types::StoredTurnItemsView; pub use types::StoredTurnStatus; pub use types::ThreadMetadataPatch; +pub use types::ThreadOccurrenceSearchPage; pub use types::ThreadPage; pub use types::ThreadPersistenceMetadata; pub use types::ThreadRelationFilter; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index f13e94559b..f0f8adbec6 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -39,10 +39,12 @@ use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; use crate::ResumeThreadParams; +use crate::SearchThreadOccurrencesParams; use crate::SearchThreadsParams; use crate::StoredModelContext; use crate::StoredThread; use crate::StoredThreadHistory; +use crate::ThreadOccurrenceSearchPage; use crate::ThreadPage; use crate::ThreadSearchPage; use crate::ThreadStore; @@ -278,6 +280,14 @@ impl LocalThreadStore { pub async fn list_items(&self, params: ListItemsParams) -> ThreadStoreResult { thread_history::list_items(self, params).await } + + /// Searches projection-backed visible messages within one paginated thread. + pub async fn search_thread_occurrences( + &self, + params: SearchThreadOccurrencesParams, + ) -> ThreadStoreResult { + thread_history::search_thread_occurrences(self, params).await + } } impl ThreadStore for LocalThreadStore { @@ -363,6 +373,13 @@ impl ThreadStore for LocalThreadStore { Box::pin(async move { search_threads::search_threads(self, params).await }) } + fn search_thread_occurrences( + &self, + params: SearchThreadOccurrencesParams, + ) -> ThreadStoreFuture<'_, ThreadOccurrenceSearchPage> { + Box::pin(LocalThreadStore::search_thread_occurrences(self, params)) + } + fn update_thread_metadata( &self, params: UpdateThreadMetadataParams, diff --git a/codex-rs/thread-store/src/local/thread_history.rs b/codex-rs/thread-store/src/local/thread_history.rs index ba1faab3bc..b09568691b 100644 --- a/codex-rs/thread-store/src/local/thread_history.rs +++ b/codex-rs/thread-store/src/local/thread_history.rs @@ -9,9 +9,11 @@ use crate::ThreadStoreError; use crate::ThreadStoreResult; mod read; +mod search; pub(super) use read::list_items; pub(super) use read::list_turns; +pub(super) use search::search_thread_occurrences; pub(super) async fn next_rollout_byte_offset( store: &LocalThreadStore, @@ -285,9 +287,11 @@ INSERT INTO thread_items ( item_id, rollout_ordinal, created_at_ms, + item_type, item_json -) VALUES (?, ?, ?, ?, ?, ?) +) VALUES (?, ?, ?, ?, ?, json_extract(?, '$.type'), ?) ON CONFLICT(thread_id, turn_id, item_id) DO UPDATE SET + item_type = excluded.item_type, item_json = excluded.item_json "#, ) @@ -296,6 +300,7 @@ ON CONFLICT(thread_id, turn_id, item_id) DO UPDATE SET .bind(item_id.as_str()) .bind(rollout_ordinal) .bind(created_at_ms) + .bind(item_json.as_str()) .bind(item_json) .execute(&mut **transaction) .await diff --git a/codex-rs/thread-store/src/local/thread_history/read.rs b/codex-rs/thread-store/src/local/thread_history/read.rs index 79b9a16ec5..a119a13bfd 100644 --- a/codex-rs/thread-store/src/local/thread_history/read.rs +++ b/codex-rs/thread-store/src/local/thread_history/read.rs @@ -36,7 +36,7 @@ struct HistoryCursor { #[derive(Clone, Deserialize, PartialEq, Eq, Serialize)] #[serde(tag = "kind", rename_all = "camelCase")] -enum CursorScope { +pub(super) enum CursorScope { Turns, Items, } @@ -191,7 +191,7 @@ WHERE thread_id = }) } -async fn validate_thread_for_paginated_reads( +pub(super) async fn validate_thread_for_paginated_reads( store: &LocalThreadStore, thread_id: ThreadId, include_archived: bool, @@ -306,7 +306,7 @@ fn page_cursors( Ok((next_cursor, backwards_cursor)) } -fn serialize_cursor( +pub(super) fn serialize_cursor( thread_id: ThreadId, scope: &CursorScope, rollout_ordinal: i64, diff --git a/codex-rs/thread-store/src/local/thread_history/search.rs b/codex-rs/thread-store/src/local/thread_history/search.rs new file mode 100644 index 0000000000..a96b341cef --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_history/search.rs @@ -0,0 +1,415 @@ +use std::borrow::Cow; + +use codex_app_server_protocol::ThreadItem; +use codex_app_server_protocol::UserInput; +use codex_protocol::ThreadId; +use codex_protocol::protocol::strip_user_message_prefix; +use futures::TryStreamExt; +use pulldown_cmark::Event; +use pulldown_cmark::Parser; +use pulldown_cmark::TagEnd; +use serde::Deserialize; +use serde::Serialize; +use sqlx::Row; + +use super::super::LocalThreadStore; +use super::read::CursorScope; +use super::read::serialize_cursor; +use super::read::validate_thread_for_paginated_reads; +use super::thread_history_error; +use crate::SearchTextRange; +use crate::SearchThreadOccurrencesParams; +use crate::StoredThreadOccurrence; +use crate::ThreadOccurrenceSearchPage; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +const SNIPPET_CONTEXT_BEFORE_CHARS: usize = 48; +const SNIPPET_CONTEXT_AFTER_CHARS: usize = 96; + +#[derive(Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +struct SearchCursor { + thread_id: ThreadId, + search_term: String, + next_rollout_ordinal: i64, + next_occurrence_index: usize, +} + +struct CandidateRow { + turn_id: String, + item_id: String, + rollout_ordinal: i64, + item_json: String, + turn_rollout_ordinal: i64, +} + +pub(in crate::local) async fn search_thread_occurrences( + store: &LocalThreadStore, + params: SearchThreadOccurrencesParams, +) -> ThreadStoreResult { + if params.search_term.trim().is_empty() { + return Err(ThreadStoreError::InvalidRequest { + message: "thread/searchOccurrences requires search_term".to_string(), + }); + } + if params.page_size == 0 { + return Err(ThreadStoreError::InvalidRequest { + message: "thread/searchOccurrences requires page_size greater than zero".to_string(), + }); + } + validate_thread_for_paginated_reads( + store, + params.thread_id, + /*include_archived*/ true, + "thread/searchOccurrences", + ) + .await?; + let cursor = parse_cursor( + params.cursor.as_deref(), + params.thread_id, + ¶ms.search_term, + )?; + let next_rollout_ordinal = cursor + .as_ref() + .map_or(0, |cursor| cursor.next_rollout_ordinal); + let matcher = LiteralMatcher::new(params.search_term.as_str()); + let pool = store.thread_history_db().await?; + let mut rows = sqlx::query( + r#" +SELECT turn_id, item_id, rollout_ordinal, item_json, turn_rollout_ordinal +FROM ( + SELECT + items.turn_id, + items.item_id, + items.rollout_ordinal, + items.item_json, + turns.rollout_ordinal AS turn_rollout_ordinal + FROM thread_items AS items + JOIN thread_turns AS turns + ON turns.thread_id = items.thread_id + AND turns.turn_id = items.turn_id + WHERE items.thread_id = ? + AND items.item_type = 'userMessage' + AND items.rollout_ordinal >= ? + + UNION ALL + + SELECT + items.turn_id, + items.item_id, + items.rollout_ordinal, + items.item_json, + turns.rollout_ordinal AS turn_rollout_ordinal + FROM thread_turns AS turns + JOIN thread_items AS items + ON items.thread_id = turns.thread_id + AND items.turn_id = turns.turn_id + AND items.item_id = turns.final_agent_item_id + WHERE turns.thread_id = ? + AND turns.final_agent_item_id IS NOT NULL + AND items.rollout_ordinal >= ? +) +ORDER BY rollout_ordinal ASC + "#, + ) + .bind(params.thread_id.to_string()) + .bind(next_rollout_ordinal) + .bind(params.thread_id.to_string()) + .bind(next_rollout_ordinal) + .fetch(pool); + + let mut items = Vec::with_capacity(params.page_size); + while let Some(row) = rows.try_next().await.map_err(thread_history_error)? { + let row = candidate_row(row)?; + let item = serde_json::from_str::(row.item_json.as_str()).map_err(|err| { + ThreadStoreError::Internal { + message: format!("failed to deserialize stored thread item: {err}"), + } + })?; + let Some(text) = searchable_text(&item) else { + continue; + }; + let first_occurrence_index = cursor + .as_ref() + .filter(|cursor| cursor.next_rollout_ordinal == row.rollout_ordinal) + .map_or(0, |cursor| cursor.next_occurrence_index); + let remaining = params + .page_size + .saturating_add(1) + .saturating_sub(items.len()); + let turn_cursor = serialize_cursor( + params.thread_id, + &CursorScope::Turns, + row.turn_rollout_ordinal, + /*include_anchor*/ true, + )?; + for (occurrence_index, matched) in matcher + .find_ranges( + text.as_ref(), + first_occurrence_index.saturating_add(remaining), + ) + .into_iter() + .enumerate() + .skip(first_occurrence_index) + { + if items.len() == params.page_size { + return Ok(ThreadOccurrenceSearchPage { + items, + next_cursor: Some(serialize_cursor_for_search(SearchCursor { + thread_id: params.thread_id, + search_term: params.search_term, + next_rollout_ordinal: row.rollout_ordinal, + next_occurrence_index: occurrence_index, + })?), + }); + } + items.push(occurrence_in_item( + row.turn_id.as_str(), + row.item_id.as_str(), + text.as_ref(), + matched, + turn_cursor.as_str(), + )); + } + } + + Ok(ThreadOccurrenceSearchPage { + items, + next_cursor: None, + }) +} + +fn candidate_row(row: sqlx::sqlite::SqliteRow) -> ThreadStoreResult { + let rollout_ordinal = row.try_get::("rollout_ordinal")?; + let turn_rollout_ordinal = row.try_get::("turn_rollout_ordinal")?; + if rollout_ordinal < 0 || turn_rollout_ordinal < 0 { + return Err(ThreadStoreError::Internal { + message: "invalid stored thread history ordinal".to_string(), + }); + } + Ok(CandidateRow { + turn_id: row.try_get("turn_id")?, + item_id: row.try_get("item_id")?, + rollout_ordinal, + item_json: row.try_get("item_json")?, + turn_rollout_ordinal, + }) +} + +fn parse_cursor( + cursor: Option<&str>, + thread_id: ThreadId, + search_term: &str, +) -> ThreadStoreResult> { + let Some(cursor) = cursor else { + return Ok(None); + }; + let cursor_value: SearchCursor = + serde_json::from_str(cursor).map_err(|_| invalid_cursor(cursor))?; + if cursor_value.thread_id != thread_id + || cursor_value.search_term != search_term + || cursor_value.next_rollout_ordinal < 0 + { + return Err(invalid_cursor(cursor)); + } + Ok(Some(cursor_value)) +} + +fn serialize_cursor_for_search(cursor: SearchCursor) -> ThreadStoreResult { + serde_json::to_string(&cursor).map_err(thread_history_error) +} + +fn invalid_cursor(cursor: &str) -> ThreadStoreError { + ThreadStoreError::InvalidRequest { + message: format!("invalid cursor: {cursor}"), + } +} + +fn searchable_text(item: &ThreadItem) -> Option> { + match item { + ThreadItem::UserMessage { content, .. } => { + let mut text_parts = content + .iter() + .filter_map(|input| match input { + UserInput::Text { text, .. } => Some(strip_user_message_prefix(text)), + UserInput::Image { .. } + | UserInput::LocalImage { .. } + | UserInput::Skill { .. } + | UserInput::Mention { .. } => None, + }) + .filter(|text| !text.is_empty()) + .peekable(); + let first = text_parts.next()?; + match text_parts.next() { + None => Some(Cow::Borrowed(first)), + Some(second) => { + let mut parts = vec![first, second]; + parts.extend(text_parts); + Some(Cow::Owned(parts.concat())) + } + } + } + ThreadItem::AgentMessage { text, .. } => { + let text = markdown_to_search_text(text); + (!text.is_empty()).then_some(Cow::Owned(text)) + } + ThreadItem::HookPrompt { .. } + | ThreadItem::Plan { .. } + | ThreadItem::Reasoning { .. } + | ThreadItem::CommandExecution { .. } + | ThreadItem::FileChange { .. } + | ThreadItem::McpToolCall { .. } + | ThreadItem::DynamicToolCall { .. } + | ThreadItem::CollabAgentToolCall { .. } + | ThreadItem::SubAgentActivity { .. } + | ThreadItem::WebSearch(_) + | ThreadItem::ImageView { .. } + | ThreadItem::Sleep(_) + | ThreadItem::ImageGeneration(_) + | ThreadItem::EnteredReviewMode { .. } + | ThreadItem::ExitedReviewMode { .. } + | ThreadItem::ContextCompaction { .. } => None, + } +} + +fn markdown_to_search_text(markdown: &str) -> String { + let mut text = String::new(); + for event in Parser::new(markdown.trim()) { + match event { + Event::Text(value) + | Event::Code(value) + | Event::Html(value) + | Event::InlineHtml(value) => text.push_str(&value), + Event::SoftBreak | Event::HardBreak | Event::Rule => text.push(' '), + Event::End( + TagEnd::Paragraph + | TagEnd::Heading(_) + | TagEnd::BlockQuote + | TagEnd::CodeBlock + | TagEnd::List(_) + | TagEnd::Item + | TagEnd::Table + | TagEnd::TableHead + | TagEnd::TableRow + | TagEnd::TableCell, + ) => text.push(' '), + Event::Start(_) + | Event::End( + TagEnd::Emphasis + | TagEnd::Strong + | TagEnd::Strikethrough + | TagEnd::Link + | TagEnd::HtmlBlock + | TagEnd::FootnoteDefinition + | TagEnd::Image + | TagEnd::MetadataBlock(_), + ) + | Event::FootnoteReference(_) + | Event::TaskListMarker(_) => {} + } + } + text.split_whitespace().collect::>().join(" ") +} + +struct LiteralMatcher { + lowercase_needle: String, +} + +impl LiteralMatcher { + fn new(needle: &str) -> Self { + Self { + lowercase_needle: needle.to_lowercase(), + } + } + + fn find_ranges(&self, text: &str, limit: usize) -> Vec> { + let lowercase_text = text.to_lowercase(); + let mut spans = Vec::with_capacity(text.chars().count()); + let mut lowercase_start = 0; + for (original_start, character) in text.char_indices() { + let lowercase_end = + lowercase_start + character.to_lowercase().map(char::len_utf8).sum::(); + spans.push(( + lowercase_start..lowercase_end, + original_start..original_start + character.len_utf8(), + )); + lowercase_start = lowercase_end; + } + + lowercase_text + .match_indices(self.lowercase_needle.as_str()) + .take(limit) + .filter_map(|(start, matched)| { + let end = start.saturating_add(matched.len()); + let original_start = spans + .iter() + .find(|(lowercase, _)| lowercase.contains(&start))? + .1 + .start; + let original_end = spans + .iter() + .find(|(lowercase, _)| lowercase.contains(&end.saturating_sub(1)))? + .1 + .end; + Some(original_start..original_end) + }) + .collect() + } +} + +fn occurrence_in_item( + turn_id: &str, + item_id: &str, + text: &str, + matched: std::ops::Range, + turn_cursor: &str, +) -> StoredThreadOccurrence { + let snippet_start = char_start_before(text, matched.start, SNIPPET_CONTEXT_BEFORE_CHARS); + let snippet_end = char_end_after(text, matched.end, SNIPPET_CONTEXT_AFTER_CHARS); + let leading_ellipsis = snippet_start > 0; + let trailing_ellipsis = snippet_end < text.len(); + let mut snippet = String::new(); + if leading_ellipsis { + snippet.push_str("... "); + } + snippet.push_str(&text[snippet_start..snippet_end]); + if trailing_ellipsis { + snippet.push_str(" ..."); + } + let snippet_match_start = + if leading_ellipsis { 4 } else { 0 } + utf16_len(&text[snippet_start..matched.start]); + let match_len = utf16_len(&text[matched]); + + StoredThreadOccurrence { + turn_id: turn_id.to_string(), + item_id: item_id.to_string(), + snippet, + snippet_match_range: SearchTextRange { + start: snippet_match_start, + end: snippet_match_start.saturating_add(match_len), + }, + turn_cursor: turn_cursor.to_string(), + } +} + +fn utf16_len(text: &str) -> u32 { + u32::try_from(text.encode_utf16().count()).unwrap_or(u32::MAX) +} + +fn char_start_before(text: &str, byte_index: usize, chars_before: usize) -> usize { + text[..byte_index] + .char_indices() + .rev() + .nth(chars_before) + .map(|(index, _)| index) + .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()) +} diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 9d12ad2e32..d596acd9e8 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -16,10 +16,12 @@ use crate::LoadThreadHistoryParams; use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; use crate::ResumeThreadParams; +use crate::SearchThreadOccurrencesParams; use crate::SearchThreadsParams; use crate::StoredModelContext; use crate::StoredThread; use crate::StoredThreadHistory; +use crate::ThreadOccurrenceSearchPage; use crate::ThreadPage; use crate::ThreadSearchPage; use crate::ThreadStoreError; @@ -122,6 +124,18 @@ pub trait ThreadStore: Any + Send + Sync { }) } + /// Searches visible message occurrences within one paginated thread. + fn search_thread_occurrences( + &self, + _params: SearchThreadOccurrencesParams, + ) -> ThreadStoreFuture<'_, ThreadOccurrenceSearchPage> { + Box::pin(async { + Err(ThreadStoreError::Unsupported { + operation: "thread/searchOccurrences", + }) + }) + } + /// Lists turns within a stored thread. fn list_turns(&self, _params: ListTurnsParams) -> ThreadStoreFuture<'_, TurnPage> { Box::pin(async { diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index c887800eda..37b02cb51a 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -425,6 +425,44 @@ pub struct ItemPage { pub backwards_cursor: Option, } +/// Parameters for searching visible message occurrences within one paginated thread. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SearchThreadOccurrencesParams { + /// Thread id to search. + pub thread_id: ThreadId, + /// Case-insensitive literal substring to find. + pub search_term: String, + /// Opaque cursor returned by a previous search call. + pub cursor: Option, + /// Maximum number of occurrences to return. + pub page_size: usize, +} + +/// UTF-16 code-unit range within `snippet`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct SearchTextRange { + pub start: u32, + pub end: u32, +} + +/// One visible message occurrence within a stored thread. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct StoredThreadOccurrence { + pub turn_id: String, + pub item_id: String, + pub snippet: String, + pub snippet_match_range: SearchTextRange, + /// Inclusive cursor accepted by `thread/turns/list` for this turn. + pub turn_cursor: String, +} + +/// A page of visible message occurrences within one stored thread. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ThreadOccurrenceSearchPage { + pub items: Vec, + pub next_cursor: Option, +} + /// Store-owned thread metadata used by list/read/resume responses. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct StoredThread {