starweaver_session/
search.rs

1//! Product-neutral session discovery contracts.
2
3use std::collections::BTreeSet;
4
5use async_trait::async_trait;
6use chrono::{DateTime, Utc};
7use serde::{Deserialize, Serialize};
8use sha2::{Digest, Sha256};
9use starweaver_core::{AgentId, RunId, SessionId};
10use starweaver_stream::{ReplayCursor, ReplayScope};
11use thiserror::Error;
12
13use crate::{RunStatus, SessionStatus};
14
15/// Read-only provider for session discovery projections.
16#[async_trait]
17pub trait SessionSearchProvider: Send + Sync {
18    /// Advertise behavior supported by this provider instance.
19    fn capabilities(&self) -> SessionSearchCapabilities;
20
21    /// Search within a host-constructed authorization scope.
22    async fn search(
23        &self,
24        scope: &SessionSearchScope,
25        query: SessionSearchQuery,
26    ) -> Result<SessionSearchPage, SessionSearchError>;
27}
28
29/// Host-constructed authorization namespace. It is deliberately separate from user input.
30#[derive(Clone, Debug, Eq, PartialEq)]
31pub struct SessionSearchScope {
32    namespace: String,
33    policy_fingerprint: String,
34}
35
36impl SessionSearchScope {
37    /// Construct a scope from a host-owned namespace and policy revision.
38    #[must_use]
39    pub fn new(namespace: impl Into<String>, policy_fingerprint: impl Into<String>) -> Self {
40        Self {
41            namespace: namespace.into(),
42            policy_fingerprint: policy_fingerprint.into(),
43        }
44    }
45
46    /// Construct the conventional single-user local-store scope.
47    #[must_use]
48    pub fn local(namespace: impl Into<String>) -> Self {
49        Self::new(namespace, "local-user-visible-v1")
50    }
51
52    /// Return a non-reversible digest suitable for cursor and mutation bindings.
53    #[must_use]
54    pub fn fingerprint(&self) -> String {
55        digest_parts(&[&self.namespace, &self.policy_fingerprint])
56    }
57}
58
59/// Text interpretation mode.
60#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
61#[serde(rename_all = "snake_case")]
62pub enum SessionSearchQueryMode {
63    /// Case-insensitive literal lexical matching.
64    #[default]
65    Literal,
66    /// Tokenizer-defined phrase matching.
67    Phrase,
68    /// Prefix matching.
69    Prefix,
70    /// Semantic similarity matching.
71    Semantic,
72}
73
74/// Searchable projection family.
75#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
76#[serde(rename_all = "snake_case")]
77pub enum SessionSearchSource {
78    /// Session title and approved typed metadata.
79    SessionMetadata,
80    /// Canonical textual input parts.
81    RunInput,
82    /// Bounded persisted run output preview.
83    RunOutputPreview,
84    /// Approved user-visible display text.
85    DisplayMessage,
86}
87
88impl SessionSearchSource {
89    /// Return all baseline source families.
90    #[must_use]
91    pub fn baseline() -> BTreeSet<Self> {
92        BTreeSet::from([
93            Self::SessionMetadata,
94            Self::RunInput,
95            Self::RunOutputPreview,
96            Self::DisplayMessage,
97        ])
98    }
99}
100
101/// Result grouping level.
102#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
103#[serde(rename_all = "snake_case")]
104pub enum SessionSearchGranularity {
105    /// Return one best hit per session.
106    #[default]
107    Session,
108    /// Return one best hit per session/run pair.
109    Run,
110    /// Return every stable projected occurrence.
111    Occurrence,
112}
113
114/// Stable ordering requested by the caller.
115#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
116#[serde(rename_all = "snake_case")]
117pub enum SessionSearchSort {
118    /// Provider selects relevance for text and updated time for browsing.
119    #[default]
120    Auto,
121    /// Relevance descending with stable updated/identity tie breakers.
122    Relevance,
123    /// Session update time descending, then session id descending.
124    UpdatedDesc,
125}
126
127/// Display visibility classes that a host may permit.
128#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
129#[serde(rename_all = "snake_case")]
130pub enum SessionSearchVisibility {
131    /// Ordinary user-visible display text.
132    Public,
133    /// Diagnostic text, only when a host explicitly permits it.
134    Diagnostic,
135    /// Internal text, only when a host explicitly permits it.
136    Internal,
137}
138
139/// Inclusive lower / exclusive upper timestamp range.
140#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
141#[serde(rename_all = "camelCase")]
142pub struct SessionSearchTimeRange {
143    /// Inclusive lower bound.
144    #[serde(default, skip_serializing_if = "Option::is_none")]
145    pub from: Option<DateTime<Utc>>,
146    /// Exclusive upper bound.
147    #[serde(default, skip_serializing_if = "Option::is_none")]
148    pub until: Option<DateTime<Utc>>,
149}
150
151/// Typed filters for search and metadata browsing.
152#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
153#[serde(rename_all = "camelCase")]
154pub struct SessionSearchFilter {
155    /// Allowed session statuses.
156    #[serde(default, skip_serializing_if = "Vec::is_empty")]
157    pub session_statuses: Vec<SessionStatus>,
158    /// Allowed run statuses.
159    #[serde(default, skip_serializing_if = "Vec::is_empty")]
160    pub run_statuses: Vec<RunStatus>,
161    /// Exact profile name.
162    #[serde(default, skip_serializing_if = "Option::is_none")]
163    pub profile: Option<String>,
164    /// Exact workspace display value.
165    #[serde(default, skip_serializing_if = "Option::is_none")]
166    pub workspace: Option<String>,
167    /// Session creation range.
168    #[serde(default, skip_serializing_if = "Option::is_none")]
169    pub created: Option<SessionSearchTimeRange>,
170    /// Session update range.
171    #[serde(default, skip_serializing_if = "Option::is_none")]
172    pub updated: Option<SessionSearchTimeRange>,
173    /// Explicit session id allowlist.
174    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
175    pub session_ids: BTreeSet<SessionId>,
176    /// Host-approved display visibility classes.
177    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
178    pub display_visibilities: BTreeSet<SessionSearchVisibility>,
179}
180
181/// A bounded search request. `text` is never a regular expression or shell expression.
182#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
183#[serde(rename_all = "camelCase")]
184pub struct SessionSearchQuery {
185    /// Optional literal text. `None` performs metadata browsing.
186    #[serde(default, skip_serializing_if = "Option::is_none", alias = "query")]
187    pub text: Option<String>,
188    /// Text interpretation mode.
189    #[serde(default)]
190    pub mode: SessionSearchQueryMode,
191    /// Typed filters.
192    #[serde(default, alias = "filters")]
193    pub filter: SessionSearchFilter,
194    /// Projection families. Empty means provider defaults.
195    #[serde(default, skip_serializing_if = "BTreeSet::is_empty")]
196    pub sources: BTreeSet<SessionSearchSource>,
197    /// Result grouping level.
198    #[serde(default)]
199    pub granularity: SessionSearchGranularity,
200    /// Stable ordering.
201    #[serde(default)]
202    pub sort: SessionSearchSort,
203    /// Maximum hits requested.
204    pub limit: u32,
205    /// Opaque provider cursor.
206    #[serde(default, skip_serializing_if = "Option::is_none", alias = "after")]
207    pub cursor: Option<String>,
208}
209
210impl Default for SessionSearchQuery {
211    fn default() -> Self {
212        Self {
213            text: None,
214            mode: SessionSearchQueryMode::Literal,
215            filter: SessionSearchFilter::default(),
216            sources: SessionSearchSource::baseline(),
217            granularity: SessionSearchGranularity::Session,
218            sort: SessionSearchSort::Auto,
219            limit: 20,
220            cursor: None,
221        }
222    }
223}
224
225impl SessionSearchQuery {
226    /// Return a stable fingerprint that intentionally excludes the pagination cursor.
227    ///
228    /// # Errors
229    /// Returns an invalid-query error if the typed query cannot be serialized.
230    pub fn fingerprint(&self) -> Result<String, SessionSearchError> {
231        let mut normalized = self.clone();
232        normalized.cursor = None;
233        if let Some(text) = normalized.text.as_mut() {
234            *text = text.trim().to_lowercase();
235        }
236        let bytes = serde_json::to_vec(&normalized)
237            .map_err(|error| SessionSearchError::InvalidQuery(error.to_string()))?;
238        Ok(hex_digest(&bytes))
239    }
240}
241
242/// Filter vocabulary advertised by a provider.
243#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
244#[serde(rename_all = "snake_case")]
245pub enum SessionSearchFilterKind {
246    /// Session status.
247    SessionStatus,
248    /// Run status.
249    RunStatus,
250    /// Profile.
251    Profile,
252    /// Workspace.
253    Workspace,
254    /// Created timestamp range.
255    CreatedTime,
256    /// Updated timestamp range.
257    UpdatedTime,
258    /// Explicit session ids.
259    SessionIds,
260    /// Display visibility policy.
261    DisplayVisibility,
262}
263
264/// Search consistency behavior.
265#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
266#[serde(rename_all = "snake_case")]
267pub enum SessionSearchConsistency {
268    /// Reads canonical state directly.
269    ReadThrough,
270    /// A materialized index is updated in the canonical transaction.
271    TransactionalIndex,
272    /// An asynchronous index may lag canonical state.
273    EventualIndex,
274}
275
276/// Provider capability advertisement.
277#[allow(clippy::struct_excessive_bools)]
278#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
279#[serde(rename_all = "camelCase")]
280pub struct SessionSearchCapabilities {
281    /// Stable provider kind, not an endpoint or topology description.
282    pub provider: String,
283    /// Supported text modes.
284    pub query_modes: BTreeSet<SessionSearchQueryMode>,
285    /// Searchable projection families.
286    pub sources: BTreeSet<SessionSearchSource>,
287    /// Supported typed filters.
288    pub filters: BTreeSet<SessionSearchFilterKind>,
289    /// Supported grouping levels.
290    pub granularities: BTreeSet<SessionSearchGranularity>,
291    /// Supported orderings.
292    pub sorts: BTreeSet<SessionSearchSort>,
293    /// Whether occurrence provenance is returned.
294    pub occurrence_locations: bool,
295    /// Whether snippets are returned.
296    pub snippets: bool,
297    /// Whether relevance scores are returned.
298    pub scores: bool,
299    /// Whether freshness watermarks are returned.
300    pub freshness_watermarks: bool,
301    /// Largest accepted page size.
302    pub max_page_size: u32,
303    /// Consistency behavior.
304    pub consistency: SessionSearchConsistency,
305}
306
307/// Completeness state for one page.
308#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
309#[serde(rename_all = "snake_case")]
310pub enum SessionSearchCoverageState {
311    /// Every requested source in the provider corpus was searched.
312    Complete,
313    /// Search succeeded against an index that may lag canonical writes.
314    EventuallyConsistent,
315    /// One or more requested sources were unavailable or bounded out.
316    Partial,
317    /// A declared fallback produced the page after primary failure.
318    Degraded,
319}
320
321/// Safe warning category.
322#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
323#[serde(rename_all = "snake_case")]
324pub enum SessionSearchWarningKind {
325    /// A compatibility display mirror was absent.
326    MissingSource,
327    /// A source could not be parsed.
328    MalformedSource,
329    /// A configured resource bound was reached.
330    LimitReached,
331    /// A source was unreadable or outside policy.
332    UnavailableSource,
333    /// A best-effort source has no authoritative freshness proof.
334    UnverifiedSource,
335    /// Results came from an explicit fallback.
336    Fallback,
337}
338
339/// Safe coverage warning. Messages must not contain backend paths or indexed content.
340#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
341#[serde(rename_all = "camelCase")]
342pub struct SessionSearchWarning {
343    /// Warning category.
344    pub kind: SessionSearchWarningKind,
345    /// Safe bounded explanation.
346    pub message: String,
347}
348
349/// Coverage and freshness for every returned page, including empty pages.
350#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
351#[serde(rename_all = "camelCase")]
352pub struct SessionSearchCoverage {
353    /// Overall state.
354    pub state: SessionSearchCoverageState,
355    /// Sources actually evaluated.
356    pub searched_sources: BTreeSet<SessionSearchSource>,
357    /// Requested sources that could not be evaluated.
358    pub unavailable_sources: BTreeSet<SessionSearchSource>,
359    /// Eventual-index watermark.
360    #[serde(default, skip_serializing_if = "Option::is_none")]
361    pub indexed_through: Option<DateTime<Utc>>,
362    /// Opaque provider generation.
363    #[serde(default, skip_serializing_if = "Option::is_none")]
364    pub generation: Option<String>,
365    /// Safe warnings.
366    #[serde(default, skip_serializing_if = "Vec::is_empty")]
367    pub warnings: Vec<SessionSearchWarning>,
368}
369
370/// Minimal canonical session projection.
371#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
372#[serde(rename_all = "camelCase")]
373pub struct SessionSearchSummary {
374    /// Session id.
375    pub session_id: SessionId,
376    /// User-facing title.
377    #[serde(default, skip_serializing_if = "Option::is_none")]
378    pub title: Option<String>,
379    /// Session status.
380    pub status: SessionStatus,
381    /// Profile.
382    #[serde(default, skip_serializing_if = "Option::is_none")]
383    pub profile: Option<String>,
384    /// Safe workspace display value.
385    #[serde(default, skip_serializing_if = "Option::is_none")]
386    pub workspace: Option<String>,
387    /// Creation time.
388    pub created_at: DateTime<Utc>,
389    /// Last update time.
390    pub updated_at: DateTime<Utc>,
391    /// Optional matching run status.
392    #[serde(default, skip_serializing_if = "Option::is_none")]
393    pub run_status: Option<RunStatus>,
394    /// Optional bounded matching run preview.
395    #[serde(default, skip_serializing_if = "Option::is_none")]
396    pub run_preview: Option<String>,
397}
398
399/// Highlight range using UTF-8 byte offsets into the returned snippet.
400#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
401#[serde(rename_all = "camelCase")]
402pub struct SessionSearchHighlight {
403    /// Inclusive byte offset.
404    pub start: usize,
405    /// Exclusive byte offset.
406    pub end: usize,
407}
408
409/// Bounded plain-text snippet.
410#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
411#[serde(rename_all = "camelCase")]
412pub struct SessionSearchSnippet {
413    /// Plain text only.
414    pub text: String,
415    /// Match ranges into `text`.
416    #[serde(default, skip_serializing_if = "Vec::is_empty")]
417    pub highlights: Vec<SessionSearchHighlight>,
418}
419
420/// Match provenance. Archive scope and source run identity remain distinct.
421#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
422#[serde(rename_all = "camelCase")]
423pub struct SessionSearchLocation {
424    /// Owning session.
425    pub session_id: SessionId,
426    /// Canonical run owning the projected document.
427    #[serde(default, skip_serializing_if = "Option::is_none")]
428    pub run_id: Option<RunId>,
429    /// Projection family.
430    pub source: SessionSearchSource,
431    /// Archive scope for display evidence.
432    #[serde(default, skip_serializing_if = "Option::is_none")]
433    pub archive_scope: Option<ReplayScope>,
434    /// Source agent for a display occurrence.
435    #[serde(default, skip_serializing_if = "Option::is_none")]
436    pub source_agent_id: Option<AgentId>,
437    /// Source run carried by a display message; it can differ from the archive owner.
438    #[serde(default, skip_serializing_if = "Option::is_none")]
439    pub source_run_id: Option<RunId>,
440    /// Display sequence.
441    #[serde(default, skip_serializing_if = "Option::is_none")]
442    pub display_sequence: Option<usize>,
443    /// Family-aware display cursor when known.
444    #[serde(default, skip_serializing_if = "Option::is_none")]
445    pub cursor: Option<ReplayCursor>,
446    /// Opaque stable occurrence identity.
447    pub document_id: String,
448}
449
450/// One discovery hit.
451#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
452#[serde(rename_all = "camelCase")]
453pub struct SessionSearchHit {
454    /// Minimal canonical session projection.
455    pub session: SessionSearchSummary,
456    /// Matching run, always interpreted together with the session id.
457    #[serde(default, skip_serializing_if = "Option::is_none")]
458    pub run_id: Option<RunId>,
459    /// Matching projection family.
460    pub source: SessionSearchSource,
461    /// Stable match provenance.
462    pub location: SessionSearchLocation,
463    /// Optional bounded plain-text snippet.
464    #[serde(default, skip_serializing_if = "Option::is_none")]
465    pub snippet: Option<SessionSearchSnippet>,
466    /// Provider-local relevance score.
467    #[serde(default, skip_serializing_if = "Option::is_none")]
468    pub score: Option<f64>,
469    /// Source timestamp when meaningful.
470    #[serde(default, skip_serializing_if = "Option::is_none")]
471    pub matched_at: Option<DateTime<Utc>>,
472}
473
474/// One page of discovery results.
475#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
476#[serde(rename_all = "camelCase")]
477pub struct SessionSearchPage {
478    /// Hits in stable provider order.
479    pub hits: Vec<SessionSearchHit>,
480    /// Opaque cursor for the next page.
481    #[serde(default, skip_serializing_if = "Option::is_none")]
482    pub next_cursor: Option<String>,
483    /// Completeness and freshness.
484    pub coverage: SessionSearchCoverage,
485}
486
487/// Required search error categories.
488#[derive(Clone, Debug, Error, Eq, PartialEq)]
489pub enum SessionSearchError {
490    /// Malformed or incompatible query.
491    #[error("invalid search query: {0}")]
492    InvalidQuery(String),
493    /// Expired or incorrectly bound cursor.
494    #[error("invalid search cursor: {0}")]
495    InvalidCursor(String),
496    /// Provider does not advertise the requested behavior.
497    #[error("unsupported search capability: {0}")]
498    Unsupported(String),
499    /// Installed provider cannot currently serve the request.
500    #[error("session search unavailable: {0}")]
501    Unavailable(String),
502    /// Scope is not permitted.
503    #[error("session search permission denied")]
504    PermissionDenied,
505    /// Safe bounded internal failure.
506    #[error("session search failed: {0}")]
507    Failed(String),
508}
509
510impl SessionSearchError {
511    /// Return the stable wire category.
512    #[must_use]
513    pub const fn category(&self) -> &'static str {
514        match self {
515            Self::InvalidQuery(_) => "invalid_query",
516            Self::InvalidCursor(_) => "invalid_cursor",
517            Self::Unsupported(_) => "unsupported",
518            Self::Unavailable(_) => "unavailable",
519            Self::PermissionDenied => "permission_denied",
520            Self::Failed(_) => "failed",
521        }
522    }
523}
524
525/// Cursor payload shared by conforming providers. Callers treat its encoded form as opaque.
526#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
527#[serde(rename_all = "camelCase")]
528pub struct SessionSearchCursorBinding {
529    /// Cursor format version.
530    pub version: u32,
531    /// Stable provider id.
532    pub provider: String,
533    /// Normalized query fingerprint.
534    pub query_fingerprint: String,
535    /// Authorization-scope fingerprint.
536    pub scope_fingerprint: String,
537    /// Provider corpus/index generation.
538    pub generation: String,
539    /// Next stable ordered offset.
540    pub offset: usize,
541    /// Last stable identity, for diagnostics and additional tie binding.
542    pub last_identity: Option<String>,
543}
544
545/// Keyed cursor encoder/validator. Encoded cursors contain no paths or backend row ids.
546#[derive(Clone, Debug)]
547pub struct SessionSearchCursorCodec {
548    key: Vec<u8>,
549}
550
551impl SessionSearchCursorCodec {
552    /// Build a codec from host/provider secret material.
553    #[must_use]
554    pub fn new(key: impl AsRef<[u8]>) -> Self {
555        Self {
556            key: key.as_ref().to_vec(),
557        }
558    }
559
560    /// Encode and authenticate a cursor binding.
561    ///
562    /// # Errors
563    /// Returns a failed error if serialization unexpectedly fails.
564    pub fn encode(
565        &self,
566        binding: &SessionSearchCursorBinding,
567    ) -> Result<String, SessionSearchError> {
568        let payload = serde_json::to_vec(binding)
569            .map_err(|error| SessionSearchError::Failed(error.to_string()))?;
570        let signature = self.signature(&payload);
571        Ok(format!("ssc1.{}.{}", hex_encode(&payload), signature))
572    }
573
574    /// Decode and authenticate an opaque cursor.
575    ///
576    /// # Errors
577    /// Returns `invalid_cursor` for malformed or unauthenticated input.
578    pub fn decode(&self, cursor: &str) -> Result<SessionSearchCursorBinding, SessionSearchError> {
579        let mut parts = cursor.split('.');
580        if parts.next() != Some("ssc1") {
581            return Err(SessionSearchError::InvalidCursor(
582                "unknown cursor format".to_string(),
583            ));
584        }
585        let payload = parts
586            .next()
587            .and_then(hex_decode)
588            .ok_or_else(|| SessionSearchError::InvalidCursor("malformed cursor".to_string()))?;
589        let signature = parts
590            .next()
591            .filter(|_| parts.next().is_none())
592            .ok_or_else(|| SessionSearchError::InvalidCursor("malformed cursor".to_string()))?;
593        if self.signature(&payload) != signature {
594            return Err(SessionSearchError::InvalidCursor(
595                "cursor authentication failed".to_string(),
596            ));
597        }
598        serde_json::from_slice(&payload)
599            .map_err(|_| SessionSearchError::InvalidCursor("malformed cursor payload".to_string()))
600    }
601
602    fn signature(&self, payload: &[u8]) -> String {
603        let mut digest = Sha256::new();
604        digest.update(b"starweaver.session-search.cursor.v1\0");
605        digest.update(&self.key);
606        digest.update([0]);
607        digest.update(payload);
608        hex_encode(&digest.finalize())
609    }
610}
611
612/// Redacted document accepted by optional index writers.
613#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
614#[serde(rename_all = "camelCase")]
615pub struct SessionSearchDocument {
616    /// Stable opaque document id.
617    pub document_id: String,
618    /// Owning session.
619    pub session_id: SessionId,
620    /// Optional owning run.
621    #[serde(default, skip_serializing_if = "Option::is_none")]
622    pub run_id: Option<RunId>,
623    /// Projection family.
624    pub source: SessionSearchSource,
625    /// Already-redacted text.
626    pub text: String,
627    /// Source content digest.
628    pub content_digest: String,
629    /// Source update time.
630    pub updated_at: DateTime<Utc>,
631}
632
633/// Idempotent index operation.
634#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
635#[serde(rename_all = "snake_case")]
636pub enum SessionSearchMutationOperation {
637    /// Upsert a session projection.
638    UpsertSession,
639    /// Upsert a run projection.
640    UpsertRun,
641    /// Upsert a display occurrence or compact batch.
642    UpsertDisplay,
643    /// Delete a run projection.
644    DeleteRun,
645    /// Tombstone an entire session.
646    TombstoneSession,
647    /// Declare a new generation/rebuild boundary.
648    ResetGeneration,
649}
650
651/// Versioned, idempotent search-index mutation.
652#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
653#[serde(rename_all = "camelCase")]
654pub struct SessionSearchMutation {
655    /// Globally stable event id within the source namespace.
656    pub event_id: String,
657    /// Monotonic source revision. Older revisions must never replace newer ones.
658    pub source_revision: u64,
659    /// Projection/redaction schema revision.
660    pub projection_version: String,
661    /// Operation.
662    pub operation: SessionSearchMutationOperation,
663    /// Host authorization-scope fingerprint, never a caller-selected tenant id.
664    pub scope_fingerprint: String,
665    /// Redacted document for upserts; omitted for tombstones/resets.
666    #[serde(default, skip_serializing_if = "Option::is_none")]
667    pub document: Option<SessionSearchDocument>,
668    /// Session targeted by delete/reset operations.
669    #[serde(default, skip_serializing_if = "Option::is_none")]
670    pub session_id: Option<SessionId>,
671    /// Run targeted by a run delete.
672    #[serde(default, skip_serializing_if = "Option::is_none")]
673    pub run_id: Option<RunId>,
674}
675
676/// Durable writer acknowledgement/watermark.
677#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
678#[serde(rename_all = "camelCase")]
679pub struct SessionSearchCheckpoint {
680    /// Accepted generation.
681    pub generation: String,
682    /// Last durably accepted event id.
683    pub event_id: String,
684    /// Indexed-through watermark when available.
685    #[serde(default, skip_serializing_if = "Option::is_none")]
686    pub indexed_through: Option<DateTime<Utc>>,
687}
688
689/// Optional mutation target used by local or external indexes.
690#[async_trait]
691pub trait SessionSearchIndexWriter: Send + Sync {
692    /// Apply mutations idempotently and durably.
693    async fn apply(
694        &self,
695        mutations: &[SessionSearchMutation],
696    ) -> Result<SessionSearchCheckpoint, SessionSearchIndexError>;
697}
698
699/// Index-publication errors.
700#[derive(Clone, Debug, Error, Eq, PartialEq)]
701pub enum SessionSearchIndexError {
702    /// Mutation is invalid or conflicts with a stable event identity.
703    #[error("invalid session search mutation: {0}")]
704    InvalidMutation(String),
705    /// Index is temporarily unavailable and the durable outbox should retry.
706    #[error("session search index unavailable: {0}")]
707    Unavailable(String),
708    /// Safe bounded failure.
709    #[error("session search index failed: {0}")]
710    Failed(String),
711}
712
713fn digest_parts(parts: &[&str]) -> String {
714    let mut digest = Sha256::new();
715    for part in parts {
716        digest.update(part.as_bytes());
717        digest.update([0]);
718    }
719    hex_encode(&digest.finalize())
720}
721
722fn hex_digest(bytes: &[u8]) -> String {
723    let mut digest = Sha256::new();
724    digest.update(bytes);
725    hex_encode(&digest.finalize())
726}
727
728fn hex_encode(bytes: &[u8]) -> String {
729    const HEX: &[u8; 16] = b"0123456789abcdef";
730    let mut output = String::with_capacity(bytes.len() * 2);
731    for byte in bytes {
732        output.push(HEX[(byte >> 4) as usize] as char);
733        output.push(HEX[(byte & 0x0f) as usize] as char);
734    }
735    output
736}
737
738fn hex_decode(value: &str) -> Option<Vec<u8>> {
739    if !value.len().is_multiple_of(2) {
740        return None;
741    }
742    value
743        .as_bytes()
744        .chunks_exact(2)
745        .map(|pair| {
746            let high = (pair[0] as char).to_digit(16)?;
747            let low = (pair[1] as char).to_digit(16)?;
748            u8::try_from((high << 4) | low).ok()
749        })
750        .collect()
751}
752
753#[cfg(test)]
754mod tests {
755    #![allow(clippy::expect_used)]
756
757    use std::{collections::BTreeMap, sync::Mutex};
758
759    use super::*;
760
761    #[test]
762    fn cursor_codec_authenticates_all_normative_bindings() {
763        let codec = SessionSearchCursorCodec::new("test-secret");
764        let binding = SessionSearchCursorBinding {
765            version: 1,
766            provider: "fake".to_string(),
767            query_fingerprint: "query-a".to_string(),
768            scope_fingerprint: "scope-a".to_string(),
769            generation: "generation-a".to_string(),
770            offset: 20,
771            last_identity: Some("document-a".to_string()),
772        };
773        let encoded = codec.encode(&binding).expect("encode cursor");
774        assert!(!encoded.contains("document-a"));
775        assert_eq!(codec.decode(&encoded).expect("decode cursor"), binding);
776
777        let mut tampered = encoded.into_bytes();
778        let last = tampered.len() - 1;
779        tampered[last] = if tampered[last] == b'a' { b'b' } else { b'a' };
780        assert!(matches!(
781            codec.decode(std::str::from_utf8(&tampered).expect("utf8")),
782            Err(SessionSearchError::InvalidCursor(_))
783        ));
784    }
785
786    #[test]
787    fn query_fingerprint_normalizes_text_and_excludes_cursor() {
788        let mut first = SessionSearchQuery {
789            text: Some("  OAuth Refresh ".to_string()),
790            cursor: Some("cursor-a".to_string()),
791            ..SessionSearchQuery::default()
792        };
793        let second = SessionSearchQuery {
794            text: Some("oauth refresh".to_string()),
795            cursor: Some("cursor-b".to_string()),
796            ..SessionSearchQuery::default()
797        };
798        assert_eq!(
799            first.fingerprint().expect("fingerprint"),
800            second.fingerprint().expect("fingerprint")
801        );
802        first.filter.profile = Some("coding".to_string());
803        assert_ne!(
804            first.fingerprint().expect("fingerprint"),
805            second.fingerprint().expect("fingerprint")
806        );
807    }
808
809    struct ConformanceProvider {
810        response: Result<SessionSearchPage, SessionSearchError>,
811    }
812
813    #[async_trait]
814    impl SessionSearchProvider for ConformanceProvider {
815        fn capabilities(&self) -> SessionSearchCapabilities {
816            SessionSearchCapabilities {
817                provider: "fake".to_string(),
818                query_modes: BTreeSet::from([SessionSearchQueryMode::Literal]),
819                sources: BTreeSet::from([SessionSearchSource::SessionMetadata]),
820                filters: BTreeSet::new(),
821                granularities: BTreeSet::from([SessionSearchGranularity::Session]),
822                sorts: BTreeSet::from([SessionSearchSort::Auto]),
823                occurrence_locations: false,
824                snippets: false,
825                scores: false,
826                freshness_watermarks: false,
827                max_page_size: 20,
828                consistency: SessionSearchConsistency::ReadThrough,
829            }
830        }
831
832        async fn search(
833            &self,
834            _scope: &SessionSearchScope,
835            _query: SessionSearchQuery,
836        ) -> Result<SessionSearchPage, SessionSearchError> {
837            self.response.clone()
838        }
839    }
840
841    fn empty_page(state: SessionSearchCoverageState) -> SessionSearchPage {
842        SessionSearchPage {
843            hits: Vec::new(),
844            next_cursor: None,
845            coverage: SessionSearchCoverage {
846                state,
847                searched_sources: BTreeSet::from([SessionSearchSource::SessionMetadata]),
848                unavailable_sources: BTreeSet::new(),
849                indexed_through: None,
850                generation: Some("fake-generation".to_string()),
851                warnings: Vec::new(),
852            },
853        }
854    }
855
856    #[tokio::test]
857    async fn fake_provider_distinguishes_empty_coverage_and_service_errors() {
858        let scope = SessionSearchScope::local("fake");
859        for state in [
860            SessionSearchCoverageState::Complete,
861            SessionSearchCoverageState::Partial,
862            SessionSearchCoverageState::Degraded,
863            SessionSearchCoverageState::EventuallyConsistent,
864        ] {
865            let provider = ConformanceProvider {
866                response: Ok(empty_page(state)),
867            };
868            let page = provider
869                .search(&scope, SessionSearchQuery::default())
870                .await
871                .expect("fake page");
872            assert!(page.hits.is_empty());
873            assert_eq!(page.coverage.state, state);
874        }
875        for error in [
876            SessionSearchError::Unsupported("mode".to_string()),
877            SessionSearchError::Unavailable("index".to_string()),
878        ] {
879            let provider = ConformanceProvider {
880                response: Err(error.clone()),
881            };
882            assert_eq!(
883                provider.search(&scope, SessionSearchQuery::default()).await,
884                Err(error)
885            );
886        }
887    }
888
889    #[derive(Default)]
890    struct ConformanceWriter {
891        revisions: Mutex<BTreeMap<String, (u64, Option<SessionSearchDocument>)>>,
892    }
893
894    #[async_trait]
895    impl SessionSearchIndexWriter for ConformanceWriter {
896        async fn apply(
897            &self,
898            mutations: &[SessionSearchMutation],
899        ) -> Result<SessionSearchCheckpoint, SessionSearchIndexError> {
900            let mut revisions = self.revisions.lock().expect("lock revisions");
901            for mutation in mutations {
902                let key = mutation
903                    .session_id
904                    .as_ref()
905                    .map_or_else(
906                        || {
907                            mutation
908                                .document
909                                .as_ref()
910                                .map_or("generation", |document| document.session_id.as_str())
911                        },
912                        SessionId::as_str,
913                    )
914                    .to_string();
915                let current = revisions.get(&key).map_or(0, |(revision, _)| *revision);
916                if mutation.source_revision < current {
917                    continue;
918                }
919                let document = match mutation.operation {
920                    SessionSearchMutationOperation::TombstoneSession
921                    | SessionSearchMutationOperation::DeleteRun => None,
922                    _ => mutation.document.clone(),
923                };
924                revisions.insert(key, (mutation.source_revision, document));
925            }
926            Ok(SessionSearchCheckpoint {
927                generation: "fake-generation".to_string(),
928                event_id: mutations
929                    .last()
930                    .map_or("none", |mutation| mutation.event_id.as_str())
931                    .to_string(),
932                indexed_through: Some(Utc::now()),
933            })
934        }
935    }
936
937    #[tokio::test]
938    async fn writer_conformance_prevents_delayed_upsert_resurrection() {
939        let writer = ConformanceWriter::default();
940        let session_id = SessionId::from_string("session_writer");
941        let document = SessionSearchDocument {
942            document_id: "doc".to_string(),
943            session_id: session_id.clone(),
944            run_id: None,
945            source: SessionSearchSource::SessionMetadata,
946            text: "safe projection".to_string(),
947            content_digest: "digest".to_string(),
948            updated_at: Utc::now(),
949        };
950        let upsert = SessionSearchMutation {
951            event_id: "upsert-1".to_string(),
952            source_revision: 1,
953            projection_version: "v1".to_string(),
954            operation: SessionSearchMutationOperation::UpsertSession,
955            scope_fingerprint: "scope".to_string(),
956            document: Some(document.clone()),
957            session_id: Some(session_id.clone()),
958            run_id: None,
959        };
960        let tombstone = SessionSearchMutation {
961            event_id: "delete-2".to_string(),
962            source_revision: 2,
963            operation: SessionSearchMutationOperation::TombstoneSession,
964            document: None,
965            ..upsert.clone()
966        };
967        writer.apply(&[tombstone]).await.expect("tombstone");
968        writer.apply(&[upsert]).await.expect("delayed upsert");
969        let revisions = writer.revisions.lock().expect("lock revisions");
970        assert_eq!(revisions[session_id.as_str()], (2, None));
971    }
972
973    #[test]
974    fn scope_fingerprint_hides_namespace_and_policy() {
975        let scope = SessionSearchScope::new("tenant-secret", "policy-secret");
976        let fingerprint = scope.fingerprint();
977        assert!(!fingerprint.contains("tenant-secret"));
978        assert!(!fingerprint.contains("policy-secret"));
979        assert_ne!(
980            fingerprint,
981            SessionSearchScope::new("other", "policy-secret").fingerprint()
982        );
983    }
984}
starweaver_session/search.rs

starweaver_session/
search.rs
