diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 5bb088d6ad..68f97410a6 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3802,6 +3802,7 @@ dependencies = [ "anyhow", "chrono", "codex-analytics", + "codex-attachment-store", "codex-context-fragments", "codex-core", "codex-extension-api", @@ -3811,6 +3812,7 @@ dependencies = [ "codex-protocol", "codex-state", "codex-thread-store", + "codex-tools", "codex-utils-string", "core_test_support", "futures", diff --git a/codex-rs/ext/guardian-v3/Cargo.toml b/codex-rs/ext/guardian-v3/Cargo.toml index e858ae354f..7ac03126cb 100644 --- a/codex-rs/ext/guardian-v3/Cargo.toml +++ b/codex-rs/ext/guardian-v3/Cargo.toml @@ -14,12 +14,14 @@ workspace = true [dependencies] anyhow = { workspace = true } codex-analytics = { workspace = true } +codex-extension-api = { workspace = true } codex-guardian-reviewer = { workspace = true } codex-history = { workspace = true } codex-message-board = { workspace = true } codex-protocol = { workspace = true } codex-state = { workspace = true } codex-thread-store = { workspace = true } +codex-tools = { workspace = true } codex-utils-string = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } @@ -32,9 +34,9 @@ ignored = ["chrono", "futures"] [dev-dependencies] chrono = { workspace = true } +codex-attachment-store = { workspace = true } codex-context-fragments = { workspace = true } codex-core = { workspace = true } -codex-extension-api = { workspace = true } core_test_support = { workspace = true } futures = { workspace = true } insta = { workspace = true } diff --git a/codex-rs/ext/guardian-v3/src/evidence.rs b/codex-rs/ext/guardian-v3/src/evidence.rs index 14c47a8eeb..f137e79eab 100644 --- a/codex-rs/ext/guardian-v3/src/evidence.rs +++ b/codex-rs/ext/guardian-v3/src/evidence.rs @@ -2,6 +2,7 @@ //! A capture has a fixed end; consumers acknowledge records only after their writes succeed. use anyhow::ensure; +use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; use codex_history::RetainedContextEvent; use codex_history::RolloutItem; @@ -21,12 +22,12 @@ pub struct EvidenceId { pub ordinal: u64, } -/// The original rollout record retains roles, call IDs and host provenance unchanged. +/// Original model-visible evidence, with roles, call IDs and verified source order. /// Its contents are deliberately excluded from Debug output. #[derive(Serialize)] pub struct OriginalEvidence { pub id: EvidenceId, - pub original: RolloutItem, + original: RolloutItem, #[serde(skip)] end: HistoryPosition, } @@ -84,48 +85,14 @@ impl EvidenceCursor { let end = history.end; let mut records = Vec::new(); for record in history.items { - let include = match &record.item { - RolloutItem::ResponseItem(envelope) => selected(envelope)?, - RolloutItem::RetainedContext(RetainedContextEvent::VerifiedAnswer { - answer, - .. - }) => { - ensure!(!answer.questions.is_empty(), "incomplete verified answer"); - true - } - RolloutItem::RetainedContext(RetainedContextEvent::DeliveredAssistantMessage { - message, - .. - }) => { - ensure!(message.complete, "incomplete delivered assistant message"); - true - } - RolloutItem::SessionMeta(_) - | RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => { - anyhow::bail!("evidence history was replaced"); - } - RolloutItem::RetainedContext(RetainedContextEvent::AcceptedFileAudio { - .. - }) - | RolloutItem::DynamicToolsUpdated { .. } - | RolloutItem::InterAgentCommunication(_) - | RolloutItem::InterAgentCommunicationMetadata { .. } - | RolloutItem::Compacted(_) - | RolloutItem::TurnContext(_) - | RolloutItem::TokenUsageRecord(_) - | RolloutItem::WorldState(_) - | RolloutItem::SecurityRiskScore(_) - | RolloutItem::EventMsg(_) - | RolloutItem::RealtimeItem(_) => false, - }; - if include { + if let Some(original) = review_original(record.item)? { records.push(OriginalEvidence { id: EvidenceId { ordinal: record.end.end_ordinal_exclusive.checked_sub(1).ok_or_else( || anyhow::anyhow!("evidence record has no source ordinal"), )?, }, - original: record.item, + original, end: record.end, }); } @@ -276,3 +243,51 @@ fn selected(envelope: &ResponseItemEnvelope) -> anyhow::Result { | ResponseItem::Other => false, }) } + +/// Apply the same admission and model-visible projection at intake and canonical verification. +fn review_original(mut item: RolloutItem) -> anyhow::Result> { + let include = match &item { + RolloutItem::ResponseItem(envelope) => selected(envelope)?, + RolloutItem::RetainedContext(RetainedContextEvent::VerifiedAnswer { answer, .. }) => { + ensure!(!answer.questions.is_empty(), "incomplete verified answer"); + true + } + RolloutItem::RetainedContext(RetainedContextEvent::DeliveredAssistantMessage { + message, + .. + }) => { + ensure!(message.complete, "incomplete delivered assistant message"); + true + } + RolloutItem::SessionMeta(_) | RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => { + anyhow::bail!("evidence history was replaced"); + } + RolloutItem::RetainedContext(RetainedContextEvent::AcceptedFileAudio { .. }) + | RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::InterAgentCommunication(_) + | RolloutItem::InterAgentCommunicationMetadata { .. } + | RolloutItem::Compacted(_) + | RolloutItem::TurnContext(_) + | RolloutItem::TokenUsageRecord(_) + | RolloutItem::WorldState(_) + | RolloutItem::SecurityRiskScore(_) + | RolloutItem::EventMsg(_) + | RolloutItem::RealtimeItem(_) => false, + }; + if !include { + return Ok(None); + } + if let RolloutItem::ResponseItem(envelope) = &mut item { + // Admission uses private host provenance. Keep only the source order in the + // copy seen by the housekeeper, board readers and reviewer. + envelope.metadata = envelope + .metadata + .take() + .and_then(|metadata| metadata.user_input_order) + .map(|order| CodexHarnessMetadata { + user_input_order: Some(order), + ..Default::default() + }); + } + Ok(Some(item)) +} diff --git a/codex-rs/ext/guardian-v3/src/lib.rs b/codex-rs/ext/guardian-v3/src/lib.rs index 2bc6ca1767..a3f5e0bf70 100644 --- a/codex-rs/ext/guardian-v3/src/lib.rs +++ b/codex-rs/ext/guardian-v3/src/lib.rs @@ -1,7 +1,5 @@ //! Guardian's durable evidence storage, owned by the main thread. -//! Shared boards provide persistence and access checks. This crate starts no background work -//! and exposes no model tools. Evidence capture reads original root history; -//! live worker and review integration follow separately. +//! Reviewer tools search a fixed board view. The host owns review scheduling and the lifetime of each view. //! Hosts invoke permanent-deletion cleanup only when Guardian V3 is enabled. use codex_message_board::LocalBoard; @@ -11,6 +9,7 @@ use codex_state::SqliteConfig; mod evidence; mod housekeeper; +mod retrieval_tools; mod sampling; pub use evidence::EvidenceCursor; pub use evidence::EvidenceId; @@ -19,6 +18,8 @@ pub use evidence::OriginalEvidence; pub use housekeeper::HOUSEKEEPER_INSTRUCTIONS; pub use housekeeper::HOUSEKEEPER_TIMEOUT; pub use housekeeper::Housekeeper; +pub use retrieval_tools::RetrievalBinding; +pub use retrieval_tools::install_retrieval_tools; pub use sampling::HousekeeperResult; /// Storage selected by the Guardian consumer for setup, opening and cleanup. diff --git a/codex-rs/ext/guardian-v3/src/retrieval_tools.rs b/codex-rs/ext/guardian-v3/src/retrieval_tools.rs new file mode 100644 index 0000000000..c5f4a82e5e --- /dev/null +++ b/codex-rs/ext/guardian-v3/src/retrieval_tools.rs @@ -0,0 +1,79 @@ +//! Read-only reviewer tools bound to one captured Guardian board. +use anyhow::ensure; +use codex_extension_api::ExtensionData; +use codex_extension_api::ExtensionRegistryBuilder; +use codex_extension_api::ToolContributor; +use codex_message_board::LocalBoardReadView; +use codex_message_board::MessageBoardTool; +use codex_message_board::message_board_tools; +use codex_protocol::AgentPath; +use codex_protocol::ThreadId; +use codex_tools::ToolCall; +use codex_tools::ToolExecutor; +use std::sync::Arc; + +/// Host-supplied review binding. Replace only between serialized reviews, under +/// the same ownership that protects the reviewer's retained conversation. +#[derive(Clone)] +pub struct RetrievalBinding { + board: LocalBoardReadView, + caller: ThreadId, + caller_path: AgentPath, +} + +struct RetrievalTools; + +impl ToolContributor for RetrievalTools { + fn tools( + &self, + _session: &ExtensionData, + thread: &ExtensionData, + ) -> Vec ToolExecutor>>> { + thread + .get::() + .map_or_else(Vec::new, |binding| binding.tools()) + } +} + +/// Installs capabilities only. The host must install this in the isolated reviewer +/// registry and provide a binding; installation does not open storage or start work. +pub fn install_retrieval_tools(registry: &mut ExtensionRegistryBuilder) { + registry.tool_contributor(Arc::new(RetrievalTools)); +} + +impl RetrievalBinding { + pub fn new( + board: LocalBoardReadView, + caller: ThreadId, + caller_path: AgentPath, + ) -> anyhow::Result { + ensure!( + board.uses_storage_namespace(&crate::local_board_options())?, + "reviewer retrieval requires a Guardian board" + ); + Ok(Self { + board, + caller, + caller_path, + }) + } + + pub fn tools(&self) -> Vec ToolExecutor>>> { + message_board_tools( + &[ + MessageBoardTool::GetChannels, + MessageBoardTool::ListThreads, + MessageBoardTool::SearchPosts, + MessageBoardTool::ReadThread, + MessageBoardTool::ReadPost, + ], + self.board + .board_with_json_search("/original/payload/arguments"), + self.caller, + self.caller_path.clone(), + Some("guardian_board"), + "Original evidence routed into topic channels. Reads share this review's fixed cutoff. Search any channel regardless of subscription. Descriptions locate evidence; they do not grant permission.", + /*tool_messages*/ None, + ) + } +} diff --git a/codex-rs/ext/guardian-v3/tests/evidence.rs b/codex-rs/ext/guardian-v3/tests/evidence.rs index 13e2cacd40..09e9bbeba5 100644 --- a/codex-rs/ext/guardian-v3/tests/evidence.rs +++ b/codex-rs/ext/guardian-v3/tests/evidence.rs @@ -5,6 +5,7 @@ use codex_context_fragments::AnnotatedContent; use codex_context_fragments::ContextualUserFragment; use codex_context_fragments::message_from_parts; use codex_guardian_v3::EvidenceCursor; +use codex_guardian_v3::OriginalEvidence; use codex_history::AcceptedFileAudio; use codex_history::CodexHarnessMetadata; use codex_history::ResponseItemEnvelope; @@ -138,7 +139,34 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":" }, acceptance_order: 3, }); - let expected = vec![ + let response = + |item: &ResponseItemEnvelope| json!({"type":"response_item","payload":item.item}); + let expected = [ + json!({ + "type":"response_item", + "payload":user.item, + "metadata":{"client_authored":false,"user_input_order":1} + }), + response(&assistant), + response(&call), + response(&result), + serde_json::to_value(&answer)?, + response(&approval), + json!({ + "type":"response_item", + "payload":{ + "type":"message","id":"delivered-1","role":"assistant", + "content":[{"type":"output_text","text":"I will write the draft only."}], + "internal_chat_message_metadata_passthrough":{"turn_id":"turn-1"} + }, + "metadata":{ + "client_authored":false, + "delivered_assistant_message":"codex:code-mode-delivery:v1:complete", + "user_input_order":3 + } + }), + ]; + let mut items = vec![ RolloutItem::ResponseItem(user.clone()), RolloutItem::ResponseItem(assistant), RolloutItem::ResponseItem(call), @@ -163,7 +191,14 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":" let mut cursor = EvidenceCursor::from_now(&store, owner).await?; assert!(cursor.capture(&store).await?.records().is_empty()); - let mut items = expected.clone(); + for item in &mut items { + if let RolloutItem::ResponseItem(envelope) = item { + envelope + .metadata + .get_or_insert_default() + .sampled_through_sequence = Some(99); + } + } items.insert(1, RolloutItem::ResponseItem(generated)); items.extend([ RolloutItem::ResponseItem(summary), @@ -188,15 +223,14 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":" .await?; store.flush_thread(owner).await?; let captured = cursor.capture(&store).await?; - let originals = captured - .records() - .iter() - .map(|record| &record.original) - .collect::>(); - assert_eq!( - serde_json::to_value(originals)?, - serde_json::to_value(&expected)? - ); + // Check the serialized board post, independently of the rollout serializer. + let board_originals = |records: &[OriginalEvidence]| -> serde_json::Result> { + records + .iter() + .map(|record| Ok(serde_json::to_value(record)?["original"].take())) + .collect() + }; + assert_eq!(board_originals(captured.records())?, expected); let mut later = user.clone(); later.item = serde_json::from_value(json!({"type":"message","id":"user-2","role":"user", @@ -215,28 +249,17 @@ Approved action: {"action":{"command":"write /workspace/report.txt"},"outcome":" // Reads do not acknowledge delivery, and a later append cannot enter the old capture. assert_eq!(cursor.pending(&captured)?.len(), expected.len()); cursor.acknowledge(&captured, /*count*/ 2)?; - assert_eq!( - serde_json::to_value( - cursor - .pending(&captured)? - .iter() - .map(|r| &r.original) - .collect::>() - )?, - serde_json::to_value(&expected[2..])? - ); + assert_eq!(board_originals(cursor.pending(&captured)?)?, expected[2..]); cursor.acknowledge(&captured, expected.len() - 2)?; assert!(cursor.pending(&captured)?.is_empty()); let latest = cursor.capture(&store).await?; assert_eq!( - serde_json::to_value( - latest - .records() - .iter() - .map(|r| &r.original) - .collect::>() - )?, - serde_json::to_value(vec![RolloutItem::ResponseItem(later.clone())])? + board_originals(latest.records())?, + [json!({ + "type":"response_item", + "payload":later.item, + "metadata":{"client_authored":false,"user_input_order":4} + })] ); assert_eq!(captured.records().len(), expected.len()); cursor.acknowledge(&latest, /*count*/ 1)?; diff --git a/codex-rs/ext/guardian-v3/tests/retrieval.rs b/codex-rs/ext/guardian-v3/tests/retrieval.rs new file mode 100644 index 0000000000..ac86377592 --- /dev/null +++ b/codex-rs/ext/guardian-v3/tests/retrieval.rs @@ -0,0 +1,168 @@ +//! Reviewer tools search routed posts from their capture, including unsubscribed channels. + +#![allow(clippy::expect_used)] + +use codex_guardian_v3::RetrievalBinding; +use codex_guardian_v3::local_board; +use codex_message_board::AppendRequest; +use codex_message_board::PageRequest; +use codex_message_board::PostDestination; +use codex_protocol::AgentPath; +use codex_protocol::ThreadId; +use codex_protocol::openai_models::TruncationPolicyConfig; +use codex_state::SqliteConfig; +use codex_tools::ToolCall; +use codex_tools::ToolCallSource; +use codex_tools::ToolName; +use codex_tools::ToolPayload; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicI64; +use std::sync::atomic::AtomicUsize; + +#[path = "../../../message-board/tests/common/mod.rs"] +mod board_host; + +#[tokio::test] +async fn searches_unsubscribed_guardian_posts_only_through_the_review_cutoff() -> anyhow::Result<()> +{ + let owner = ThreadId::new(); + let dir = tempfile::tempdir()?; + let host = Arc::new(board_host::Host { + clock: AtomicI64::default(), + fail_clock: AtomicBool::default(), + agent_path_calls: AtomicUsize::default(), + members: [(owner, AgentPath::root())].into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::default(), + notifications: Mutex::default(), + }); + let board = local_board( + SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into()?), + owner.into(), + ) + .read_write() + .open(host) + .await?; + let routed = |ordinal: u64, text: &str| { + json!({ + "id": {"ordinal": ordinal}, + "original": {"type": "response_item", "payload": { + "type": "message", "role": "user", + "content": [{"type": "input_text", "text": text}] + }} + }) + .to_string() + }; + let earlier = routed(1, r#"Do not publish "prod" from C:\work\draft."#); + let first = board + .append( + owner, + AppendRequest { + request_id: "first".into(), + destination: PostDestination::NewChannel("release".into()), + text: earlier.clone(), + }, + ) + .await?; + let call = json!({ + "id": {"ordinal": 2}, + "original": {"type": "response_item", "payload": { + "type": "function_call", + "name": "exec_command", + "call_id": "call-2", + "arguments": r#"{"recipient":"Alice","cmd":"publish \"prod\" from C:\\work\\draft","paths":["C:\\tmp","C:\\work\\plan-\u00c9"]}"# + }} + }) + .to_string(); + let second = board + .append( + owner, + AppendRequest { + request_id: "second".into(), + destination: PostDestination::Channel("release".into()), + text: call.clone(), + }, + ) + .await?; + let binding = RetrievalBinding::new(board.read_view(owner).await?, owner, AgentPath::root())?; + board + .append( + owner, + AppendRequest { + request_id: "later".into(), + destination: PostDestination::Channel("release".into()), + text: routed(3, r#"Publish "prod" from C:\work\draft now."#), + }, + ) + .await?; + assert!( + board + .subscribed_channels(owner, PageRequest::default()) + .await? + .results + .is_empty() + ); + + let tools = binding.tools(); + let tool_name = ToolName::namespaced("guardian_board", "search_posts"); + let search = tools + .iter() + .find(|tool| tool.tool_name() == tool_name) + .expect("board search"); + let both = [(&second, &call), (&first, &earlier)]; + for (query, expected) in [ + (r#"publish "prod""#, both.as_slice()), + (r"C:\work\draft", both.as_slice()), + (r"C:\work\plan-é", &both[..1]), + ("plan-é", &both[..1]), + (r#""recipient":"Alice""#, &both[..1]), + (r#""type":"function_call""#, &both[..1]), + ("\t", &both[..0]), + ] { + let payload = ToolPayload::Function { + arguments: json!({"query":query}).to_string(), + }; + let result = search + .handle(ToolCall { + turn_id: "review".into(), + call_id: "search".into(), + tool_name: tool_name.clone(), + model: "test".into(), + codex_turn_metadata: None, + truncation_policy: TruncationPolicyConfig::bytes(/*limit*/ 8_000).into(), + source: ToolCallSource::Direct, + conversation_history: Default::default(), + turn_item_emitter: Arc::new(codex_tools::NoopTurnItemEmitter), + attachment_store: Arc::new(codex_attachment_store::InlineAttachmentStore), + ephemeral: false, + environments: Vec::new(), + payload: payload.clone(), + }) + .await?; + let output = result.code_mode_result(&payload); + assert_eq!(result.log_output(), output.to_string()); + let page: codex_message_board::Page = + serde_json::from_value(output)?; + assert_eq!( + page, + codex_message_board::Page { + results: expected + .iter() + .copied() + .map(|(metadata, text)| codex_message_board::PostPreview { + metadata: metadata.clone(), + text_preview: text.clone(), + n_chars: text.chars().count(), + truncated: false, + }) + .collect(), + next_cursor: None, + } + ); + } + Ok(()) +} diff --git a/codex-rs/message-board/src/local.rs b/codex-rs/message-board/src/local.rs index ea652a23db..1ec1ae7186 100644 --- a/codex-rs/message-board/src/local.rs +++ b/codex-rs/message-board/src/local.rs @@ -45,6 +45,7 @@ use uuid::Uuid; pub(crate) mod catalog; pub(crate) mod connection; +mod json_search; mod lifecycle; pub(crate) mod options; mod paging; @@ -91,6 +92,7 @@ pub struct LocalAgentMessageBoard { pool: Arc, host: Arc, mode: BoardMode, + nested_json_pointer: Option<&'static str>, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -114,9 +116,17 @@ impl LocalAgentMessageBoard { pool: pool::open(sqlite, options).await?, host, mode: BoardMode::Live { access: None }, + nested_json_pointer: None, }) } + /// Whether this already-open handle uses the requested storage namespace. + pub fn uses_storage_namespace(&self, options: &LocalBoardOptions) -> Result { + let filename = options.database_file()?; + Ok(self.pool.connect_options().get_filename().file_name() + == Some(std::ffi::OsStr::new(&filename))) + } + /// Captures a read-only catalog and post boundary without retaining a SQLite transaction. /// Later appends and metadata edits are excluded; deletion may make captured posts unavailable. pub async fn read_view(&self, caller: ThreadId) -> Result { diff --git a/codex-rs/message-board/src/local/catalog.rs b/codex-rs/message-board/src/local/catalog.rs index e953b7595e..863cbd6142 100644 --- a/codex-rs/message-board/src/local/catalog.rs +++ b/codex-rs/message-board/src/local/catalog.rs @@ -5,6 +5,7 @@ use super::BoardMode; use super::LocalAgentMessageBoard; use super::connection::Access; use super::invalid; +use super::options::LocalBoardOptions; use super::paging::Window; use super::storage_error; use crate::AgentMessageBoard; @@ -29,14 +30,27 @@ use uuid::Uuid; /// transaction open. Clones share the catalog, cutoff and underlying connection pool. #[derive(Clone)] pub struct LocalBoardReadView { - board: Arc, + board: Arc, view: Arc, } impl LocalBoardReadView { /// Reads use the captured boundary; every mutation, including subscriptions, is denied. pub fn board(&self) -> Arc { - Arc::clone(&self.board) + self.board.clone() + } + + /// Reads this capture with literal search over JSON posts and a string-encoded JSON field. + /// The JSON Pointer identifies the field to decode. + pub fn board_with_json_search(&self, pointer: &'static str) -> Arc { + let mut board = self.board.as_ref().clone(); + board.nested_json_pointer = Some(pointer); + Arc::new(board) + } + + /// Whether the captured database belongs to the caller's requested storage namespace. + pub fn uses_storage_namespace(&self, options: &LocalBoardOptions) -> Result { + self.board.uses_storage_namespace(options) } /// Complete catalog captured with the post boundary, ordered by newest activity first. diff --git a/codex-rs/message-board/src/local/json_search.rs b/codex-rs/message-board/src/local/json_search.rs new file mode 100644 index 0000000000..3cef557565 --- /dev/null +++ b/codex-rs/message-board/src/local/json_search.rs @@ -0,0 +1,81 @@ +//! Literal search over decoded JSON fields in a fixed read view. + +use super::StoredPost; +use super::paging::Window; +use super::storage_error; +use caseless::default_case_fold_str; +use codex_protocol::error::Result; +use futures::TryStreamExt; +use serde_json::Value; +use sqlx::QueryBuilder; +use sqlx::Sqlite; +use sqlx::SqliteConnection; + +pub(super) async fn page( + mut sql: QueryBuilder, + conn: &mut SqliteConnection, + pointer: &str, + needle: &str, + window: &Window, +) -> Result> { + // Plain text matches the folded column directly. Also consider rows in which + // JSON escaping could hide a match, including Unicode escapes for ASCII text. + let escape = if needle + .bytes() + .any(|b| b < 0x20 || matches!(b, b'"' | b'/' | b'\\')) + { + r"\" + } else { + r"\u" + }; + sql.push(" AND (instr(body_search,") + .push_bind(needle) + .push(")>0 OR instr(body_search,") + .push_bind(escape) + .push(")>0) ORDER BY timestamp DESC,seq DESC"); + let mut rows = sql.build_query_scalar::().fetch(conn); + let mut skip = window.offset(); + let mut posts = Vec::new(); + while let Some(row) = rows.try_next().await.map_err(storage_error)? { + let post: StoredPost = serde_json::from_str(&row).map_err(storage_error)?; + if !matches_post(&post.text, pointer, needle) { + continue; + } + if skip > 0 { + skip -= 1; + continue; + } + posts.push(post); + if posts.len() > window.limit { + break; + } + } + Ok(posts) +} + +fn matches_post(text: &str, pointer: &str, needle: &str) -> bool { + if default_case_fold_str(text).contains(needle) { + return true; + } + let Ok(post) = serde_json::from_str::(text) else { + return false; + }; + // The arguments string is searchable both as posted JSON and as decoded values. + contains_literal(&post, needle) + || post + .pointer(pointer) + .and_then(Value::as_str) + .and_then(|raw| serde_json::from_str(raw).ok()) + .is_some_and(|arguments| contains_literal(&arguments, needle)) +} + +fn contains_literal(value: &Value, needle: &str) -> bool { + match value { + Value::String(text) => default_case_fold_str(text).contains(needle), + Value::Array(items) => items.iter().any(|item| contains_literal(item, needle)), + Value::Object(fields) => fields.iter().any(|(key, item)| { + default_case_fold_str(key).contains(needle) || contains_literal(item, needle) + }), + Value::Number(_) | Value::Bool(_) | Value::Null => value.to_string().contains(needle), + } +} diff --git a/codex-rs/message-board/src/local/queries.rs b/codex-rs/message-board/src/local/queries.rs index cae86e6e5d..a13250cf1c 100644 --- a/codex-rs/message-board/src/local/queries.rs +++ b/codex-rs/message-board/src/local/queries.rs @@ -6,6 +6,7 @@ use super::StoredPost; use super::catalog; use super::connection::Access; use super::invalid; +use super::json_search; use super::paging::Window; use super::paging::decode_posts; use super::paging::direction; @@ -428,11 +429,6 @@ impl AgentMessageBoard for LocalAgentMessageBoard { if let Some(author) = query.author { sql.push(" AND author=").push_bind(author.to_string()); } - if let Some(text) = query.query { - sql.push(" AND instr(body_search,") - .push_bind(default_case_fold_str(&text)) - .push(")>0"); - } if let Some((timestamp, seq)) = after { sql.push(" AND (timestamp,seq)>(") .push_bind(timestamp) @@ -440,16 +436,29 @@ impl AgentMessageBoard for LocalAgentMessageBoard { .push_bind(seq) .push(")"); } - sql.push(" ORDER BY timestamp DESC,seq DESC LIMIT ") - .push_bind((window.limit + 1) as i64) - .push(" OFFSET ") - .push_bind(window.offset()); - let posts = decode_posts( - sql.build_query_scalar::() - .fetch_all(&mut *tx) - .await - .map_err(storage_error)?, - )?; + let needle = query.query.as_deref().map(default_case_fold_str); + let posts = if let (Some(pointer), Some(needle)) = ( + self.nested_json_pointer, + needle.as_deref().filter(|text| !text.is_empty()), + ) { + json_search::page(sql, &mut tx, pointer, needle, &window).await? + } else { + if let Some(needle) = needle { + sql.push(" AND instr(body_search,") + .push_bind(needle) + .push(")>0"); + } + sql.push(" ORDER BY timestamp DESC,seq DESC LIMIT ") + .push_bind((window.limit + 1) as i64) + .push(" OFFSET ") + .push_bind(window.offset()); + decode_posts( + sql.build_query_scalar::() + .fetch_all(&mut *tx) + .await + .map_err(storage_error)?, + )? + }; let chars = (query.max_chars_per_post.get() as usize) .min(MAX_READ_CHARS / posts.len().min(window.limit).max(1)); window.finish(posts.into_iter().map(|post| preview(post, chars)).collect())