khive_runtime/operations/
read_helpers.rs

1use khive_score::DeterministicScore;
2use khive_storage::note::Note;
3use khive_storage::types::{Direction, SqlStatement, SqlValue};
4use khive_storage::{EdgeRelation, Entity, Event};
5use uuid::Uuid;
6
7use crate::error::{RuntimeError, RuntimeResult};
8#[cfg(doc)]
9use crate::runtime::KhiveRuntime;
10
11/// A note search result with UUID, salience-weighted RRF score, and display text.
12#[derive(Clone, Debug)]
13pub struct NoteSearchHit {
14    pub note_id: Uuid,
15    pub score: DeterministicScore,
16    pub rank_score_kind: crate::RankScoreKind,
17    pub signals: crate::SearchSignals,
18    pub source: crate::SearchSource,
19    pub title: Option<String>,
20    pub snippet: Option<String>,
21}
22
23pub(super) fn salience_weighted_rank(
24    score: DeterministicScore,
25    salience: Option<f64>,
26) -> DeterministicScore {
27    const SCALE_RAW: i128 = 1_i128 << 32;
28    let salience = DeterministicScore::from_f64(salience.unwrap_or(0.5));
29    let weight_raw = SCALE_RAW / 2 + i128::from(salience.to_raw()) / 2;
30    // Match khive-score's fixed-point multiplication and saturation without
31    // converting the derived weight or ranking score back to floating point.
32    let weighted_raw = i128::from(score.to_raw()) * weight_raw / SCALE_RAW;
33    DeterministicScore::from_raw(weighted_raw.clamp(
34        i128::from(DeterministicScore::NEG_INF.to_raw()),
35        i128::from(DeterministicScore::MAX.to_raw()),
36    ) as i64)
37}
38
39/// Result of [`KhiveRuntime::search_notes_outcome`]: the fused hits — text
40/// hits alone when the vector arm failed — plus the vector arm's error, if
41/// any. Mirrors [`crate::HybridSearchOutcome`] for the note substrate.
42#[derive(Clone, Debug)]
43pub struct NoteSearchOutcome {
44    pub hits: Vec<NoteSearchHit>,
45    pub vector_error: Option<String>,
46}
47
48/// Re-insert hyphens at canonical UUID positions (8-4-4-4-12) into a
49/// hyphen-free hex prefix, so a `LIKE '<pattern>%'` scan against the
50/// hyphenated `id` column matches correctly. Prefixes that already
51/// contain a hyphen are passed through unchanged. No-op for len <= 8
52/// (already correct). Input longer than 32 hex chars is NOT truncated: the
53/// extra hex chars are appended past the canonical 12-char final segment
54/// with no further hyphen, so the resulting `LIKE` pattern requires literal
55/// characters beyond position 36 that no real (36-char) UUID string can
56/// ever have — the scan naturally fails closed instead of silently
57/// resolving `<valid-32-hex><extra-hex>` to the valid UUID.
58pub fn hex_prefix_to_uuid_pattern(prefix: &str) -> String {
59    if prefix.contains('-') {
60        return prefix.to_string();
61    }
62    const BOUNDARIES: [usize; 4] = [8, 13, 18, 23]; // post-hyphen-insertion offsets
63    let mut out = String::with_capacity(36);
64    for c in prefix.chars() {
65        if BOUNDARIES.contains(&out.len()) {
66            out.push('-');
67        }
68        out.push(c);
69    }
70    out
71}
72
73/// Return the inclusive lower and exclusive upper bounds for a UUID prefix
74/// stored as a canonical lowercase UUID string under SQLite's BINARY collation.
75///
76/// Compact hexadecimal prefixes and prefixes of the canonical dashed spelling
77/// are accepted. The returned bounds are canonicalized to lowercase; malformed
78/// dashed spellings and prefixes longer than one UUID fail closed. The upper
79/// bound is the shortest lexicographic successor, so a trailing run of `f`
80/// digits carries into the preceding digit. `g` is the exclusive sentinel when
81/// the prefix is all `f`, because canonical UUID strings contain only `0`-`f`.
82pub fn uuid_prefix_bounds(prefix: &str) -> Option<(String, String)> {
83    const HYPHEN_POSITIONS: [usize; 4] = [8, 13, 18, 23];
84
85    let compact = if prefix.contains('-') {
86        if prefix.len() > 36 {
87            return None;
88        }
89        let mut compact = String::with_capacity(32);
90        for (index, byte) in prefix.bytes().enumerate() {
91            if HYPHEN_POSITIONS.contains(&index) {
92                if byte != b'-' {
93                    return None;
94                }
95            } else if byte.is_ascii_hexdigit() {
96                compact.push(char::from(byte.to_ascii_lowercase()));
97            } else {
98                return None;
99            }
100        }
101        compact
102    } else {
103        if prefix.is_empty()
104            || prefix.len() > 32
105            || !prefix.bytes().all(|byte| byte.is_ascii_hexdigit())
106        {
107            return None;
108        }
109        prefix.to_ascii_lowercase()
110    };
111
112    if compact.is_empty() || compact.len() > 32 {
113        return None;
114    }
115
116    let lower = hex_prefix_to_uuid_pattern(&compact);
117    let mut successor = compact.into_bytes();
118    let mut carried_past_start = true;
119    for index in (0..successor.len()).rev() {
120        let next = match successor[index] {
121            b'0'..=b'8' | b'a'..=b'e' => Some(successor[index] + 1),
122            b'9' => Some(b'a'),
123            b'f' => None,
124            _ => return None,
125        };
126        if let Some(next) = next {
127            successor[index] = next;
128            successor.truncate(index + 1);
129            carried_past_start = false;
130            break;
131        }
132    }
133
134    let upper = if carried_past_start {
135        "g".to_string()
136    } else {
137        let compact_upper = String::from_utf8(successor).ok()?;
138        hex_prefix_to_uuid_pattern(&compact_upper)
139    };
140    Some((lower, upper))
141}
142
143pub(super) fn resolve_prefix_statement(
144    table: &str,
145    has_deleted_at: bool,
146    include_deleted: bool,
147    namespaces: Option<&[String]>,
148    lower: &str,
149    upper: &str,
150) -> SqlStatement {
151    let namespace_clause = namespaces.map(|namespaces| {
152        let placeholders: Vec<String> = (0..namespaces.len())
153            .map(|index| format!("?{}", index + 3))
154            .collect();
155        // The unary `+` keeps the namespace predicate as a row filter but out of
156        // index selection: SQLite will not use `idx_*_namespace` for it, so the
157        // plan seeks the primary-key range instead of scanning every row of the
158        // caller's namespaces. khive-db does not run `ANALYZE`, so the planner
159        // has no statistics showing that a namespace is usually not selective.
160        format!(" AND +namespace IN ({})", placeholders.join(", "))
161    });
162    let deleted_filter = if has_deleted_at && !include_deleted {
163        " AND deleted_at IS NULL"
164    } else {
165        ""
166    };
167    let mut params = vec![
168        SqlValue::Text(lower.to_owned()),
169        SqlValue::Text(upper.to_owned()),
170    ];
171    if let Some(namespaces) = namespaces {
172        params.extend(
173            namespaces
174                .iter()
175                .map(|namespace| SqlValue::Text(namespace.clone())),
176        );
177    }
178
179    SqlStatement {
180        sql: format!(
181            "SELECT id FROM {table} \
182             WHERE id >= ?1 AND id < ?2{namespace_clause}{deleted_filter} ORDER BY id LIMIT 2",
183            namespace_clause = namespace_clause.as_deref().unwrap_or("")
184        ),
185        params,
186        label: Some("resolve_prefix".into()),
187    }
188}
189
190fn text_preview(text: &str, max_chars: usize) -> Option<String> {
191    let trimmed = text.trim();
192    if trimmed.is_empty() {
193        None
194    } else {
195        Some(trimmed.chars().take(max_chars).collect())
196    }
197}
198
199/// Symmetric relations (`competes_with`, `composed_with`) are stored with a
200/// canonical source (lower UUID wins), so a directed `Out` or `In` query may
201/// miss results. When the relations filter is non-empty and contains **only**
202/// symmetric relations, override direction to `Both` so callers always see all
203/// edges for these relations regardless of storage canonicalization.
204pub(super) fn normalize_symmetric_direction(
205    direction: Direction,
206    relations: Option<&[EdgeRelation]>,
207) -> Direction {
208    let Some(rels) = relations else {
209        return direction;
210    };
211    if rels.is_empty() {
212        return direction;
213    }
214    let all_symmetric = rels
215        .iter()
216        .all(|r| matches!(r, EdgeRelation::CompetesWith | EdgeRelation::ComposedWith));
217    if all_symmetric {
218        Direction::Both
219    } else {
220        direction
221    }
222}
223
224/// Stable tie-break rank for [`Direction`] — `Out` before `In` — used to make
225/// the both-direction sort/dedup key total over self-loop edges. A self-loop
226/// (`source_id == target_id == node_id`) produces two `UNION ALL` rows with
227/// the same `(node_id, edge_id)` but opposite directions; without direction in
228/// the key, sort-then-dedup collapses them to one and drops the direction
229/// parity a separate `Out` call plus a separate `In` call would preserve.
230pub(super) fn direction_sort_rank(direction: &Direction) -> u8 {
231    match direction {
232        Direction::Out => 0,
233        Direction::In => 1,
234        Direction::Both => 2,
235    }
236}
237
238pub(super) fn note_title(note: &Note) -> Option<String> {
239    note.name
240        .clone()
241        .filter(|s| !s.trim().is_empty())
242        .or_else(|| Some(format!("[{}]", note.kind.as_str())))
243}
244
245pub(super) fn note_snippet(note: &Note) -> Option<String> {
246    text_preview(&note.content, 200)
247}
248
249/// Result of resolving a UUID to its substrate kind.
250#[derive(Clone, Debug)]
251pub enum Resolved {
252    Entity(Entity),
253    Note(Note),
254    Event(Event),
255    /// A record owned by a pack's private tables.
256    ///
257    /// `pack` identifies the owning pack by name, `kind` is the pack-local
258    /// record type (e.g. "domain", "atom"), and `data` is the full record as
259    /// a JSON Value. Pack-private records are not valid edge endpoints,
260    /// annotates sources, or task context entities.
261    PackRecord {
262        pack: String,
263        kind: String,
264        data: serde_json::Value,
265    },
266}
267
268/// Await every spawned multi-model embed task in `join_set`, returning one
269/// vector per model (in model order) on full success.
270///
271/// `join_set` entries are `(model_index, embed_result)` — the index lets
272/// completion order (which is arrival order, not spawn order) be reassembled
273/// into the caller's model order. On the first failure (an embed error or a
274/// task panic), every remaining handle is aborted and detached so the error
275/// return is not gated on a sibling reaching a cancellation point. A sibling
276/// already inside synchronous native inference may finish that call in the
277/// background. Embed calls are counted when issued, before the provider
278/// await, so detached completion cannot change the operation's usage count.
279/// Each task owns cloned runtime/provider state and only computes an embedding;
280/// storage writes remain in the parent after this drain succeeds.
281pub(super) async fn drain_embed_join_set<T: Send + 'static>(
282    mut join_set: tokio::task::JoinSet<(usize, RuntimeResult<T>)>,
283    model_count: usize,
284) -> RuntimeResult<Vec<T>> {
285    let mut vectors: Vec<Option<T>> = (0..model_count).map(|_| None).collect();
286
287    while let Some(joined) = join_set.join_next().await {
288        match joined {
289            Ok((idx, Ok(vector))) => vectors[idx] = Some(vector),
290            Ok((_idx, Err(e))) => {
291                join_set.abort_all();
292                return Err(e);
293            }
294            Err(join_err) => {
295                join_set.abort_all();
296                return Err(RuntimeError::Internal(format!(
297                    "embed task panicked: {join_err}"
298                )));
299            }
300        }
301    }
302
303    Ok(vectors
304        .into_iter()
305        .map(|v| v.expect("every model index observed exactly once by join_set drain"))
306        .collect())
307}
khive_runtime/operations/read_helpers.rs

khive_runtime/operations/
read_helpers.rs
