hyphae_engine/retrieval_proof/
verify.rs

1// SPDX-License-Identifier: Apache-2.0
2
3use std::{
4    fs::{File, Metadata, OpenOptions},
5    io::{Read, Write},
6    path::Path,
7    time::{Duration, Instant},
8};
9
10use hyphae_query::Record;
11use hyphae_retrieval::{
12    DurableVectorRecord, ExactRetrievalError, ExactRetrievalLimits, ExactRetrievalRequest,
13    HybridRequest, LexicalLimits, fuse_hybrid, retrieve_exact, retrieve_lexical,
14};
15use hyphae_storage::{SnapshotContents, load_snapshot_with_timeout};
16
17use super::{
18    ExactRetrievalProof, ExactRetrievalVerificationReport, HybridRetrievalProof,
19    HybridRetrievalVerificationReport, LexicalRetrievalProof, LexicalRetrievalVerificationReport,
20    MAX_RETRIEVAL_PROOF_BYTES, RetrievalProofAnchor, RetrievalProofError,
21    RetrievalVerificationLimits, decode_hybrid_proof, decode_lexical_proof, decode_proof,
22    encode_hybrid_proof, encode_lexical_proof, encode_proof,
23};
24use crate::decode_document;
25
26const PROOF_READ_BUFFER_BYTES: usize = 64 * 1024;
27
28/// Writes a canonical exact-retrieval proof to a new synchronized file.
29///
30/// Existing paths are never replaced.
31///
32/// # Errors
33///
34/// Returns a proof encoding, create, write, or synchronization error.
35pub fn write_exact_retrieval_proof(
36    path: impl AsRef<Path>,
37    proof: &ExactRetrievalProof,
38) -> Result<(), RetrievalProofError> {
39    let encoded = encode_proof(proof)?;
40    let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
41    file.write_all(&encoded)?;
42    file.sync_all()?;
43    Ok(())
44}
45
46/// Writes a canonical lexical-retrieval proof to a new synchronized file.
47///
48/// # Errors
49///
50/// Returns a proof encoding, create, write, or synchronization error.
51pub fn write_lexical_retrieval_proof(
52    path: impl AsRef<Path>,
53    proof: &LexicalRetrievalProof,
54) -> Result<(), RetrievalProofError> {
55    write_new(path, &encode_lexical_proof(proof)?)
56}
57
58/// Writes a canonical hybrid-retrieval proof to a new synchronized file.
59///
60/// # Errors
61///
62/// Returns a proof encoding, create, write, or synchronization error.
63pub fn write_hybrid_retrieval_proof(
64    path: impl AsRef<Path>,
65    proof: &HybridRetrievalProof,
66) -> Result<(), RetrievalProofError> {
67    write_new(path, &encode_hybrid_proof(proof)?)
68}
69
70/// Reads and verifies one canonical exact-retrieval proof under a byte limit.
71///
72/// # Errors
73///
74/// Returns an I/O, resource-limit, framing, canonicality, checksum, or digest
75/// error.
76pub fn read_exact_retrieval_proof(
77    path: impl AsRef<Path>,
78    maximum_bytes: u64,
79) -> Result<ExactRetrievalProof, RetrievalProofError> {
80    let mut no_deadline = || Ok(());
81    decode_proof(&read_bounded(path, maximum_bytes, &mut no_deadline)?)
82}
83
84/// Reads and verifies one canonical lexical-retrieval proof.
85///
86/// # Errors
87///
88/// Returns an I/O, resource-limit, framing, canonicality, checksum, or digest
89/// error.
90pub fn read_lexical_retrieval_proof(
91    path: impl AsRef<Path>,
92    maximum_bytes: u64,
93) -> Result<LexicalRetrievalProof, RetrievalProofError> {
94    let mut no_deadline = || Ok(());
95    decode_lexical_proof(&read_bounded(path, maximum_bytes, &mut no_deadline)?)
96}
97
98/// Reads and verifies one canonical hybrid-retrieval proof.
99///
100/// # Errors
101///
102/// Returns an I/O, resource-limit, framing, canonicality, checksum, or digest
103/// error.
104pub fn read_hybrid_retrieval_proof(
105    path: impl AsRef<Path>,
106    maximum_bytes: u64,
107) -> Result<HybridRetrievalProof, RetrievalProofError> {
108    let mut no_deadline = || Ok(());
109    decode_hybrid_proof(&read_bounded(path, maximum_bytes, &mut no_deadline)?)
110}
111
112/// Verifies an exact-retrieval proof completely offline against a trusted
113/// anchor and canonical format-2 snapshot witness.
114///
115/// # Errors
116///
117/// Returns an error for corruption, wrong trust anchor, wrong witness,
118/// unsupported semantics, resource exhaustion, timeout, or replay mismatch.
119/// No partial outcome is accepted.
120pub fn verify_exact_retrieval_proof(
121    proof_path: impl AsRef<Path>,
122    snapshot_path: impl AsRef<Path>,
123    expected_anchor_digest: [u8; 32],
124    limits: &RetrievalVerificationLimits,
125) -> Result<ExactRetrievalVerificationReport, RetrievalProofError> {
126    let started = Instant::now();
127    let mut check_read_deadline = || check_timeout(started, limits);
128    let proof = decode_proof(&read_bounded(
129        proof_path,
130        limits.proof_bytes,
131        &mut check_read_deadline,
132    )?)?;
133    check_timeout(started, limits)?;
134
135    let anchor_digest = proof.anchor_digest();
136    if anchor_digest != expected_anchor_digest {
137        return Err(RetrievalProofError::AnchorMismatch);
138    }
139
140    let snapshot = load_snapshot_before_deadline(snapshot_path, limits, started)?;
141    check_timeout(started, limits)?;
142    if snapshot.info.disk_format_version != 2 {
143        return Err(RetrievalProofError::SnapshotFormatMismatch);
144    }
145    if RetrievalProofAnchor::from_snapshot(&snapshot.info) != *proof.anchor() {
146        return Err(RetrievalProofError::SnapshotAnchorMismatch);
147    }
148
149    let Some(definition) = snapshot
150        .vector_spaces
151        .iter()
152        .find(|definition| definition.name == proof.request().vector_space)
153    else {
154        return Err(RetrievalProofError::Invalid {
155            reason: "proof references an unknown vector space",
156        });
157    };
158    definition.validate_vector(&proof.request().query)?;
159
160    let candidates = materialize_exact_candidates(&snapshot, proof.request(), limits, || {
161        check_timeout(started, limits)
162    })?;
163
164    let remaining = limits
165        .timeout
166        .checked_sub(started.elapsed())
167        .ok_or(RetrievalProofError::TimedOut)?;
168    if remaining.is_zero() {
169        return Err(RetrievalProofError::TimedOut);
170    }
171    let execution_limits = ExactRetrievalLimits {
172        max_candidates: limits.max_candidates,
173        max_candidate_bytes: limits.max_candidate_bytes,
174        max_returned: limits.max_returned,
175        timeout: remaining,
176    };
177    let actual = retrieve_exact(&candidates, proof.request(), &execution_limits)?;
178    if &actual != proof.outcome() {
179        return Err(RetrievalProofError::ReexecutionMismatch);
180    }
181    check_timeout(started, limits)?;
182
183    Ok(ExactRetrievalVerificationReport {
184        anchor: proof.anchor().clone(),
185        anchor_digest,
186        proof_digest: proof.proof_digest(),
187        outcome: actual,
188    })
189}
190
191/// Verifies a lexical-retrieval proof completely offline.
192///
193/// # Errors
194///
195/// Returns an error for corruption, wrong anchor/witness, unsupported
196/// semantics, resource exhaustion, timeout, or replay mismatch.
197pub fn verify_lexical_retrieval_proof(
198    proof_path: impl AsRef<Path>,
199    snapshot_path: impl AsRef<Path>,
200    expected_anchor_digest: [u8; 32],
201    limits: &RetrievalVerificationLimits,
202) -> Result<LexicalRetrievalVerificationReport, RetrievalProofError> {
203    let started = Instant::now();
204    let mut check_read_deadline = || check_timeout(started, limits);
205    let proof = decode_lexical_proof(&read_bounded(
206        proof_path,
207        limits.proof_bytes,
208        &mut check_read_deadline,
209    )?)?;
210    let snapshot = load_bound_snapshot(
211        snapshot_path,
212        proof.anchor(),
213        expected_anchor_digest,
214        limits,
215        started,
216    )?;
217    let definition = snapshot
218        .lexical_indexes
219        .iter()
220        .find(|definition| definition.name == proof.request().index)
221        .ok_or(RetrievalProofError::Invalid {
222            reason: "proof references an unknown lexical index",
223        })?;
224    let records = decode_records(&snapshot, started, limits)?;
225    let remaining = remaining_timeout(started, limits)?;
226    let execution_limits = LexicalLimits {
227        max_documents: limits.max_documents,
228        max_tokens: limits.max_tokens,
229        max_candidates: limits.max_lexical_candidates,
230        max_returned: limits.max_lexical_returned,
231        timeout: remaining,
232    };
233    let actual = retrieve_lexical(&records, definition, proof.request(), &execution_limits)?;
234    if &actual != proof.outcome() {
235        return Err(RetrievalProofError::ReexecutionMismatch);
236    }
237    check_timeout(started, limits)?;
238    Ok(LexicalRetrievalVerificationReport {
239        anchor: proof.anchor().clone(),
240        anchor_digest: proof.anchor_digest(),
241        proof_digest: proof.proof_digest(),
242        outcome: actual,
243    })
244}
245
246/// Verifies a hybrid-retrieval proof completely offline.
247///
248/// # Errors
249///
250/// Returns an error for corruption, wrong anchor/witness, unsupported
251/// semantics, resource exhaustion, timeout, branch mismatch, or fusion
252/// mismatch.
253pub fn verify_hybrid_retrieval_proof(
254    proof_path: impl AsRef<Path>,
255    snapshot_path: impl AsRef<Path>,
256    expected_anchor_digest: [u8; 32],
257    limits: &RetrievalVerificationLimits,
258) -> Result<HybridRetrievalVerificationReport, RetrievalProofError> {
259    let started = Instant::now();
260    let mut check_read_deadline = || check_timeout(started, limits);
261    let proof = decode_hybrid_proof(&read_bounded(
262        proof_path,
263        limits.proof_bytes,
264        &mut check_read_deadline,
265    )?)?;
266    let snapshot = load_bound_snapshot(
267        snapshot_path,
268        proof.anchor(),
269        expected_anchor_digest,
270        limits,
271        started,
272    )?;
273    let definition = snapshot
274        .lexical_indexes
275        .iter()
276        .find(|definition| definition.name == proof.lexical_request().index)
277        .ok_or(RetrievalProofError::Invalid {
278            reason: "proof references an unknown lexical index",
279        })?;
280    let records = decode_records(&snapshot, started, limits)?;
281    let lexical = retrieve_lexical(
282        &records,
283        definition,
284        proof.lexical_request(),
285        &LexicalLimits {
286            max_documents: limits.max_documents,
287            max_tokens: limits.max_tokens,
288            max_candidates: limits.max_lexical_candidates,
289            max_returned: limits.max_lexical_returned,
290            timeout: remaining_timeout(started, limits)?,
291        },
292    )?;
293    if &lexical != proof.lexical_outcome() {
294        return Err(RetrievalProofError::ReexecutionMismatch);
295    }
296    let vector = replay_exact(&snapshot, proof.vector_request(), started, limits)?;
297    if &vector != proof.vector_outcome() {
298        return Err(RetrievalProofError::ReexecutionMismatch);
299    }
300    if proof.fusion_request().limit > limits.max_hybrid_returned {
301        return Err(RetrievalProofError::Invalid {
302            reason: "hybrid result limit exceeds verifier policy",
303        });
304    }
305    let fusion_request = HybridRequest {
306        lexical_weight: proof.fusion_request().lexical_weight,
307        vector_weight: proof.fusion_request().vector_weight,
308        limit: proof.fusion_request().limit,
309    };
310    let actual = fuse_hybrid(&lexical, &vector, &fusion_request)?;
311    if &actual != proof.outcome() {
312        return Err(RetrievalProofError::ReexecutionMismatch);
313    }
314    check_timeout(started, limits)?;
315    Ok(HybridRetrievalVerificationReport {
316        anchor: proof.anchor().clone(),
317        anchor_digest: proof.anchor_digest(),
318        proof_digest: proof.proof_digest(),
319        outcome: actual,
320    })
321}
322
323fn write_new(path: impl AsRef<Path>, encoded: &[u8]) -> Result<(), RetrievalProofError> {
324    let mut file = OpenOptions::new().create_new(true).write(true).open(path)?;
325    file.write_all(encoded)?;
326    file.sync_all()?;
327    Ok(())
328}
329
330fn read_bounded(
331    path: impl AsRef<Path>,
332    maximum_bytes: u64,
333    check_deadline: &mut impl FnMut() -> Result<(), RetrievalProofError>,
334) -> Result<Vec<u8>, RetrievalProofError> {
335    check_deadline()?;
336    let path = path.as_ref();
337    let path_metadata = std::fs::metadata(path)?;
338    check_deadline()?;
339    ensure_regular_proof_file(&path_metadata)?;
340
341    let file = File::open(path)?;
342    check_deadline()?;
343    let initial_metadata = file.metadata()?;
344    check_deadline()?;
345    ensure_regular_proof_file(&initial_metadata)?;
346
347    read_open_bounded(
348        file,
349        &initial_metadata,
350        maximum_bytes.min(MAX_RETRIEVAL_PROOF_BYTES),
351        check_deadline,
352    )
353}
354
355fn read_open_bounded(
356    mut file: File,
357    initial_metadata: &Metadata,
358    maximum_bytes: u64,
359    check_deadline: &mut impl FnMut() -> Result<(), RetrievalProofError>,
360) -> Result<Vec<u8>, RetrievalProofError> {
361    let initial_length = initial_metadata.len();
362    if initial_length > maximum_bytes {
363        return Err(RetrievalProofError::ProofLimitExceeded {
364            actual: initial_length,
365            maximum: maximum_bytes,
366        });
367    }
368    let capacity =
369        usize::try_from(initial_length).map_err(|_| RetrievalProofError::LengthOverflow)?;
370    let mut encoded = Vec::with_capacity(capacity);
371    let mut remaining = maximum_bytes
372        .checked_add(1)
373        .ok_or(RetrievalProofError::LengthOverflow)?;
374    let mut buffer = vec![0_u8; PROOF_READ_BUFFER_BYTES];
375    while remaining > 0 {
376        check_deadline()?;
377        let read_length = usize::try_from(remaining.min(PROOF_READ_BUFFER_BYTES as u64))
378            .map_err(|_| RetrievalProofError::LengthOverflow)?;
379        let read = file.read(&mut buffer[..read_length])?;
380        check_deadline()?;
381        if read == 0 {
382            break;
383        }
384        encoded.extend_from_slice(&buffer[..read]);
385        remaining = remaining
386            .checked_sub(u64::try_from(read).map_err(|_| RetrievalProofError::LengthOverflow)?)
387            .ok_or(RetrievalProofError::LengthOverflow)?;
388    }
389
390    let final_metadata = file.metadata()?;
391    check_deadline()?;
392    ensure_regular_proof_file(&final_metadata)?;
393    let actual = u64::try_from(encoded.len()).map_err(|_| RetrievalProofError::LengthOverflow)?;
394    let observed = actual.max(final_metadata.len());
395    if observed > maximum_bytes {
396        return Err(RetrievalProofError::ProofLimitExceeded {
397            actual: observed,
398            maximum: maximum_bytes,
399        });
400    }
401    if actual != initial_length || final_metadata.len() != initial_length {
402        return Err(RetrievalProofError::Invalid {
403            reason: "proof changed while being read",
404        });
405    }
406    Ok(encoded)
407}
408
409fn ensure_regular_proof_file(metadata: &Metadata) -> Result<(), RetrievalProofError> {
410    if metadata.is_file() {
411        Ok(())
412    } else {
413        Err(RetrievalProofError::Invalid {
414            reason: "proof path is not a regular file",
415        })
416    }
417}
418
419fn load_bound_snapshot(
420    path: impl AsRef<Path>,
421    anchor: &RetrievalProofAnchor,
422    expected_anchor_digest: [u8; 32],
423    limits: &RetrievalVerificationLimits,
424    started: Instant,
425) -> Result<SnapshotContents, RetrievalProofError> {
426    if anchor.digest() != expected_anchor_digest {
427        return Err(RetrievalProofError::AnchorMismatch);
428    }
429    let snapshot = load_snapshot_before_deadline(path, limits, started)?;
430    check_timeout(started, limits)?;
431    if snapshot.info.disk_format_version != 2 {
432        return Err(RetrievalProofError::SnapshotFormatMismatch);
433    }
434    if RetrievalProofAnchor::from_snapshot(&snapshot.info) != *anchor {
435        return Err(RetrievalProofError::SnapshotAnchorMismatch);
436    }
437    Ok(snapshot)
438}
439
440fn load_snapshot_before_deadline(
441    path: impl AsRef<Path>,
442    limits: &RetrievalVerificationLimits,
443    started: Instant,
444) -> Result<SnapshotContents, RetrievalProofError> {
445    match load_snapshot_with_timeout(path, &limits.snapshot, remaining_timeout(started, limits)?) {
446        Err(error) if error.is_timeout() => Err(RetrievalProofError::TimedOut),
447        Err(error) => Err(error.into()),
448        Ok(snapshot) => Ok(snapshot),
449    }
450}
451
452fn decode_records(
453    snapshot: &SnapshotContents,
454    started: Instant,
455    limits: &RetrievalVerificationLimits,
456) -> Result<Vec<Record>, RetrievalProofError> {
457    if u64::try_from(snapshot.entries.len()).unwrap_or(u64::MAX) > limits.max_documents {
458        return Err(RetrievalProofError::Invalid {
459            reason: "snapshot document count exceeds verifier policy",
460        });
461    }
462    snapshot
463        .entries
464        .iter()
465        .map(|entry| {
466            check_timeout(started, limits)?;
467            Ok(Record {
468                key: entry.key.clone(),
469                value: decode_document(&entry.value)?,
470            })
471        })
472        .collect()
473}
474
475fn replay_exact(
476    snapshot: &SnapshotContents,
477    request: &ExactRetrievalRequest,
478    started: Instant,
479    limits: &RetrievalVerificationLimits,
480) -> Result<hyphae_retrieval::ExactRetrievalOutcome, RetrievalProofError> {
481    let definition = snapshot
482        .vector_spaces
483        .iter()
484        .find(|definition| definition.name == request.vector_space)
485        .ok_or(RetrievalProofError::Invalid {
486            reason: "proof references an unknown vector space",
487        })?;
488    definition.validate_vector(&request.query)?;
489    let candidates =
490        materialize_exact_candidates(snapshot, request, limits, || check_timeout(started, limits))?;
491    Ok(retrieve_exact(
492        &candidates,
493        request,
494        &ExactRetrievalLimits {
495            max_candidates: limits.max_candidates,
496            max_candidate_bytes: limits.max_candidate_bytes,
497            max_returned: limits.max_returned,
498            timeout: remaining_timeout(started, limits)?,
499        },
500    )?)
501}
502
503fn materialize_exact_candidates(
504    snapshot: &SnapshotContents,
505    request: &ExactRetrievalRequest,
506    limits: &RetrievalVerificationLimits,
507    mut check_deadline: impl FnMut() -> Result<(), RetrievalProofError>,
508) -> Result<Vec<DurableVectorRecord>, RetrievalProofError> {
509    let matching = || {
510        snapshot
511            .vectors
512            .iter()
513            .filter(|vector| vector.space == request.vector_space)
514    };
515    let mut candidate_count = 0_u64;
516    let mut candidate_bytes = 0_u64;
517    for vector in matching() {
518        check_deadline()?;
519        if candidate_count >= limits.max_candidates {
520            return Err(ExactRetrievalError::CandidateBudgetExceeded {
521                maximum: limits.max_candidates,
522            }
523            .into());
524        }
525        candidate_count =
526            candidate_count
527                .checked_add(1)
528                .ok_or(ExactRetrievalError::CandidateBudgetExceeded {
529                    maximum: limits.max_candidates,
530                })?;
531
532        let vector_bytes = u64::try_from(vector.vector.as_slice().len())
533            .ok()
534            .and_then(|elements| elements.checked_mul(2))
535            .ok_or(ExactRetrievalError::CandidateByteBudgetExceeded {
536                maximum: limits.max_candidate_bytes,
537            })?;
538        let bytes = u64::try_from(vector.key.len())
539            .ok()
540            .and_then(|key_bytes| key_bytes.checked_add(vector_bytes))
541            .ok_or(ExactRetrievalError::CandidateByteBudgetExceeded {
542                maximum: limits.max_candidate_bytes,
543            })?;
544        candidate_bytes = candidate_bytes.checked_add(bytes).ok_or(
545            ExactRetrievalError::CandidateByteBudgetExceeded {
546                maximum: limits.max_candidate_bytes,
547            },
548        )?;
549        if candidate_bytes > limits.max_candidate_bytes {
550            return Err(ExactRetrievalError::CandidateByteBudgetExceeded {
551                maximum: limits.max_candidate_bytes,
552            }
553            .into());
554        }
555    }
556    check_deadline()?;
557
558    let capacity = usize::try_from(candidate_count).map_err(|_| {
559        ExactRetrievalError::CandidateBudgetExceeded {
560            maximum: limits.max_candidates,
561        }
562    })?;
563    let mut candidates = Vec::with_capacity(capacity);
564    for vector in matching() {
565        check_deadline()?;
566        candidates.push(DurableVectorRecord {
567            key: vector.key.clone(),
568            vector: vector.vector.clone(),
569        });
570    }
571    check_deadline()?;
572    Ok(candidates)
573}
574
575fn remaining_timeout(
576    started: Instant,
577    limits: &RetrievalVerificationLimits,
578) -> Result<Duration, RetrievalProofError> {
579    let remaining = limits
580        .timeout
581        .checked_sub(started.elapsed())
582        .ok_or(RetrievalProofError::TimedOut)?;
583    if remaining.is_zero() {
584        Err(RetrievalProofError::TimedOut)
585    } else {
586        Ok(remaining)
587    }
588}
589
590fn check_timeout(
591    started: Instant,
592    limits: &RetrievalVerificationLimits,
593) -> Result<(), RetrievalProofError> {
594    if started.elapsed() >= limits.timeout {
595        Err(RetrievalProofError::TimedOut)
596    } else {
597        Ok(())
598    }
599}
600
601#[cfg(test)]
602mod tests {
603    use std::{
604        collections::BTreeMap, error::Error, fs, io::Write as _, path::PathBuf, time::Duration,
605    };
606
607    use hyphae_core::{Q15Vector, VectorSpaceDefinition, VectorSpaceName};
608    use hyphae_query::{FieldPath, Record, Value};
609    use hyphae_retrieval::{
610        ExactRetrievalError, ExactRetrievalLimits, ExactRetrievalOutcome, ExactRetrievalRequest,
611        HybridRequest, LexicalField, LexicalIndexDefinition, LexicalLimits, LexicalRequest,
612    };
613    use hyphae_storage::{SnapshotReadLimits, load_snapshot};
614    use uuid::Uuid;
615
616    use super::{
617        PROOF_READ_BUFFER_BYTES, RetrievalVerificationLimits, materialize_exact_candidates,
618        read_exact_retrieval_proof, read_open_bounded, verify_exact_retrieval_proof,
619        verify_hybrid_retrieval_proof, verify_lexical_retrieval_proof, write_exact_retrieval_proof,
620        write_hybrid_retrieval_proof, write_lexical_retrieval_proof,
621    };
622    use crate::{
623        HyphaeEngine, MAX_RETRIEVAL_PROOF_BYTES, RetrievalProofError,
624        write_exact_retrieval_proof as write,
625    };
626
627    struct TestDirectory {
628        path: PathBuf,
629    }
630
631    impl TestDirectory {
632        fn create(name: &str) -> Result<Self, Box<dyn Error>> {
633            let path = std::env::temp_dir()
634                .join(format!("hyphae-retrieval-proof-{name}-{}", Uuid::now_v7()));
635            fs::create_dir_all(&path)?;
636            Ok(Self { path })
637        }
638    }
639
640    impl Drop for TestDirectory {
641        fn drop(&mut self) {
642            let _ignored = fs::remove_dir_all(&self.path);
643        }
644    }
645
646    #[test]
647    fn retrieval_proof_reader_enforces_the_canonical_hard_limit() -> Result<(), Box<dyn Error>> {
648        let temporary = TestDirectory::create("oversized-proof")?;
649        let proof_path = temporary.path.join("oversized.hyrproof");
650        let file = fs::File::create(&proof_path)?;
651        file.set_len(MAX_RETRIEVAL_PROOF_BYTES + 1)?;
652        drop(file);
653
654        assert!(matches!(
655            read_exact_retrieval_proof(&proof_path, u64::MAX),
656            Err(RetrievalProofError::ProofLimitExceeded {
657                actual,
658                maximum: MAX_RETRIEVAL_PROOF_BYTES,
659            }) if actual == MAX_RETRIEVAL_PROOF_BYTES + 1
660        ));
661        Ok(())
662    }
663
664    #[test]
665    fn retrieval_proof_reader_detects_same_handle_growth() -> Result<(), Box<dyn Error>> {
666        let temporary = TestDirectory::create("growing-proof")?;
667        let proof_path = temporary.path.join("growing.hyrproof");
668        fs::write(&proof_path, b"initial")?;
669        let file = fs::File::open(&proof_path)?;
670        let initial_metadata = file.metadata()?;
671        let mut writer = fs::OpenOptions::new().append(true).open(&proof_path)?;
672        writer.write_all(b"-growth")?;
673        writer.sync_all()?;
674        drop(writer);
675
676        let mut no_deadline = || Ok(());
677        assert!(matches!(
678            read_open_bounded(file, &initial_metadata, 1024, &mut no_deadline),
679            Err(RetrievalProofError::Invalid {
680                reason: "proof changed while being read",
681            })
682        ));
683        Ok(())
684    }
685
686    #[test]
687    fn retrieval_proof_reader_checks_deadline_between_chunks() -> Result<(), Box<dyn Error>> {
688        let temporary = TestDirectory::create("timed-proof")?;
689        let proof_path = temporary.path.join("timed.hyrproof");
690        fs::write(&proof_path, vec![0_u8; PROOF_READ_BUFFER_BYTES * 2])?;
691        let file = fs::File::open(&proof_path)?;
692        let initial_metadata = file.metadata()?;
693        let mut checks = 0_u8;
694        let mut deadline = || {
695            checks += 1;
696            if checks == 2 {
697                Err(RetrievalProofError::TimedOut)
698            } else {
699                Ok(())
700            }
701        };
702
703        assert!(matches!(
704            read_open_bounded(
705                file,
706                &initial_metadata,
707                MAX_RETRIEVAL_PROOF_BYTES,
708                &mut deadline,
709            ),
710            Err(RetrievalProofError::TimedOut)
711        ));
712        assert_eq!(checks, 2);
713        Ok(())
714    }
715
716    #[test]
717    fn exact_candidate_preflight_rejects_count_and_bytes_before_materialization()
718    -> Result<(), Box<dyn Error>> {
719        let temporary = TestDirectory::create("candidate-preflight")?;
720        let (artifact, _) = create_artifact(&temporary.path.join("data"))?;
721        let snapshot = load_snapshot(&artifact.snapshot.path, &SnapshotReadLimits::default())?;
722
723        let count_limits = RetrievalVerificationLimits {
724            max_candidates: 1,
725            ..RetrievalVerificationLimits::default()
726        };
727        let mut count_checks = 0_u8;
728        let count_result = materialize_exact_candidates(
729            &snapshot,
730            artifact.proof.request(),
731            &count_limits,
732            || {
733                count_checks += 1;
734                Ok(())
735            },
736        );
737        assert!(matches!(
738            count_result,
739            Err(RetrievalProofError::Retrieval { source })
740                if *source
741                    == ExactRetrievalError::CandidateBudgetExceeded { maximum: 1 }
742        ));
743        assert_eq!(count_checks, 2);
744
745        let byte_limits = RetrievalVerificationLimits {
746            max_candidate_bytes: 9,
747            ..RetrievalVerificationLimits::default()
748        };
749        let mut byte_checks = 0_u8;
750        let byte_result =
751            materialize_exact_candidates(&snapshot, artifact.proof.request(), &byte_limits, || {
752                byte_checks += 1;
753                Ok(())
754            });
755        assert!(matches!(
756            byte_result,
757            Err(RetrievalProofError::Retrieval { source })
758                if *source
759                    == ExactRetrievalError::CandidateByteBudgetExceeded { maximum: 9 }
760        ));
761        assert_eq!(byte_checks, 2);
762        Ok(())
763    }
764
765    #[test]
766    fn exact_candidate_materialization_checks_its_deadline() -> Result<(), Box<dyn Error>> {
767        let temporary = TestDirectory::create("candidate-deadline")?;
768        let (artifact, _) = create_artifact(&temporary.path.join("data"))?;
769        let snapshot = load_snapshot(&artifact.snapshot.path, &SnapshotReadLimits::default())?;
770        let limits = RetrievalVerificationLimits::default();
771        let mut checks = 0_u8;
772
773        let result =
774            materialize_exact_candidates(&snapshot, artifact.proof.request(), &limits, || {
775                checks += 1;
776                if checks == 4 {
777                    Err(RetrievalProofError::TimedOut)
778                } else {
779                    Ok(())
780                }
781            });
782
783        assert!(matches!(result, Err(RetrievalProofError::TimedOut)));
784        assert_eq!(checks, 4);
785        Ok(())
786    }
787
788    fn request(space: VectorSpaceName) -> Result<ExactRetrievalRequest, Box<dyn Error>> {
789        Ok(ExactRetrievalRequest {
790            vector_space: space,
791            query: Q15Vector::new(vec![32_767, 0])?,
792            limit: 2,
793            minimum_score_nanos: -1_000_000_000,
794            minimum_margin_nanos: 0,
795        })
796    }
797
798    fn create_artifact(
799        root: &std::path::Path,
800    ) -> Result<(crate::ExactRetrievalProofArtifact, ExactRetrievalLimits), Box<dyn Error>> {
801        let space = VectorSpaceName::new("semantic")?;
802        let mut opened = HyphaeEngine::open(root)?;
803        opened.engine.define_vector_space(
804            Uuid::now_v7(),
805            VectorSpaceDefinition::cosine(space.clone(), 2)?,
806        )?;
807        opened.engine.put_vectors(
808            Uuid::now_v7(),
809            &space,
810            &[
811                (b"alpha".to_vec(), Q15Vector::new(vec![32_767, 0])?),
812                (b"beta".to_vec(), Q15Vector::new(vec![0, 32_767])?),
813            ],
814        )?;
815        let limits = ExactRetrievalLimits {
816            max_candidates: 10,
817            max_candidate_bytes: 1024,
818            max_returned: 10,
819            timeout: Duration::from_secs(1),
820        };
821        let artifact = opened
822            .engine
823            .retrieve_exact_with_proof(&request(space)?, &limits)?;
824        Ok((artifact, limits))
825    }
826
827    fn lexical_record(key: &[u8], title: &str, body: &str) -> Record {
828        Record::new(
829            key,
830            Value::Object(BTreeMap::from([
831                ("title".into(), Value::String(title.into())),
832                ("body".into(), Value::String(body.into())),
833            ])),
834        )
835    }
836
837    #[allow(clippy::type_complexity)]
838    fn create_multimodal_engine(
839        root: &std::path::Path,
840    ) -> Result<
841        (
842            HyphaeEngine,
843            LexicalRequest,
844            LexicalLimits,
845            ExactRetrievalRequest,
846            ExactRetrievalLimits,
847            HybridRequest,
848        ),
849        Box<dyn Error>,
850    > {
851        let lexical_name = VectorSpaceName::new("content")?;
852        let vector_space = VectorSpaceName::new("semantic")?;
853        let mut opened = HyphaeEngine::open(root)?;
854        opened.engine.put_records(
855            Uuid::now_v7(),
856            &[
857                lexical_record(b"alpha", "Durable memory", "offline agent memory"),
858                lexical_record(b"beta", "Fast search", "exact vector retrieval"),
859            ],
860        )?;
861        opened.engine.define_lexical_index(
862            Uuid::now_v7(),
863            LexicalIndexDefinition::new(
864                lexical_name.clone(),
865                vec![
866                    LexicalField {
867                        path: FieldPath::field("body"),
868                        weight_micros: 1_000_000,
869                    },
870                    LexicalField {
871                        path: FieldPath::field("title"),
872                        weight_micros: 2_000_000,
873                    },
874                ],
875            )?,
876        )?;
877        opened.engine.define_vector_space(
878            Uuid::now_v7(),
879            VectorSpaceDefinition::cosine(vector_space.clone(), 2)?,
880        )?;
881        opened.engine.put_vectors(
882            Uuid::now_v7(),
883            &vector_space,
884            &[
885                (b"alpha".to_vec(), Q15Vector::new(vec![32_767, 0])?),
886                (b"beta".to_vec(), Q15Vector::new(vec![0, 32_767])?),
887            ],
888        )?;
889        Ok((
890            opened.engine,
891            LexicalRequest {
892                index: lexical_name,
893                query: "durable memory".into(),
894                limit: 2,
895            },
896            LexicalLimits {
897                max_documents: 10,
898                max_tokens: 1_000,
899                max_candidates: 10,
900                max_returned: 10,
901                timeout: Duration::from_secs(2),
902            },
903            ExactRetrievalRequest {
904                vector_space,
905                query: Q15Vector::new(vec![32_767, 0])?,
906                limit: 2,
907                minimum_score_nanos: -1_000_000_000,
908                minimum_margin_nanos: 0,
909            },
910            ExactRetrievalLimits {
911                max_candidates: 10,
912                max_candidate_bytes: 1_024,
913                max_returned: 10,
914                timeout: Duration::from_secs(2),
915            },
916            HybridRequest {
917                lexical_weight: 1,
918                vector_weight: 1,
919                limit: 2,
920            },
921        ))
922    }
923
924    #[test]
925    fn exact_proof_verifies_after_originating_directory_is_deleted() -> Result<(), Box<dyn Error>> {
926        let temporary = TestDirectory::create("offline")?;
927        let data = temporary.path.join("data");
928        let portable = temporary.path.join("portable");
929        fs::create_dir_all(&portable)?;
930        let (artifact, _execution_limits) = create_artifact(&data)?;
931        let proof_path = portable.join("result.hyrproof");
932        let witness_path = portable.join("witness.hysnap");
933        write_exact_retrieval_proof(&proof_path, &artifact.proof)?;
934        fs::copy(&artifact.snapshot.path, &witness_path)?;
935        fs::remove_dir_all(&data)?;
936
937        let report = verify_exact_retrieval_proof(
938            &proof_path,
939            &witness_path,
940            artifact.proof.anchor_digest(),
941            &RetrievalVerificationLimits::default(),
942        )?;
943        assert_eq!(report.outcome, artifact.proof.outcome().clone());
944        Ok(())
945    }
946
947    #[test]
948    fn self_consistently_rehashed_request_and_outcome_edits_are_rejected()
949    -> Result<(), Box<dyn Error>> {
950        let temporary = TestDirectory::create("tamper")?;
951        let (artifact, _) = create_artifact(&temporary.path.join("data"))?;
952
953        for mutation in 0..5 {
954            let mut forged = artifact.proof.clone();
955            match mutation {
956                0 => forged.request.query = Q15Vector::new(vec![0, 32_767])?,
957                1 => {
958                    let ExactRetrievalOutcome::Matches { matches, .. } = &mut forged.outcome else {
959                        return Err("expected matches".into());
960                    };
961                    matches[0].score_nanos -= 1;
962                }
963                2 => {
964                    let ExactRetrievalOutcome::Matches { matches, .. } = &mut forged.outcome else {
965                        return Err("expected matches".into());
966                    };
967                    matches.swap(0, 1);
968                }
969                3 => {
970                    let ExactRetrievalOutcome::Matches { matches, .. } = &mut forged.outcome else {
971                        return Err("expected matches".into());
972                    };
973                    matches[0].key = b"forged".to_vec();
974                }
975                4 => forged.request.vector_space = VectorSpaceName::new("other")?,
976                _ => unreachable!(),
977            }
978            let proof_path = temporary.path.join(format!("forged-{mutation}.hyrproof"));
979            if write(&proof_path, &forged).is_err() {
980                continue;
981            }
982            let result = verify_exact_retrieval_proof(
983                &proof_path,
984                &artifact.snapshot.path,
985                artifact.proof.anchor_digest(),
986                &RetrievalVerificationLimits::default(),
987            );
988            assert!(result.is_err(), "mutation {mutation} unexpectedly verified");
989        }
990        Ok(())
991    }
992
993    #[test]
994    fn wrong_witness_anchor_and_semantics_are_rejected() -> Result<(), Box<dyn Error>> {
995        let temporary = TestDirectory::create("binding")?;
996        let (artifact, _) = create_artifact(&temporary.path.join("data"))?;
997        let proof_path = temporary.path.join("proof.hyrproof");
998        write(&proof_path, &artifact.proof)?;
999
1000        assert!(matches!(
1001            verify_exact_retrieval_proof(
1002                &proof_path,
1003                &artifact.snapshot.path,
1004                [0; 32],
1005                &RetrievalVerificationLimits::default(),
1006            ),
1007            Err(RetrievalProofError::AnchorMismatch)
1008        ));
1009
1010        let mut encoded = fs::read(&proof_path)?;
1011        encoded[14..16].copy_from_slice(&99_u16.to_le_bytes());
1012        let semantics_path = temporary.path.join("semantics.hyrproof");
1013        fs::write(&semantics_path, encoded)?;
1014        assert!(matches!(
1015            super::read_exact_retrieval_proof(&semantics_path, u64::MAX),
1016            Err(RetrievalProofError::UnsupportedSemantics { found: 99, .. })
1017        ));
1018        Ok(())
1019    }
1020
1021    #[test]
1022    fn lexical_and_hybrid_proofs_verify_without_the_originating_directory()
1023    -> Result<(), Box<dyn Error>> {
1024        let temporary = TestDirectory::create("multimodal-offline")?;
1025        let data = temporary.path.join("data");
1026        let portable = temporary.path.join("portable");
1027        fs::create_dir_all(&portable)?;
1028        let (
1029            engine,
1030            lexical_request,
1031            lexical_limits,
1032            vector_request,
1033            vector_limits,
1034            hybrid_request,
1035        ) = create_multimodal_engine(&data)?;
1036        let lexical = engine.retrieve_lexical_with_proof(&lexical_request, &lexical_limits)?;
1037        let hybrid = engine.retrieve_hybrid_with_proof(
1038            &lexical_request,
1039            &lexical_limits,
1040            &vector_request,
1041            &vector_limits,
1042            &hybrid_request,
1043        )?;
1044        let lexical_proof = portable.join("lexical.hyrproof");
1045        let hybrid_proof = portable.join("hybrid.hyrproof");
1046        let witness = portable.join("witness.hysnap");
1047        write_lexical_retrieval_proof(&lexical_proof, &lexical.proof)?;
1048        write_hybrid_retrieval_proof(&hybrid_proof, &hybrid.proof)?;
1049        fs::copy(&hybrid.snapshot.path, &witness)?;
1050        drop(engine);
1051        fs::remove_dir_all(&data)?;
1052
1053        let lexical_report = verify_lexical_retrieval_proof(
1054            &lexical_proof,
1055            &witness,
1056            lexical.proof.anchor_digest(),
1057            &RetrievalVerificationLimits::default(),
1058        )?;
1059        let hybrid_report = verify_hybrid_retrieval_proof(
1060            &hybrid_proof,
1061            &witness,
1062            hybrid.proof.anchor_digest(),
1063            &RetrievalVerificationLimits::default(),
1064        )?;
1065        assert_eq!(lexical_report.outcome, lexical.proof.outcome().clone());
1066        assert_eq!(hybrid_report.outcome, hybrid.proof.outcome().clone());
1067        Ok(())
1068    }
1069
1070    #[test]
1071    fn lexical_and_hybrid_self_consistent_tampering_fails_reexecution() -> Result<(), Box<dyn Error>>
1072    {
1073        let temporary = TestDirectory::create("multimodal-tamper")?;
1074        let (
1075            engine,
1076            lexical_request,
1077            lexical_limits,
1078            vector_request,
1079            vector_limits,
1080            hybrid_request,
1081        ) = create_multimodal_engine(&temporary.path.join("data"))?;
1082        let lexical = engine.retrieve_lexical_with_proof(&lexical_request, &lexical_limits)?;
1083        let hybrid = engine.retrieve_hybrid_with_proof(
1084            &lexical_request,
1085            &lexical_limits,
1086            &vector_request,
1087            &vector_limits,
1088            &hybrid_request,
1089        )?;
1090
1091        let mut forged_lexical = lexical.proof.clone();
1092        let hyphae_retrieval::LexicalOutcome::Matches { matches, .. } = &mut forged_lexical.outcome
1093        else {
1094            return Err("expected lexical matches".into());
1095        };
1096        matches[0].score_nanos -= 1;
1097        matches[0].terms[0].score_nanos -= 1;
1098        let lexical_path = temporary.path.join("forged-lexical.hyrproof");
1099        write_lexical_retrieval_proof(&lexical_path, &forged_lexical)?;
1100        assert!(
1101            verify_lexical_retrieval_proof(
1102                &lexical_path,
1103                &lexical.snapshot.path,
1104                lexical.proof.anchor_digest(),
1105                &RetrievalVerificationLimits::default(),
1106            )
1107            .is_err()
1108        );
1109
1110        let mut forged_hybrid = hybrid.proof.clone();
1111        forged_hybrid.vector_request.query = Q15Vector::new(vec![0, 32_767])?;
1112        let hybrid_path = temporary.path.join("forged-hybrid.hyrproof");
1113        write_hybrid_retrieval_proof(&hybrid_path, &forged_hybrid)?;
1114        assert!(
1115            verify_hybrid_retrieval_proof(
1116                &hybrid_path,
1117                &hybrid.snapshot.path,
1118                hybrid.proof.anchor_digest(),
1119                &RetrievalVerificationLimits::default(),
1120            )
1121            .is_err()
1122        );
1123        Ok(())
1124    }
1125}
hyphae_engine/retrieval_proof/verify.rs

hyphae_engine/retrieval_proof/
verify.rs
