mirror of
https://github.com/openai/codex.git
synced 2026-10-18 17:18:52 +00:00
Add occurrence search for paginated threads (#33907)
## What changed - Add the experimental `thread/searchOccurrences` app-server method for case-insensitive literal search across visible user messages and final assistant messages without replaying the thread. - Return paginated snippets with UTF-16 match ranges and `thread/turns/list` cursors for navigating to each matching turn. - Index persisted item types to efficiently select searchable messages from paginated thread history. ## Testing - Add an app-server integration test covering pagination, steering messages, final-answer selection, Markdown normalization, and UTF-16 offsets. GitOrigin-RevId: e80160e0f83eb4e5db8de862307085093bee362b
This commit is contained in:
Owen Lin
Generated
-1
@@ -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\":{}}",
|
||||
|
||||
Generated
+2
-17
@@ -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"
|
||||
|
||||
+1
-1
@@ -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"
|
||||
|
||||
@@ -50,6 +50,8 @@ const EXPERIMENTAL_CLIENT_METHOD_DEPENDENCY_TYPES: &[&str] = &[
|
||||
"RemoteControlClient",
|
||||
"RemoteControlClientsListOrder",
|
||||
"ThreadBackgroundTerminal",
|
||||
"ThreadSearchOccurrence",
|
||||
"ThreadSearchTextRange",
|
||||
];
|
||||
const SPECIAL_DEFINITIONS: &[&str] = &[
|
||||
"ClientNotification",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -1249,6 +1249,58 @@ pub struct ThreadSearchResponse {
|
||||
pub backwards_cursor: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
/// Optional occurrence page size.
|
||||
#[ts(optional = nullable)]
|
||||
pub limit: Option<u32>,
|
||||
}
|
||||
|
||||
/// 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<ThreadSearchOccurrence>,
|
||||
/// Opaque cursor to continue after the last returned occurrence.
|
||||
pub next_cursor: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Default, JsonSchema, TS)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
#[ts(export_to = "v2/")]
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -710,6 +710,15 @@ impl ThreadRequestProcessor {
|
||||
.map(|response| Some(response.into()))
|
||||
}
|
||||
|
||||
pub(crate) async fn thread_search_occurrences(
|
||||
&self,
|
||||
params: ThreadSearchOccurrencesParams,
|
||||
) -> Result<Option<ClientResponsePayload>, 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<ThreadSearchOccurrencesResponse, JSONRPCErrorError> {
|
||||
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,
|
||||
|
||||
@@ -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<i64> {
|
||||
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,
|
||||
|
||||
@@ -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<_>>(),
|
||||
vec!["user-1", "user-1", "user-1"]
|
||||
);
|
||||
assert_eq!(
|
||||
data.iter()
|
||||
.map(|occurrence| occurrence.turn_id.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
vec!["turn-1", "turn-1", "turn-1"]
|
||||
);
|
||||
assert_eq!(
|
||||
data.iter()
|
||||
.map(|occurrence| occurrence.snippet_match_range.start)
|
||||
.collect::<Vec<_>>(),
|
||||
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<_>>(),
|
||||
vec!["user-1", "steer-1", "final-1"]
|
||||
);
|
||||
assert_eq!(
|
||||
data.iter()
|
||||
.map(|occurrence| occurrence.turn_id.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
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()?;
|
||||
|
||||
@@ -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';
|
||||
@@ -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 }
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<ItemPage> {
|
||||
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<ThreadOccurrenceSearchPage> {
|
||||
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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<ThreadOccurrenceSearchPage> {
|
||||
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::<ThreadItem>(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<CandidateRow> {
|
||||
let rollout_ordinal = row.try_get::<i64, _>("rollout_ordinal")?;
|
||||
let turn_rollout_ordinal = row.try_get::<i64, _>("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<Option<SearchCursor>> {
|
||||
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<String> {
|
||||
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<Cow<'_, str>> {
|
||||
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::<Vec<_>>().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<std::ops::Range<usize>> {
|
||||
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::<usize>();
|
||||
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<usize>,
|
||||
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())
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -425,6 +425,44 @@ pub struct ItemPage {
|
||||
pub backwards_cursor: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
/// 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<StoredThreadOccurrence>,
|
||||
pub next_cursor: Option<String>,
|
||||
}
|
||||
|
||||
/// Store-owned thread metadata used by list/read/resume responses.
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct StoredThread {
|
||||
|
||||
Reference in New Issue
Block a user