trusty-common 0.61.0

//! Pure helper functions and constants for the dream loop.
//!
//! Why: Extracted from dream.rs to keep each file under the 500-SLOC cap
//! (#607). These are stateless helpers shared by the Dreamer passes.
//! What: `extract_keywords`, `is_low_quality_content`, `now_secs`,
//! `merged_drawer`, `pick_survivor`, `persist_merge`, `rebuild_index_from_drawers`, content blocklist, stop-words.
//! Test: Indirectly via dream cycle and closet tests.

use crate::memory_core::embed::Embedder;
use crate::memory_core::palace::Drawer;
use crate::memory_core::retrieval::{
    ChunkPlanner, PalaceHandle, embed_chunks_within, embed_planned_within, shared_embedder,
};
use crate::memory_core::timeouts;
use anyhow::{Context, Result};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use uuid::Uuid;

/// Stop-word filter for closet keyword extraction.
// #9279: two-letter tokens are now kept, so the common two-letter words that
// were not already here join the list. Keep the two-letter words in step with
// `STOPWORDS` in trusty-memory's `tools/recall_rulings_floor.rs`.
pub(crate) const STOP_WORDS: &[&str] = &[
    "the", "a", "an", "is", "are", "was", "were", "be", "been", "being", "of", "in", "on", "at",
    "to", "for", "with", "and", "or", "but", "not", "no", "yes", "i", "you", "he", "she", "it",
    "we", "they", "this", "that", "these", "those", "as", "by", "from", "into", "over", "under",
    "if", "then", "than", "so", "do", "does", "did", "have", "has", "had", "will", "would",
    "shall", "should", "can", "could", "may", "might", "must", "about", "any", "all", "some",
    "more", "most", "such", "ok", "up", "my", "us", "me", "am", "go", "id", "vs", "hi", "oh",
];

/// Shortest keyword, in characters, that closets and recall keep (#9279).
///
/// Why: ids such as `e1`, `v2`, `#42` and `fe` are two characters long. A
/// three-byte floor dropped every one of them, so a query naming an id could
/// never match the drawer that holds it.
pub(crate) const MIN_KEYWORD_CHARS: usize = 2;

/// Normalize one whitespace-delimited word into a keyword token.
///
/// What: keeps alphanumeric characters only, lowercased, so `#42` becomes
/// `42` and `E1:` becomes `e1`. Shared by [`extract_keywords`] and the recall
/// id-token match, so both sides compare the same form.
pub(crate) fn normalize_keyword(raw: &str) -> String {
    raw.chars()
        .filter(|c| c.is_alphanumeric())
        .flat_map(|c| c.to_lowercase())
        .collect()
}

/// Extract keyword tokens from a drawer's content.
///
/// Why: Closets are a lightweight pre-computed index; we want stable, deduped
/// keyword tokens so the dream cycle's index is reproducible.
/// What: Normalizes each word with [`normalize_keyword`], drops stop-words and
/// tokens shorter than [`MIN_KEYWORD_CHARS`] characters, and dedups within a
/// single drawer.
/// Test: `extract_keywords_keeps_short_id_tokens`, `closet_refresh_builds_index`.
pub fn extract_keywords(content: &str) -> Vec<String> {
    let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
    let mut out: Vec<String> = Vec::new();
    for raw in content.split_whitespace() {
        let token = normalize_keyword(raw);
        // #9279: count characters, not bytes, and keep two-character ids.
        if token.chars().count() < MIN_KEYWORD_CHARS {
            continue;
        }
        if STOP_WORDS.iter().any(|s| *s == token) {
            continue;
        }
        if seen.insert(token.clone()) {
            out.push(token);
        }
    }
    out
}

/// Returns true when `content` should be dropped by the content-quality
/// prune pass.
///
/// Why: Centralises the "is this drawer noise?" decision so the prune pass
/// and its tests share one rule. The rule mirrors the write-path gate
/// (`trusty-memory::tools::blocklist_gate` plus a minimum word-count
/// floor) so a drawer that wouldn't be written today is also a drawer
/// that should not survive the next dream cycle.
/// Issue #2442: previously carried its own `CONTENT_BLOCKLIST` copy checked
/// via `str::contains` (substring-anywhere), which could retroactively prune
/// a drawer that merely QUOTED an auto-capture phrase mid-prose — thinning
/// the recall surface the same way the write-path over-match did. Now
/// delegates to [`crate::memory_core::filter::blocklist_match`], the single
/// prefix-anchored source of truth shared with the write path.
/// What: Trims leading whitespace, then returns true iff the trimmed content
/// is FRAMED by a known blocklist pattern (see `blocklist_match`), OR the
/// whitespace-delimited word count is strictly less than `min_words`. An
/// empty `content` (zero words) is always low-quality whenever
/// `min_words >= 1`.
/// Test: `dream_content_prune_drops_blocklist_drawer`,
/// `dream_content_prune_drops_short_drawer`,
/// `dream_content_prune_keeps_good_drawer`.
pub(crate) fn is_low_quality_content(content: &str, min_words: usize) -> bool {
    if crate::memory_core::filter::blocklist_match(content).is_some() {
        return true;
    }
    let word_count = content.split_whitespace().count();
    word_count < min_words
}

/// Largest prefix of `s` that fits in `max_bytes` and ends on a UTF-8 char
/// boundary.
///
/// Why (#5187): the semantic pass caps its log preview of drawer text at a
/// fixed byte count. Applied as a raw byte offset, the cap panics the moment it
/// lands inside a multi-byte `char` (`assertion failed: self.is_char_boundary`),
/// which killed a `tokio-rt-worker` in the shipped `com.trusty.memory` daemon.
/// Drawer content is arbitrary user text, so CJK, Cyrillic, emoji, and
/// accented Latin all reach the cap.
/// What: rounds `max_bytes` DOWN to the nearest char boundary via
/// `str::floor_char_boundary`, so the result never exceeds `max_bytes` and
/// never splits a `char`. Returns all of `s` when `s` already fits.
/// Test: `char_safe_prefix_stops_below_a_multibyte_char`.
pub(crate) fn char_safe_prefix(s: &str, max_bytes: usize) -> &str {
    &s[..s.floor_char_boundary(max_bytes)]
}

/// Current unix timestamp in seconds. Saturates to 0 on clock errors.
pub(crate) fn now_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Tag that marks a drawer as an owner ruling (#9172).
pub(crate) const RULING_TAG: &str = "ruling";

/// Largest content, in bytes, a dedup merge may write into its survivor.
///
/// Why (#9172): without the old 500-byte cut, a near-duplicate that keeps
/// coming back is appended to the same survivor on every cycle, and the drawer
/// grows without bound. #9275: [`persist_merge`] re-embeds the survivor as a
/// full chunk group, so the bound now caps the chunk count a merge can add.
/// What: [`merged_drawer`] refuses a merge that would append past this bound.
/// Test: `a_merge_is_skipped_past_the_byte_bound`,
/// `dedup_survivor_tests::a_merge_past_the_byte_bound_keeps_both_drawers`.
pub(crate) const MERGE_MAX_BYTES: usize = 4 * 1024;

/// Text a merge puts between the survivor's body and the loser's.
const MERGE_SEPARATOR: &str = "\n\nAlso: ";

/// The survivor `loser` leaves behind when a dedup merge folds it in.
///
/// Why (#9172): the merge used to cut the combined text at 500 bytes, which
/// dropped the loser's text and, past the cap, the survivor's own. A dedup
/// merge must not lose text, and must not grow a drawer without bound.
/// What: a copy of `survivor` whose content is its own body plus
/// `"\n\nAlso: " + loser` — skipped when the survivor already contains the
/// loser's text (exact duplicates) — with the higher importance and the union
/// of both tag sets. Nothing is truncated. `None` when the appended content
/// would exceed [`MERGE_MAX_BYTES`]; the caller then keeps both drawers. A
/// merge that appends nothing adds no bytes and is not bounded.
/// Test: `dedup_survivor_tests::a_dedup_merge_survives_a_palace_reopen`,
/// `a_merge_keeps_multibyte_loser_text_whole`,
/// `merge_into_keeps_the_content_hash_in_step`,
/// `a_merge_is_skipped_past_the_byte_bound`.
pub(crate) fn merged_drawer(survivor: &Drawer, loser: &Drawer) -> Option<Drawer> {
    let mut merged = survivor.clone();
    if !survivor.content().contains(loser.content()) {
        // #9172: past the bound the merge is skipped, never truncated.
        let len = survivor.content().len() + MERGE_SEPARATOR.len() + loser.content().len();
        if len > MERGE_MAX_BYTES {
            return None;
        }
        // #5902: `set_content`, so the content digest moves with the body.
        merged.set_content(format!(
            "{}{MERGE_SEPARATOR}{}",
            survivor.content(),
            loser.content()
        ));
    }
    merged.importance = merged.importance.max(loser.importance);
    for tag in &loser.tags {
        if !merged.tags.contains(tag) {
            merged.tags.push(tag.clone());
        }
    }
    Some(merged)
}

/// Order a near-duplicate pair as `(survivor, loser)`, or refuse to merge it.
///
/// Why (#9172): importance alone let an older note replace the current one.
/// A drawer holding a `fact_key` slot is the live fact for that slot; deleting
/// it releases the slot.
/// What: returns `None` when both drawers hold a slot. Otherwise the survivor
/// is the drawer that ranks higher on, in order: holds a slot, carries
/// [`RULING_TAG`], newer `created_at`, higher importance. A full tie keeps `a`.
/// Test: `dedup_survivor_tests::a_newer_status_note_survives_an_older_higher_importance_duplicate`,
/// `dedup_survivor_tests::a_slot_holder_or_ruling_outlives_a_more_important_duplicate`.
pub(crate) fn pick_survivor<'a>(a: &'a Drawer, b: &'a Drawer) -> Option<(&'a Drawer, &'a Drawer)> {
    if a.is_tier_c() && b.is_tier_c() {
        return None;
    }
    let rank = |d: &Drawer| (d.is_tier_c(), d.tags.iter().any(|t| t == RULING_TAG));
    let order = rank(a)
        .cmp(&rank(b))
        .then(a.created_at.cmp(&b.created_at))
        .then(a.importance.total_cmp(&b.importance));
    Some(if order.is_lt() { (b, a) } else { (a, b) })
}

/// Merge the pair `a`/`b` and persist the survivor; returns `(survivor, loser)`.
///
/// Why (#9172): the merge changed only the in-memory table, so the loser's
/// text was lost at the next open while its row was already deleted.
/// #9275: a merge that appends text left the survivor's old, shorter vector
/// group in place, so health read it short and its tail was not findable.
/// What: under the palace write mutex and then the commit-order guard (the
/// write pipeline's lock order), reads both drawers from the live table,
/// picks the survivor with [`pick_survivor`], and builds [`merged_drawer`].
/// When the merge changed the text, it embeds the new text with
/// `embed_chunks_within` (the remember path's bounded chunked embed, under
/// `embed_timeout` per chunk) and replaces the survivor's vector group with
/// `upsert_chunks`. Then it writes the drawer to redb and mirrors it into the
/// table. Every failure commits nothing: an embed, split or upsert error
/// returns before redb is touched, and a redb error after the vector swap
/// restores the survivor's own group (logged if that fails too). Returns
/// `Ok(None)` when either drawer is gone, the pair must not merge, or the
/// merged text would pass [`MERGE_MAX_BYTES`]. The caller deletes the loser
/// only after `Ok(Some(_))`: a failed persist leaves both drawers intact.
/// Test: `dedup_survivor_tests::a_dedup_merge_survives_a_palace_reopen`,
/// `dedup_survivor_tests::a_failed_merge_persist_keeps_the_duplicate`,
/// `chunk_key_tests::a_dedup_merge_reembeds_the_survivor_as_a_full_chunk_group`.
pub(crate) async fn persist_merge(
    handle: &Arc<PalaceHandle>,
    a: Uuid,
    b: Uuid,
    embedder: &(dyn Embedder + Send + Sync),
    embed_timeout: Duration,
) -> Result<Option<(Uuid, Uuid)>> {
    let _write_guard = timeouts::lock_with_timeout(
        &handle.write_mutex,
        timeouts::write_lock_timeout(),
        handle.id.as_str(),
    )
    .await?;
    let _order = handle.commit_mutex.lock().await;
    let (a, b) = {
        let drawers = handle.drawers.read();
        let find = |id: Uuid| drawers.iter().find(|d| d.id == id).cloned();
        (find(a), find(b))
    };
    let (Some(a), Some(b)) = (a, b) else {
        return Ok(None);
    };
    let Some((survivor, loser)) = pick_survivor(&a, &b) else {
        return Ok(None);
    };
    // #9172: an over-bound merge is skipped; both drawers stay whole.
    let Some(merged) = merged_drawer(survivor, loser) else {
        tracing::info!(
            palace = %handle.id, survivor = %survivor.id, loser = %loser.id,
            "dream dedup: merge would pass {MERGE_MAX_BYTES} bytes; both drawers kept"
        );
        return Ok(None);
    };
    // #9275: re-embed before anything is committed; a failure keeps both.
    let reembed = merged.content() != survivor.content();
    if reembed {
        replace_group(handle, embedder, merged.id, merged.content(), embed_timeout)
            .await
            .with_context(|| format!("re-embed merged dedup survivor {}", merged.id))?;
    }
    if let Err(e) = handle.kg.upsert_drawer(&merged).await {
        if reembed {
            restore_group(handle, embedder, survivor, embed_timeout).await;
        }
        return Err(e).with_context(|| format!("persist merged dedup survivor {}", merged.id));
    }
    let pair = (survivor.id, loser.id);
    if let Some(row) = handle.drawers.write().iter_mut().find(|d| d.id == pair.0) {
        *row = merged;
    }
    Ok(Some(pair))
}

/// Embed `text` as a full chunk group and make it `id`'s whole group.
async fn replace_group(
    handle: &PalaceHandle,
    embedder: &(dyn Embedder + Send + Sync),
    id: Uuid,
    text: &str,
    embed_timeout: Duration,
) -> Result<()> {
    let vecs = embed_chunks_within(embedder, text, embed_timeout).await?;
    handle.vector_store.upsert_chunks(id, vecs).await
}

/// Put `survivor`'s own group back after a failed merge commit (#9275).
async fn restore_group(
    handle: &PalaceHandle,
    embedder: &(dyn Embedder + Send + Sync),
    survivor: &Drawer,
    embed_timeout: Duration,
) {
    let restored = replace_group(
        handle,
        embedder,
        survivor.id,
        survivor.content(),
        embed_timeout,
    )
    .await;
    if let Err(e) = restored {
        tracing::error!(
            palace = %handle.id, drawer = %survivor.id,
            "#9275: merge not committed and the survivor's vectors not restored: {e:#}"
        );
    }
}

/// What one index rebuild did (#9275).
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub(crate) struct RebuildReport {
    /// Drawers stored again as a whole chunk group.
    pub(crate) rebuilt: usize,
    /// Drawers whose split, embed or upsert failed. Each is logged and left
    /// with no vector, which `embed_health` reports as missing.
    pub(crate) skipped: usize,
}

/// Reset the vector index and re-upsert every drawer from the in-memory
/// drawer table. Returns a [`RebuildReport`].
///
/// Why: When the HNSW index accumulates orphans we can't address through
/// `key_map` (pre-fix data, partial writes, schema migrations), the cheapest
/// correct fix is to throw away the index and rebuild from the authoritative
/// drawer table.
/// What: Snapshots drawers, resolves the embedder and its chunk planner, calls
/// `UsearchStore::reset` to truncate the index, then re-embeds and re-upserts
/// each drawer. #9275: a planner that cannot be built returns the error before
/// the reset, so the index is untouched; one drawer that fails to split, embed
/// or upsert is logged and counted in `skipped`, and the rebuild goes on to
/// the next drawer. Respects the budget by
/// stopping early — incomplete rebuilds are still safe (the next cycle picks
/// up where this one left off). #6208: the snapshot→reset→re-upsert runs under
/// the palace write mutex so a concurrent `remember` cannot land a vector that
/// `reset()` then wipes without re-adding.
/// Test: `rebuild_index_keeps_every_chunk_of_a_long_drawer`,
/// `a_rebuild_skips_a_drawer_that_will_not_split_and_rebuilds_the_rest`,
/// `a_rebuild_whose_planner_cannot_be_built_leaves_the_index_untouched`.
pub(crate) async fn rebuild_index_from_drawers(
    handle: &Arc<PalaceHandle>,
    started: std::time::Instant,
    budget: Duration,
) -> Result<RebuildReport> {
    rebuild_index_with(handle, None, started, budget).await
}

/// [`rebuild_index_from_drawers`] with an explicit embedder; `None` resolves
/// the shared one before the reset (#9275 test seam).
/// Test: `rebuild_stopping_at_the_budget_leaves_only_complete_groups`.
pub(crate) async fn rebuild_index_with(
    handle: &Arc<PalaceHandle>,
    embedder: Option<&Arc<dyn Embedder + Send + Sync>>,
    started: std::time::Instant,
    budget: Duration,
) -> Result<RebuildReport> {
    // #6208: hold the write mutex across the whole rebuild. `reset()` truncates
    // the entire index; a `remember` that upserts between the snapshot and the
    // re-upsert loop would be dropped by the reset and never re-added, because
    // only the snapshotted drawers are re-embedded. Guarding snapshot→reset→
    // re-upsert makes a mid-flight remember impossible during this region — its
    // vector is either in the snapshot (re-added) or lands after release (never
    // reset). The sole caller, `compact_pass`, releases its own guard before
    // calling this, so there is no reentrancy.
    let _write_guard = timeouts::lock_with_timeout(
        &handle.write_mutex,
        timeouts::write_lock_timeout(),
        handle.id.as_str(),
    )
    .await?;
    let snapshot: Vec<Drawer> = handle.drawers.read().clone();
    let reset = || {
        handle
            .vector_store
            .reset()
            .context("reset vector index for rebuild")
    };
    if snapshot.is_empty() {
        reset()?;
        return Ok(RebuildReport::default());
    }

    // #9275: the embedder and planner come before the reset; if either cannot
    // be built, the index is left as it is.
    let embedder = match embedder {
        Some(e) => Arc::clone(e),
        None => shared_embedder()
            .await
            .context("acquire shared embedder for dream rebuild")?,
    };
    let planner = ChunkPlanner::for_embedder(embedder.as_ref())
        .context("build the chunk planner for dream rebuild")?;
    reset()?;

    let per_chunk = timeouts::embed_batch_timeout();
    let mut report = RebuildReport::default();
    for drawer in snapshot.iter() {
        if started.elapsed() >= budget {
            break;
        }
        // #9275: every chunk of the drawer, stored as one whole group, so a
        // budget stop never leaves a partial group. One failing drawer is
        // skipped, never the end of the rebuild.
        match rebuild_one(
            handle,
            embedder.as_ref(),
            planner.as_ref(),
            drawer,
            per_chunk,
        )
        .await
        {
            Ok(()) => report.rebuilt += 1,
            Err(e) => {
                tracing::warn!(
                    palace = %handle.id, drawer = %drawer.id,
                    "#9275: dream rebuild skipped a drawer; it has no vector: {e:#}"
                );
                report.skipped += 1;
            }
        }
    }
    Ok(report)
}

/// Split, embed and store one drawer's whole chunk group for the rebuild.
async fn rebuild_one(
    handle: &PalaceHandle,
    embedder: &(dyn Embedder + Send + Sync),
    planner: Option<&ChunkPlanner>,
    drawer: &Drawer,
    per_chunk: Duration,
) -> Result<()> {
    let chunks: Vec<String> = match planner {
        Some(p) => p.split(drawer.content())?,
        None => vec![drawer.content()],
    }
    .into_iter()
    .map(str::to_string)
    .collect();
    let vecs = embed_planned_within(embedder, &chunks, per_chunk).await?;
    handle.vector_store.upsert_chunks(drawer.id, vecs).await
}

/// Rebuild closets: simple whitespace tokenization, stop-word filter,
/// keyword -> drawer ids. Returns the number of keywords indexed.
///
/// Why: Centralises closet rebuild logic so both `Dreamer::refresh_closets`
/// and future callers share one implementation.
/// What: Snapshots the drawer table, tokenizes each drawer's content via
/// `extract_keywords`, and builds a fresh `HashMap<String, Vec<Uuid>>`.
/// Test: `closet_refresh_builds_index`.
pub(crate) fn build_closet_index(drawers: &[Drawer]) -> HashMap<String, Vec<Uuid>> {
    let mut new_index: HashMap<String, Vec<Uuid>> = HashMap::new();
    for drawer in drawers.iter() {
        for kw in extract_keywords(drawer.content()) {
            new_index.entry(kw).or_default().push(drawer.id);
        }
    }
    new_index
}