Skip to main content

rac_engine/
index_store.rs

1//! Persistent memory-mapped index store (ADR-104) — port of
2//! `services/index_store.py` per `rust/spec/index-store-format.md`.
3//!
4//! Store byte-identity is the parity surface: for the same corpus bytes this
5//! writer must produce a segment directory byte-identical to the oracle's.
6//! Every reader-side failure degrades to a miss (`None`), never an answer
7//! change; every writer-side failure degrades to "not written" (ADR-080).
8
9use std::collections::BTreeMap;
10use std::fs;
11use std::io::Write as _;
12use std::path::{Path, PathBuf};
13
14use memmap2::Mmap;
15use serde_json::Value;
16
17use crate::derived::DerivedIndex;
18use crate::index_format::{
19    encode_segment, segment_payload, write_indexed, IndexFormatError, IndexedSegment, Reader,
20    Writer,
21};
22use crate::pycompat::py_casefold;
23use crate::relationships::Relationship;
24use crate::resolve::{FieldTokens, IndexEntry};
25use crate::walk::find_markdown_files;
26
27// The scorable field families in the exact BM25F iteration order (ADR-078).
28pub const FIELDS: [&str; 6] = ["id", "title", "path", "heading", "body", "tags"];
29
30pub const STORE_DIRNAME: &str = "store";
31pub const STORE_LAYOUT_VERSION: &str = "v1";
32
33const SEG_HEADER: &str = "header.seg";
34const SEG_ENTRIES: &str = "entries.seg";
35const SEG_SECTIONS: &str = "sections.seg";
36const SEG_TOKENS: &str = "tokens.seg";
37const SEG_TERMDICT: &str = "termdict.seg";
38const SEG_POSTINGS: &str = "postings.seg";
39const SEG_RELATIONSHIPS: &str = "relationships.seg";
40const SEG_LIVE: &str = "live.seg";
41const SEG_SCOPE: &str = "scope.seg";
42const SEG_PORTFOLIO: &str = "portfolio.seg";
43const SEG_ALIASMAP: &str = "aliasmap.seg";
44const SEG_PATHMAP: &str = "pathmap.seg";
45
46const ALL_SEGMENTS: [&str; 12] = [
47    SEG_HEADER,
48    SEG_ENTRIES,
49    SEG_SECTIONS,
50    SEG_TOKENS,
51    SEG_TERMDICT,
52    SEG_POSTINGS,
53    SEG_RELATIONSHIPS,
54    SEG_LIVE,
55    SEG_SCOPE,
56    SEG_PORTFOLIO,
57    SEG_ALIASMAP,
58    SEG_PATHMAP,
59];
60
61/// The pinned scoring-constant fingerprint (spec/index-store-format.md §3.1).
62/// Must track `resolve`'s BM25F constants; the golden-vector test pins the
63/// exact string against the oracle's `scoring_fingerprint()`.
64pub fn scoring_fingerprint() -> &'static str {
65    "id=4.0|title=3.0|path=2.0|heading=1.5|body=1.0|tags=2.5|k1=1.2|b=0.75|rrf=60|graph=0.5"
66}
67
68// ---------------------------------------------------------------------------
69// Corpus hash (spec §6)
70// ---------------------------------------------------------------------------
71
72/// SHA-256 of a file's bytes; unreadable files hash a stable sentinel.
73pub fn content_hash(path: &Path) -> String {
74    match fs::read(path) {
75        Ok(bytes) => crate::sha256::hexdigest(&bytes),
76        Err(_) => crate::sha256::hexdigest(b"\x00rac-unreadable-artifact"),
77    }
78}
79
80/// `corpus_content_hash(directory, recursive)` — fold of the sorted
81/// `(rel_posix, content_hash)` pairs.
82pub fn corpus_content_hash(directory: &str, recursive: bool) -> String {
83    let mut hasher = crate::sha256::Sha256::new();
84    for entry in find_markdown_files(directory, recursive) {
85        let rel = entry.components.join("/");
86        hasher.update(rel.as_bytes());
87        hasher.update(b"\0");
88        hasher.update(content_hash(&entry.abs).as_bytes());
89        hasher.update(b"\0");
90    }
91    hasher.hexdigest()
92}
93
94// ---------------------------------------------------------------------------
95// Store layout paths
96// ---------------------------------------------------------------------------
97
98pub fn store_root(cache_dir: &Path) -> PathBuf {
99    cache_dir.join(STORE_DIRNAME).join(STORE_LAYOUT_VERSION)
100}
101
102pub fn store_dir(cache_dir: &Path, corpus_hash: &str) -> PathBuf {
103    store_root(cache_dir).join(corpus_hash)
104}
105
106// ---------------------------------------------------------------------------
107// Writer — one DerivedIndex -> a directory of segment files, atomically.
108// ---------------------------------------------------------------------------
109
110fn encode_segments(
111    corpus_hash: &str,
112    bundle_version: &str,
113    derived: &DerivedIndex,
114) -> Result<Vec<(&'static str, Vec<u8>)>, IndexFormatError> {
115    let entries = &derived.index_entries;
116    let field_tokens = &derived.field_tokens;
117
118    // Global vocabulary -> sorted term dictionary -> term id (code-point
119    // order — BTreeMap keys iterate sorted).
120    let mut term_id: BTreeMap<&str, u32> = BTreeMap::new();
121    for fields in field_tokens {
122        for name in FIELDS {
123            for token in fields.get(name) {
124                term_id.insert(token.as_str(), 0);
125            }
126        }
127    }
128    let termdict: Vec<&str> = term_id.keys().copied().collect();
129    for (i, term) in termdict.iter().enumerate() {
130        *term_id.get_mut(*term).expect("present") = i as u32;
131    }
132
133    let mut length_sums = [0u64; 6];
134    let mut entry_rows: Vec<Vec<u8>> = Vec::with_capacity(entries.len());
135    let mut section_rows: Vec<Vec<u8>> = Vec::with_capacity(entries.len());
136    let mut token_rows: Vec<Vec<u8>> = Vec::with_capacity(entries.len());
137    let mut postings_lists: Vec<Vec<u32>> = vec![Vec::new(); termdict.len()];
138    // Casefolded identifier -> ascending docids (consecutive-dup guarded).
139    let mut alias_docids: BTreeMap<String, Vec<u32>> = BTreeMap::new();
140
141    for (docid, entry) in entries.iter().enumerate() {
142        let docid = docid as u32;
143        let fields = &field_tokens[docid as usize];
144        let lengths: Vec<u64> = FIELDS
145            .iter()
146            .map(|name| fields.get(name).len() as u64)
147            .collect();
148        for (i, value) in lengths.iter().enumerate() {
149            length_sums[i] += value;
150        }
151
152        for alias in &entry.aliases {
153            let docids = alias_docids.entry(py_casefold(alias)).or_default();
154            if docids.last() != Some(&docid) {
155                docids.push(docid);
156            }
157        }
158
159        let mut row = Writer::new();
160        row.text(&entry.id)?;
161        row.text(&entry.artifact_type)?;
162        row.opt_text(entry.title.as_deref())?;
163        row.text(&entry.path)?;
164        row.text_list(&entry.aliases)?;
165        row.text_list(&entry.tags)?;
166        row.u32(entry.inbound_count.max(0) as u64)?;
167        for value in &lengths {
168            row.u32(*value)?;
169        }
170        entry_rows.push(row.payload());
171
172        let mut sec = Writer::new();
173        sec.u32(entry.search_sections.len() as u64)?;
174        for section in &entry.search_sections {
175            sec.text(&section.heading)?;
176            sec.text_list(&section.lines)?;
177        }
178        section_rows.push(sec.payload());
179
180        let mut doc_term_ids: std::collections::BTreeSet<u32> = std::collections::BTreeSet::new();
181        let mut tok = Writer::new();
182        for name in FIELDS {
183            let ids: Vec<u32> = fields
184                .get(name)
185                .iter()
186                .map(|token| term_id[token.as_str()])
187                .collect();
188            tok.u32_list(&ids)?;
189            doc_term_ids.extend(ids);
190        }
191        token_rows.push(tok.payload());
192        for tid in doc_term_ids {
193            postings_lists[tid as usize].push(docid);
194        }
195    }
196
197    let n_entries = entries.len() as u64;
198    let n_terms = termdict.len() as u64;
199
200    let mut out: Vec<(&'static str, Vec<u8>)> = Vec::with_capacity(12);
201    out.push((SEG_ENTRIES, encode_segment(&write_indexed(&entry_rows)?)));
202    drop(entry_rows);
203    out.push((SEG_SECTIONS, encode_segment(&write_indexed(&section_rows)?)));
204    drop(section_rows);
205    out.push((SEG_TOKENS, encode_segment(&write_indexed(&token_rows)?)));
206    drop(token_rows);
207
208    let postings_rows: Vec<Vec<u8>> = postings_lists
209        .iter()
210        .map(|docids| {
211            let mut w = Writer::new();
212            w.u32_list(docids)?;
213            Ok(w.payload())
214        })
215        .collect::<Result<_, IndexFormatError>>()?;
216    drop(postings_lists);
217    out.push((SEG_POSTINGS, encode_segment(&write_indexed(&postings_rows)?)));
218    drop(postings_rows);
219
220    let termdict_rows: Vec<Vec<u8>> = termdict
221        .iter()
222        .map(|term| {
223            let mut w = Writer::new();
224            w.text(term)?;
225            Ok(w.payload())
226        })
227        .collect::<Result<_, IndexFormatError>>()?;
228    out.push((SEG_TERMDICT, encode_segment(&write_indexed(&termdict_rows)?)));
229    drop(termdict_rows);
230
231    let aliasmap_rows: Vec<Vec<u8>> = alias_docids
232        .iter()
233        .map(|(key, docids)| {
234            let mut w = Writer::new();
235            w.text(key)?;
236            w.u32_list(docids)?;
237            Ok(w.payload())
238        })
239        .collect::<Result<_, IndexFormatError>>()?;
240    drop(alias_docids);
241    out.push((SEG_ALIASMAP, encode_segment(&write_indexed(&aliasmap_rows)?)));
242    drop(aliasmap_rows);
243
244    // Path map: rows sorted by path STRING (docids index walk order).
245    let mut path_pairs: Vec<(&str, u32)> = entries
246        .iter()
247        .enumerate()
248        .map(|(docid, entry)| (entry.path.as_str(), docid as u32))
249        .collect();
250    path_pairs.sort();
251    let pathmap_rows: Vec<Vec<u8>> = path_pairs
252        .iter()
253        .map(|(path, docid)| {
254            let mut w = Writer::new();
255            w.text(path)?;
256            w.u32(u64::from(*docid))?;
257            Ok(w.payload())
258        })
259        .collect::<Result<_, IndexFormatError>>()?;
260    out.push((SEG_PATHMAP, encode_segment(&write_indexed(&pathmap_rows)?)));
261    drop(pathmap_rows);
262
263    let mut relationships = Writer::new();
264    relationships.u32(derived.relationships.len() as u64)?;
265    for rel in &derived.relationships {
266        relationships.text(&rel.source_path)?;
267        relationships.text(&rel.relationship)?;
268        relationships.text(&rel.target)?;
269        relationships.opt_text(rel.resolved_path.as_deref())?;
270        relationships.opt_text(rel.issue.as_deref())?;
271    }
272    out.push((SEG_RELATIONSHIPS, encode_segment(&relationships.payload())));
273
274    let mut live = Writer::new();
275    live.text_list(&derived.live_decision_paths)?;
276    out.push((SEG_LIVE, encode_segment(&live.payload())));
277
278    let mut scope = Writer::new();
279    scope.u32(derived.scope_rows.len() as u64)?;
280    for row in &derived.scope_rows {
281        scope.text(&row.id)?;
282        scope.text(&row.title)?;
283        scope.text(&row.status)?;
284        scope.text(&row.path)?;
285        scope.text_list(&row.scope_entries)?;
286    }
287    out.push((SEG_SCOPE, encode_segment(&scope.payload())));
288
289    // The one JSON-in-binary blob: `json.dumps(summary, ensure_ascii=False)`.
290    let mut portfolio = Writer::new();
291    portfolio.text(&crate::pyjson::dumps_compact(&derived.portfolio_summary))?;
292    out.push((SEG_PORTFOLIO, encode_segment(&portfolio.payload())));
293
294    let mut header = Writer::new();
295    header.text(corpus_hash)?;
296    header.text(bundle_version)?;
297    header.text(scoring_fingerprint())?;
298    header.u32(n_entries)?;
299    for value in length_sums {
300        header.u32(value)?;
301    }
302    header.u32(n_terms)?;
303    out.push((SEG_HEADER, encode_segment(&header.payload())));
304
305    Ok(out)
306}
307
308fn write_file_synced_measured(
309    path: &Path,
310    payload: &[u8],
311    timing: bool,
312    write_duration: &mut std::time::Duration,
313    sync_duration: &mut std::time::Duration,
314) -> std::io::Result<()> {
315    let write_started = timing.then(std::time::Instant::now);
316    let mut file = fs::OpenOptions::new()
317        .write(true)
318        .create(true)
319        .truncate(true)
320        .open(path)?;
321    file.write_all(payload)?;
322    if let Some(started) = write_started {
323        *write_duration += started.elapsed();
324    }
325    let sync_started = timing.then(std::time::Instant::now);
326    let result = file.sync_all();
327    if let Some(started) = sync_started {
328        *sync_duration += started.elapsed();
329    }
330    result
331}
332
333fn write_file_synced(path: &Path, payload: &[u8]) -> std::io::Result<()> {
334    let mut write_duration = std::time::Duration::ZERO;
335    let mut sync_duration = std::time::Duration::ZERO;
336    write_file_synced_measured(
337        path,
338        payload,
339        false,
340        &mut write_duration,
341        &mut sync_duration,
342    )
343}
344
345fn fsync_dir(path: &Path) {
346    if let Ok(dir) = fs::File::open(path) {
347        let _ = dir.sync_all();
348    }
349}
350
351fn remove_tree(path: &Path) {
352    let _ = fs::remove_dir_all(path);
353}
354
355/// Temp-dir suffix entropy: pid plus a few clock-derived bytes (never mapped
356/// into any payload — mirrors the oracle's pid+urandom temp names).
357fn temp_suffix() -> String {
358    let nanos = std::time::SystemTime::now()
359        .duration_since(std::time::UNIX_EPOCH)
360        .map(|d| d.subsec_nanos())
361        .unwrap_or(0);
362    format!("{}-{:08x}", std::process::id(), nanos)
363}
364
365/// Write the store for `corpus_hash` atomically; return whether it landed.
366pub fn write_store(
367    cache_dir: &Path,
368    corpus_hash: &str,
369    bundle_version: &str,
370    derived: &DerivedIndex,
371) -> bool {
372    let root = store_root(cache_dir);
373    let final_dir = root.join(corpus_hash);
374    if final_dir.is_dir() {
375        // Content addressing: a same-hash store is byte-equivalent within one
376        // format. Probe readability with the full open; replace when bad.
377        if MmapIndexReader::open(&final_dir, corpus_hash, bundle_version).is_ok() {
378            return true;
379        }
380        remove_tree(&final_dir);
381    }
382    let encode_started = crate::timing::start();
383    let Ok(segments) = encode_segments(corpus_hash, bundle_version, derived) else {
384        crate::timing::emit_since("store.encode", encode_started, &[("success", 0)]);
385        return false;
386    };
387    crate::timing::emit_since(
388        "store.encode",
389        encode_started,
390        &[("success", 1), ("segments", segments.len() as u64)],
391    );
392    let tmp = root.join(format!(".{corpus_hash}.tmp-{}", temp_suffix()));
393    let timing = crate::timing::enabled();
394    let mut write_duration = std::time::Duration::ZERO;
395    let mut sync_duration = std::time::Duration::ZERO;
396    let mut write_all = || -> std::io::Result<()> {
397        fs::create_dir_all(&tmp)?;
398        for (name, payload) in &segments {
399            write_file_synced_measured(
400                &tmp.join(name),
401                payload,
402                timing,
403                &mut write_duration,
404                &mut sync_duration,
405            )?;
406        }
407        let sync_started = timing.then(std::time::Instant::now);
408        fsync_dir(&tmp);
409        if let Some(started) = sync_started {
410            sync_duration += started.elapsed();
411        }
412        Ok(())
413    };
414    let write_result = write_all();
415    crate::timing::emit(
416        "store.segment_write",
417        write_duration,
418        &[("segments", segments.len() as u64)],
419    );
420    crate::timing::emit(
421        "store.segment_sync",
422        sync_duration,
423        &[("segments", segments.len() as u64)],
424    );
425    if write_result.is_err() {
426        remove_tree(&tmp);
427        return false;
428    }
429    match fs::rename(&tmp, &final_dir) {
430        Ok(()) => true,
431        Err(_) => {
432            // Populated by a concurrent writer (identical content), or the
433            // rename failed: discard and report the store's presence honestly.
434            remove_tree(&tmp);
435            final_dir.is_dir()
436        }
437    }
438}
439
440/// Best-effort removal of a store directory (used to clear a corrupt one).
441pub fn remove_store(cache_dir: &Path, corpus_hash: &str) {
442    remove_tree(&store_dir(cache_dir, corpus_hash));
443}
444
445// ---------------------------------------------------------------------------
446// Reader — mmap the segments, validate on open, point-access the rows.
447// ---------------------------------------------------------------------------
448
449/// Memory-mapped reader over one corpus-hash store directory (the base).
450pub struct MmapIndexReader {
451    maps: Vec<Mmap>, // ALL_SEGMENTS order; payload = &map[18..] after gates
452    pub doc_count: u32,
453    pub field_length_sums: [u64; 6],
454    pub term_count: u32,
455}
456
457fn seg_index(name: &str) -> usize {
458    ALL_SEGMENTS.iter().position(|s| *s == name).expect("known segment")
459}
460
461impl MmapIndexReader {
462    pub fn open(
463        directory: &Path,
464        corpus_hash: &str,
465        bundle_version: &str,
466    ) -> Result<Self, IndexFormatError> {
467        let mut maps = Vec::with_capacity(ALL_SEGMENTS.len());
468        for name in ALL_SEGMENTS {
469            let path = directory.join(name);
470            let file = fs::File::open(&path)
471                .map_err(|e| IndexFormatError(format!("cannot open {name}: {e}")))?;
472            let len = file
473                .metadata()
474                .map_err(|e| IndexFormatError(format!("cannot stat {name}: {e}")))?
475                .len();
476            if len == 0 {
477                return Err(IndexFormatError(format!("empty segment: {name}")));
478            }
479            let map = unsafe { Mmap::map(&file) }
480                .map_err(|e| IndexFormatError(format!("cannot map {name}: {e}")))?;
481            segment_payload(&map)?; // framing gates: magic, version, length
482            maps.push(map);
483        }
484        let mut reader = Self {
485            maps,
486            doc_count: 0,
487            field_length_sums: [0; 6],
488            term_count: 0,
489        };
490        reader.read_header(corpus_hash, bundle_version)?;
491        Ok(reader)
492    }
493
494    fn payload(&self, name: &str) -> &[u8] {
495        segment_payload(&self.maps[seg_index(name)]).expect("validated on open")
496    }
497
498    fn read_header(
499        &mut self,
500        corpus_hash: &str,
501        bundle_version: &str,
502    ) -> Result<(), IndexFormatError> {
503        let payload = segment_payload(&self.maps[seg_index(SEG_HEADER)])?;
504        let mut reader = Reader::new(payload);
505        let stored_hash = reader.text()?;
506        let stored_bundle = reader.text()?;
507        let stored_fingerprint = reader.text()?;
508        let doc_count = reader.u32()?;
509        let mut sums = [0u64; 6];
510        for slot in &mut sums {
511            *slot = u64::from(reader.u32()?);
512        }
513        let term_count = reader.u32()?;
514        if stored_hash != corpus_hash {
515            return Err(IndexFormatError("store corpus-hash mismatch".into()));
516        }
517        if stored_bundle != bundle_version {
518            return Err(IndexFormatError("store bundle-version mismatch".into()));
519        }
520        if stored_fingerprint != scoring_fingerprint() {
521            return Err(IndexFormatError("store scoring-constant mismatch".into()));
522        }
523        self.doc_count = doc_count;
524        self.field_length_sums = sums;
525        self.term_count = term_count;
526        Ok(())
527    }
528
529    fn indexed(&self, name: &str) -> Result<IndexedSegment<'_>, IndexFormatError> {
530        IndexedSegment::new(self.payload(name))
531    }
532
533    /// The lightweight identity row (no sections, no inbound).
534    pub fn identity_entry(&self, docid: u32) -> Result<IndexEntry, IndexFormatError> {
535        let mut reader = self.indexed(SEG_ENTRIES)?.row(docid)?;
536        let id = reader.text()?;
537        let artifact_type = reader.text()?;
538        let title = reader.opt_text()?;
539        let path = reader.text()?;
540        let aliases = reader.text_list()?;
541        let tags = reader.text_list()?;
542        Ok(IndexEntry {
543            id,
544            artifact_type,
545            title,
546            path,
547            aliases,
548            search_sections: Vec::new(),
549            inbound_count: 0,
550            tags,
551        })
552    }
553
554    /// The full index row: identity plus searchable sections and inbound.
555    pub fn full_entry(&self, docid: u32) -> Result<IndexEntry, IndexFormatError> {
556        let mut reader = self.indexed(SEG_ENTRIES)?.row(docid)?;
557        let id = reader.text()?;
558        let artifact_type = reader.text()?;
559        let title = reader.opt_text()?;
560        let path = reader.text()?;
561        let aliases = reader.text_list()?;
562        let tags = reader.text_list()?;
563        let inbound = reader.u32()?;
564        let sections = self.read_sections(docid)?;
565        Ok(IndexEntry {
566            id,
567            artifact_type,
568            title,
569            path,
570            aliases,
571            search_sections: sections,
572            inbound_count: i64::from(inbound),
573            tags,
574        })
575    }
576
577    fn read_sections(
578        &self,
579        docid: u32,
580    ) -> Result<Vec<crate::markdown::SearchSection>, IndexFormatError> {
581        let mut reader = self.indexed(SEG_SECTIONS)?.row(docid)?;
582        let count = reader.u32()?;
583        let mut sections = Vec::with_capacity(count.min(1 << 16) as usize);
584        for _ in 0..count {
585            sections.push(crate::markdown::SearchSection {
586                heading: reader.text()?,
587                lines: reader.text_list()?,
588            });
589        }
590        Ok(sections)
591    }
592
593    pub fn entry_path(&self, docid: u32) -> Result<String, IndexFormatError> {
594        let mut reader = self.indexed(SEG_ENTRIES)?.row(docid)?;
595        reader.text()?; // id
596        reader.text()?; // type
597        reader.opt_text()?; // title
598        reader.text() // path
599    }
600
601    /// The first non-empty line of this document's `## Status` section.
602    /// Status is already present in the searchable-section segment, so
603    /// grounding can determine lifecycle without reopening and reparsing the
604    /// Markdown file.
605    pub fn entry_status(&self, docid: u32) -> Result<String, IndexFormatError> {
606        let mut reader = self.indexed(SEG_SECTIONS)?.row(docid)?;
607        let count = reader.u32()?;
608        for _ in 0..count {
609            let is_status = {
610                let heading = reader.text_ref()?;
611                crate::pycompat::py_casefold(crate::pycompat::py_strip(heading)) == "status"
612            };
613            let line_count = reader.u32()?;
614            if is_status {
615                let mut status = String::new();
616                for _ in 0..line_count {
617                    let line = crate::pycompat::py_strip(reader.text_ref()?);
618                    if status.is_empty() && !line.is_empty() {
619                        status = line.to_string();
620                    }
621                }
622                return Ok(status);
623            }
624            for _ in 0..line_count {
625                reader.text_ref()?;
626            }
627        }
628        Ok(String::new())
629    }
630
631    /// Per-field token counts for one doc, FIELDS order.
632    pub fn field_lengths(&self, docid: u32) -> Result<[u64; 6], IndexFormatError> {
633        let mut reader = self.indexed(SEG_ENTRIES)?.row(docid)?;
634        reader.text()?;
635        reader.text()?;
636        reader.opt_text()?;
637        reader.text()?;
638        reader.text_list()?; // aliases
639        reader.text_list()?; // tags
640        reader.u32()?; // inbound
641        let mut lengths = [0u64; 6];
642        for slot in &mut lengths {
643            *slot = u64::from(reader.u32()?);
644        }
645        Ok(lengths)
646    }
647
648    /// The six forward token-id sequences of one doc, FIELDS order.
649    pub fn forward_token_ids(&self, docid: u32) -> Result<[Vec<u32>; 6], IndexFormatError> {
650        let mut reader = self.indexed(SEG_TOKENS)?.row(docid)?;
651        Ok([
652            reader.u32_list()?,
653            reader.u32_list()?,
654            reader.u32_list()?,
655            reader.u32_list()?,
656            reader.u32_list()?,
657            reader.u32_list()?,
658        ])
659    }
660
661    pub fn term_at(&self, term_id: u32) -> Result<String, IndexFormatError> {
662        self.indexed(SEG_TERMDICT)?.row(term_id)?.text()
663    }
664
665    /// Reconstruct one doc's per-field token vectors in document order.
666    pub fn field_tokens(&self, docid: u32) -> Result<FieldTokens, IndexFormatError> {
667        let ids = self.forward_token_ids(docid)?;
668        let terms = self.indexed(SEG_TERMDICT)?;
669        let resolve = |ids: &[u32]| -> Result<Vec<String>, IndexFormatError> {
670            ids.iter().map(|&i| terms.row(i)?.text()).collect()
671        };
672        Ok(FieldTokens {
673            id: resolve(&ids[0])?,
674            title: resolve(&ids[1])?,
675            path: resolve(&ids[2])?,
676            heading: resolve(&ids[3])?,
677            body: resolve(&ids[4])?,
678            tags: resolve(&ids[5])?,
679        })
680    }
681
682    fn bisect_left(&self, target: &str) -> Result<u32, IndexFormatError> {
683        let segment = self.indexed(SEG_TERMDICT)?;
684        let (mut lo, mut hi) = (0u32, segment.count());
685        while lo < hi {
686            let mid = (lo + hi) / 2;
687            if segment.row(mid)?.text()?.as_str() < target {
688                lo = mid + 1;
689            } else {
690                hi = mid;
691            }
692        }
693        Ok(lo)
694    }
695
696    /// The `[lo, hi)` term-id range of every indexed term `term` prefixes.
697    pub fn prefix_range(&self, term: &str) -> Result<(u32, u32), IndexFormatError> {
698        if term.is_empty() {
699            return Ok((0, 0));
700        }
701        let lo = self.bisect_left(term)?;
702        // Successor: last char's code point incremented (Python chr(ord+1)).
703        let mut chars: Vec<char> = term.chars().collect();
704        let last = chars.pop().expect("non-empty");
705        let successor_last = char::from_u32(last as u32 + 1);
706        let hi = match successor_last {
707            Some(c) => {
708                chars.push(c);
709                let successor: String = chars.into_iter().collect();
710                self.bisect_left(&successor)?
711            }
712            None => self.indexed(SEG_TERMDICT)?.count(),
713        };
714        Ok((lo, hi))
715    }
716
717    /// The ascending docids that hold `term_id` in any field.
718    pub fn postings(&self, term_id: u32) -> Result<Vec<u32>, IndexFormatError> {
719        self.indexed(SEG_POSTINGS)?.row(term_id)?.u32_list()
720    }
721
722    /// Distinct docids matching `term` under the prefix predicate (ADR-037).
723    pub fn prefix_docids(
724        &self,
725        term: &str,
726    ) -> Result<std::collections::BTreeSet<u32>, IndexFormatError> {
727        let (lo, hi) = self.prefix_range(term)?;
728        let mut result = std::collections::BTreeSet::new();
729        for term_id in lo..hi {
730            result.extend(self.postings(term_id)?);
731        }
732        Ok(result)
733    }
734
735    /// The ascending docids whose identity set holds `wanted` (already
736    /// casefolded by the caller) — binary search over the alias map.
737    pub fn alias_docids(&self, wanted: &str) -> Result<Vec<u32>, IndexFormatError> {
738        let segment = self.indexed(SEG_ALIASMAP)?;
739        let (mut lo, mut hi) = (0u32, segment.count());
740        while lo < hi {
741            let mid = (lo + hi) / 2;
742            let mut reader = segment.row(mid)?;
743            let key = reader.text()?;
744            match key.as_str().cmp(wanted) {
745                std::cmp::Ordering::Less => lo = mid + 1,
746                std::cmp::Ordering::Greater => hi = mid,
747                std::cmp::Ordering::Equal => return reader.u32_list(),
748            }
749        }
750        Ok(Vec::new())
751    }
752
753    /// The docid whose stored path equals `path`, or None — binary search.
754    pub fn docid_for_path(&self, path: &str) -> Result<Option<u32>, IndexFormatError> {
755        let segment = self.indexed(SEG_PATHMAP)?;
756        let (mut lo, mut hi) = (0u32, segment.count());
757        while lo < hi {
758            let mid = (lo + hi) / 2;
759            let mut reader = segment.row(mid)?;
760            let key = reader.text()?;
761            match key.as_str().cmp(path) {
762                std::cmp::Ordering::Less => lo = mid + 1,
763                std::cmp::Ordering::Greater => hi = mid,
764                std::cmp::Ordering::Equal => return Ok(Some(reader.u32()?)),
765            }
766        }
767        Ok(None)
768    }
769
770    pub fn relationships(&self) -> Result<Vec<Relationship>, IndexFormatError> {
771        let mut reader = Reader::new(self.payload(SEG_RELATIONSHIPS));
772        let count = reader.u32()?;
773        let mut result = Vec::with_capacity(count.min(1 << 20) as usize);
774        for _ in 0..count {
775            result.push(Relationship {
776                source_path: reader.text()?,
777                relationship: reader.text()?,
778                target: reader.text()?,
779                resolved_path: reader.opt_text()?,
780                issue: reader.opt_text()?,
781            });
782        }
783        Ok(result)
784    }
785
786    pub fn live_decision_paths(&self) -> Result<Vec<String>, IndexFormatError> {
787        Reader::new(self.payload(SEG_LIVE)).text_list()
788    }
789
790    pub fn scope_rows(&self) -> Result<Vec<crate::retrieve::ScopeRow>, IndexFormatError> {
791        let mut reader = Reader::new(self.payload(SEG_SCOPE));
792        let count = reader.u32()?;
793        let mut rows = Vec::with_capacity(count.min(1 << 20) as usize);
794        for _ in 0..count {
795            rows.push(crate::retrieve::ScopeRow {
796                id: reader.text()?,
797                title: reader.text()?,
798                status: reader.text()?,
799                path: reader.text()?,
800                scope_entries: reader.text_list()?,
801            });
802        }
803        Ok(rows)
804    }
805
806    /// The portfolio summary parsed back from its stored JSON text.
807    pub fn portfolio_summary(&self) -> Result<Value, IndexFormatError> {
808        let text = Reader::new(self.payload(SEG_PORTFOLIO)).text()?;
809        serde_json::from_str(&text)
810            .map_err(|e| IndexFormatError(format!("portfolio segment is not valid JSON: {e}")))
811    }
812}
813
814/// Open the store for `corpus_hash`, or `None` on any miss (never fatal).
815pub fn open_store(
816    cache_dir: &Path,
817    corpus_hash: &str,
818    bundle_version: &str,
819) -> Option<MmapIndexReader> {
820    let directory = store_dir(cache_dir, corpus_hash);
821    if !directory.is_dir() {
822        return None;
823    }
824    MmapIndexReader::open(&directory, corpus_hash, bundle_version).ok()
825}
826
827// ---------------------------------------------------------------------------
828// Per-file validation-result store (`.vseg`, ADR-106) — codec only here;
829// the incremental-validate seam consumes it (INDEX-PLAN B4).
830// ---------------------------------------------------------------------------
831
832pub const VALIDATE_STORE_DIRNAME: &str = "validate";
833pub const VALIDATE_LAYOUT_VERSION: &str = "v1";
834
835/// One file's cached validation result plus its freshness stat proxy.
836#[derive(Debug, Clone, PartialEq)]
837pub struct ValidationCacheRow {
838    pub size: u64,
839    pub mtime_ns: u64,
840    pub content_hash: String,
841    pub artifact_type: String,
842    pub status: String,
843    pub issues: Vec<CachedIssue>,
844}
845
846/// A path-free cached issue (`Issue` without location context).
847#[derive(Debug, Clone, PartialEq)]
848pub struct CachedIssue {
849    pub severity: String,
850    pub code: String,
851    pub message: String,
852    pub line: Option<u32>,
853}
854
855pub fn validate_store_root(cache_dir: &Path) -> PathBuf {
856    cache_dir
857        .join(VALIDATE_STORE_DIRNAME)
858        .join(VALIDATE_LAYOUT_VERSION)
859}
860
861fn validate_store_path(cache_dir: &Path, root_key: &str) -> PathBuf {
862    validate_store_root(cache_dir).join(format!("{root_key}.vseg"))
863}
864
865/// Encode the `.vseg` payload — rows in insertion order.
866pub fn encode_validation_store(
867    config_hash: &str,
868    rows: &[(String, ValidationCacheRow)],
869) -> Result<Vec<u8>, IndexFormatError> {
870    let mut writer = Writer::new();
871    writer.text(config_hash)?;
872    writer.u32(rows.len() as u64)?;
873    for (rel, row) in rows {
874        writer.text(rel)?;
875        writer.u64(row.size);
876        writer.u64(row.mtime_ns);
877        writer.text(&row.content_hash)?;
878        writer.text(&row.artifact_type)?;
879        writer.text(&row.status)?;
880        writer.u32(row.issues.len() as u64)?;
881        for issue in &row.issues {
882            writer.text(&issue.severity)?;
883            writer.text(&issue.code)?;
884            writer.text(&issue.message)?;
885            match issue.line {
886                None => {
887                    writer.u32(0)?;
888                    writer.u32(0)?;
889                }
890                Some(line) => {
891                    writer.u32(1)?;
892                    writer.u32(u64::from(line))?;
893                }
894            }
895        }
896    }
897    Ok(encode_segment(&writer.payload()))
898}
899
900/// Decode a `.vseg` payload; `None` on config mismatch (a miss).
901pub fn decode_validation_store(
902    payload: &[u8],
903    config_hash: &str,
904) -> Result<Option<Vec<(String, ValidationCacheRow)>>, IndexFormatError> {
905    let mut reader = Reader::new(payload);
906    if reader.text()? != config_hash {
907        return Ok(None);
908    }
909    let count = reader.u32()?;
910    let mut rows = Vec::with_capacity(count.min(1 << 20) as usize);
911    for _ in 0..count {
912        let rel = reader.text()?;
913        let size = reader.u64()?;
914        let mtime_ns = reader.u64()?;
915        let content_hash = reader.text()?;
916        let artifact_type = reader.text()?;
917        let status = reader.text()?;
918        let issue_count = reader.u32()?;
919        let mut issues = Vec::with_capacity(issue_count.min(1 << 16) as usize);
920        for _ in 0..issue_count {
921            let severity = reader.text()?;
922            let code = reader.text()?;
923            let message = reader.text()?;
924            let has_line = reader.u32()?;
925            let line_value = reader.u32()?;
926            issues.push(CachedIssue {
927                severity,
928                code,
929                message,
930                line: if has_line != 0 { Some(line_value) } else { None },
931            });
932        }
933        rows.push((
934            rel,
935            ValidationCacheRow {
936                size,
937                mtime_ns,
938                content_hash,
939                artifact_type,
940                status,
941                issues,
942            },
943        ));
944    }
945    Ok(Some(rows))
946}
947
948/// Load the per-file validation rows for a corpus root, or `None` on a miss.
949pub fn open_validation_store(
950    cache_dir: &Path,
951    root_key: &str,
952    config_hash: &str,
953) -> Option<Vec<(String, ValidationCacheRow)>> {
954    let data = fs::read(validate_store_path(cache_dir, root_key)).ok()?;
955    let payload = segment_payload(&data).ok()?;
956    decode_validation_store(payload, config_hash).ok()?
957}
958
959/// Write the per-file validation rows atomically; return whether it landed.
960pub fn write_validation_store(
961    cache_dir: &Path,
962    root_key: &str,
963    config_hash: &str,
964    rows: &[(String, ValidationCacheRow)],
965) -> bool {
966    let Ok(payload) = encode_validation_store(config_hash, rows) else {
967        return false;
968    };
969    atomic_write(
970        &validate_store_root(cache_dir),
971        root_key,
972        &validate_store_path(cache_dir, root_key),
973        &payload,
974    )
975}
976
977// ---------------------------------------------------------------------------
978// Per-root freshness-manifest store (`.fseg`, ADR-112)
979// ---------------------------------------------------------------------------
980
981pub const MANIFEST_DIRNAME: &str = "manifest";
982pub const MANIFEST_LAYOUT_VERSION: &str = "v1";
983const MANIFEST_FORMAT_VERSION: u32 = 1;
984
985/// The freshness proxy for one file: content hash plus the stat pair.
986#[derive(Debug, Clone, PartialEq)]
987pub struct FileState {
988    pub content_hash: String,
989    pub size: u64,
990    pub mtime_ns: u64,
991}
992
993pub fn manifest_store_root(cache_dir: &Path) -> PathBuf {
994    cache_dir.join(MANIFEST_DIRNAME).join(MANIFEST_LAYOUT_VERSION)
995}
996
997/// `Path(directory).resolve()` — absolutise against the cwd and normalise,
998/// canonicalising the longest existing prefix (Python resolves symlinks for
999/// the part of the path that exists and keeps the nonexistent tail).
1000pub fn py_resolve(directory: &str) -> PathBuf {
1001    let path = Path::new(directory);
1002    let absolute = if path.is_absolute() {
1003        path.to_path_buf()
1004    } else {
1005        std::env::current_dir()
1006            .unwrap_or_else(|_| PathBuf::from("/"))
1007            .join(path)
1008    };
1009    // Lexically normalise `.` and `..`, then canonicalise the longest
1010    // existing prefix so symlinked ancestors resolve as Python's do.
1011    let mut parts: Vec<std::ffi::OsString> = Vec::new();
1012    for comp in absolute.components() {
1013        use std::path::Component;
1014        match comp {
1015            Component::CurDir => {}
1016            Component::ParentDir => {
1017                parts.pop();
1018            }
1019            Component::RootDir | Component::Prefix(_) => {}
1020            Component::Normal(p) => parts.push(p.to_os_string()),
1021        }
1022    }
1023    let mut resolved = PathBuf::from("/");
1024    let mut tail: Vec<std::ffi::OsString> = Vec::new();
1025    let mut existing = PathBuf::from("/");
1026    for (i, part) in parts.iter().enumerate() {
1027        existing.push(part);
1028        if tail.is_empty() && existing.exists() {
1029            continue;
1030        }
1031        if tail.is_empty() {
1032            // first nonexistent component: canonicalise what exists so far
1033            let prefix = {
1034                let mut p = PathBuf::from("/");
1035                for q in &parts[..i] {
1036                    p.push(q);
1037                }
1038                p
1039            };
1040            resolved = fs::canonicalize(&prefix).unwrap_or(prefix);
1041        }
1042        tail.push(part.clone());
1043    }
1044    if tail.is_empty() {
1045        fs::canonicalize(&existing).unwrap_or(existing)
1046    } else {
1047        for part in tail {
1048            resolved.push(part);
1049        }
1050        resolved
1051    }
1052}
1053
1054/// A stable key for one corpus root in one recursion mode.
1055pub fn manifest_root_key(directory: &str, recursive: bool) -> String {
1056    let mode = if recursive { "recursive" } else { "top-level" };
1057    let seed = format!("{}\0{mode}", py_resolve(directory).display());
1058    crate::sha256::hexdigest(seed.as_bytes())
1059}
1060
1061fn manifest_store_path(cache_dir: &Path, root_key: &str) -> PathBuf {
1062    manifest_store_root(cache_dir).join(format!("{root_key}.fseg"))
1063}
1064
1065/// Encode the `.fseg` manifest — rows in insertion (scan) order.
1066pub fn encode_freshness_manifest(
1067    manifest: &[(String, FileState)],
1068) -> Result<Vec<u8>, IndexFormatError> {
1069    let mut writer = Writer::new();
1070    writer.u32(u64::from(MANIFEST_FORMAT_VERSION))?;
1071    writer.u32(manifest.len() as u64)?;
1072    for (rel, state) in manifest {
1073        writer.text(rel)?;
1074        writer.u64(state.size);
1075        writer.u64(state.mtime_ns);
1076        writer.text(&state.content_hash)?;
1077    }
1078    Ok(encode_segment(&writer.payload()))
1079}
1080
1081/// Load the persisted stat manifest for a corpus root, or `None` on a miss.
1082pub fn open_freshness_manifest(
1083    cache_dir: &Path,
1084    root_key: &str,
1085) -> Option<Vec<(String, FileState)>> {
1086    let data = fs::read(manifest_store_path(cache_dir, root_key)).ok()?;
1087    let payload = segment_payload(&data).ok()?;
1088    let mut reader = Reader::new(payload);
1089    if reader.u32().ok()? != MANIFEST_FORMAT_VERSION {
1090        return None;
1091    }
1092    let count = reader.u32().ok()?;
1093    let mut manifest = Vec::with_capacity(count.min(1 << 20) as usize);
1094    for _ in 0..count {
1095        let rel = reader.text().ok()?;
1096        let size = reader.u64().ok()?;
1097        let mtime_ns = reader.u64().ok()?;
1098        let content_hash = reader.text().ok()?;
1099        manifest.push((
1100            rel,
1101            FileState {
1102                content_hash,
1103                size,
1104                mtime_ns,
1105            },
1106        ));
1107    }
1108    Some(manifest)
1109}
1110
1111/// Write the stat manifest atomically; return whether it landed.
1112pub fn write_freshness_manifest(
1113    cache_dir: &Path,
1114    root_key: &str,
1115    manifest: &[(String, FileState)],
1116) -> bool {
1117    let Ok(payload) = encode_freshness_manifest(manifest) else {
1118        return false;
1119    };
1120    atomic_write(
1121        &manifest_store_root(cache_dir),
1122        root_key,
1123        &manifest_store_path(cache_dir, root_key),
1124        &payload,
1125    )
1126}
1127
1128/// Shared temp-file + rename atomic write for the single-file stores.
1129fn atomic_write(root: &Path, key: &str, target: &Path, payload: &[u8]) -> bool {
1130    if fs::create_dir_all(root).is_err() {
1131        return false;
1132    }
1133    let tmp = root.join(format!(".{key}.tmp-{}", temp_suffix()));
1134    if write_file_synced(&tmp, payload).is_err() {
1135        let _ = fs::remove_file(&tmp);
1136        return false;
1137    }
1138    if fs::rename(&tmp, target).is_err() {
1139        let _ = fs::remove_file(&tmp);
1140        return false;
1141    }
1142    true
1143}