openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-07 20:30:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add read-only evidence retrieval tools for Guardian reviewers (#59994)

Browse Source 
## What changed

- Expose `guardian_board` tools to list channels and threads, search posts, and read evidence through a captured read-only view. Require a Guardian storage namespace and keep reads within the review's fixed cutoff, including searches across unsubscribed channels.
- Support literal, case-insensitive searches over decoded JSON fields and nested function-call arguments so quotes, Windows paths, and Unicode escapes do not hide matching evidence.
- Strip response-envelope harness metadata from captured evidence except for `user_input_order`, preserving message payloads and source ordering.

## Testing

Add retrieval coverage for unsubscribed channels, fixed cutoffs, and escaped JSON searches. Update evidence tests to verify serialized payloads retain source order while excluding other harness metadata.

GitOrigin-RevId: 3128e4a2d59d1f71f2cce0bc6b1fbdf7e2aba287
This commit is contained in:
felixxia-oai
2026-11-06 21:54:55 +00:00
committed by copyberry
parent 632eee03e9
commit e4bd7871b1
11 changed files with 492 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -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",
codex-rs/ext/guardian-v3/Cargo.toml
+3 -1
View File
@@ -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 }
codex-rs/ext/guardian-v3/src/evidence.rs
+53 -38
View File
@@ -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<bool> {
| ResponseItem::Other => false,
})
}
/// Apply the same admission and model-visible projection at intake and canonical verification.
fn review_original(mut item: RolloutItem) -> anyhow::Result<Option<RolloutItem>> {
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))
}
codex-rs/ext/guardian-v3/src/lib.rs
+4 -3
View File
@@ -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.
codex-rs/ext/guardian-v3/src/retrieval_tools.rs
+79
View File
@@ -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<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> {
thread
.get::<RetrievalBinding>()
.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<C: Sync>(registry: &mut ExtensionRegistryBuilder<C>) {
registry.tool_contributor(Arc::new(RetrievalTools));
}
impl RetrievalBinding {
pub fn new(
board: LocalBoardReadView,
caller: ThreadId,
caller_path: AgentPath,
) -> anyhow::Result<Self> {
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<Arc<dyn for<'call> ToolExecutor<ToolCall<'call>>>> {
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,
)
}
}
codex-rs/ext/guardian-v3/tests/evidence.rs
+52 -29
View File
@@ -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::<Vec<_>>();
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<Vec<_>> {
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::<Vec<_>>()
)?,
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::<Vec<_>>()
)?,
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)?;
codex-rs/ext/guardian-v3/tests/retrieval.rs
+168
View File
@@ -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<codex_message_board::PostPreview> =
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(())
}
codex-rs/message-board/src/local.rs
+10
View File
@@ -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<SqlitePool>,
host: Arc<dyn MessageBoardHost>,
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<bool> {
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<catalog::LocalBoardReadView> {
codex-rs/message-board/src/local/catalog.rs
+16 -2
View File
@@ -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<dyn AgentMessageBoard>,
board: Arc<LocalAgentMessageBoard>,
view: Arc<ReadView>,
}
impl LocalBoardReadView {
/// Reads use the captured boundary; every mutation, including subscriptions, is denied.
pub fn board(&self) -> Arc<dyn AgentMessageBoard> {
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<dyn AgentMessageBoard> {
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<bool> {
self.board.uses_storage_namespace(options)
}
/// Complete catalog captured with the post boundary, ordered by newest activity first.
codex-rs/message-board/src/local/json_search.rs
+81
View File
@@ -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<Sqlite>,
conn: &mut SqliteConnection,
pointer: &str,
needle: &str,
window: &Window,
) -> Result<Vec<StoredPost>> {
// 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::<String>().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::<Value>(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),
}
}
codex-rs/message-board/src/local/queries.rs
+24 -15
View File
@@ -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::<String>()
.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::<String>()
.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())