diff --git a/loomem-core/src/tantivy_index.rs b/loomem-core/src/tantivy_index.rs index 058f17d..43c310d 100644 --- a/loomem-core/src/tantivy_index.rs +++ b/loomem-core/src/tantivy_index.rs @@ -260,7 +260,7 @@ mod lexical_parse_tests { "search_with_stream({q:?}) errored" ); assert!( - idx.search_with_date_range(q, 0, 2_000, 5).is_ok(), + idx.search_with_date_range(q, 0, 2_000, None, 5).is_ok(), "search_with_date_range({q:?}) errored" ); assert!( @@ -739,11 +739,12 @@ impl TantivyIndex { &self, query_text: &str, entity: &str, + stream: Option<&str>, limit: usize, ) -> Result> { debug!( - "Searching with entity filter: query='{}', entity='{}', limit={}", - query_text, entity, limit + "Searching with entity filter: query='{}', entity='{}', stream={:?}, limit={}", + query_text, entity, stream, limit ); let searcher = self.reader.searcher(); @@ -770,77 +771,21 @@ impl TantivyIndex { .parse_query(&sanitize_query(entity)) .context("Failed to parse entity filter")?; - // Combine with boolean query (AND) - let combined_query = tantivy::query::BooleanQuery::new(vec![ - (tantivy::query::Occur::Must, Box::new(content_query)), - (tantivy::query::Occur::Must, Box::new(entity_query)), - ]); + // Combine with boolean query (AND). The stream predicate is pushed at + // the source so the entity lane is stream-scoped exactly like + // `search_with_stream`; `None` leaves it unfiltered (callers that + // already pre-scope, and the backward-compatible default). + let mut clauses: Vec<(Occur, Box)> = + vec![(Occur::Must, content_query), (Occur::Must, entity_query)]; + self.push_stream_clause(&mut clauses, stream); + let combined_query = BooleanQuery::new(clauses); // Execute search let top_docs = searcher .search(&combined_query, &TopDocs::with_limit(limit)) .context("Failed to execute search with entity filter")?; - // Convert results - let mut results = Vec::new(); - for (_score, doc_address) in top_docs { - let retrieved_doc: tantivy::TantivyDocument = searcher - .doc(doc_address) - .context("Failed to retrieve document")?; - - let id = retrieved_doc - .get_first(self.id_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let content = retrieved_doc - .get_first(self.content_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let user_id = retrieved_doc - .get_first(self.user_id_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let app_id = retrieved_doc - .get_first(self.app_id_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let level = retrieved_doc - .get_first(self.level_field) - .and_then(|v| v.as_i64()) - .unwrap_or(0) as i32; - - let timestamp = retrieved_doc - .get_first(self.timestamp_field) - .and_then(|v| v.as_i64()) - .unwrap_or(0); - - let stream = retrieved_doc - .get_first(self.stream_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - results.push(SearchResult { - id, - content, - user_id, - app_id, - level, - timestamp, - stream, - score: _score, - }); - } - - Ok(results) + self.collect_results(&searcher, top_docs) } pub fn merge_segments(&mut self) -> Result<()> { @@ -1029,11 +974,12 @@ impl TantivyIndex { query_text: &str, start_ts: i64, end_ts: i64, + stream: Option<&str>, limit: usize, ) -> Result> { debug!( - "Searching with date range: query='{}', start={}, end={}, limit={}", - query_text, start_ts, end_ts, limit + "Searching with date range: query='{}', start={}, end={}, stream={:?}, limit={}", + query_text, start_ts, end_ts, stream, limit ); let searcher = self.reader.searcher(); @@ -1047,10 +993,13 @@ impl TantivyIndex { std::ops::Bound::Included(tantivy::Term::from_field_i64(self.event_date_field, end_ts)); let range_query = RangeQuery::new(lower, upper); - // If query text is empty, only use date range - let combined_query: Box = if query_text.trim().is_empty() { - Box::new(range_query) - } else { + // Assemble MUST clauses. Content is optional (date-only when the query + // text is empty). The stream predicate is pushed at the source so the + // date lane is stream-scoped exactly like `search_with_stream`; `None` + // leaves it unfiltered (backward-compatible for pre-scoped callers). + let empty_query = query_text.trim().is_empty(); + let mut clauses: Vec<(Occur, Box)> = Vec::new(); + if !empty_query { // Parse content query: entities/relations as weak tiebreakers let mut query_parser = QueryParser::for_index( &self.index, @@ -1066,80 +1015,21 @@ impl TantivyIndex { let Some((content_query, _)) = parse_lexical_query(&query_parser, query_text) else { return Ok(Vec::new()); }; - - // Combine with AND - Box::new(BooleanQuery::new(vec![ - (Occur::Must, Box::new(content_query)), - (Occur::Must, Box::new(range_query)), - ])) - }; + clauses.push((Occur::Must, content_query)); + } + clauses.push((Occur::Must, Box::new(range_query))); + self.push_stream_clause(&mut clauses, stream); + let combined_query = BooleanQuery::new(clauses); // Execute search let top_docs = searcher - .search(&*combined_query, &TopDocs::with_limit(limit)) + .search(&combined_query, &TopDocs::with_limit(limit)) .context("Failed to execute search with date range")?; - // Convert results - let mut results = Vec::new(); - for (_score, doc_address) in top_docs { - let retrieved_doc: tantivy::TantivyDocument = searcher - .doc(doc_address) - .context("Failed to retrieve document")?; - - let id = retrieved_doc - .get_first(self.id_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let content = retrieved_doc - .get_first(self.content_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let user_id = retrieved_doc - .get_first(self.user_id_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let app_id = retrieved_doc - .get_first(self.app_id_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - let level = retrieved_doc - .get_first(self.level_field) - .and_then(|v| v.as_i64()) - .unwrap_or(0) as i32; - - let timestamp = retrieved_doc - .get_first(self.timestamp_field) - .and_then(|v| v.as_i64()) - .unwrap_or(0); - - let stream = retrieved_doc - .get_first(self.stream_field) - .and_then(|v| v.as_str()) - .unwrap_or("") - .to_string(); - - results.push(SearchResult { - id, - content, - user_id, - app_id, - level, - timestamp, - stream, - score: _score, - }); - } + let mut results = self.collect_results(&searcher, top_docs)?; // Sort by timestamp descending if no text query - if query_text.trim().is_empty() { + if empty_query { results.sort_by_key(|b| std::cmp::Reverse(b.timestamp)); } @@ -1272,6 +1162,27 @@ impl TantivyIndex { Ok(ids) } + /// Append a MUST `stream` term clause when `stream` is `Some`, mirroring + /// the stream predicate in [`Self::search_with_stream`]. Shared by the date + /// and entity lanes so every BM25 lane scopes at the source. `None` is a + /// no-op — the caller either pre-scoped or wants the global index. + fn push_stream_clause( + &self, + clauses: &mut Vec<(Occur, Box)>, + stream: Option<&str>, + ) { + if let Some(stream) = stream { + let term = tantivy::Term::from_field_text(self.stream_field, stream); + clauses.push(( + Occur::Must, + Box::new(tantivy::query::TermQuery::new( + term, + tantivy::schema::IndexRecordOption::Basic, + )), + )); + } + } + /// Append the `source_agent` include (MUST) and `exclude_source_agents` /// (MUST_NOT) term clauses shared by [`Self::search_with_agent`] and /// [`Self::ids_matching_agent`]. @@ -1588,7 +1499,7 @@ mod agent_search_tests { .unwrap(); idx.commit().unwrap(); - let entity_hits = idx.search_with_entity("zeta", "zeta", 10).unwrap(); + let entity_hits = idx.search_with_entity("zeta", "zeta", None, 10).unwrap(); let p = AgentSearchParams { query_text: "zeta", stream: None, @@ -1613,3 +1524,168 @@ mod agent_search_tests { ); } } + +#[cfg(test)] +mod stream_scoping_tests { + //! BM25 stream-scoping parity (cycle: bm25-stream-scoping-parity). The date + //! and entity lanes push the `stream` predicate at index-query time, exactly + //! like `search_with_stream`, so a stream-scoped query never surfaces a + //! twin chunk that lives in another stream. `stream = None` keeps the + //! pre-existing unscoped behaviour (backward compatibility for callers that + //! already pre-scope). Handler-level e2e coverage lives in + //! `loomem-server/src/handlers/search.rs`. + use super::*; + use crate::config::TantivyConfig; + use tempfile::TempDir; + + fn cfg() -> TantivyConfig { + TantivyConfig { + enabled: true, + heap_size_mb: 16, + drift_warn_pct: 5.0, + auto_rebuild_on_drift: false, + } + } + + /// Doc with an explicit stream, entities and event_date so both the date and + /// entity lanes can be exercised with real filters. + fn doc( + id: &str, + stream: &str, + level: i32, + content: &str, + entities: &str, + event_date: i64, + ) -> TextDocument { + TextDocument { + id: id.to_string(), + content: content.to_string(), + user_id: "default".to_string(), + app_id: "default".to_string(), + level, + timestamp: 1_000, + stream: stream.to_string(), + entities: Some(entities.to_string()), + relations: None, + event_date: Some(event_date), + source_agent: None, + } + } + + /// Two content/entity/event_date twins in different streams, so only the + /// stream predicate can tell them apart. + fn seeded() -> (TempDir, TantivyIndex) { + let tmp = TempDir::new().expect("tempdir"); + let mut idx = TantivyIndex::open(tmp.path().join("tantivy"), &cfg()).expect("open"); + idx.index_document(doc( + "a1", + "streamA", + 0, + "budget review notes", + "Anna", + 1_000, + )) + .unwrap(); + idx.index_document(doc( + "b1", + "streamB", + 0, + "budget review notes", + "Anna", + 1_000, + )) + .unwrap(); + idx.commit().unwrap(); + (tmp, idx) + } + + fn ids(results: &[SearchResult]) -> std::collections::HashSet { + results.iter().map(|r| r.id.clone()).collect() + } + + fn one(id: &str) -> std::collections::HashSet { + [id.to_string()].into_iter().collect() + } + + #[test] + fn date_lane_scopes_to_stream() { + let (_tmp, idx) = seeded(); + let got = idx + .search_with_date_range("budget", 0, 2_000, Some("streamA"), 10) + .unwrap(); + assert_eq!( + ids(&got), + one("a1"), + "date lane scoped to streamA must not return the streamB twin" + ); + } + + #[test] + fn entity_lane_scopes_to_stream() { + let (_tmp, idx) = seeded(); + let got = idx + .search_with_entity("budget", "Anna", Some("streamA"), 10) + .unwrap(); + assert_eq!( + ids(&got), + one("a1"), + "entity lane scoped to streamA must not return the streamB twin" + ); + } + + #[test] + fn none_stream_returns_union_both_lanes() { + // Backward compatibility: `None` is unfiltered, so both twins return. + let (_tmp, idx) = seeded(); + let both: std::collections::HashSet = + ["a1", "b1"].iter().map(|s| s.to_string()).collect(); + let date = idx + .search_with_date_range("budget", 0, 2_000, None, 10) + .unwrap(); + assert_eq!(ids(&date), both, "unscoped date lane returns the union"); + let entity = idx.search_with_entity("budget", "Anna", None, 10).unwrap(); + assert_eq!(ids(&entity), both, "unscoped entity lane returns the union"); + } + + #[test] + fn date_lane_empty_query_scopes_to_stream() { + // Date-only path (no content clause) still scopes at the source. + let (_tmp, idx) = seeded(); + let got = idx + .search_with_date_range("", 0, 2_000, Some("streamB"), 10) + .unwrap(); + assert_eq!(ids(&got), one("b1")); + } + + #[test] + fn date_lane_excludes_high_score_l1_twin_in_other_stream() { + // Invariant vs ranking: an L1-consolidated streamB chunk whose content + // matches the query far more strongly cannot enter a streamA-scoped date + // query's pool at all — scoping at the source pre-empts any later + // ranking-stage reorder (e.g. the handler's L1 x1.5 boost). Fails if the + // date lane queries the global index and relies on a downstream gate. + let tmp = TempDir::new().expect("tempdir"); + let mut idx = TantivyIndex::open(tmp.path().join("tantivy"), &cfg()).expect("open"); + idx.index_document(doc("a1", "streamA", 0, "budget", "Anna", 1_000)) + .unwrap(); + idx.index_document(doc( + "b_l1", + "streamB", + 1, + "budget budget budget review", + "Anna", + 1_000, + )) + .unwrap(); + idx.commit().unwrap(); + + let got = idx + .search_with_date_range("budget", 0, 2_000, Some("streamA"), 10) + .unwrap(); + assert_eq!( + ids(&got), + one("a1"), + "the high-BM25 L1 streamB twin must never surface in a streamA date query" + ); + } +} diff --git a/loomem-core/tests/integration_test.rs b/loomem-core/tests/integration_test.rs index 7c32799..dca94d4 100644 --- a/loomem-core/tests/integration_test.rs +++ b/loomem-core/tests/integration_test.rs @@ -205,7 +205,7 @@ fn test_entity_tagged_search() -> Result<()> { tantivy.commit()?; // Entity search — should prefer the tagged chunk - let results = tantivy.search_with_entity("budget", "Anna", 10)?; + let results = tantivy.search_with_entity("budget", "Anna", None, 10)?; assert!(!results.is_empty(), "Expected results for entity search"); assert_eq!( results[0].id, "id-entity", diff --git a/loomem-server/src/handlers/search.rs b/loomem-server/src/handlers/search.rs index 921e280..e7d6bc6 100644 --- a/loomem-server/src/handlers/search.rs +++ b/loomem-server/src/handlers/search.rs @@ -289,21 +289,21 @@ impl<'a> Bm25Leaf<'a> { limit, } } - /// Entity branch: content + entity filter. - fn entity(query: &'a str, entity: &'a str, limit: usize) -> Self { + /// Entity branch: content + entity filter, stream-scoped at the source. + fn entity(query: &'a str, stream: Option<&'a str>, entity: &'a str, limit: usize) -> Self { Self { query, - stream: None, + stream, entity: Some(entity), date_range: None, limit, } } - /// Date branch: content + `event_date` range filter. - fn date(query: &'a str, range: (i64, i64), limit: usize) -> Self { + /// Date branch: content + `event_date` range filter, stream-scoped at the source. + fn date(query: &'a str, stream: Option<&'a str>, range: (i64, i64), limit: usize) -> Self { Self { query, - stream: None, + stream, entity: None, date_range: Some(range), limit, @@ -335,9 +335,9 @@ fn bm25_leaf( limit: leaf.limit, }) } else if let Some((start_ts, end_ts)) = leaf.date_range { - tantivy.search_with_date_range(leaf.query, start_ts, end_ts, leaf.limit) + tantivy.search_with_date_range(leaf.query, start_ts, end_ts, leaf.stream, leaf.limit) } else if let Some(entity) = leaf.entity { - tantivy.search_with_entity(leaf.query, entity, leaf.limit) + tantivy.search_with_entity(leaf.query, entity, leaf.stream, leaf.limit) } else if let Some(stream) = leaf.stream { tantivy.search_with_stream(leaf.query, stream, leaf.limit) } else { @@ -345,6 +345,52 @@ fn bm25_leaf( } } +/// Run one stream's BM25 leaf via `run_leaf` for every scoped stream and union +/// the hits with max-score-wins merge — the multi-stream union semantics shared +/// by all branches (cycle: BM25 stream-scoping parity). `run_leaf` builds and +/// runs the branch-specific leaf (plain / date / entity) for the given stream, +/// pushing the `stream` predicate at index-query time via `bm25_leaf`; it +/// returns owned results, so the stream scope stays at the source rather than +/// being reconstructed post-hoc: +/// - single stream → one scoped leaf (fast path, no merge); +/// - many streams → per-stream leaves merged by id, highest score wins; +/// - `None` → one unscoped leaf (backward-compatible with pre-scoped callers). +fn bm25_over_streams( + streams: Option<&[String]>, + run_leaf: F, +) -> anyhow::Result> +where + F: Fn(Option<&str>) -> anyhow::Result>, +{ + match streams { + Some(streams) if streams.len() == 1 => run_leaf(Some(&streams[0])), + Some(streams) => { + let mut merged_map: std::collections::HashMap = + std::collections::HashMap::new(); + for s in streams { + for r in run_leaf(Some(s))? { + merged_map + .entry(r.id.clone()) + .and_modify(|e| { + if r.score > e.score { + e.score = r.score; + } + }) + .or_insert(r); + } + } + let mut merged: Vec<_> = merged_map.into_values().collect(); + merged.sort_by(|a, b| { + b.score + .partial_cmp(&a.score) + .unwrap_or(std::cmp::Ordering::Equal) + }); + Ok(merged) + } + None => run_leaf(None), + } +} + /// The agent-matching chunk id set for the vector path, or `None` when no agent /// filter is set (embeddings left untouched). One Tantivy full-collect query /// (`ids_matching_agent`) replaces #257's per-embedding `get_chunk`. Fails open @@ -1392,62 +1438,37 @@ async fn bm25_retrieve( let tantivy = state.tantivy.lock().await; let oq: &str = &ctx.original_query_stemmed; + // Every branch routes through `bm25_over_streams`, so the stream predicate + // is applied at index-query time (per-stream union for multi-stream + // callers) rather than post-hoc — parity with the plain/vector/graph lanes. + let streams = ctx.stream_list.as_deref(); + let bm25_results = if let Some(ref date_filter) = ctx.date_filter { let (start_ts, end_ts) = match date_filter { DateFilter::Range(start, end) => (*start, *end), }; - bm25_leaf( - &tantivy, - payload, - Bm25Leaf::date(oq, (start_ts, end_ts), ctx.limit * 2), - )? + bm25_over_streams(streams, |s| { + bm25_leaf( + &tantivy, + payload, + Bm25Leaf::date(oq, s, (start_ts, end_ts), ctx.limit * 2), + ) + })? } else if let Some(ref entity) = payload.entity { - bm25_leaf( - &tantivy, - payload, - Bm25Leaf::entity(oq, entity, ctx.limit * 2), - )? + bm25_over_streams(streams, |s| { + bm25_leaf( + &tantivy, + payload, + Bm25Leaf::entity(oq, s, entity, ctx.limit * 2), + ) + })? } else { - let stream_filter: Option> = ctx.stream_list.clone(); - let search_query = |query: &str, lim: usize| -> anyhow::Result> { - match &stream_filter { - Some(streams) if streams.len() == 1 => bm25_leaf( - &tantivy, - payload, - Bm25Leaf::plain(query, Some(&streams[0]), lim), - ), - Some(streams) => { - let mut merged_map: std::collections::HashMap< - String, - loomem_core::SearchResult, - > = std::collections::HashMap::new(); - for s in streams { - let results = - bm25_leaf(&tantivy, payload, Bm25Leaf::plain(query, Some(s), lim))?; - for r in results { - merged_map - .entry(r.id.clone()) - .and_modify(|e| { - if r.score > e.score { - e.score = r.score; - } - }) - .or_insert(r); - } - } - let mut merged: Vec<_> = merged_map.into_values().collect(); - merged.sort_by(|a, b| { - b.score - .partial_cmp(&a.score) - .unwrap_or(std::cmp::Ordering::Equal) - }); - Ok(merged) - } - None => bm25_leaf(&tantivy, payload, Bm25Leaf::plain(query, None, lim)), - } + bm25_over_streams(streams, |s| { + bm25_leaf(&tantivy, payload, Bm25Leaf::plain(query, s, lim)) + }) }; let original_results = search_query(&ctx.original_query_stemmed, ctx.limit * 2)?; @@ -2600,6 +2621,20 @@ fn filter_and_truncate( }); } + // Fail-closed stream-scoping guard (defense-in-depth). The retrieval lanes + // now apply the stream predicate at index-query time, but this `retain` + // enforces the invariant one last time before top-K truncation so a future + // lane that forgets the predicate cannot leak cross-stream chunks. Mirrors + // the graph lane's `streams.contains(&chunk.stream)` gate — but fail-CLOSED: + // a result whose chunk is missing, unreadable, or carries no in-scope stream + // attribution is DROPPED, not passed through. `None` scope = unscoped query, + // so the guard is a no-op (backward compatible). + if let Some(ref streams) = ctx.stream_list { + hybrid_results.retain(|r| { + matches!(state.store.get_chunk(&r.id), Ok(Some(chunk)) if streams.contains(&chunk.stream)) + }); + } + let total_results_before_topk = hybrid_results.len(); let final_top_k = if matches!(ctx.complexity, QueryComplexity::Aggregation) { payload.top_k.unwrap_or(30) @@ -4585,3 +4620,219 @@ mod agent_filter_tests { ); } } + +#[cfg(test)] +mod stream_scoping_e2e_tests { + //! BM25 stream-scoping parity — handler-level e2e (cycle: + //! bm25-stream-scoping-parity). Drives `search_handler` end-to-end through + //! the `time_filter` (date) and `entity` retrieval branches with two content + //! twins living in different streams, and asserts a stream-scoped query + //! returns only the in-scope twin. The non-vacuous check re-runs the same + //! query scoped to the OTHER stream to prove the twin is genuinely + //! retrievable — the isolation is not just "returns nothing". + use super::*; + use crate::auth::{AuthContext, KeyScope}; + use loomem_core::storage::{Chunk, UserRole}; + use loomem_core::{SourceTag, TextDocument}; + + /// A chunk in an explicit stream. `1_700_000_000` = 2023-11-14, so a + /// 2023 date window covers it. + fn chunk(id: &str, stream: &str, content: &str) -> Chunk { + Chunk { + id: id.to_string(), + content: content.to_string(), + stream: stream.to_string(), + level: 0, + score: 1.0, + timestamp: 1_700_000_000, + consolidated: false, + dormant: false, + in_progress: false, + prompt_version: None, + source_ids: None, + last_decay: None, + metadata: None, + importance: None, + persistent: false, + last_implicit_boost: None, + access_count: 0, + source: Some(SourceTag::from_agent("tester")), + created_by: None, + updated_at: None, + valid_from: None, + valid_until: None, + is_latest: true, + superseded_by: None, + supersedes_id: None, + root_memory_id: None, + version: 1, + memory_type: None, + extraction_meta: None, + deleted_at: None, + trust_level: None, + ingester_user_id: None, + alpha: 1.0, + beta: 1.0, + harmful_count: 0, + n_ratings: 0, + last_rated_at: None, + provenance_role: loomem_core::storage::ProvenanceRole::Claim, + } + } + + fn index_doc(c: &Chunk, entities: Option<&str>) -> TextDocument { + TextDocument { + id: c.id.clone(), + content: c.content.clone(), + user_id: "default".into(), + app_id: "default".into(), + level: 0, + timestamp: c.timestamp as i64, + stream: c.stream.clone(), + entities: entities.map(|e| e.to_string()), + relations: None, + event_date: Some(c.timestamp as i64), + source_agent: Some("tester".into()), + } + } + + /// Seed one chunk into both RocksDB and Tantivy (with an optional entity tag). + async fn seed_one( + state: &std::sync::Arc, + id: &str, + stream: &str, + content: &str, + entities: Option<&str>, + ) { + let c = chunk(id, stream, content); + state.store.store_chunk(&c).unwrap(); + let mut tv = state.tantivy.lock().await; + tv.upsert_document(index_doc(&c, entities)).unwrap(); + tv.commit().unwrap(); + } + + /// Twins: identical content + entity, one in each stream. + async fn seed_twins(state: &std::sync::Arc) { + seed_one(state, "a1", "streamA", "budget review notes", Some("Anna")).await; + seed_one(state, "b1", "streamB", "budget review notes", Some("Anna")).await; + } + + /// Admin auth can reach any valid stream (single-user model), so the + /// `streams` filter in the request is what actually scopes the query. + fn admin_auth() -> AuthContext { + AuthContext::single_stream( + "streamA", + UserRole::Admin, + KeyScope::Shared, + Some("admin".into()), + true, + ) + } + + fn base_req() -> SearchRequest { + SearchRequest { + query: "budget".into(), + user_id: None, + top_k: Some(10), + stream: None, + streams: None, + entity: None, + date_from: None, + date_to: None, + valid_at: None, + dry_run: false, + filters: None, + include_superseded: false, + trace: false, + fact_type: None, + subject: None, + min_confidence: None, + include_associations: false, + source_agent: None, + exclude_source_agents: None, + scope: None, + debug_query_classification: false, + debug_signal_breakdown: false, + debug_channels: false, + } + } + + async fn run(state: &std::sync::Arc, payload: SearchRequest) -> Vec { + let resp = search_handler( + axum::extract::State(state.clone()), + axum::Extension(admin_auth()), + axum::Json(payload), + ) + .await + .expect("search_handler"); + resp.0.results.into_iter().map(|r| r.id).collect() + } + + #[tokio::test] + async fn time_filter_scopes_to_stream() { + let (_app, state) = crate::tests::make_test_app(); + seed_twins(&state).await; + + let mut payload = base_req(); + payload.streams = Some(vec!["streamA".into()]); + payload.date_from = Some("2023-01-01".into()); + payload.date_to = Some("2023-12-31".into()); + + let ids = run(&state, payload).await; + assert!( + ids.iter().all(|id| id == "a1"), + "time_filter scoped to streamA must not surface the streamB twin, got {ids:?}" + ); + assert!( + ids.contains(&"a1".to_string()), + "streamA twin must be present" + ); + } + + #[tokio::test] + async fn entity_filter_scopes_to_stream() { + let (_app, state) = crate::tests::make_test_app(); + seed_twins(&state).await; + + let mut payload = base_req(); + payload.streams = Some(vec!["streamA".into()]); + payload.entity = Some("Anna".into()); + + let ids = run(&state, payload).await; + assert!( + ids.iter().all(|id| id == "a1"), + "entity filter scoped to streamA must not surface the streamB twin, got {ids:?}" + ); + assert!( + ids.contains(&"a1".to_string()), + "streamA twin must be present" + ); + } + + #[tokio::test] + async fn scoped_twin_is_still_retrievable_in_its_own_stream() { + // Non-vacuous: the streamB twin dropped above IS retrievable when the + // query is scoped to its own stream — isolation, not a dead pipeline. + let (_app, state) = crate::tests::make_test_app(); + seed_twins(&state).await; + + let mut date_req = base_req(); + date_req.streams = Some(vec!["streamB".into()]); + date_req.date_from = Some("2023-01-01".into()); + date_req.date_to = Some("2023-12-31".into()); + let date_ids = run(&state, date_req).await; + assert!( + date_ids.contains(&"b1".to_string()), + "streamB twin must be retrievable via time_filter scoped to streamB, got {date_ids:?}" + ); + + let mut entity_req = base_req(); + entity_req.streams = Some(vec!["streamB".into()]); + entity_req.entity = Some("Anna".into()); + let entity_ids = run(&state, entity_req).await; + assert!( + entity_ids.contains(&"b1".to_string()), + "streamB twin must be retrievable via entity filter scoped to streamB, got {entity_ids:?}" + ); + } +}