a3s_code_core/workspace/
source_snapshot.rs

1//! Typed workspace source-snapshot authority (KRN-4).
2//!
3//! One immutable value binds every revision that derived workspace data
4//! depends on: the manifest scan revision, the eligible file content
5//! identity, the eligibility policy revision, the document/LSP revision,
6//! the chunk catalog revision, and the derived index generation. Search
7//! hits, symbol results, model context items, and evidence facts can carry
8//! one snapshot digest, and a result bound to an older snapshot can never be
9//! presented as current against a newer live snapshot.
10//!
11//! This is an identity and boundary layer. The manifest remains the scan
12//! authority, the catalog remains the chunk authority, and each index
13//! remains a rebuildable derived cache; none of them gains a second opinion
14//! about revisions.
15
16use crate::workspace::manifest::LocalWorkspaceManifestSnapshot;
17use serde::{Deserialize, Serialize};
18use thiserror::Error;
19
20pub const WORKSPACE_SOURCE_SNAPSHOT_SCHEMA_V1: &str = "a3s.code.workspace-source-snapshot.v1";
21pub const WORKSPACE_SOURCE_SNAPSHOT_DIGEST_DOMAIN_V1: &str =
22    "a3s.code.workspace-source-snapshot.identity.v1";
23/// Digest domain binding the eligible file set of one manifest revision.
24pub const WORKSPACE_SOURCE_CONTENT_DOMAIN_V1: &str =
25    "a3s.code.workspace-source-snapshot.content.v1";
26const MAX_ROOT_BYTES: usize = 1024;
27const MAX_ELIGIBLE_FILES: u64 = u32::MAX as u64;
28
29#[derive(Debug, Clone, PartialEq, Eq, Error)]
30pub enum WorkspaceSourceSnapshotError {
31    #[error("workspace source snapshot schema is unsupported")]
32    UnsupportedSchema,
33    #[error("workspace source snapshot field `{0}` is invalid")]
34    InvalidField(&'static str),
35    #[error("workspace source snapshot digest `{0}` is invalid")]
36    InvalidDigest(&'static str),
37    #[error("workspace source snapshot serialization failed: {0}")]
38    Serialization(String),
39}
40
41/// The single revision authority for one workspace at one point in time.
42#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
43#[serde(rename_all = "camelCase", deny_unknown_fields)]
44pub struct WorkspaceSourceSnapshotV1 {
45    pub schema: String,
46    /// Normalized workspace root (display form, bounded, no NULs).
47    pub root: String,
48    /// Manifest scan revision (`LocalWorkspaceManifestSnapshot::version`).
49    pub manifest_version: u64,
50    /// Eligible file count covered by this snapshot.
51    pub eligible_files: u64,
52    /// Domain-separated digest binding the eligible file identity set
53    /// (normalized path, size, modified time) at this manifest revision.
54    pub content_digest: String,
55    /// Eligibility policy revision applied while selecting files.
56    pub eligibility_revision: u64,
57    /// Highest document/LSP revision settled into this snapshot.
58    pub document_revision: u64,
59    /// Chunk catalog revision the derived indexes were built from.
60    pub catalog_revision: u64,
61    /// Derived index generation published for this snapshot.
62    pub index_generation: u64,
63    /// Logical observation time in milliseconds since the epoch.
64    pub observed_at_ms: u64,
65    /// Canonical digest over every identity field above.
66    pub snapshot_digest: String,
67}
68
69impl WorkspaceSourceSnapshotV1 {
70    #[allow(clippy::too_many_arguments)]
71    pub fn new(
72        root: impl Into<String>,
73        manifest_version: u64,
74        eligible_files: u64,
75        content_digest: impl Into<String>,
76        eligibility_revision: u64,
77        document_revision: u64,
78        catalog_revision: u64,
79        index_generation: u64,
80        observed_at_ms: u64,
81    ) -> Result<Self, WorkspaceSourceSnapshotError> {
82        let mut snapshot = Self {
83            schema: WORKSPACE_SOURCE_SNAPSHOT_SCHEMA_V1.to_owned(),
84            root: root.into(),
85            manifest_version,
86            eligible_files,
87            content_digest: content_digest.into(),
88            eligibility_revision,
89            document_revision,
90            catalog_revision,
91            index_generation,
92            observed_at_ms,
93            snapshot_digest: String::new(),
94        };
95        snapshot.validate_without_digest()?;
96        snapshot.snapshot_digest = snapshot.expected_digest()?;
97        Ok(snapshot)
98    }
99
100    /// Derive the snapshot identity from a manifest snapshot plus the
101    /// derived-data revisions that were admitted against it.
102    ///
103    /// `content_digest` must be computed with
104    /// [`workspace_content_digest`] over the same
105    /// eligible file set the manifest revision describes, so two callers that
106    /// observe the same files derive the same identity.
107    pub fn from_manifest(
108        manifest: &LocalWorkspaceManifestSnapshot,
109        content_digest: impl Into<String>,
110        eligibility_revision: u64,
111        document_revision: u64,
112        catalog_revision: u64,
113        index_generation: u64,
114    ) -> Result<Self, WorkspaceSourceSnapshotError> {
115        Self::new(
116            manifest.root.display().to_string(),
117            manifest.version,
118            u64::try_from(manifest.files.len())
119                .map_err(|_| WorkspaceSourceSnapshotError::InvalidField("eligible_files"))?,
120            content_digest,
121            eligibility_revision,
122            document_revision,
123            catalog_revision,
124            index_generation,
125            manifest.scanned_at_ms,
126        )
127    }
128
129    pub fn validate(&self) -> Result<(), WorkspaceSourceSnapshotError> {
130        self.validate_without_digest()?;
131        validate_digest("snapshot_digest", &self.snapshot_digest)?;
132        if self.snapshot_digest != self.expected_digest()? {
133            return Err(WorkspaceSourceSnapshotError::InvalidDigest(
134                "snapshot_digest",
135            ));
136        }
137        Ok(())
138    }
139
140    fn expected_digest(&self) -> Result<String, WorkspaceSourceSnapshotError> {
141        #[derive(Serialize)]
142        struct Identity<'a> {
143            schema: &'a str,
144            root: &'a str,
145            manifest_version: u64,
146            eligible_files: u64,
147            content_digest: &'a str,
148            eligibility_revision: u64,
149            document_revision: u64,
150            catalog_revision: u64,
151            index_generation: u64,
152            observed_at_ms: u64,
153        }
154        let bytes = serde_json::to_vec(&Identity {
155            schema: &self.schema,
156            root: &self.root,
157            manifest_version: self.manifest_version,
158            eligible_files: self.eligible_files,
159            content_digest: &self.content_digest,
160            eligibility_revision: self.eligibility_revision,
161            document_revision: self.document_revision,
162            catalog_revision: self.catalog_revision,
163            index_generation: self.index_generation,
164            observed_at_ms: self.observed_at_ms,
165        })
166        .map_err(|error| WorkspaceSourceSnapshotError::Serialization(error.to_string()))?;
167        Ok(digest_bytes(
168            WORKSPACE_SOURCE_SNAPSHOT_DIGEST_DOMAIN_V1,
169            &bytes,
170        ))
171    }
172
173    fn validate_without_digest(&self) -> Result<(), WorkspaceSourceSnapshotError> {
174        if self.schema != WORKSPACE_SOURCE_SNAPSHOT_SCHEMA_V1 {
175            return Err(WorkspaceSourceSnapshotError::UnsupportedSchema);
176        }
177        if self.root.is_empty()
178            || self.root.len() > MAX_ROOT_BYTES
179            || self.root.contains('\0')
180            || self.root.lines().count() != 1
181        {
182            return Err(WorkspaceSourceSnapshotError::InvalidField("root"));
183        }
184        if self.eligible_files > MAX_ELIGIBLE_FILES {
185            return Err(WorkspaceSourceSnapshotError::InvalidField("eligible_files"));
186        }
187        validate_digest("content_digest", &self.content_digest)?;
188        if self.observed_at_ms == 0 {
189            return Err(WorkspaceSourceSnapshotError::InvalidField("observed_at_ms"));
190        }
191        Ok(())
192    }
193
194    /// Whether every identity revision of `self` is at least `other`'s.
195    ///
196    /// Derived work may only move forward: a newer snapshot never re-binds to
197    /// older manifest, document, catalog, or index revisions.
198    pub fn revision_at_least(&self, other: &Self) -> bool {
199        self.manifest_version >= other.manifest_version
200            && self.eligibility_revision >= other.eligibility_revision
201            && self.document_revision >= other.document_revision
202            && self.catalog_revision >= other.catalog_revision
203            && self.index_generation >= other.index_generation
204    }
205
206    /// Whether a result produced at `self` may still be presented as current
207    /// against `live`.
208    ///
209    /// A manifest change (new scan revision or different eligible content)
210    /// always invalidates: stale results cannot cross a source snapshot
211    /// boundary. Derived-revision advances alone (a rebuilt index or a newer
212    /// settled document) do not invalidate older results bound to the same
213    /// manifest content, because the source they were derived from is
214    /// unchanged.
215    pub fn is_current_against(&self, live: &Self) -> bool {
216        self.root == live.root
217            && self.manifest_version == live.manifest_version
218            && self.content_digest == live.content_digest
219            && self.eligibility_revision == live.eligibility_revision
220            && self.catalog_revision <= live.catalog_revision
221    }
222
223    /// Whether a result produced at `self` is stale against `live`.
224    pub fn is_stale_against(&self, live: &Self) -> bool {
225        !self.is_current_against(live)
226    }
227}
228
229impl<'de> Deserialize<'de> for WorkspaceSourceSnapshotV1 {
230    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
231    where
232        D: serde::Deserializer<'de>,
233    {
234        #[derive(Deserialize)]
235        #[serde(rename_all = "camelCase", deny_unknown_fields)]
236        struct Wire {
237            schema: String,
238            root: String,
239            manifest_version: u64,
240            eligible_files: u64,
241            content_digest: String,
242            eligibility_revision: u64,
243            document_revision: u64,
244            catalog_revision: u64,
245            index_generation: u64,
246            observed_at_ms: u64,
247            snapshot_digest: String,
248        }
249
250        let wire = Wire::deserialize(deserializer)?;
251        let value = Self {
252            schema: wire.schema,
253            root: wire.root,
254            manifest_version: wire.manifest_version,
255            eligible_files: wire.eligible_files,
256            content_digest: wire.content_digest,
257            eligibility_revision: wire.eligibility_revision,
258            document_revision: wire.document_revision,
259            catalog_revision: wire.catalog_revision,
260            index_generation: wire.index_generation,
261            observed_at_ms: wire.observed_at_ms,
262            snapshot_digest: wire.snapshot_digest,
263        };
264        value.validate().map_err(serde::de::Error::custom)?;
265        Ok(value)
266    }
267}
268
269/// Compute the content identity of one manifest snapshot's eligible files.
270///
271/// The digest binds normalized path, size, and modified time in sorted path
272/// order, so any observed change to the eligible set produces a different
273/// identity. Two snapshots of the same content derive the same digest
274/// regardless of scan order.
275pub fn workspace_content_digest(manifest: &LocalWorkspaceManifestSnapshot) -> String {
276    let mut entries: Vec<(&str, u64, Option<u64>)> = manifest
277        .files
278        .iter()
279        .map(|file| (file.path.as_str(), file.size, file.modified_ms))
280        .collect();
281    entries.sort_unstable_by(|left, right| left.0.cmp(right.0));
282    let mut bytes = Vec::new();
283    for (path, size, modified_ms) in entries {
284        bytes.extend_from_slice(path.as_bytes());
285        bytes.push(0);
286        bytes.extend_from_slice(&size.to_le_bytes());
287        bytes.push(0);
288        bytes.extend_from_slice(&modified_ms.unwrap_or(0).to_le_bytes());
289        bytes.push(0);
290    }
291    digest_bytes(WORKSPACE_SOURCE_CONTENT_DOMAIN_V1, &bytes)
292}
293
294fn validate_digest(field: &'static str, value: &str) -> Result<(), WorkspaceSourceSnapshotError> {
295    if value.len() != 71
296        || !value.starts_with("sha256:")
297        || !value[7..]
298            .bytes()
299            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
300    {
301        return Err(WorkspaceSourceSnapshotError::InvalidDigest(field));
302    }
303    Ok(())
304}
305
306fn digest_bytes(domain: &str, bytes: &[u8]) -> String {
307    use sha2::{Digest, Sha256};
308    let mut hasher = Sha256::new();
309    hasher.update(domain.as_bytes());
310    hasher.update([0]);
311    hasher.update(bytes);
312    format!("sha256:{:x}", hasher.finalize())
313}
314
315#[cfg(test)]
316mod tests {
317    use super::*;
318    use crate::workspace::manifest::{LocalWorkspaceFile, LocalWorkspaceFileStatus};
319
320    fn manifest(version: u64, files: Vec<LocalWorkspaceFile>) -> LocalWorkspaceManifestSnapshot {
321        LocalWorkspaceManifestSnapshot {
322            version,
323            root: std::path::PathBuf::from("/tmp/ws"),
324            files,
325            scanned_at_ms: 1_000 + version,
326        }
327    }
328
329    fn file(path: &str, size: u64) -> LocalWorkspaceFile {
330        LocalWorkspaceFile {
331            path: path.to_owned(),
332            size,
333            modified_ms: Some(42),
334            language: Some("rust".to_owned()),
335            status: LocalWorkspaceFileStatus::Tracked,
336            binary: false,
337            generated: false,
338        }
339    }
340
341    fn snapshot(manifest_version: u64) -> WorkspaceSourceSnapshotV1 {
342        let m = manifest(manifest_version, vec![file("src/main.rs", 10)]);
343        WorkspaceSourceSnapshotV1::from_manifest(&m, workspace_content_digest(&m), 1, 2, 3, 4)
344            .unwrap()
345    }
346
347    #[test]
348    fn identity_is_deterministic_and_order_independent() {
349        let a = manifest(7, vec![file("a.rs", 1), file("b.rs", 2)]);
350        let b = manifest(7, vec![file("b.rs", 2), file("a.rs", 1)]);
351        assert_eq!(workspace_content_digest(&a), workspace_content_digest(&b));
352
353        let first =
354            WorkspaceSourceSnapshotV1::from_manifest(&a, workspace_content_digest(&a), 1, 2, 3, 4)
355                .unwrap();
356        assert!(first.validate().is_ok());
357        let encoded = serde_json::to_string(&first).unwrap();
358        let decoded: WorkspaceSourceSnapshotV1 = serde_json::from_str(&encoded).unwrap();
359        assert_eq!(decoded, first);
360
361        let mut unknown = serde_json::to_value(&first).unwrap();
362        unknown
363            .as_object_mut()
364            .unwrap()
365            .insert("future".to_owned(), serde_json::json!(true));
366        assert!(serde_json::from_value::<WorkspaceSourceSnapshotV1>(unknown).is_err());
367    }
368
369    #[test]
370    fn tampering_fails_closed() {
371        let mut forged = snapshot(7);
372        forged.catalog_revision = 99;
373        assert!(forged.validate().is_err());
374    }
375
376    #[test]
377    fn stale_results_cannot_cross_a_source_boundary() {
378        let base = snapshot(7);
379        // Same manifest content with a rebuilt index stays current.
380        let rebuilt = WorkspaceSourceSnapshotV1::from_manifest(
381            &manifest(7, vec![file("src/main.rs", 10)]),
382            workspace_content_digest(&manifest(7, vec![file("src/main.rs", 10)])),
383            1,
384            2,
385            3,
386            9,
387        )
388        .unwrap();
389        assert!(base.is_current_against(&rebuilt));
390        assert!(!base.is_stale_against(&rebuilt));
391
392        // A new manifest revision invalidates prior results.
393        let advanced = snapshot(8);
394        assert!(base.is_stale_against(&advanced));
395        assert!(!base.is_current_against(&advanced));
396
397        // Changed eligible content at the same revision invalidates.
398        let changed = {
399            let m = manifest(7, vec![file("src/main.rs", 11)]);
400            WorkspaceSourceSnapshotV1::from_manifest(&m, workspace_content_digest(&m), 1, 2, 3, 4)
401                .unwrap()
402        };
403        assert!(base.is_stale_against(&changed));
404
405        // Revisions only move forward.
406        assert!(rebuilt.revision_at_least(&base));
407        assert!(!base.revision_at_least(&advanced));
408    }
409
410    #[test]
411    fn roots_cannot_be_confused() {
412        let other_root = {
413            let m = manifest(7, vec![file("src/main.rs", 10)]);
414            let mut s = WorkspaceSourceSnapshotV1::from_manifest(
415                &m,
416                workspace_content_digest(&m),
417                1,
418                2,
419                3,
420                4,
421            )
422            .unwrap();
423            s.root = "/tmp/other".to_owned();
424            s.snapshot_digest = s.expected_digest().unwrap();
425            s
426        };
427        assert!(snapshot(7).is_stale_against(&other_root));
428    }
429}
430
431#[cfg(test)]
432mod qualification {
433    use super::*;
434    use crate::workspace::manifest::{LocalWorkspaceFile, LocalWorkspaceFileStatus};
435    use crate::workspace::retrieval::{
436        ChunkCatalogLimits, ChunkingConfig, WorkspaceChunkCatalog, WorkspaceIndexError,
437    };
438    use crate::workspace::WorkspacePath;
439
440    fn catalog() -> std::sync::Arc<WorkspaceChunkCatalog> {
441        WorkspaceChunkCatalog::new(
442            ChunkingConfig::default(),
443            ChunkCatalogLimits {
444                max_files: 64,
445                max_chunks: 512,
446                max_text_bytes: 1024 * 1024,
447                max_index_bytes: 8 * 1024 * 1024,
448            },
449        )
450        .unwrap()
451    }
452
453    fn manifest_for(
454        catalog_snapshot: &crate::workspace::retrieval::ChunkCatalogSnapshot,
455    ) -> LocalWorkspaceManifestSnapshot {
456        let files = catalog_snapshot
457            .paths()
458            .into_iter()
459            .map(|path| LocalWorkspaceFile {
460                path,
461                size: 32,
462                modified_ms: Some(7),
463                language: Some("rust".to_owned()),
464                status: LocalWorkspaceFileStatus::Tracked,
465                binary: false,
466                generated: false,
467            })
468            .collect();
469        LocalWorkspaceManifestSnapshot {
470            version: catalog_snapshot.source_revision(),
471            root: std::path::PathBuf::from("/tmp/ws"),
472            files,
473            scanned_at_ms: 10_000,
474        }
475    }
476
477    /// KRN-4 exit-gate slice: a search hit's chunk identity, the catalog
478    /// revision it was served from, and the source snapshot that admitted it
479    /// are one traceable chain, and a concurrent edit invalidates prior
480    /// results instead of yielding a false current answer.
481    #[test]
482    fn retrieval_results_trace_to_one_snapshot_and_edits_invalidate() {
483        let catalog = catalog();
484        let path = WorkspacePath::from_normalized("src/main.rs");
485        catalog
486            .replace_file(&path, Some("rust"), 1, "fn main() {}\n")
487            .unwrap();
488        let admitted = catalog.snapshot().unwrap();
489        let manifest = manifest_for(&admitted);
490        let snapshot = WorkspaceSourceSnapshotV1::from_manifest(
491            &manifest,
492            workspace_content_digest(&manifest),
493            1,
494            0,
495            admitted.revision(),
496            1,
497        )
498        .unwrap();
499
500        // The served chunk carries the catalog's content digest; the
501        // snapshot binds the same catalog revision the hit was served from.
502        let chunk_digest = admitted
503            .content_digest(&path)
504            .expect("admitted file carries a content digest");
505        assert!(chunk_digest.starts_with("sha256:"));
506        assert_eq!(admitted.revision(), snapshot.catalog_revision);
507        assert_eq!(admitted.source_revision(), snapshot.manifest_version);
508        assert_eq!(
509            admitted.eligible_file_count() as u64,
510            snapshot.eligible_files
511        );
512
513        // A rebuilt index generation against the same source stays current.
514        let rebuilt = WorkspaceSourceSnapshotV1::from_manifest(
515            &manifest,
516            workspace_content_digest(&manifest),
517            1,
518            0,
519            admitted.revision(),
520            2,
521        )
522        .unwrap();
523        assert!(snapshot.is_current_against(&rebuilt));
524
525        // A concurrent edit advances the source revision; results bound to
526        // the old snapshot are stale and cannot be presented as current.
527        catalog
528            .replace_file(&path, Some("rust"), 2, "fn main() { changed }\n")
529            .unwrap();
530        let edited = catalog.snapshot().unwrap();
531        assert!(edited.source_revision() > admitted.source_revision());
532        let edited_manifest = manifest_for(&edited);
533        let live = WorkspaceSourceSnapshotV1::from_manifest(
534            &edited_manifest,
535            workspace_content_digest(&edited_manifest),
536            1,
537            0,
538            edited.revision(),
539            1,
540        )
541        .unwrap();
542        assert!(snapshot.is_stale_against(&live));
543        assert!(live.revision_at_least(&snapshot));
544
545        let _ = std::any::type_name::<WorkspaceIndexError>();
546    }
547}
a3s_code_core/workspace/source_snapshot.rs

a3s_code_core/workspace/
source_snapshot.rs
