Skip to main content

wire/
object_graph.rs

1// SPDX-License-Identifier: Apache-2.0
2use std::collections::{HashSet, VecDeque};
3
4use objects::{
5    object::{
6        ContentHash, SemanticEntryKind, SemanticIndexRoot, SemanticTreeNode, State,
7        StateAttachment, StateAttachmentBody, StateAttachmentId, StateAttachmentKind, StateId,
8        TreeEntryTarget,
9    },
10    store::{ObjectStore, pack::ObjectType as PackObjectType},
11};
12use serde::{Deserialize, Serialize};
13
14use crate::{ProtocolError, Result};
15
16#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
17pub enum ObjectId {
18    Hash(ContentHash),
19    StateId(StateId),
20    StateAttachment {
21        state: StateId,
22        id: StateAttachmentId,
23        /// The attachment's kind, a pure projection of its body
24        /// ([`StateAttachmentBody::kind`]). Carried through the wire so
25        /// descriptors self-describe their kind; the dedup/identity key is
26        /// still `(state, id)`, and kind is coherent under `Eq`/`Hash` because
27        /// it is a deterministic function of the same record.
28        kind: StateAttachmentKind,
29    },
30}
31
32#[derive(Debug, Clone, Serialize, Deserialize)]
33pub struct ObjectInfo {
34    pub id: ObjectId,
35    pub obj_type: ObjectType,
36    pub size: u64,
37    pub delta_base: Option<ContentHash>,
38}
39
40#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
41pub struct PlannedObject {
42    pub id: ObjectId,
43    pub obj_type: ObjectType,
44}
45
46#[derive(Debug, Clone)]
47pub struct StateClosureTransferObjects {
48    pub planned_objects: Vec<PlannedObject>,
49    pub full_objects: Option<Vec<ObjectInfo>>,
50}
51
52#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
53pub enum ObjectType {
54    Blob,
55    Tree,
56    State,
57    Action,
58    /// A `RedactionsBlob` sidecar — the rmp-encoded record(s) declaring
59    /// that a specific blob has been redacted (and possibly purged) by
60    /// an authorized operator. Keyed on the wire by `ObjectId::Hash` of
61    /// the *redacted blob*, since `Repository`'s sidecar store is
62    /// indexed that way.
63    Redaction,
64    /// A `StateVisibilityBlob` sidecar — the rmp-encoded record(s)
65    /// declaring a non-public audience tier for a specific state. Keyed
66    /// on the wire by `ObjectId::StateId` of the state, since the
67    /// per-state sidecar store is indexed that way. Like `Redaction`, it
68    /// is a sidecar record that lives outside the content-addressed pack
69    /// and ships via the per-object transfer path, not the pack.
70    StateVisibility,
71    StateAttachment,
72}
73
74#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
75pub enum ObjectTypeBucket {
76    Blob,
77    Tree,
78    State,
79    Action,
80    Redaction,
81    StateVisibility,
82    StateAttachment,
83}
84
85impl ObjectType {
86    pub fn wire_name(self) -> &'static str {
87        match self {
88            ObjectType::Blob => "blob",
89            ObjectType::Tree => "tree",
90            ObjectType::State => "state",
91            ObjectType::Action => "action",
92            ObjectType::Redaction => "redaction",
93            ObjectType::StateVisibility => "state_visibility",
94            ObjectType::StateAttachment => "state_attachment",
95        }
96    }
97
98    pub fn from_wire(value: &str) -> Result<Self> {
99        match value {
100            "blob" => Ok(ObjectType::Blob),
101            "tree" => Ok(ObjectType::Tree),
102            "state" => Ok(ObjectType::State),
103            "action" => Ok(ObjectType::Action),
104            "redaction" => Ok(ObjectType::Redaction),
105            "state_visibility" => Ok(ObjectType::StateVisibility),
106            "state_attachment" => Ok(ObjectType::StateAttachment),
107            _ => Err(ProtocolError::InvalidState(format!(
108                "unknown object type: {value}"
109            ))),
110        }
111    }
112
113    /// Whether this object type can ride the content-addressed native pack in
114    /// general (the historical, direction-agnostic predicate). Sidecar records
115    /// (`Redaction`, `StateVisibility`) live outside `.heddle/objects/` and can
116    /// never be packed; everything else can.
117    ///
118    /// Prefer [`ObjectType::packable_for_push`] /
119    /// [`ObjectType::packable_for_pull`] at transfer sites: packability is
120    /// direction-dependent for `StateAttachment` (see those methods). This
121    /// method retains the pull/general semantics so pack construction,
122    /// have-set, and local-copy planners are unchanged.
123    pub fn packable(self) -> bool {
124        !matches!(self, ObjectType::Redaction | ObjectType::StateVisibility)
125    }
126
127    /// Whether this object type may ride the client→server **push** pack.
128    ///
129    /// `StateAttachment` is deliberately excluded on push even though it is a
130    /// content-addressed object: a deployed server rejects any pack-carried
131    /// attachment as a forgery-prevention measure (weft#549). Pushed
132    /// attachments must instead ride the out-of-pack **sidecar** lane (the same
133    /// lane as `Redaction`/`StateVisibility`), where the server can verify them
134    /// per-kind at finalize while the pack itself stays forgery-sealed. Every
135    /// other type keeps its [`ObjectType::packable`] answer.
136    pub fn packable_for_push(self) -> bool {
137        self.packable() && !matches!(self, ObjectType::StateAttachment)
138    }
139
140    /// Whether this object type may ride the server→client **pull/clone** pack.
141    ///
142    /// Identical to [`ObjectType::packable`]: attachments are carried in the
143    /// pull pack exactly as today. The push/pull split exists solely so the
144    /// push direction can exclude `StateAttachment` without disturbing the
145    /// working pull carriage.
146    pub fn packable_for_pull(self) -> bool {
147        self.packable()
148    }
149
150    pub fn pack_object_type(self) -> Result<PackObjectType> {
151        match self {
152            ObjectType::Blob => Ok(PackObjectType::Blob),
153            ObjectType::Tree => Ok(PackObjectType::Tree),
154            ObjectType::State => Ok(PackObjectType::State),
155            ObjectType::Action => Ok(PackObjectType::Action),
156            ObjectType::StateAttachment => Ok(PackObjectType::StateAttachment),
157            ObjectType::Redaction => Err(ProtocolError::InvalidState(
158                "Redaction sidecar records cannot be packed into the content-addressed object pack"
159                    .to_string(),
160            )),
161            ObjectType::StateVisibility => Err(ProtocolError::InvalidState(
162                "StateVisibility sidecar records cannot be packed into the content-addressed object pack"
163                    .to_string(),
164            )),
165        }
166    }
167
168    pub fn bucket(self) -> ObjectTypeBucket {
169        match self {
170            ObjectType::Blob => ObjectTypeBucket::Blob,
171            ObjectType::Tree => ObjectTypeBucket::Tree,
172            ObjectType::State => ObjectTypeBucket::State,
173            ObjectType::Action => ObjectTypeBucket::Action,
174            ObjectType::Redaction => ObjectTypeBucket::Redaction,
175            ObjectType::StateVisibility => ObjectTypeBucket::StateVisibility,
176            ObjectType::StateAttachment => ObjectTypeBucket::StateAttachment,
177        }
178    }
179}
180
181#[derive(Debug, Clone, Default)]
182pub struct StateClosureOptions {
183    pub depth: Option<u32>,
184    pub exclude_states: Vec<StateId>,
185}
186
187pub fn enumerate_state_closure(
188    store: &impl ObjectStore,
189    state_id: StateId,
190) -> Result<Vec<ObjectInfo>> {
191    enumerate_state_closure_with_options(store, state_id, StateClosureOptions::default())
192}
193
194pub fn enumerate_state_closure_with_options(
195    store: &impl ObjectStore,
196    state_id: StateId,
197    options: StateClosureOptions,
198) -> Result<Vec<ObjectInfo>> {
199    let mut out = Vec::new();
200    walk_state_closure(store, state_id, options, |event| {
201        if let Some(info) = object_info_from_event(store, event)? {
202            out.push(info);
203        }
204        Ok(())
205    })?;
206
207    Ok(out)
208}
209
210pub fn enumerate_state_closure_plan(
211    store: &impl ObjectStore,
212    state_id: StateId,
213) -> Result<Vec<PlannedObject>> {
214    enumerate_state_closure_plan_with_options(store, state_id, StateClosureOptions::default())
215}
216
217pub fn enumerate_state_closure_plan_with_options(
218    store: &impl ObjectStore,
219    state_id: StateId,
220    options: StateClosureOptions,
221) -> Result<Vec<PlannedObject>> {
222    let mut out = Vec::new();
223    walk_state_closure(store, state_id, options, |event| {
224        if let Some(object) = planned_object_from_event(store, event)? {
225            out.push(object);
226        }
227        Ok(())
228    })?;
229
230    Ok(out)
231}
232
233pub fn enumerate_state_closure_transfer_with_options(
234    store: &impl ObjectStore,
235    state_id: StateId,
236    options: StateClosureOptions,
237    full_descriptor_object_threshold: usize,
238) -> Result<StateClosureTransferObjects> {
239    let mut planned_objects = Vec::new();
240    let mut full_objects = Some(Vec::new());
241
242    walk_state_closure(store, state_id, options, |event| {
243        if let Some(object) = planned_object_from_event(store, event)? {
244            planned_objects.push(object);
245        }
246
247        if full_objects.is_some() && planned_objects.len() > full_descriptor_object_threshold {
248            full_objects = None;
249        }
250        if let Some(objects) = full_objects.as_mut()
251            && let Some(info) = object_info_from_event(store, event)?
252        {
253            objects.push(info);
254        }
255
256        Ok(())
257    })?;
258
259    Ok(StateClosureTransferObjects {
260        planned_objects,
261        full_objects,
262    })
263}
264
265/// Enumerate a transfer delta while treating `boundary_states` as complete
266/// server-held roots. Unlike [`StateClosureOptions::exclude_states`], this does
267/// not expand each boundary's tree/history to build a hash exclusion set: the
268/// walk stops as soon as it reaches the boundary state. Objects reused by the
269/// new tip may still be advertised, and the receiver's have-set filters them.
270/// This keeps incremental push planning proportional to the new history.
271pub fn enumerate_state_closure_transfer_from_boundaries(
272    store: &impl ObjectStore,
273    state_id: StateId,
274    boundary_states: &[StateId],
275    full_descriptor_object_threshold: usize,
276) -> Result<StateClosureTransferObjects> {
277    let mut planned_objects = Vec::new();
278    let mut full_objects = Some(Vec::new());
279    let excluded_states = boundary_states.iter().copied().collect();
280
281    walk_state_closure_with_exclusions(
282        store,
283        state_id,
284        None,
285        excluded_states,
286        HashSet::new(),
287        |event| {
288            if let Some(object) = planned_object_from_event(store, event)? {
289                planned_objects.push(object);
290            }
291
292            if full_objects.is_some() && planned_objects.len() > full_descriptor_object_threshold {
293                full_objects = None;
294            }
295            if let Some(objects) = full_objects.as_mut()
296                && let Some(info) = object_info_from_event(store, event)?
297            {
298                objects.push(info);
299            }
300
301            Ok(())
302        },
303    )?;
304
305    Ok(StateClosureTransferObjects {
306        planned_objects,
307        full_objects,
308    })
309}
310
311#[derive(Debug, Clone, Copy, PartialEq, Eq)]
312enum BlobSource {
313    Tree,
314    StateMetadata,
315}
316
317#[derive(Debug, Clone, Copy)]
318enum StateClosureEvent<'a> {
319    State {
320        id: StateId,
321        state: &'a State,
322    },
323    Tree {
324        hash: ContentHash,
325        tree: &'a objects::object::Tree,
326    },
327    Blob {
328        hash: ContentHash,
329        source: BlobSource,
330    },
331    Redaction {
332        blob: ContentHash,
333    },
334    StateVisibility {
335        state: StateId,
336    },
337    StateAttachment {
338        state: StateId,
339        attachment: &'a StateAttachment,
340    },
341    ExcludedState {
342        id: StateId,
343    },
344    ExcludedHash {
345        hash: ContentHash,
346    },
347}
348
349fn walk_state_closure(
350    store: &impl ObjectStore,
351    state_id: StateId,
352    options: StateClosureOptions,
353    visit: impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
354) -> Result<()> {
355    let (excluded_states, excluded_hashes) = collect_excluded(store, &options.exclude_states)?;
356
357    walk_state_closure_with_exclusions(
358        store,
359        state_id,
360        options.depth,
361        excluded_states,
362        excluded_hashes,
363        visit,
364    )
365}
366
367fn walk_state_closure_with_exclusions(
368    store: &impl ObjectStore,
369    state_id: StateId,
370    max_depth: Option<u32>,
371    excluded_states: HashSet<StateId>,
372    excluded_hashes: HashSet<ContentHash>,
373    mut visit: impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
374) -> Result<()> {
375    let mut seen_states: HashSet<StateId> = HashSet::new();
376    let mut seen_hashes: HashSet<ContentHash> = HashSet::new();
377    let mut queue: VecDeque<(StateId, u32)> = VecDeque::new();
378    queue.push_back((state_id, 0));
379
380    while let Some((id, depth)) = queue.pop_front() {
381        if excluded_states.contains(&id) {
382            visit(StateClosureEvent::ExcludedState { id })?;
383            continue;
384        }
385        if !seen_states.insert(id) {
386            continue;
387        }
388
389        let state = store
390            .get_state(&id)?
391            .ok_or_else(|| ProtocolError::ObjectNotFound(id.to_string()))?;
392
393        visit(StateClosureEvent::State { id, state: &state })?;
394        if store.has_state_visibility_for_state(&id)? {
395            visit(StateClosureEvent::StateVisibility { state: id })?;
396        }
397        for attachment in store.list_state_attachments(&id)? {
398            visit(StateClosureEvent::StateAttachment {
399                state: id,
400                attachment: &attachment,
401            })?;
402            match attachment.body {
403                StateAttachmentBody::Context(root) => walk_tree_closure_filtered(
404                    store,
405                    root,
406                    &excluded_hashes,
407                    &mut seen_hashes,
408                    &mut visit,
409                )?,
410                StateAttachmentBody::RiskSignals(hash)
411                | StateAttachmentBody::ReviewSignatures(hash)
412                | StateAttachmentBody::Discussions(hash)
413                | StateAttachmentBody::StructuredConflicts(hash) => walk_blob_filtered(
414                    store,
415                    hash,
416                    BlobSource::StateMetadata,
417                    &excluded_hashes,
418                    &mut seen_hashes,
419                    &mut visit,
420                )?,
421                StateAttachmentBody::SemanticIndex(root) => walk_semantic_index_closure(
422                    store,
423                    root,
424                    &excluded_hashes,
425                    &mut seen_hashes,
426                    &mut visit,
427                )?,
428                StateAttachmentBody::Signature(_) => {}
429            }
430        }
431
432        if max_depth.map(|max| depth < max).unwrap_or(true) {
433            for parent in &state.parents {
434                queue.push_back((*parent, depth + 1));
435            }
436        }
437
438        walk_tree_closure_filtered(
439            store,
440            state.tree,
441            &excluded_hashes,
442            &mut seen_hashes,
443            &mut visit,
444        )?;
445        if let Some(provenance_root) = state.provenance {
446            walk_tree_closure_filtered(
447                store,
448                provenance_root,
449                &excluded_hashes,
450                &mut seen_hashes,
451                &mut visit,
452            )?;
453        }
454    }
455
456    Ok(())
457}
458
459fn walk_tree_closure_filtered(
460    store: &impl ObjectStore,
461    tree_hash: ContentHash,
462    excluded: &HashSet<ContentHash>,
463    seen: &mut HashSet<ContentHash>,
464    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
465) -> Result<()> {
466    if excluded.contains(&tree_hash) {
467        visit(StateClosureEvent::ExcludedHash { hash: tree_hash })?;
468        return Ok(());
469    }
470    if !seen.insert(tree_hash) {
471        return Ok(());
472    }
473
474    let tree = store
475        .get_tree(&tree_hash)?
476        .ok_or_else(|| ProtocolError::ObjectNotFound(tree_hash.to_hex()))?;
477
478    visit(StateClosureEvent::Tree {
479        hash: tree_hash,
480        tree: &tree,
481    })?;
482
483    for entry in tree.entries() {
484        match entry.target() {
485            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
486                walk_blob_filtered(store, *hash, BlobSource::Tree, excluded, seen, visit)?;
487            }
488            TreeEntryTarget::Tree { hash } => {
489                walk_tree_closure_filtered(store, *hash, excluded, seen, visit)?;
490            }
491            TreeEntryTarget::Gitlink { .. } => {}
492            // Native child-spool edge: its target lives in a separate spool
493            // object graph, not this store, so it is not walked here.
494            TreeEntryTarget::Spoollink { .. } => {}
495        }
496    }
497
498    Ok(())
499}
500
501fn walk_blob_filtered(
502    store: &impl ObjectStore,
503    blob_hash: ContentHash,
504    source: BlobSource,
505    excluded: &HashSet<ContentHash>,
506    seen: &mut HashSet<ContentHash>,
507    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
508) -> Result<()> {
509    if excluded.contains(&blob_hash) {
510        visit(StateClosureEvent::ExcludedHash { hash: blob_hash })?;
511        return Ok(());
512    }
513    if !seen.insert(blob_hash) {
514        return Ok(());
515    }
516    visit(StateClosureEvent::Blob {
517        hash: blob_hash,
518        source,
519    })?;
520    if store.has_redactions_for_blob(&blob_hash)? {
521        visit(StateClosureEvent::Redaction { blob: blob_hash })?;
522    }
523    Ok(())
524}
525
526/// Walk the merkle semantic-index closure rooted at `root_hash` (a
527/// `SemanticIndexRoot` blob), emitting every reachable semantic node blob.
528///
529/// All semantic-index nodes (root, tree nodes, file nodes) are stored as
530/// ordinary content-addressed blobs, so replication just enumerates them.
531/// Opaque entries point back at raw source blobs already covered by the state's
532/// tree closure, so they are not re-walked here.
533///
534/// Iterative (explicit stack) so a crafted deep `SemanticTreeNode` chain in a
535/// pushed state can't overflow the stack. A missing or undecodable node in the
536/// closure is a HARD failure (`ObjectNotFound`/`Serialization`) — a partial or
537/// corrupt semantic closure must never be shipped silently.
538fn walk_semantic_index_closure(
539    store: &impl ObjectStore,
540    root_hash: ContentHash,
541    excluded: &HashSet<ContentHash>,
542    seen: &mut HashSet<ContentHash>,
543    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
544) -> Result<()> {
545    // Stack of (node_hash, is_tree_node): the root and dir nodes must be decoded
546    // to enumerate their children; file/opaque leaves are emitted only.
547    let mut stack: Vec<ContentHash> = vec![root_hash];
548    while let Some(node_hash) = stack.pop() {
549        if !emit_semantic_blob(store, node_hash, excluded, seen, visit)? {
550            continue; // excluded (present on the far side) or already seen.
551        }
552        let blob = store
553            .get_blob(&node_hash)?
554            .ok_or_else(|| ProtocolError::ObjectNotFound(node_hash.to_hex()))?;
555        // The root and tree nodes decode to the same `SemanticTreeNode.entries`
556        // shape after the root's one indirection; walk uniformly by decoding
557        // every interior node as a tree node, tolerating the root shape.
558        let node = decode_semantic_container(&blob, node_hash)?;
559        for child in node {
560            match child {
561                SemanticChild::Interior(hash) => stack.push(hash),
562                SemanticChild::Leaf(hash) => {
563                    // Emit the leaf (file node) blob; it has no children.
564                    emit_semantic_blob(store, hash, excluded, seen, visit)?;
565                }
566            }
567        }
568    }
569    Ok(())
570}
571
572/// The two child kinds encountered walking the semantic closure: an interior
573/// node to descend into, or a leaf (file node) blob to emit.
574enum SemanticChild {
575    Interior(ContentHash),
576    Leaf(ContentHash),
577}
578
579/// Decode a semantic-closure node — the root (which points at a tree node) or a
580/// tree node — into the child hashes to walk. Missing/corrupt is a hard error.
581fn decode_semantic_container(
582    blob: &objects::object::Blob,
583    node_hash: ContentHash,
584) -> Result<Vec<SemanticChild>> {
585    // Try the root shape first (it has an extra `tree` indirection), then the
586    // tree-node shape. Content-addressed hashes make this unambiguous in
587    // practice; a blob that decodes as neither is corrupt.
588    if let Ok(root) = SemanticIndexRoot::decode(blob.content()) {
589        return Ok(vec![SemanticChild::Interior(root.tree)]);
590    }
591    let node = SemanticTreeNode::decode(blob.content())
592        .map_err(|err| ProtocolError::Serialization(format!("semantic node {node_hash}: {err}")))?;
593    Ok(node
594        .entries
595        .iter()
596        .filter_map(|entry| match entry.kind {
597            SemanticEntryKind::Dir => Some(SemanticChild::Interior(entry.node)),
598            SemanticEntryKind::File => Some(SemanticChild::Leaf(entry.node)),
599            // Opaque `node` is the raw source blob, already in the tree closure.
600            SemanticEntryKind::Opaque => None,
601        })
602        .collect())
603}
604
605/// Emit a semantic node blob as state metadata. Returns `true` when the blob
606/// was newly visited (so the caller may descend into it), `false` when it was
607/// excluded (present on the far side) or already seen. A blob that is neither
608/// excluded nor present is a HARD failure — the closure must be complete.
609fn emit_semantic_blob(
610    store: &impl ObjectStore,
611    hash: ContentHash,
612    excluded: &HashSet<ContentHash>,
613    seen: &mut HashSet<ContentHash>,
614    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
615) -> Result<bool> {
616    if excluded.contains(&hash) {
617        visit(StateClosureEvent::ExcludedHash { hash })?;
618        return Ok(false);
619    }
620    if !seen.insert(hash) {
621        return Ok(false);
622    }
623    if store.get_blob(&hash)?.is_none() {
624        return Err(ProtocolError::ObjectNotFound(hash.to_hex()));
625    }
626    visit(StateClosureEvent::Blob {
627        hash,
628        source: BlobSource::StateMetadata,
629    })?;
630    Ok(true)
631}
632
633/// Collect every semantic-index node blob hash reachable from `root_hash` into
634/// `excluded`, for the have-set computation. Iterative + tolerant (a broken
635/// have-set index is not fatal — it just means fewer objects are marked
636/// already-present, which is safe over-fetching, not corruption).
637fn collect_semantic_hashes(
638    store: &impl ObjectStore,
639    root_hash: ContentHash,
640    excluded: &mut HashSet<ContentHash>,
641) -> Result<()> {
642    let mut stack: Vec<ContentHash> = vec![root_hash];
643    while let Some(node_hash) = stack.pop() {
644        if !excluded.insert(node_hash) {
645            continue;
646        }
647        let Some(blob) = store.get_blob(&node_hash)? else {
648            continue;
649        };
650        let children = match decode_semantic_container(&blob, node_hash) {
651            Ok(children) => children,
652            Err(_) => continue,
653        };
654        for child in children {
655            match child {
656                SemanticChild::Interior(hash) => stack.push(hash),
657                SemanticChild::Leaf(hash) => {
658                    excluded.insert(hash);
659                }
660            }
661        }
662    }
663    Ok(())
664}
665
666fn object_info_from_event(
667    store: &impl ObjectStore,
668    event: StateClosureEvent<'_>,
669) -> Result<Option<ObjectInfo>> {
670    match event {
671        StateClosureEvent::State { id, state } => {
672            let state_bytes = rmp_serde::to_vec_named(state)?;
673            Ok(Some(ObjectInfo {
674                id: ObjectId::StateId(id),
675                obj_type: ObjectType::State,
676                size: state_bytes.len() as u64,
677                delta_base: None,
678            }))
679        }
680        StateClosureEvent::Tree { hash, tree } => {
681            let tree_bytes = rmp_serde::to_vec_named(tree)?;
682            Ok(Some(ObjectInfo {
683                id: ObjectId::Hash(hash),
684                obj_type: ObjectType::Tree,
685                size: tree_bytes.len() as u64,
686                delta_base: None,
687            }))
688        }
689        StateClosureEvent::Blob { hash, .. } => {
690            let blob = store
691                .get_blob(&hash)?
692                .ok_or_else(|| ProtocolError::ObjectNotFound(hash.to_hex()))?;
693            Ok(Some(ObjectInfo {
694                id: ObjectId::Hash(hash),
695                obj_type: ObjectType::Blob,
696                size: blob.size() as u64,
697                delta_base: None,
698            }))
699        }
700        StateClosureEvent::Redaction { blob } => Ok(store
701            .get_redactions_bytes_for_blob(&blob)?
702            .map(|bytes| ObjectInfo {
703                id: ObjectId::Hash(blob),
704                obj_type: ObjectType::Redaction,
705                size: bytes.len() as u64,
706                delta_base: None,
707            })),
708        StateClosureEvent::StateVisibility { state } => Ok(store
709            .get_state_visibility_bytes_for_state(&state)?
710            .map(|bytes| ObjectInfo {
711                id: ObjectId::StateId(state),
712                obj_type: ObjectType::StateVisibility,
713                size: bytes.len() as u64,
714                delta_base: None,
715            })),
716        StateClosureEvent::StateAttachment { state, attachment } => {
717            let bytes = rmp_serde::to_vec_named(attachment)?;
718            Ok(Some(ObjectInfo {
719                id: ObjectId::StateAttachment {
720                    state,
721                    id: attachment.id(),
722                    kind: attachment.body.kind(),
723                },
724                obj_type: ObjectType::StateAttachment,
725                size: bytes.len() as u64,
726                delta_base: None,
727            }))
728        }
729        StateClosureEvent::ExcludedState { id } => {
730            let _ = id;
731            Ok(None)
732        }
733        StateClosureEvent::ExcludedHash { hash } => {
734            let _ = hash;
735            Ok(None)
736        }
737    }
738}
739
740fn planned_object_from_event(
741    store: &impl ObjectStore,
742    event: StateClosureEvent<'_>,
743) -> Result<Option<PlannedObject>> {
744    match event {
745        StateClosureEvent::State { id, .. } => Ok(Some(PlannedObject {
746            id: ObjectId::StateId(id),
747            obj_type: ObjectType::State,
748        })),
749        StateClosureEvent::Tree { hash, .. } => Ok(Some(PlannedObject {
750            id: ObjectId::Hash(hash),
751            obj_type: ObjectType::Tree,
752        })),
753        StateClosureEvent::Blob { hash, source } => {
754            if source == BlobSource::StateMetadata && store.get_blob(&hash)?.is_none() {
755                return Err(ProtocolError::ObjectNotFound(hash.to_hex()));
756            }
757            Ok(Some(PlannedObject {
758                id: ObjectId::Hash(hash),
759                obj_type: ObjectType::Blob,
760            }))
761        }
762        StateClosureEvent::Redaction { blob } => Ok(Some(PlannedObject {
763            id: ObjectId::Hash(blob),
764            obj_type: ObjectType::Redaction,
765        })),
766        StateClosureEvent::StateVisibility { state } => Ok(Some(PlannedObject {
767            id: ObjectId::StateId(state),
768            obj_type: ObjectType::StateVisibility,
769        })),
770        StateClosureEvent::StateAttachment { state, attachment } => Ok(Some(PlannedObject {
771            id: ObjectId::StateAttachment {
772                state,
773                id: attachment.id(),
774                kind: attachment.body.kind(),
775            },
776            obj_type: ObjectType::StateAttachment,
777        })),
778        StateClosureEvent::ExcludedState { id } => {
779            let _ = id;
780            Ok(None)
781        }
782        StateClosureEvent::ExcludedHash { hash } => {
783            let _ = hash;
784            Ok(None)
785        }
786    }
787}
788
789pub fn missing_blobs_in_tree(
790    store: &impl ObjectStore,
791    tree_hash: ContentHash,
792) -> Result<Vec<ContentHash>> {
793    let mut missing = Vec::new();
794    collect_missing_blobs_recursive(store, &tree_hash, &mut missing)?;
795    Ok(missing)
796}
797
798fn collect_missing_blobs_recursive(
799    store: &impl ObjectStore,
800    tree_hash: &ContentHash,
801    missing: &mut Vec<ContentHash>,
802) -> Result<()> {
803    let Some(tree) = store.get_tree(tree_hash).map_err(|err| {
804        ProtocolError::InvalidState(format!(
805            "load tree {} while collecting lazy hydration missing blobs: {err}",
806            tree_hash.to_hex()
807        ))
808    })?
809    else {
810        return Ok(());
811    };
812
813    for entry in tree.entries() {
814        match entry.target() {
815            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
816                if !store.has_blob(hash).map_err(|err| {
817                    ProtocolError::InvalidState(format!(
818                        "check blob {} while collecting lazy hydration missing blobs: {err}",
819                        hash.to_hex()
820                    ))
821                })? {
822                    missing.push(*hash);
823                }
824            }
825            TreeEntryTarget::Tree { hash } => {
826                collect_missing_blobs_recursive(store, hash, missing)?;
827            }
828            TreeEntryTarget::Gitlink { .. } => {}
829            // Native child-spool edge: its target lives in a separate spool
830            // object graph, not this store, so it is not walked here.
831            TreeEntryTarget::Spoollink { .. } => {}
832        }
833    }
834    Ok(())
835}
836
837fn collect_excluded(
838    store: &impl ObjectStore,
839    roots: &[StateId],
840) -> Result<(HashSet<StateId>, HashSet<ContentHash>)> {
841    if roots.is_empty() {
842        return Ok((HashSet::new(), HashSet::new()));
843    }
844
845    let mut excluded_states: HashSet<StateId> = HashSet::new();
846    let mut excluded_hashes: HashSet<ContentHash> = HashSet::new();
847    let mut queue: VecDeque<StateId> = VecDeque::new();
848
849    for id in roots {
850        queue.push_back(*id);
851    }
852
853    while let Some(id) = queue.pop_front() {
854        if !excluded_states.insert(id) {
855            continue;
856        }
857
858        let state = match store.get_state(&id)? {
859            Some(state) => state,
860            None => continue,
861        };
862
863        for parent in &state.parents {
864            queue.push_back(*parent);
865        }
866
867        collect_tree_hashes(store, state.tree, &mut excluded_hashes)?;
868        if let Some(provenance_root) = state.provenance {
869            collect_tree_hashes(store, provenance_root, &mut excluded_hashes)?;
870        }
871        for attachment in store.list_state_attachments(&id)? {
872            match attachment.body {
873                StateAttachmentBody::Context(root) => {
874                    collect_tree_hashes(store, root, &mut excluded_hashes)?
875                }
876                StateAttachmentBody::RiskSignals(hash)
877                | StateAttachmentBody::ReviewSignatures(hash)
878                | StateAttachmentBody::Discussions(hash)
879                | StateAttachmentBody::StructuredConflicts(hash) => {
880                    excluded_hashes.insert(hash);
881                }
882                StateAttachmentBody::SemanticIndex(root) => {
883                    collect_semantic_hashes(store, root, &mut excluded_hashes)?;
884                }
885                StateAttachmentBody::Signature(_) => {}
886            }
887        }
888    }
889
890    Ok((excluded_states, excluded_hashes))
891}
892
893fn collect_tree_hashes(
894    store: &impl ObjectStore,
895    tree_hash: ContentHash,
896    excluded: &mut HashSet<ContentHash>,
897) -> Result<()> {
898    if !excluded.insert(tree_hash) {
899        return Ok(());
900    }
901
902    let tree = match store.get_tree(&tree_hash)? {
903        Some(tree) => tree,
904        None => return Ok(()),
905    };
906
907    for entry in tree.entries() {
908        match entry.target() {
909            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
910                excluded.insert(*hash);
911            }
912            TreeEntryTarget::Tree { hash } => {
913                collect_tree_hashes(store, *hash, excluded)?;
914            }
915            TreeEntryTarget::Gitlink { .. } => {}
916            // Native child-spool edge: its target lives in a separate spool
917            // object graph, not this store, so it is not walked here.
918            TreeEntryTarget::Spoollink { .. } => {}
919        }
920    }
921
922    Ok(())
923}
924
925pub fn is_ancestor(
926    store: &impl ObjectStore,
927    ancestor: StateId,
928    descendant: StateId,
929) -> Result<bool> {
930    if ancestor == descendant {
931        return Ok(true);
932    }
933
934    let mut seen: HashSet<StateId> = HashSet::new();
935    let mut queue: VecDeque<StateId> = VecDeque::new();
936    queue.push_back(descendant);
937
938    while let Some(id) = queue.pop_front() {
939        if !seen.insert(id) {
940            continue;
941        }
942        let state = match store.get_state(&id)? {
943            Some(s) => s,
944            None => return Ok(false),
945        };
946        for parent in state.parents {
947            if parent == ancestor {
948                return Ok(true);
949            }
950            queue.push_back(parent);
951        }
952    }
953
954    Ok(false)
955}
956
957#[cfg(test)]
958mod tests {
959    use std::{
960        collections::HashSet,
961        sync::atomic::{AtomicUsize, Ordering},
962    };
963
964    use chrono::Utc;
965    use objects::{
966        object::{
967            Action, ActionId, Attribution, Blob, ContentHash, Discussion, DiscussionResolution,
968            DiscussionTurn, DiscussionsBlob, Principal, Redaction, State, StateAttachment,
969            StateAttachmentBody, StateId, StateVisibility, SymbolAnchor, Tree, TreeEntry,
970            VisibilityTier,
971        },
972        store::{ObjectStore, Result as StoreResult},
973    };
974    use repo::Repository;
975    use sley::ObjectId as GitObjectId;
976    use tempfile::TempDir;
977
978    use super::{
979        ObjectId, ObjectInfo, ObjectType, PlannedObject, StateClosureOptions,
980        enumerate_state_closure_plan_with_options,
981        enumerate_state_closure_transfer_from_boundaries,
982        enumerate_state_closure_transfer_with_options, enumerate_state_closure_with_options,
983        missing_blobs_in_tree,
984    };
985
986    fn pairs_from_full(objects: &[ObjectInfo]) -> HashSet<(ObjectId, ObjectType)> {
987        objects
988            .iter()
989            .map(|info| (info.id.clone(), info.obj_type))
990            .collect()
991    }
992
993    fn pairs_from_plan(objects: &[PlannedObject]) -> HashSet<(ObjectId, ObjectType)> {
994        objects
995            .iter()
996            .map(|info| (info.id.clone(), info.obj_type))
997            .collect()
998    }
999
1000    fn object_info_fingerprint(
1001        objects: &[ObjectInfo],
1002    ) -> Vec<(ObjectId, ObjectType, u64, Option<ContentHash>)> {
1003        objects
1004            .iter()
1005            .map(|info| (info.id.clone(), info.obj_type, info.size, info.delta_base))
1006            .collect()
1007    }
1008
1009    fn assert_plan_parity(
1010        repo: &Repository,
1011        state_id: StateId,
1012        options: StateClosureOptions,
1013    ) -> HashSet<(ObjectId, ObjectType)> {
1014        let full =
1015            enumerate_state_closure_with_options(repo.store(), state_id, options.clone()).unwrap();
1016        let plan =
1017            enumerate_state_closure_plan_with_options(repo.store(), state_id, options).unwrap();
1018
1019        let full_pairs = pairs_from_full(&full);
1020        let plan_pairs = pairs_from_plan(&plan);
1021        assert_eq!(full_pairs, plan_pairs);
1022        full_pairs
1023    }
1024
1025    fn assert_contains_object(
1026        objects: &HashSet<(ObjectId, ObjectType)>,
1027        id: ObjectId,
1028        obj_type: ObjectType,
1029    ) {
1030        assert!(
1031            objects.contains(&(id.clone(), obj_type)),
1032            "expected closure to contain {id:?} as {obj_type:?}: {objects:?}"
1033        );
1034    }
1035
1036    struct CountingStore<'a, S> {
1037        inner: &'a S,
1038        state_reads: AtomicUsize,
1039    }
1040
1041    impl<'a, S> CountingStore<'a, S> {
1042        fn new(inner: &'a S) -> Self {
1043            Self {
1044                inner,
1045                state_reads: AtomicUsize::new(0),
1046            }
1047        }
1048
1049        fn state_reads(&self) -> usize {
1050            self.state_reads.load(Ordering::SeqCst)
1051        }
1052    }
1053
1054    impl<S: ObjectStore> ObjectStore for CountingStore<'_, S> {
1055        fn get_blob(&self, hash: &ContentHash) -> StoreResult<Option<Blob>> {
1056            self.inner.get_blob(hash)
1057        }
1058
1059        fn put_blob(&self, blob: &Blob) -> StoreResult<ContentHash> {
1060            self.inner.put_blob(blob)
1061        }
1062
1063        fn has_blob(&self, hash: &ContentHash) -> StoreResult<bool> {
1064            self.inner.has_blob(hash)
1065        }
1066
1067        fn get_tree(&self, hash: &ContentHash) -> StoreResult<Option<Tree>> {
1068            self.inner.get_tree(hash)
1069        }
1070
1071        fn put_tree(&self, tree: &Tree) -> StoreResult<ContentHash> {
1072            self.inner.put_tree(tree)
1073        }
1074
1075        fn has_tree(&self, hash: &ContentHash) -> StoreResult<bool> {
1076            self.inner.has_tree(hash)
1077        }
1078
1079        fn get_state(&self, id: &StateId) -> StoreResult<Option<State>> {
1080            self.state_reads.fetch_add(1, Ordering::SeqCst);
1081            self.inner.get_state(id)
1082        }
1083
1084        fn put_state(&self, state: &State) -> StoreResult<()> {
1085            self.inner.put_state(state)
1086        }
1087
1088        fn has_state(&self, id: &StateId) -> StoreResult<bool> {
1089            self.inner.has_state(id)
1090        }
1091
1092        fn list_states(&self) -> StoreResult<Vec<StateId>> {
1093            self.inner.list_states()
1094        }
1095
1096        fn get_action(&self, id: &ActionId) -> StoreResult<Option<Action>> {
1097            self.inner.get_action(id)
1098        }
1099
1100        fn put_action(&self, action: &mut Action) -> StoreResult<ActionId> {
1101            self.inner.put_action(action)
1102        }
1103
1104        fn list_actions(&self) -> StoreResult<Vec<ActionId>> {
1105            self.inner.list_actions()
1106        }
1107
1108        fn list_blobs(&self) -> StoreResult<Vec<ContentHash>> {
1109            self.inner.list_blobs()
1110        }
1111
1112        fn list_trees(&self) -> StoreResult<Vec<ContentHash>> {
1113            self.inner.list_trees()
1114        }
1115    }
1116
1117    fn test_attribution() -> Attribution {
1118        Attribution::human(Principal::new("Graph Tester", "graph@example.com"))
1119    }
1120
1121    #[test]
1122    fn lean_closure_planner_matches_object_info_ids_and_types() {
1123        let temp = TempDir::new().unwrap();
1124        let repo = Repository::init_default(temp.path()).unwrap();
1125        std::fs::create_dir_all(temp.path().join("src")).unwrap();
1126        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1127        std::fs::write(temp.path().join("src/lib.rs"), "pub fn hi() {}\n").unwrap();
1128        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1129
1130        let full = enumerate_state_closure_with_options(
1131            repo.store(),
1132            state.state_id,
1133            StateClosureOptions::default(),
1134        )
1135        .unwrap();
1136        let lean = enumerate_state_closure_plan_with_options(
1137            repo.store(),
1138            state.state_id,
1139            StateClosureOptions::default(),
1140        )
1141        .unwrap();
1142
1143        let full_pairs = full
1144            .into_iter()
1145            .map(|info| (info.id, info.obj_type))
1146            .collect::<std::collections::HashSet<_>>();
1147        let lean_pairs = lean
1148            .into_iter()
1149            .map(|info| (info.id, info.obj_type))
1150            .collect::<std::collections::HashSet<_>>();
1151
1152        assert_eq!(full_pairs, lean_pairs);
1153        assert!(
1154            full_pairs
1155                .iter()
1156                .any(|(id, _)| matches!(id, ObjectId::StateId(_)))
1157        );
1158    }
1159
1160    #[test]
1161    fn transfer_boundary_stops_at_server_head_without_walking_its_history() {
1162        let temp = TempDir::new().unwrap();
1163        let repo = Repository::init_default(temp.path()).unwrap();
1164        let path = temp.path().join("story.txt");
1165
1166        std::fs::write(&path, "base\n").unwrap();
1167        let base = repo.snapshot(Some("base".to_string()), None).unwrap();
1168        std::fs::write(&path, "middle\n").unwrap();
1169        let middle = repo.snapshot(Some("middle".to_string()), None).unwrap();
1170        std::fs::write(&path, "tip\n").unwrap();
1171        let tip = repo.snapshot(Some("tip".to_string()), None).unwrap();
1172
1173        let counting = CountingStore::new(repo.store());
1174        let transfer = enumerate_state_closure_transfer_from_boundaries(
1175            &counting,
1176            tip.state_id,
1177            &[middle.state_id],
1178            512,
1179        )
1180        .unwrap();
1181        let states = transfer
1182            .planned_objects
1183            .iter()
1184            .filter_map(|object| match object.id {
1185                ObjectId::StateId(state) if object.obj_type == ObjectType::State => Some(state),
1186                _ => None,
1187            })
1188            .collect::<Vec<_>>();
1189
1190        assert_eq!(states, vec![tip.state_id]);
1191        assert!(!states.contains(&middle.state_id));
1192        assert!(!states.contains(&base.state_id));
1193        assert_eq!(
1194            counting.state_reads(),
1195            1,
1196            "the advertised server boundary must stop the walk before reading old states"
1197        );
1198    }
1199
1200    #[test]
1201    fn transfer_projection_matches_full_and_plan_on_mixed_state_closure_fixture() {
1202        let temp = TempDir::new().unwrap();
1203        let repo = Repository::init_default(temp.path()).unwrap();
1204
1205        let excluded_blob = repo
1206            .store()
1207            .put_blob(&Blob::from("excluded"))
1208            .expect("put excluded blob");
1209        let excluded_tree_hash = repo
1210            .store()
1211            .put_tree(&Tree::from_entries(vec![
1212                TreeEntry::file("excluded.txt", excluded_blob, false).unwrap(),
1213            ]))
1214            .expect("put excluded tree");
1215        let excluded_parent = State::new(excluded_tree_hash, Vec::new(), test_attribution());
1216        repo.store()
1217            .put_state(&excluded_parent)
1218            .expect("put excluded parent");
1219
1220        let redacted_blob = repo
1221            .store()
1222            .put_blob(&Blob::from("secret"))
1223            .expect("put redacted blob");
1224        let nested_blob = repo
1225            .store()
1226            .put_blob(&Blob::from("nested"))
1227            .expect("put nested blob");
1228        let symlink_blob = repo
1229            .store()
1230            .put_blob(&Blob::from("target"))
1231            .expect("put symlink blob");
1232        let context_blob = repo
1233            .store()
1234            .put_blob(&Blob::from("context"))
1235            .expect("put context blob");
1236        let provenance_blob = repo
1237            .store()
1238            .put_blob(&Blob::from("provenance"))
1239            .expect("put provenance blob");
1240        let risk_blob = repo
1241            .store()
1242            .put_blob(&Blob::from("risk"))
1243            .expect("put risk blob");
1244        let review_blob = repo
1245            .store()
1246            .put_blob(&Blob::from("review"))
1247            .expect("put review blob");
1248        let discussions_blob = repo
1249            .store()
1250            .put_blob(&Blob::from("discussion"))
1251            .expect("put discussion blob");
1252        let conflicts_blob = repo
1253            .store()
1254            .put_blob(&Blob::from("conflicts"))
1255            .expect("put conflicts blob");
1256
1257        let nested_tree_hash = repo
1258            .store()
1259            .put_tree(&Tree::from_entries(vec![
1260                TreeEntry::file("nested.txt", nested_blob, false).unwrap(),
1261                TreeEntry::symlink("latest", symlink_blob).unwrap(),
1262            ]))
1263            .expect("put nested tree");
1264        let context_tree_hash = repo
1265            .store()
1266            .put_tree(&Tree::from_entries(vec![
1267                TreeEntry::file("context.txt", context_blob, false).unwrap(),
1268            ]))
1269            .expect("put context tree");
1270        let provenance_tree_hash = repo
1271            .store()
1272            .put_tree(&Tree::from_entries(vec![
1273                TreeEntry::file("lineage.txt", provenance_blob, false).unwrap(),
1274            ]))
1275            .expect("put provenance tree");
1276        let gitlink_target: GitObjectId = "0303030303030303030303030303030303030303"
1277            .parse()
1278            .expect("git oid");
1279        let root_tree_hash = repo
1280            .store()
1281            .put_tree(&Tree::from_entries(vec![
1282                TreeEntry::file("secret.txt", redacted_blob, false).unwrap(),
1283                TreeEntry::directory("nested", nested_tree_hash).unwrap(),
1284                TreeEntry::gitlink("vendor", gitlink_target).unwrap(),
1285            ]))
1286            .expect("put root tree");
1287        let state = State::new(
1288            root_tree_hash,
1289            vec![excluded_parent.state_id],
1290            test_attribution(),
1291        )
1292        .with_provenance(provenance_tree_hash);
1293        repo.store().put_state(&state).expect("put state");
1294        for body in [
1295            StateAttachmentBody::Context(context_tree_hash),
1296            StateAttachmentBody::RiskSignals(risk_blob),
1297            StateAttachmentBody::ReviewSignatures(review_blob),
1298            StateAttachmentBody::Discussions(discussions_blob),
1299            StateAttachmentBody::StructuredConflicts(conflicts_blob),
1300        ] {
1301            repo.put_state_attachment(&StateAttachment {
1302                state_id: state.id(),
1303                body,
1304                attribution: state.attribution.clone(),
1305                created_at: Utc::now(),
1306                supersedes: None,
1307            })
1308            .unwrap();
1309        }
1310
1311        repo.put_redaction(Redaction {
1312            redacted_blob,
1313            state: state.state_id,
1314            path: "secret.txt".to_string(),
1315            reason: "test leak".to_string(),
1316            redactor: Principal::new("Tester", "tester@example.test"),
1317            redacted_at: Utc::now(),
1318            signature: None,
1319            purged_at: None,
1320            supersedes: None,
1321        })
1322        .expect("put redaction");
1323        repo.put_state_visibility(StateVisibility {
1324            state: state.state_id,
1325            tier: VisibilityTier::Restricted {
1326                scope_label: "security".to_string(),
1327            },
1328            embargo_until: None,
1329            declarer: Principal::new("Tester", "tester@example.test"),
1330            declared_at: Utc::now(),
1331            signature: None,
1332            supersedes: None,
1333        })
1334        .expect("put visibility");
1335
1336        let options = StateClosureOptions {
1337            depth: None,
1338            exclude_states: vec![excluded_parent.state_id],
1339        };
1340        let transfer = enumerate_state_closure_transfer_with_options(
1341            repo.store(),
1342            state.state_id,
1343            options.clone(),
1344            512,
1345        )
1346        .expect("transfer projection");
1347
1348        let full =
1349            enumerate_state_closure_with_options(repo.store(), state.state_id, options.clone())
1350                .expect("full closure");
1351        let plan = enumerate_state_closure_plan_with_options(repo.store(), state.state_id, options)
1352            .expect("plan closure");
1353        assert_eq!(
1354            transfer
1355                .full_objects
1356                .as_deref()
1357                .map(object_info_fingerprint),
1358            Some(object_info_fingerprint(&full))
1359        );
1360        assert_eq!(transfer.planned_objects, plan);
1361
1362        let full_pairs = pairs_from_full(&full);
1363        assert_eq!(full_pairs, pairs_from_plan(&plan));
1364        assert_contains_object(
1365            &full_pairs,
1366            ObjectId::StateId(state.state_id),
1367            ObjectType::State,
1368        );
1369        assert_contains_object(
1370            &full_pairs,
1371            ObjectId::StateId(state.state_id),
1372            ObjectType::StateVisibility,
1373        );
1374        assert_contains_object(&full_pairs, ObjectId::Hash(redacted_blob), ObjectType::Blob);
1375        assert_contains_object(
1376            &full_pairs,
1377            ObjectId::Hash(redacted_blob),
1378            ObjectType::Redaction,
1379        );
1380        for hash in [
1381            root_tree_hash,
1382            nested_tree_hash,
1383            context_tree_hash,
1384            provenance_tree_hash,
1385        ] {
1386            assert_contains_object(&full_pairs, ObjectId::Hash(hash), ObjectType::Tree);
1387        }
1388        for hash in [
1389            nested_blob,
1390            symlink_blob,
1391            context_blob,
1392            provenance_blob,
1393            risk_blob,
1394            review_blob,
1395            discussions_blob,
1396            conflicts_blob,
1397        ] {
1398            assert_contains_object(&full_pairs, ObjectId::Hash(hash), ObjectType::Blob);
1399        }
1400        assert!(!full_pairs.contains(&(
1401            ObjectId::StateId(excluded_parent.state_id),
1402            ObjectType::State
1403        )));
1404        assert!(!full_pairs.contains(&(ObjectId::Hash(excluded_tree_hash), ObjectType::Tree)));
1405        assert!(!full_pairs.contains(&(ObjectId::Hash(excluded_blob), ObjectType::Blob)));
1406    }
1407
1408    #[test]
1409    fn transfer_projection_reads_root_state_once_on_small_transfer() {
1410        let temp = TempDir::new().unwrap();
1411        let repo = Repository::init_default(temp.path()).unwrap();
1412        let blob = repo
1413            .store()
1414            .put_blob(&Blob::from("hello\n"))
1415            .expect("put blob");
1416        let tree_hash = repo
1417            .store()
1418            .put_tree(&Tree::from_entries(vec![
1419                TreeEntry::file("README.md", blob, false).unwrap(),
1420            ]))
1421            .expect("put tree");
1422        let state = State::new(tree_hash, Vec::new(), test_attribution());
1423        repo.store().put_state(&state).expect("put state");
1424        let store = CountingStore::new(repo.store());
1425
1426        let transfer = enumerate_state_closure_transfer_with_options(
1427            &store,
1428            state.state_id,
1429            StateClosureOptions::default(),
1430            512,
1431        )
1432        .expect("transfer projection");
1433
1434        assert!(
1435            !transfer.planned_objects.is_empty(),
1436            "lean projection should be available"
1437        );
1438        assert!(transfer.full_objects.is_some());
1439        assert_eq!(
1440            store.state_reads(),
1441            1,
1442            "small transfer projection must not read the root state through a second closure walk"
1443        );
1444    }
1445
1446    #[test]
1447    fn transfer_projection_drops_full_descriptors_after_threshold() {
1448        let temp = TempDir::new().unwrap();
1449        let repo = Repository::init_default(temp.path()).unwrap();
1450        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1451        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1452
1453        let transfer = enumerate_state_closure_transfer_with_options(
1454            repo.store(),
1455            state.state_id,
1456            StateClosureOptions::default(),
1457            0,
1458        )
1459        .expect("transfer projection");
1460
1461        assert!(
1462            !transfer.planned_objects.is_empty(),
1463            "lean projection should still be available over the threshold"
1464        );
1465        assert!(transfer.full_objects.is_none());
1466    }
1467
1468    #[test]
1469    fn depth_and_exclude_options_match_between_full_and_plan() {
1470        let temp = TempDir::new().unwrap();
1471        let repo = Repository::init_default(temp.path()).unwrap();
1472        let path = temp.path().join("story.txt");
1473
1474        std::fs::write(&path, "base\n").unwrap();
1475        let base = repo.snapshot(Some("base".to_string()), None).unwrap();
1476        std::fs::write(&path, "middle\n").unwrap();
1477        let middle = repo.snapshot(Some("middle".to_string()), None).unwrap();
1478        std::fs::write(&path, "tip\n").unwrap();
1479        let tip = repo.snapshot(Some("tip".to_string()), None).unwrap();
1480
1481        let depth_zero = assert_plan_parity(
1482            &repo,
1483            tip.state_id,
1484            StateClosureOptions {
1485                depth: Some(0),
1486                exclude_states: Vec::new(),
1487            },
1488        );
1489        assert!(depth_zero.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
1490        assert!(!depth_zero.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
1491        assert!(!depth_zero.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
1492
1493        let depth_one = assert_plan_parity(
1494            &repo,
1495            tip.state_id,
1496            StateClosureOptions {
1497                depth: Some(1),
1498                exclude_states: Vec::new(),
1499            },
1500        );
1501        assert!(depth_one.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
1502        assert!(depth_one.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
1503        assert!(!depth_one.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
1504
1505        let exclude_middle = assert_plan_parity(
1506            &repo,
1507            tip.state_id,
1508            StateClosureOptions {
1509                depth: None,
1510                exclude_states: vec![middle.state_id],
1511            },
1512        );
1513        assert!(exclude_middle.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
1514        assert!(!exclude_middle.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
1515        assert!(!exclude_middle.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
1516    }
1517
1518    #[test]
1519    fn shared_tree_and_blob_references_are_emitted_once() {
1520        let temp = TempDir::new().unwrap();
1521        let repo = Repository::init_default(temp.path()).unwrap();
1522
1523        let shared_blob = Blob::from("shared contents\n");
1524        let shared_blob_hash = repo.store().put_blob(&shared_blob).unwrap();
1525        let shared_tree = Tree::from_entries(vec![
1526            TreeEntry::file("shared.txt", shared_blob_hash, false).unwrap(),
1527        ]);
1528        let shared_tree_hash = repo.store().put_tree(&shared_tree).unwrap();
1529        let root = Tree::from_entries(vec![
1530            TreeEntry::directory("left", shared_tree_hash).unwrap(),
1531            TreeEntry::directory("right", shared_tree_hash).unwrap(),
1532        ]);
1533        let root_hash = repo.store().put_tree(&root).unwrap();
1534        let state = State::new(root_hash, Vec::new(), test_attribution());
1535        repo.store().put_state(&state).unwrap();
1536
1537        let full = enumerate_state_closure_with_options(
1538            repo.store(),
1539            state.state_id,
1540            StateClosureOptions::default(),
1541        )
1542        .unwrap();
1543        let plan = enumerate_state_closure_plan_with_options(
1544            repo.store(),
1545            state.state_id,
1546            StateClosureOptions::default(),
1547        )
1548        .unwrap();
1549
1550        assert_eq!(
1551            pairs_from_full(&full),
1552            pairs_from_plan(&plan),
1553            "full and lean closure enumerators must dedup the same objects"
1554        );
1555
1556        assert_eq!(
1557            full.iter()
1558                .filter(|info| info.id == ObjectId::Hash(root_hash)
1559                    && info.obj_type == ObjectType::Tree)
1560                .count(),
1561            1
1562        );
1563        assert_eq!(
1564            full.iter()
1565                .filter(|info| info.id == ObjectId::Hash(shared_tree_hash)
1566                    && info.obj_type == ObjectType::Tree)
1567                .count(),
1568            1
1569        );
1570        assert_eq!(
1571            full.iter()
1572                .filter(|info| info.id == ObjectId::Hash(shared_blob_hash)
1573                    && info.obj_type == ObjectType::Blob)
1574                .count(),
1575            1
1576        );
1577    }
1578
1579    #[test]
1580    fn state_closure_skips_gitlink_targets() {
1581        let temp = TempDir::new().unwrap();
1582        let repo = Repository::init_default(temp.path()).unwrap();
1583        let target: GitObjectId = "0303030303030303030303030303030303030303"
1584            .parse()
1585            .expect("git oid");
1586        let root = Tree::from_entries(vec![
1587            TreeEntry::gitlink("vendor", target).expect("gitlink entry"),
1588        ]);
1589        let root_hash = repo.store().put_tree(&root).unwrap();
1590        let state = State::new(root_hash, Vec::new(), test_attribution());
1591        repo.store().put_state(&state).unwrap();
1592
1593        let full = enumerate_state_closure_with_options(
1594            repo.store(),
1595            state.state_id,
1596            StateClosureOptions::default(),
1597        )
1598        .unwrap();
1599        let plan = enumerate_state_closure_plan_with_options(
1600            repo.store(),
1601            state.state_id,
1602            StateClosureOptions::default(),
1603        )
1604        .unwrap();
1605
1606        assert_eq!(pairs_from_full(&full), pairs_from_plan(&plan));
1607        assert!(
1608            !full.iter().any(|info| info.obj_type == ObjectType::Blob),
1609            "gitlinks carry foreign Git commit ids, not Heddle blob dependencies: {full:?}"
1610        );
1611        assert!(full.iter().any(|info| {
1612            info.id == ObjectId::Hash(root_hash) && info.obj_type == ObjectType::Tree
1613        }));
1614    }
1615
1616    #[test]
1617    fn missing_blobs_in_tree_skips_gitlinks_and_walks_nested_side_paths() {
1618        let temp = TempDir::new().unwrap();
1619        let repo = Repository::init_default(temp.path()).unwrap();
1620        let present_blob = repo
1621            .store()
1622            .put_blob(&Blob::from("already local"))
1623            .expect("put present blob");
1624        let missing_nested = ContentHash::from_bytes([7; 32]);
1625        let missing_symlink = ContentHash::from_bytes([8; 32]);
1626        let nested_tree = Tree::from_entries(vec![
1627            TreeEntry::file("remote.txt", missing_nested, false).unwrap(),
1628            TreeEntry::symlink("remote-link", missing_symlink).unwrap(),
1629        ]);
1630        let nested_tree_hash = repo
1631            .store()
1632            .put_tree(&nested_tree)
1633            .expect("put nested tree");
1634        let gitlink_target: GitObjectId = "0404040404040404040404040404040404040404"
1635            .parse()
1636            .expect("git oid");
1637        let root = Tree::from_entries(vec![
1638            TreeEntry::file("local.txt", present_blob, false).unwrap(),
1639            TreeEntry::directory("nested", nested_tree_hash).unwrap(),
1640            TreeEntry::gitlink("vendor", gitlink_target).unwrap(),
1641        ]);
1642        let root_hash = repo.store().put_tree(&root).expect("put root tree");
1643
1644        let missing = missing_blobs_in_tree(repo.store(), root_hash).expect("missing blobs");
1645
1646        assert_eq!(
1647            missing.into_iter().collect::<HashSet<_>>(),
1648            HashSet::from([missing_nested, missing_symlink])
1649        );
1650    }
1651
1652    /// Once a redaction is declared for a blob in a snapshot, the
1653    /// state closure must include an `ObjectType::Redaction` entry
1654    /// keyed on that blob's hash — that's the wire-side signal the
1655    /// receiver replays.
1656    #[test]
1657    fn enumerate_state_closure_emits_redaction_for_redacted_blob() {
1658        let temp = TempDir::new().unwrap();
1659        let repo = Repository::init_default(temp.path()).unwrap();
1660        std::fs::write(temp.path().join("secret.toml"), "api_token = \"x\"\n").unwrap();
1661        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1662
1663        // Find the blob hash for secret.toml by walking the snapshot's tree.
1664        let tree = repo
1665            .store()
1666            .get_tree(&state.tree)
1667            .unwrap()
1668            .expect("tree present");
1669        let blob_hash = tree
1670            .iter()
1671            .find(|e| e.name() == "secret.toml")
1672            .expect("entry present")
1673            .blob_hash()
1674            .expect("secret.toml is a blob");
1675
1676        let redaction = Redaction {
1677            redacted_blob: blob_hash,
1678            state: state.state_id,
1679            path: "secret.toml".to_string(),
1680            reason: "test leak".to_string(),
1681            redactor: Principal {
1682                name: "Tester".into(),
1683                email: "tester@heddle.sh".into(),
1684            },
1685            redacted_at: Utc::now(),
1686            signature: None,
1687            purged_at: None,
1688            supersedes: None,
1689        };
1690        repo.put_redaction(redaction).unwrap();
1691
1692        let full = enumerate_state_closure_with_options(
1693            repo.store(),
1694            state.state_id,
1695            StateClosureOptions::default(),
1696        )
1697        .unwrap();
1698        let plan = enumerate_state_closure_plan_with_options(
1699            repo.store(),
1700            state.state_id,
1701            StateClosureOptions::default(),
1702        )
1703        .unwrap();
1704
1705        assert!(
1706            full.iter()
1707                .any(|info| info.obj_type == ObjectType::Redaction
1708                    && info.id == ObjectId::Hash(blob_hash)),
1709            "full closure must include a Redaction entry for the redacted blob"
1710        );
1711        assert!(
1712            plan.iter()
1713                .any(|p| p.obj_type == ObjectType::Redaction && p.id == ObjectId::Hash(blob_hash)),
1714            "plan closure must include a Redaction entry for the redacted blob"
1715        );
1716    }
1717
1718    #[test]
1719    fn enumerate_state_closure_emits_state_visibility_for_visible_state() {
1720        let temp = TempDir::new().unwrap();
1721        let repo = Repository::init_default(temp.path()).unwrap();
1722        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1723        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1724
1725        repo.put_state_visibility(StateVisibility {
1726            state: state.state_id,
1727            tier: VisibilityTier::Restricted {
1728                scope_label: "security-embargo".into(),
1729            },
1730            embargo_until: None,
1731            declarer: Principal {
1732                name: "Tester".into(),
1733                email: "tester@heddle.sh".into(),
1734            },
1735            declared_at: Utc::now(),
1736            signature: None,
1737            supersedes: None,
1738        })
1739        .unwrap();
1740
1741        let full = enumerate_state_closure_with_options(
1742            repo.store(),
1743            state.state_id,
1744            StateClosureOptions::default(),
1745        )
1746        .unwrap();
1747        let plan = enumerate_state_closure_plan_with_options(
1748            repo.store(),
1749            state.state_id,
1750            StateClosureOptions::default(),
1751        )
1752        .unwrap();
1753
1754        assert!(
1755            full.iter()
1756                .any(|info| info.obj_type == ObjectType::StateVisibility
1757                    && info.id == ObjectId::StateId(state.state_id)),
1758            "full closure must include a StateVisibility entry for the visible state"
1759        );
1760        assert!(
1761            plan.iter()
1762                .any(|p| p.obj_type == ObjectType::StateVisibility
1763                    && p.id == ObjectId::StateId(state.state_id)),
1764            "plan closure must include a StateVisibility entry for the visible state"
1765        );
1766    }
1767
1768    #[test]
1769    fn enumerate_state_closure_emits_state_metadata_blobs() {
1770        let temp = TempDir::new().unwrap();
1771        let repo = Repository::init_default(temp.path()).unwrap();
1772        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1773        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1774
1775        let principal = Principal::new("Tester", "tester@example.test");
1776        let discussion_bytes = DiscussionsBlob::new(vec![Discussion {
1777            id: "disc-1".to_string(),
1778            anchor: SymbolAnchor::new("src/lib.rs", "answer"),
1779            opened_against_state: state.state_id,
1780            opened_at: 1_782_400_000,
1781            thread_ref: None,
1782            turns: vec![DiscussionTurn {
1783                author: principal,
1784                body: "Should this sync?".to_string(),
1785                posted_at: 1_782_400_000,
1786            }],
1787            resolution: DiscussionResolution::Open,
1788            body_changed_since_open: false,
1789            orphaned: false,
1790            visibility: VisibilityTier::default(),
1791            resolved_annotation_id: None,
1792        }])
1793        .encode()
1794        .expect("encode discussions");
1795        let discussion_hash = repo
1796            .store()
1797            .put_blob(&Blob::new(discussion_bytes))
1798            .expect("put discussions blob");
1799        let risk_hash = repo
1800            .store()
1801            .put_blob(&Blob::from_slice(b"risk signals"))
1802            .expect("put risk blob");
1803        let review_hash = repo
1804            .store()
1805            .put_blob(&Blob::from_slice(b"review signatures"))
1806            .expect("put review blob");
1807        let conflicts_hash = repo
1808            .store()
1809            .put_blob(&Blob::from_slice(b"structured conflicts"))
1810            .expect("put conflicts blob");
1811        for body in [
1812            StateAttachmentBody::RiskSignals(risk_hash),
1813            StateAttachmentBody::ReviewSignatures(review_hash),
1814            StateAttachmentBody::Discussions(discussion_hash),
1815            StateAttachmentBody::StructuredConflicts(conflicts_hash),
1816        ] {
1817            repo.put_state_attachment(&StateAttachment {
1818                state_id: state.id(),
1819                body,
1820                attribution: state.attribution.clone(),
1821                created_at: Utc::now(),
1822                supersedes: None,
1823            })
1824            .unwrap();
1825        }
1826
1827        let full = enumerate_state_closure_with_options(
1828            repo.store(),
1829            state.state_id,
1830            StateClosureOptions::default(),
1831        )
1832        .unwrap();
1833        let plan = enumerate_state_closure_plan_with_options(
1834            repo.store(),
1835            state.state_id,
1836            StateClosureOptions::default(),
1837        )
1838        .unwrap();
1839
1840        for metadata_hash in [risk_hash, review_hash, discussion_hash, conflicts_hash] {
1841            assert!(
1842                full.iter().any(|info| info.obj_type == ObjectType::Blob
1843                    && info.id == ObjectId::Hash(metadata_hash)),
1844                "full closure must include state metadata blob {metadata_hash}"
1845            );
1846            assert!(
1847                plan.iter().any(
1848                    |p| p.obj_type == ObjectType::Blob && p.id == ObjectId::Hash(metadata_hash)
1849                ),
1850                "plan closure must include state metadata blob {metadata_hash}"
1851            );
1852        }
1853    }
1854
1855    /// The push/pull packability split routes `StateAttachment` off the push
1856    /// pack (weft#549 forgery seal) while leaving pull carriage and every other
1857    /// type untouched.
1858    #[test]
1859    fn packable_predicates_split_state_attachment_by_direction() {
1860        // Sidecar records are never packable in either direction.
1861        for sidecar in [ObjectType::Redaction, ObjectType::StateVisibility] {
1862            assert!(!sidecar.packable_for_push(), "{sidecar:?} push");
1863            assert!(!sidecar.packable_for_pull(), "{sidecar:?} pull");
1864        }
1865        // Content-addressed objects ride the pack in both directions.
1866        for packable in [
1867            ObjectType::Blob,
1868            ObjectType::Tree,
1869            ObjectType::State,
1870            ObjectType::Action,
1871        ] {
1872            assert!(packable.packable_for_push(), "{packable:?} push");
1873            assert!(packable.packable_for_pull(), "{packable:?} pull");
1874        }
1875        // The attachment record: excluded from the push pack, kept on pull.
1876        assert!(!ObjectType::StateAttachment.packable_for_push());
1877        assert!(ObjectType::StateAttachment.packable_for_pull());
1878    }
1879
1880    /// A pushed state's semantic-index attachment RECORD must be excluded from
1881    /// the push pack (it rides the sidecar lane) while its semantic-index
1882    /// content blobs still ride the pack in both directions, and the same
1883    /// record stays packable server->client on pull.
1884    #[test]
1885    fn semantic_index_attachment_excluded_from_push_pack_but_kept_for_pull() {
1886        use std::collections::BTreeMap;
1887
1888        use objects::object::{SemanticIndexRoot, SemanticTreeNode};
1889
1890        let temp = TempDir::new().unwrap();
1891        let repo = Repository::init_default(temp.path()).unwrap();
1892        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
1893        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();
1894
1895        // Minimal valid semantic-index fixture: an empty tree node under a root.
1896        let (node, node_digest) = SemanticTreeNode::new(Vec::new());
1897        let node_hash = repo
1898            .store()
1899            .put_blob(&Blob::new(node.encode().unwrap()))
1900            .expect("put semantic tree node");
1901        let root = SemanticIndexRoot::new(1, BTreeMap::new(), node_hash, node_digest);
1902        let root_hash = repo
1903            .store()
1904            .put_blob(&Blob::new(root.encode().unwrap()))
1905            .expect("put semantic index root");
1906        repo.put_state_attachment(&StateAttachment {
1907            state_id: state.state_id,
1908            body: StateAttachmentBody::SemanticIndex(root_hash),
1909            attribution: test_attribution(),
1910            created_at: Utc::now(),
1911            supersedes: None,
1912        })
1913        .unwrap();
1914
1915        let plan = enumerate_state_closure_plan_with_options(
1916            repo.store(),
1917            state.state_id,
1918            StateClosureOptions::default(),
1919        )
1920        .unwrap();
1921
1922        // Every StateAttachment record (at least the semantic index authored
1923        // above) is push-excluded and pull-included.
1924        let attachments: Vec<_> = plan
1925            .iter()
1926            .filter(|p| p.obj_type == ObjectType::StateAttachment)
1927            .collect();
1928        assert!(
1929            !attachments.is_empty(),
1930            "closure must contain the semantic-index attachment record"
1931        );
1932        for attachment in &attachments {
1933            assert!(matches!(attachment.id, ObjectId::StateAttachment { .. }));
1934            // Push: excluded from the pack (sidecar lane). Pull: kept in pack.
1935            assert!(
1936                !attachment.obj_type.packable_for_push(),
1937                "attachment record must be excluded from the push pack"
1938            );
1939            assert!(
1940                attachment.obj_type.packable_for_pull(),
1941                "attachment record must stay in the pull pack"
1942            );
1943        }
1944
1945        // The semantic-index CONTENT blobs are ordinary content-addressed
1946        // objects and still ride the pack in both directions — only the
1947        // attachment record is sidecar'd on push.
1948        for content in [root_hash, node_hash] {
1949            let obj = plan
1950                .iter()
1951                .find(|p| p.id == ObjectId::Hash(content))
1952                .unwrap_or_else(|| panic!("semantic content blob {content} in closure"));
1953            assert_eq!(obj.obj_type, ObjectType::Blob);
1954            assert!(obj.obj_type.packable_for_push());
1955            assert!(obj.obj_type.packable_for_pull());
1956        }
1957
1958        // Partitioning the plan by push-packability puts the record on the
1959        // sidecar side and never in the pack side.
1960        let (push_pack, push_sidecar): (Vec<_>, Vec<_>) =
1961            plan.iter().partition(|p| p.obj_type.packable_for_push());
1962        assert!(
1963            push_sidecar
1964                .iter()
1965                .any(|p| p.obj_type == ObjectType::StateAttachment),
1966            "attachment record routed to the push sidecar partition"
1967        );
1968        assert!(
1969            !push_pack
1970                .iter()
1971                .any(|p| p.obj_type == ObjectType::StateAttachment),
1972            "attachment record must not be in the push pack partition"
1973        );
1974    }
1975}