polyc-query 2026.8.3

//! The participation-scoped search index
//! (docs/proposals/participation-scoped-agent-search.md).
//!
//! An agent asked to find something in the conversations its caller took part
//! in cannot replay each one per call: a heavy user's participation set is
//! plausibly thousands of small partitions, because the chat edge derives a
//! conversation id per thread rather than per channel. This module is the
//! maintained projection that makes such a search bounded.
//!
//! # Append-only segments per conversation, in the layout `DataFusion` reads
//!
//! `docs/reference/datafusion-data-layer.md` ("Cold storage: Parquet
//! projections") already settled the storage question: `ListingTable` reads
//! Parquet over an object store with zero custom code, including row-group and
//! page-index statistics pruning and Bloom-filter pruning for high-selectivity
//! equality predicates, and the per-conversation layout means a session's
//! catalog registers only the conversations it is already authorized for.
//!
//! So the projection is a directory per conversation holding a row per
//! (committed message, distinct term), written in the layout a `ListingTable`
//! reads: Hive-style `conversation_id=` directories, a Bloom filter and sorted
//! page statistics on `term_hash`. A search will prune on `term_hash` equality
//! natively and open only the files that survive — turning "replay every
//! partition" into "prune most, scan few".
//!
//! Within that directory a publish APPENDS a segment covering only the
//! positions it just indexed, rather than rewriting one file per conversation.
//! A rewrite costs O(N) rows per committed turn and therefore O(N²) over a
//! conversation's life — the same quadratic the Parquet layout was chosen to
//! escape, reintroduced one level down. `ListingTable` reads every segment in
//! the directory as one table, so the read side is unchanged, and
//! [`store::SearchProjection::compact`] folds the segments back to one once
//! either compaction trigger fires (see [`worker`]).
//!
//! That read path is the search port, a later change in this stack. This
//! module writes the layout and reads whole files; see
//! [`store`]'s own doc for exactly what prunes today and what does not.
//!
//! An earlier draft built a bespoke key-value store here, on the stated
//! grounds that "there is no range-queryable durable store in this
//! workspace." That was wrong, and every mechanism it hand-rolled has a
//! native counterpart: a membership filter is a Bloom filter, a postings
//! record is columnar rows, a delta-varint codec is Parquet's own encoding,
//! and candidate ordering is query planning.
//!
//! # Why document frequency is load-bearing here
//!
//! The live lexical core admits an entry at term overlap >= 1 and performs no
//! stopword removal. A natural-language query — "where did we decide the
//! timeout" — therefore contains terms present in essentially every
//! conversation, so an unfiltered membership test passes the ENTIRE scope and
//! the mechanism collapses back into the per-call replay it exists to avoid.
//! The governing variable is term document frequency, not the filter's
//! false-positive rate. [`SearchIndexConfig::max_term_document_frequency`] is
//! what bounds it, and it is runtime configuration rather than a constant
//! because no corpus exists to calibrate it against before launch.
//!
//! # Derived, never authoritative
//!
//! Every file here is reconstructible from the event log alone, and one that
//! disagrees with its journal must be discarded rather than trusted. The
//! footer carries what makes that checkable — the source incarnation, the term
//! key's identity, and the format version. The key and format are verified on
//! every coverage read; the incarnation is verified against the live journal by
//! [`store::SearchProjection::verified_coverage`], which is the read a search
//! must go through and the reason this design needs no startup barrier.
//!
//! # What a Container wires, and what is still missing
//!
//! [`SearchIndex`] is this module's whole outward surface: it opens the
//! projection, hands back the [`marks::CommitMarks`] handle the Container
//! drives from the durable commit feed, and owns the [`worker`] loop the
//! Container supervises. The three go together by construction — a marks
//! handle driven without a running worker fills a dirty set nothing drains,
//! and a worker without a reconcile can never clear the degraded flag that
//! overflow sets (see [`marks::CommitMarks`]'s own doc).
//!
//! The `SearchSnapshot` port that READS what the worker publishes lands in a
//! later change, so the index is maintained but not yet queried.

#![allow(
    dead_code,
    reason = "the store's read side and the retrieval bounds land before the search port that consumes them"
)]

pub(crate) mod marks;
pub(crate) mod project;
pub(crate) mod store;
pub(crate) mod terms;
pub(crate) mod worker;

use std::path::PathBuf;
use std::sync::Arc;

use crate::journal::PartitionJournal;
use tokio_util::sync::CancellationToken;

/// Raised when the search index's projection cannot be opened.
///
/// Carries the store's failure as text rather than the `pub(crate)` store
/// error itself: the on-disk layout is this module's business, and a Container
/// can only ever log this and refuse to start.
#[derive(Debug, thiserror::Error)]
#[error("open the search index projection at {}: {reason}", root.display())]
pub struct SearchIndexError {
    /// Where the Container asked for the projection to live.
    root: PathBuf,
    /// What the store reported.
    reason: String,
}

/// Where a Container's term key came from.
///
/// Not a detail the index can shrug at. Every stored term hash is computed under
/// the key, and the footer records the key's identity — so a key the Container
/// MINTED rather than read back invalidates every segment in the projection at
/// once. Detection already worked: coverage under a foreign key identity reads
/// as unreadable. Repair did not. Nothing sweeps on a key change, so a
/// conversation that never commits another turn keeps its orphaned segments and
/// its participants' search refuses forever, with no degraded flag and therefore
/// no reconcile to notice.
///
/// The mint is not hypothetical. A dev custody directory on a non-persistent
/// path loses its keys on every reschedule, and a Kubernetes Secret that is
/// deleted or missing its data key reads as absent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TermKeyOrigin {
    /// The key was read back from durable custody, so the segments already on
    /// disk were written under it.
    Stored,
    /// The key was newly minted, so nothing on disk was written under it and
    /// the whole projection has to be rebuilt.
    Minted,
}

/// The maintained search index, ready for a Container to register and
/// supervise.
///
/// # Take the marks handle, then supervise the worker
///
/// [`SearchIndex::marks`] borrows and [`SearchIndex::run`] consumes, so the
/// only order the two can be called in is the correct one: start driving the
/// handle from the commit feed first, so no commit lands unmarked, then hand
/// the loop to the task set that notices when it stops. Both sides share one
/// dirty set, so there is no way to fill a different set from the one the
/// worker drains.
///
/// What the types do NOT prevent is taking the handle and dropping the index:
/// driving that handle leaves a set nothing empties, which fills to its bound
/// and then refuses every search. The Container is what pairs them, and this
/// is stated rather than claimed away.
///
/// The worker is a plain future rather than something that spawns itself. A
/// perpetual loop belongs in the Container's supervised set, where a task that
/// dies takes the process down with it — a detached spawn would leave the
/// index silently frozen behind a watermark that keeps claiming coverage.
///
/// # Everything environmental arrives from the Container
///
/// This crate is a Component and reads no configuration of its own: the
/// projection root, the per-deployment term key, and the host to replay
/// through are all constructor arguments.
pub struct SearchIndex {
    /// The set the marks handle fills and the worker drains. Held here so
    /// [`SearchIndex::marks`] can hand out a second reference to the SAME
    /// set the worker already owns — two sets would mean marks nothing drains.
    dirty: Arc<marks::DirtySet>,
    worker: worker::SearchIndexWorker,
}

impl SearchIndex {
    /// Open the projection at `root` and build the index over it.
    ///
    /// `root` must be a subtree no other component enumerates. The event-log
    /// host lists partitions by looking for `<name>_data` directories directly
    /// under its own storage root, so a sibling directory there is disjoint
    /// from every partition — the same placement the credential, persona, and
    /// web-session stores already use.
    ///
    /// `term_key` is 32 bytes of per-deployment secret. Every stored term hash
    /// is computed under it, so replacing it invalidates every segment — which
    /// the footer's key identity detects and refuses, rather than silently
    /// answering zero hits (`search_index::terms`).
    ///
    /// Detection is not repair, which is what `origin` is for. A
    /// [`TermKeyOrigin::Minted`] key means every segment on disk is orphaned, so
    /// this starts the index degraded: the worker's first pass then sweeps the
    /// deployment and rebuilds it. Without that the orphaned segments sit there
    /// refusing every search forever, because nothing else revisits a
    /// conversation that has stopped committing turns.
    ///
    /// # Errors
    ///
    /// Returns [`SearchIndexError`] when the projection root cannot be
    /// created. A Container should treat that as fatal: an index that cannot
    /// be written is one whose coverage no reader can verify.
    pub fn open(
        root: PathBuf,
        term_key: [u8; 32],
        origin: TermKeyOrigin,
        journal: Arc<dyn PartitionJournal>,
    ) -> Result<Self, SearchIndexError> {
        let projection =
            store::SearchProjection::open(root.clone()).map_err(|err| SearchIndexError {
                root,
                reason: err.to_string(),
            })?;
        let dirty = Arc::new(marks::DirtySet::default());
        if origin == TermKeyOrigin::Minted {
            tracing::warn!(
                "search index: the term key was minted, so every stored segment is unreadable \
                 under it; the index refuses until a reconcile rebuilds the deployment"
            );
            dirty.degrade();
        }
        let worker = worker::SearchIndexWorker::new(
            projection,
            journal,
            Arc::clone(&dirty),
            terms::TermKey::new(term_key),
        );
        Ok(Self { dirty, worker })
    }

    /// The handle that marks conversations dirty as commits and mutations
    /// reach this index.
    ///
    /// Start driving it BEFORE anything can commit, so no committed turn lands
    /// unmarked. It only marks — every replay happens on [`SearchIndex::run`]'s
    /// loop.
    #[must_use]
    pub fn marks(&self) -> Arc<marks::CommitMarks> {
        Arc::new(marks::CommitMarks::new(Arc::clone(&self.dirty)))
    }

    /// Drain and apply on a fixed interval until `shutdown` is cancelled.
    ///
    /// Put this in the Container's supervised task set. It returns only on
    /// shutdown, so a return from anywhere else is a task that stopped and
    /// should take the process with it.
    ///
    /// # What a restart loses, and what does not recover it yet
    ///
    /// The dirty set lives in memory. A shutdown drains what it can, but every
    /// mark still outstanding when the process exits is gone, and there is no
    /// startup sweep to rediscover it — the design record dropped that barrier
    /// because it costs O(every partition in the deployment) per boot. So a
    /// conversation that committed a turn the worker had not yet indexed comes
    /// back behind its stored watermark, and stays there until its next
    /// committed turn re-marks it.
    ///
    /// `SearchProjection::verified_coverage` does NOT catch this. It checks the
    /// event at the stored watermark against the live journal, and a shorter
    /// prefix's incarnation still matches — the events below the watermark did
    /// not move, so a stale-but-honest watermark verifies. What would catch it
    /// is a read-side comparison of the watermark against the conversation's
    /// last committed turn boundary, which is a coverage question rather than an
    /// identity one. This change does not ship that.
    ///
    /// A lost mark is merely stale for ordinary content, and that is the trade
    /// being made. It is NOT merely stale for an excision, where the discarded
    /// mark would leave removed text searchable — so that one obligation is
    /// recorded durably instead of trusted to this set: the journal holds the
    /// marker and the footer records how far the index has scanned for one, and
    /// `verified_coverage` refuses on the difference. See
    /// `store::Coverage::excision_scanned_through`.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation is observed between conversations, never inside one. Work
    /// the interrupted pass did not reach is marked dirty again before this
    /// returns, so nothing is lost while the process lives.
    pub async fn run(self, shutdown: CancellationToken) {
        self.worker.run(shutdown).await;
    }
}

/// Tunables that govern how much work a search may do.
///
/// Both bounds are runtime configuration rather than constants, deliberately.
/// The deployment has no participation corpus to calibrate them against before
/// launch, so the honest position is that the first real numbers arrive from
/// production — `partitions_read_per_search` is the metric that supplies them
/// — and tuning must then be a config change rather than a rebuild of the
/// layer that is hardest to rebase.
#[derive(Debug, Clone, Copy)]
pub(crate) struct SearchIndexConfig {
    /// Drop a query term present in more than this fraction of the in-scope
    /// conversations.
    ///
    /// Computed from the membership records already being tested, so it costs
    /// nothing extra and adapts per caller: a term common in ONE person's
    /// history is filtered for them without appearing on any global stopword
    /// list. This diverges from same-conversation `conversation_find`, which
    /// weighs every term equally, and the divergence is documented at the tool
    /// surface rather than left to be discovered.
    pub(crate) max_term_document_frequency: f64,

    /// The most partitions one search may read postings for.
    ///
    /// Retrieval is score-bounded rather than cap-and-refuse: a membership
    /// record over-estimates and never under-estimates overlap, so
    /// record-derived overlap is a strict upper bound on the true score.
    /// Reading candidates in upper-bound order means this cap bites the LEAST
    /// promising partitions rather than producing a blanket refusal.
    pub(crate) max_partitions_read: usize,
}

impl Default for SearchIndexConfig {
    /// Starting points, not measured values.
    ///
    /// A term in more than half the scope carries almost no selectivity, and
    /// 256 partition reads is roughly the point past which a search stops
    /// feeling interactive. Both are first guesses that production is expected
    /// to move.
    fn default() -> Self {
        Self {
            max_term_document_frequency: 0.5,
            max_partitions_read: 256,
        }
    }
}

/// Where a conversation's matching terms actually are.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct PostingsRecord {
    /// Every committed message indexed for this partition, ascending by
    /// position.
    pub(crate) messages: Vec<IndexedMessage>,
}

/// One committed message's searchable terms. Carries no message text — the
/// text is read back from the journal when a hit is actually fetched.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct IndexedMessage {
    /// The message's durable journal position.
    pub(crate) position: u64,
    /// The turn this message belongs to; ranking deduplicates to one hit per
    /// turn, so this is the grouping key.
    pub(crate) turn_id: String,
    /// This message's distinct keyed term hashes, ascending.
    pub(crate) term_hashes: Vec<u32>,
}

impl IndexedMessage {
    /// How many of `query_hashes` this message carries.
    pub(crate) fn overlap(&self, query_hashes: &[u32]) -> usize {
        query_hashes
            .iter()
            .filter(|hash| self.term_hashes.binary_search(hash).is_ok())
            .count()
    }
}

#[cfg(test)]
mod tests;