wire/

object_graph.rs

// SPDX-License-Identifier: Apache-2.0
use std::collections::{HashSet, VecDeque};

use objects::{
    object::{
        BindingDelta, ContentHash, SemanticEntryKind, SemanticIndexRoot, SemanticTreeNode, State,
        StateAttachment, StateAttachmentBody, StateAttachmentId, StateAttachmentKind, StateId,
        TreeEntryTarget,
    },
    store::{ObjectStore, pack::ObjectType as PackObjectType},
};
use serde::{Deserialize, Serialize};

use crate::{ProtocolError, Result};

#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum ObjectId {
    Hash(ContentHash),
    StateId(StateId),
    StateAttachment {
        state: StateId,
        id: StateAttachmentId,
        /// The attachment's kind, a pure projection of its body
        /// ([`StateAttachmentBody::kind`]). Carried through the wire so
        /// descriptors self-describe their kind; the dedup/identity key is
        /// still `(state, id)`, and kind is coherent under `Eq`/`Hash` because
        /// it is a deterministic function of the same record.
        kind: StateAttachmentKind,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ObjectInfo {
    pub id: ObjectId,
    pub obj_type: ObjectType,
    pub size: u64,
    pub delta_base: Option<ContentHash>,
}

#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct PlannedObject {
    pub id: ObjectId,
    pub obj_type: ObjectType,
}

#[derive(Debug, Clone)]
pub struct StateClosureTransferObjects {
    pub planned_objects: Vec<PlannedObject>,
    pub full_objects: Option<Vec<ObjectInfo>>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum ObjectType {
    Blob,
    Tree,
    State,
    Action,
    /// A `RedactionsBlob` sidecar — the rmp-encoded record(s) declaring
    /// that a specific blob has been redacted (and possibly purged) by
    /// an authorized operator. Keyed on the wire by `ObjectId::Hash` of
    /// the *redacted blob*, since `Repository`'s sidecar store is
    /// indexed that way.
    Redaction,
    /// A `StateVisibilityBlob` sidecar — the rmp-encoded record(s)
    /// declaring a non-public audience tier for a specific state. Keyed
    /// on the wire by `ObjectId::StateId` of the state, since the
    /// per-state sidecar store is indexed that way. Like `Redaction`, it
    /// is a sidecar record that lives outside the content-addressed pack
    /// and ships via the per-object transfer path, not the pack.
    StateVisibility,
    StateAttachment,
    /// A content-addressed `KeyBindingRegistry` together with each binding's
    /// revocation/liveness overlay. Hosted materializers append this object to
    /// a closure and carry it out of pack because the native pack format has no
    /// key-binding record kind.
    KeyBinding,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ObjectTypeBucket {
    Blob,
    Tree,
    State,
    Action,
    Redaction,
    StateVisibility,
    StateAttachment,
    KeyBinding,
}

impl ObjectType {
    pub fn wire_name(self) -> &'static str {
        match self {
            ObjectType::Blob => "blob",
            ObjectType::Tree => "tree",
            ObjectType::State => "state",
            ObjectType::Action => "action",
            ObjectType::Redaction => "redaction",
            ObjectType::StateVisibility => "state_visibility",
            ObjectType::StateAttachment => "state_attachment",
            ObjectType::KeyBinding => "key_binding",
        }
    }

    pub fn from_wire(value: &str) -> Result<Self> {
        match value {
            "blob" => Ok(ObjectType::Blob),
            "tree" => Ok(ObjectType::Tree),
            "state" => Ok(ObjectType::State),
            "action" => Ok(ObjectType::Action),
            "redaction" => Ok(ObjectType::Redaction),
            "state_visibility" => Ok(ObjectType::StateVisibility),
            "state_attachment" => Ok(ObjectType::StateAttachment),
            "key_binding" => Ok(ObjectType::KeyBinding),
            _ => Err(ProtocolError::InvalidState(format!(
                "unknown object type: {value}"
            ))),
        }
    }

    /// Whether this object type can ride the content-addressed native pack in
    /// general (the historical, direction-agnostic predicate). Sidecar records
    /// (`Redaction`, `StateVisibility`) live outside `.heddle/objects/` and can
    /// never be packed; everything else can.
    ///
    /// Prefer [`ObjectType::packable_for_push`] /
    /// [`ObjectType::packable_for_pull`] at transfer sites: packability is
    /// direction-dependent for `StateAttachment` (see those methods). This
    /// method retains the pull/general semantics so pack construction,
    /// have-set, and local-copy planners are unchanged.
    pub fn packable(self) -> bool {
        !matches!(
            self,
            ObjectType::Redaction | ObjectType::StateVisibility | ObjectType::KeyBinding
        )
    }

    /// Whether this object type may ride the client→server **push** pack.
    ///
    /// `StateAttachment` is deliberately excluded on push even though it is a
    /// content-addressed object: a deployed server rejects any pack-carried
    /// attachment as a forgery-prevention measure (weft#549). Pushed
    /// attachments must instead ride the out-of-pack **sidecar** lane (the same
    /// lane as `Redaction`/`StateVisibility`), where the server can verify them
    /// per-kind at finalize while the pack itself stays forgery-sealed. Every
    /// other type keeps its [`ObjectType::packable`] answer.
    pub fn packable_for_push(self) -> bool {
        self.packable() && !matches!(self, ObjectType::StateAttachment)
    }

    /// Whether this object type may ride the server→client **pull/clone** pack.
    ///
    /// Identical to [`ObjectType::packable`]: attachments are carried in the
    /// pull pack exactly as today. The push/pull split exists solely so the
    /// push direction can exclude `StateAttachment` without disturbing the
    /// working pull carriage.
    pub fn packable_for_pull(self) -> bool {
        self.packable()
    }

    pub fn pack_object_type(self) -> Result<PackObjectType> {
        match self {
            ObjectType::Blob => Ok(PackObjectType::Blob),
            ObjectType::Tree => Ok(PackObjectType::Tree),
            ObjectType::State => Ok(PackObjectType::State),
            ObjectType::Action => Ok(PackObjectType::Action),
            ObjectType::StateAttachment => Ok(PackObjectType::StateAttachment),
            ObjectType::Redaction => Err(ProtocolError::InvalidState(
                "Redaction sidecar records cannot be packed into the content-addressed object pack"
                    .to_string(),
            )),
            ObjectType::StateVisibility => Err(ProtocolError::InvalidState(
                "StateVisibility sidecar records cannot be packed into the content-addressed object pack"
                    .to_string(),
            )),
            ObjectType::KeyBinding => Err(ProtocolError::InvalidState(
                "KeyBinding registry objects cannot be packed into the content-addressed object pack"
                    .to_string(),
            )),
        }
    }

    pub fn bucket(self) -> ObjectTypeBucket {
        match self {
            ObjectType::Blob => ObjectTypeBucket::Blob,
            ObjectType::Tree => ObjectTypeBucket::Tree,
            ObjectType::State => ObjectTypeBucket::State,
            ObjectType::Action => ObjectTypeBucket::Action,
            ObjectType::Redaction => ObjectTypeBucket::Redaction,
            ObjectType::StateVisibility => ObjectTypeBucket::StateVisibility,
            ObjectType::StateAttachment => ObjectTypeBucket::StateAttachment,
            ObjectType::KeyBinding => ObjectTypeBucket::KeyBinding,
        }
    }
}

#[derive(Debug, Clone, Default)]
pub struct StateClosureOptions {
    pub depth: Option<u32>,
    pub exclude_states: Vec<StateId>,
}

pub fn enumerate_state_closure(
    store: &impl ObjectStore,
    state_id: StateId,
) -> Result<Vec<ObjectInfo>> {
    enumerate_state_closure_with_options(store, state_id, StateClosureOptions::default())
}

pub fn enumerate_state_closure_with_options(
    store: &impl ObjectStore,
    state_id: StateId,
    options: StateClosureOptions,
) -> Result<Vec<ObjectInfo>> {
    let mut out = Vec::new();
    walk_state_closure(store, state_id, options, |event| {
        if let Some(info) = object_info_from_event(store, event)? {
            out.push(info);
        }
        Ok(())
    })?;

    Ok(out)
}

pub fn enumerate_state_closure_plan(
    store: &impl ObjectStore,
    state_id: StateId,
) -> Result<Vec<PlannedObject>> {
    enumerate_state_closure_plan_with_options(store, state_id, StateClosureOptions::default())
}

pub fn enumerate_state_closure_plan_with_options(
    store: &impl ObjectStore,
    state_id: StateId,
    options: StateClosureOptions,
) -> Result<Vec<PlannedObject>> {
    let mut out = Vec::new();
    walk_state_closure(store, state_id, options, |event| {
        if let Some(object) = planned_object_from_event(store, event)? {
            out.push(object);
        }
        Ok(())
    })?;

    Ok(out)
}

pub fn enumerate_state_closure_transfer_with_options(
    store: &impl ObjectStore,
    state_id: StateId,
    options: StateClosureOptions,
    full_descriptor_object_threshold: usize,
) -> Result<StateClosureTransferObjects> {
    let mut planned_objects = Vec::new();
    let mut full_objects = Some(Vec::new());

    walk_state_closure(store, state_id, options, |event| {
        if let Some(object) = planned_object_from_event(store, event)? {
            planned_objects.push(object);
        }

        if full_objects.is_some() && planned_objects.len() > full_descriptor_object_threshold {
            full_objects = None;
        }
        if let Some(objects) = full_objects.as_mut()
            && let Some(info) = object_info_from_event(store, event)?
        {
            objects.push(info);
        }

        Ok(())
    })?;

    Ok(StateClosureTransferObjects {
        planned_objects,
        full_objects,
    })
}

/// Enumerate a transfer delta while treating `boundary_states` as complete
/// server-held roots. Unlike [`StateClosureOptions::exclude_states`], this does
/// not expand each boundary's tree/history to build a hash exclusion set: the
/// walk stops as soon as it reaches the boundary state. Objects reused by the
/// new tip may still be advertised, and the receiver's have-set filters them.
/// This keeps incremental push planning proportional to the new history.
pub fn enumerate_state_closure_transfer_from_boundaries(
    store: &impl ObjectStore,
    state_id: StateId,
    boundary_states: &[StateId],
    full_descriptor_object_threshold: usize,
) -> Result<StateClosureTransferObjects> {
    let mut planned_objects = Vec::new();
    let mut full_objects = Some(Vec::new());
    let excluded_states = boundary_states.iter().copied().collect();

    walk_state_closure_with_exclusions(
        store,
        state_id,
        None,
        excluded_states,
        HashSet::new(),
        |event| {
            if let Some(object) = planned_object_from_event(store, event)? {
                planned_objects.push(object);
            }

            if full_objects.is_some() && planned_objects.len() > full_descriptor_object_threshold {
                full_objects = None;
            }
            if let Some(objects) = full_objects.as_mut()
                && let Some(info) = object_info_from_event(store, event)?
            {
                objects.push(info);
            }

            Ok(())
        },
    )?;

    Ok(StateClosureTransferObjects {
        planned_objects,
        full_objects,
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BlobSource {
    Tree,
    StateMetadata,
}

#[derive(Debug, Clone, Copy)]
enum StateClosureEvent<'a> {
    State {
        id: StateId,
        state: &'a State,
    },
    Tree {
        hash: ContentHash,
        tree: &'a objects::object::Tree,
    },
    Blob {
        hash: ContentHash,
        source: BlobSource,
    },
    Redaction {
        blob: ContentHash,
    },
    StateVisibility {
        state: StateId,
    },
    StateAttachment {
        state: StateId,
        attachment: &'a StateAttachment,
    },
    ExcludedState {
        id: StateId,
    },
    ExcludedHash {
        hash: ContentHash,
    },
}

fn walk_state_closure(
    store: &impl ObjectStore,
    state_id: StateId,
    options: StateClosureOptions,
    visit: impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
) -> Result<()> {
    let (excluded_states, excluded_hashes) = collect_excluded(store, &options.exclude_states)?;

    walk_state_closure_with_exclusions(
        store,
        state_id,
        options.depth,
        excluded_states,
        excluded_hashes,
        visit,
    )
}

fn walk_state_closure_with_exclusions(
    store: &impl ObjectStore,
    state_id: StateId,
    max_depth: Option<u32>,
    excluded_states: HashSet<StateId>,
    excluded_hashes: HashSet<ContentHash>,
    mut visit: impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
) -> Result<()> {
    let mut seen_states: HashSet<StateId> = HashSet::new();
    let mut seen_hashes: HashSet<ContentHash> = HashSet::new();
    let mut queue: VecDeque<(StateId, u32)> = VecDeque::new();
    queue.push_back((state_id, 0));

    while let Some((id, depth)) = queue.pop_front() {
        if excluded_states.contains(&id) {
            visit(StateClosureEvent::ExcludedState { id })?;
            continue;
        }
        if !seen_states.insert(id) {
            continue;
        }

        let state = store
            .get_state(&id)?
            .ok_or_else(|| ProtocolError::ObjectNotFound(id.to_string()))?;

        visit(StateClosureEvent::State { id, state: &state })?;
        if store.has_state_visibility_for_state(&id)? {
            visit(StateClosureEvent::StateVisibility { state: id })?;
        }
        for attachment in store.list_state_attachments(&id)? {
            visit(StateClosureEvent::StateAttachment {
                state: id,
                attachment: &attachment,
            })?;
            match attachment.body {
                StateAttachmentBody::Context(root) => walk_tree_closure_filtered(
                    store,
                    root,
                    &excluded_hashes,
                    &mut seen_hashes,
                    &mut visit,
                )?,
                StateAttachmentBody::RiskSignals(hash)
                | StateAttachmentBody::ReviewSignatures(hash)
                | StateAttachmentBody::Discussions(hash)
                | StateAttachmentBody::StructuredConflicts(hash) => walk_blob_filtered(
                    store,
                    hash,
                    BlobSource::StateMetadata,
                    &excluded_hashes,
                    &mut seen_hashes,
                    &mut visit,
                )?,
                StateAttachmentBody::SemanticIndex(root) => walk_semantic_index_closure(
                    store,
                    root,
                    &excluded_hashes,
                    &mut seen_hashes,
                    &mut visit,
                )?,
                StateAttachmentBody::Signature(_) => {}
            }
        }

        if max_depth.map(|max| depth < max).unwrap_or(true) {
            for parent in &state.parents {
                queue.push_back((*parent, depth + 1));
            }
        }

        walk_tree_closure_filtered(
            store,
            state.tree,
            &excluded_hashes,
            &mut seen_hashes,
            &mut visit,
        )?;
        if let Some(provenance_root) = state.provenance {
            walk_tree_closure_filtered(
                store,
                provenance_root,
                &excluded_hashes,
                &mut seen_hashes,
                &mut visit,
            )?;
        }
    }

    Ok(())
}

fn walk_tree_closure_filtered(
    store: &impl ObjectStore,
    tree_hash: ContentHash,
    excluded: &HashSet<ContentHash>,
    seen: &mut HashSet<ContentHash>,
    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
) -> Result<()> {
    if excluded.contains(&tree_hash) {
        visit(StateClosureEvent::ExcludedHash { hash: tree_hash })?;
        return Ok(());
    }
    if !seen.insert(tree_hash) {
        return Ok(());
    }

    let tree = store
        .get_tree(&tree_hash)?
        .ok_or_else(|| ProtocolError::ObjectNotFound(tree_hash.to_hex()))?;

    visit(StateClosureEvent::Tree {
        hash: tree_hash,
        tree: &tree,
    })?;

    for entry in tree.entries() {
        match entry.target() {
            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
                walk_blob_filtered(store, *hash, BlobSource::Tree, excluded, seen, visit)?;
            }
            TreeEntryTarget::Tree { hash } => {
                walk_tree_closure_filtered(store, *hash, excluded, seen, visit)?;
            }
            TreeEntryTarget::Gitlink { .. } => {}
            // Native child-spool edge: its target lives in a separate spool
            // object graph, not this store, so it is not walked here.
            TreeEntryTarget::Spoollink { .. } => {}
        }
    }

    Ok(())
}

fn walk_blob_filtered(
    store: &impl ObjectStore,
    blob_hash: ContentHash,
    source: BlobSource,
    excluded: &HashSet<ContentHash>,
    seen: &mut HashSet<ContentHash>,
    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
) -> Result<()> {
    if excluded.contains(&blob_hash) {
        visit(StateClosureEvent::ExcludedHash { hash: blob_hash })?;
        return Ok(());
    }
    if !seen.insert(blob_hash) {
        return Ok(());
    }
    visit(StateClosureEvent::Blob {
        hash: blob_hash,
        source,
    })?;
    if store.has_redactions_for_blob(&blob_hash)? {
        visit(StateClosureEvent::Redaction { blob: blob_hash })?;
    }
    Ok(())
}

/// Walk the merkle semantic-index closure rooted at `root_hash` (a
/// `SemanticIndexRoot` blob), emitting every reachable semantic node blob.
///
/// All semantic-index nodes (root, tree nodes, file nodes) are stored as
/// ordinary content-addressed blobs, so replication just enumerates them.
/// Opaque entries point back at raw source blobs already covered by the state's
/// tree closure, so they are not re-walked here.
///
/// Iterative (explicit stack) so a crafted deep `SemanticTreeNode` chain in a
/// pushed state can't overflow the stack. A missing or undecodable node in the
/// closure is a HARD failure (`ObjectNotFound`/`Serialization`) — a partial or
/// corrupt semantic closure must never be shipped silently.
fn walk_semantic_index_closure(
    store: &impl ObjectStore,
    root_hash: ContentHash,
    excluded: &HashSet<ContentHash>,
    seen: &mut HashSet<ContentHash>,
    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
) -> Result<()> {
    // Stack of (node_hash, is_tree_node): the root and dir nodes must be decoded
    // to enumerate their children; file/opaque leaves are emitted only.
    let mut stack: Vec<ContentHash> = vec![root_hash];
    while let Some(node_hash) = stack.pop() {
        if !emit_semantic_blob(store, node_hash, excluded, seen, visit)? {
            continue; // excluded (present on the far side) or already seen.
        }
        let blob = store
            .get_blob(&node_hash)?
            .ok_or_else(|| ProtocolError::ObjectNotFound(node_hash.to_hex()))?;
        // The root and tree nodes decode to the same `SemanticTreeNode.entries`
        // shape after the root's one indirection; walk uniformly by decoding
        // every interior node as a tree node, tolerating the root shape.
        let node = decode_semantic_container(&blob, node_hash)?;
        for child in node {
            match child {
                SemanticChild::Interior(hash) => stack.push(hash),
                SemanticChild::Leaf(hash) => {
                    // Emit the leaf (file node) blob; it has no children.
                    emit_semantic_blob(store, hash, excluded, seen, visit)?;
                }
                SemanticChild::BindingDelta(hash) => {
                    // Binding deltas are state-scoped attachments. Emit this
                    // state's direct delta; ancestor states contribute their
                    // own deltas when (and only when) the state walk reaches
                    // them. Following `delta.parent` here would cross a
                    // shallow-clone boundary.
                    emit_binding_delta(store, hash, excluded, seen, visit)?;
                }
            }
        }
    }
    Ok(())
}

/// Child kinds encountered while walking a state's semantic closure.
enum SemanticChild {
    Interior(ContentHash),
    Leaf(ContentHash),
    BindingDelta(ContentHash),
}

/// Decode a semantic-closure node — the root (which points at a tree node) or a
/// tree node — into the child hashes to walk. Missing/corrupt is a hard error.
fn decode_semantic_container(
    blob: &objects::object::Blob,
    node_hash: ContentHash,
) -> Result<Vec<SemanticChild>> {
    // Try the root shape first (it has an extra `tree` indirection), then the
    // tree-node shape. Content-addressed hashes make this unambiguous in
    // practice; a blob that decodes as neither is corrupt.
    if let Ok(root) = SemanticIndexRoot::decode(blob.content()) {
        let mut children = vec![SemanticChild::Interior(root.tree)];
        if let Some(binding_delta) = root.binding_delta {
            children.push(SemanticChild::BindingDelta(binding_delta));
        }
        return Ok(children);
    }
    let node = SemanticTreeNode::decode(blob.content())
        .map_err(|err| ProtocolError::Serialization(format!("semantic node {node_hash}: {err}")))?;
    Ok(node
        .entries
        .iter()
        .filter_map(|entry| match entry.kind {
            SemanticEntryKind::Dir => Some(SemanticChild::Interior(entry.node)),
            SemanticEntryKind::File => Some(SemanticChild::Leaf(entry.node)),
            // Opaque `node` is the raw source blob, already in the tree closure.
            SemanticEntryKind::Opaque => None,
        })
        .collect())
}

fn emit_binding_delta(
    store: &impl ObjectStore,
    hash: ContentHash,
    excluded: &HashSet<ContentHash>,
    seen: &mut HashSet<ContentHash>,
    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
) -> Result<()> {
    if !emit_semantic_blob(store, hash, excluded, seen, visit)? {
        return Ok(());
    }
    let blob = store
        .get_blob(&hash)?
        .ok_or_else(|| ProtocolError::ObjectNotFound(hash.to_hex()))?;
    BindingDelta::decode(blob.content()).map_err(|err| {
        ProtocolError::Serialization(format!("semantic binding delta {hash}: {err}"))
    })?;
    Ok(())
}

/// Emit a semantic node blob as state metadata. Returns `true` when the blob
/// was newly visited (so the caller may descend into it), `false` when it was
/// excluded (present on the far side) or already seen. A blob that is neither
/// excluded nor present is a HARD failure — the closure must be complete.
fn emit_semantic_blob(
    store: &impl ObjectStore,
    hash: ContentHash,
    excluded: &HashSet<ContentHash>,
    seen: &mut HashSet<ContentHash>,
    visit: &mut impl for<'event> FnMut(StateClosureEvent<'event>) -> Result<()>,
) -> Result<bool> {
    if excluded.contains(&hash) {
        visit(StateClosureEvent::ExcludedHash { hash })?;
        return Ok(false);
    }
    if !seen.insert(hash) {
        return Ok(false);
    }
    if store.get_blob(&hash)?.is_none() {
        return Err(ProtocolError::ObjectNotFound(hash.to_hex()));
    }
    visit(StateClosureEvent::Blob {
        hash,
        source: BlobSource::StateMetadata,
    })?;
    Ok(true)
}

/// Collect every semantic-index node blob hash reachable from `root_hash` into
/// `excluded`, for the have-set computation. Iterative + tolerant (a broken
/// have-set index is not fatal — it just means fewer objects are marked
/// already-present, which is safe over-fetching, not corruption).
fn collect_semantic_hashes(
    store: &impl ObjectStore,
    root_hash: ContentHash,
    excluded: &mut HashSet<ContentHash>,
) -> Result<()> {
    let mut stack: Vec<ContentHash> = vec![root_hash];
    while let Some(node_hash) = stack.pop() {
        if !excluded.insert(node_hash) {
            continue;
        }
        let Some(blob) = store.get_blob(&node_hash)? else {
            continue;
        };
        let children = match decode_semantic_container(&blob, node_hash) {
            Ok(children) => children,
            Err(_) => continue,
        };
        for child in children {
            match child {
                SemanticChild::Interior(hash) => stack.push(hash),
                SemanticChild::Leaf(hash) => {
                    excluded.insert(hash);
                }
                SemanticChild::BindingDelta(hash) => {
                    excluded.insert(hash);
                }
            }
        }
    }
    Ok(())
}

fn object_info_from_event(
    store: &impl ObjectStore,
    event: StateClosureEvent<'_>,
) -> Result<Option<ObjectInfo>> {
    match event {
        StateClosureEvent::State { id, state } => {
            let state_bytes = rmp_serde::to_vec_named(state)?;
            Ok(Some(ObjectInfo {
                id: ObjectId::StateId(id),
                obj_type: ObjectType::State,
                size: state_bytes.len() as u64,
                delta_base: None,
            }))
        }
        StateClosureEvent::Tree { hash, tree } => {
            let tree_bytes = rmp_serde::to_vec_named(tree)?;
            Ok(Some(ObjectInfo {
                id: ObjectId::Hash(hash),
                obj_type: ObjectType::Tree,
                size: tree_bytes.len() as u64,
                delta_base: None,
            }))
        }
        StateClosureEvent::Blob { hash, .. } => {
            let blob = store
                .get_blob(&hash)?
                .ok_or_else(|| ProtocolError::ObjectNotFound(hash.to_hex()))?;
            Ok(Some(ObjectInfo {
                id: ObjectId::Hash(hash),
                obj_type: ObjectType::Blob,
                size: blob.size() as u64,
                delta_base: None,
            }))
        }
        StateClosureEvent::Redaction { blob } => Ok(store
            .get_redactions_bytes_for_blob(&blob)?
            .map(|bytes| ObjectInfo {
                id: ObjectId::Hash(blob),
                obj_type: ObjectType::Redaction,
                size: bytes.len() as u64,
                delta_base: None,
            })),
        StateClosureEvent::StateVisibility { state } => Ok(store
            .get_state_visibility_bytes_for_state(&state)?
            .map(|bytes| ObjectInfo {
                id: ObjectId::StateId(state),
                obj_type: ObjectType::StateVisibility,
                size: bytes.len() as u64,
                delta_base: None,
            })),
        StateClosureEvent::StateAttachment { state, attachment } => {
            let bytes = rmp_serde::to_vec_named(attachment)?;
            Ok(Some(ObjectInfo {
                id: ObjectId::StateAttachment {
                    state,
                    id: attachment.id(),
                    kind: attachment.body.kind(),
                },
                obj_type: ObjectType::StateAttachment,
                size: bytes.len() as u64,
                delta_base: None,
            }))
        }
        StateClosureEvent::ExcludedState { id } => {
            let _ = id;
            Ok(None)
        }
        StateClosureEvent::ExcludedHash { hash } => {
            let _ = hash;
            Ok(None)
        }
    }
}

fn planned_object_from_event(
    store: &impl ObjectStore,
    event: StateClosureEvent<'_>,
) -> Result<Option<PlannedObject>> {
    match event {
        StateClosureEvent::State { id, .. } => Ok(Some(PlannedObject {
            id: ObjectId::StateId(id),
            obj_type: ObjectType::State,
        })),
        StateClosureEvent::Tree { hash, .. } => Ok(Some(PlannedObject {
            id: ObjectId::Hash(hash),
            obj_type: ObjectType::Tree,
        })),
        StateClosureEvent::Blob { hash, source } => {
            if source == BlobSource::StateMetadata && store.get_blob(&hash)?.is_none() {
                return Err(ProtocolError::ObjectNotFound(hash.to_hex()));
            }
            Ok(Some(PlannedObject {
                id: ObjectId::Hash(hash),
                obj_type: ObjectType::Blob,
            }))
        }
        StateClosureEvent::Redaction { blob } => Ok(Some(PlannedObject {
            id: ObjectId::Hash(blob),
            obj_type: ObjectType::Redaction,
        })),
        StateClosureEvent::StateVisibility { state } => Ok(Some(PlannedObject {
            id: ObjectId::StateId(state),
            obj_type: ObjectType::StateVisibility,
        })),
        StateClosureEvent::StateAttachment { state, attachment } => Ok(Some(PlannedObject {
            id: ObjectId::StateAttachment {
                state,
                id: attachment.id(),
                kind: attachment.body.kind(),
            },
            obj_type: ObjectType::StateAttachment,
        })),
        StateClosureEvent::ExcludedState { id } => {
            let _ = id;
            Ok(None)
        }
        StateClosureEvent::ExcludedHash { hash } => {
            let _ = hash;
            Ok(None)
        }
    }
}

pub fn missing_blobs_in_tree(
    store: &impl ObjectStore,
    tree_hash: ContentHash,
) -> Result<Vec<ContentHash>> {
    let mut missing = Vec::new();
    collect_missing_blobs_recursive(store, &tree_hash, &mut missing)?;
    Ok(missing)
}

fn collect_missing_blobs_recursive(
    store: &impl ObjectStore,
    tree_hash: &ContentHash,
    missing: &mut Vec<ContentHash>,
) -> Result<()> {
    let Some(tree) = store.get_tree(tree_hash).map_err(|err| {
        ProtocolError::InvalidState(format!(
            "load tree {} while collecting lazy hydration missing blobs: {err}",
            tree_hash.to_hex()
        ))
    })?
    else {
        return Ok(());
    };

    for entry in tree.entries() {
        match entry.target() {
            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
                if !store.has_blob(hash).map_err(|err| {
                    ProtocolError::InvalidState(format!(
                        "check blob {} while collecting lazy hydration missing blobs: {err}",
                        hash.to_hex()
                    ))
                })? {
                    missing.push(*hash);
                }
            }
            TreeEntryTarget::Tree { hash } => {
                collect_missing_blobs_recursive(store, hash, missing)?;
            }
            TreeEntryTarget::Gitlink { .. } => {}
            // Native child-spool edge: its target lives in a separate spool
            // object graph, not this store, so it is not walked here.
            TreeEntryTarget::Spoollink { .. } => {}
        }
    }
    Ok(())
}

fn collect_excluded(
    store: &impl ObjectStore,
    roots: &[StateId],
) -> Result<(HashSet<StateId>, HashSet<ContentHash>)> {
    if roots.is_empty() {
        return Ok((HashSet::new(), HashSet::new()));
    }

    let mut excluded_states: HashSet<StateId> = HashSet::new();
    let mut excluded_hashes: HashSet<ContentHash> = HashSet::new();
    let mut queue: VecDeque<StateId> = VecDeque::new();

    for id in roots {
        queue.push_back(*id);
    }

    while let Some(id) = queue.pop_front() {
        if !excluded_states.insert(id) {
            continue;
        }

        let state = match store.get_state(&id)? {
            Some(state) => state,
            None => continue,
        };

        for parent in &state.parents {
            queue.push_back(*parent);
        }

        collect_tree_hashes(store, state.tree, &mut excluded_hashes)?;
        if let Some(provenance_root) = state.provenance {
            collect_tree_hashes(store, provenance_root, &mut excluded_hashes)?;
        }
        for attachment in store.list_state_attachments(&id)? {
            match attachment.body {
                StateAttachmentBody::Context(root) => {
                    collect_tree_hashes(store, root, &mut excluded_hashes)?
                }
                StateAttachmentBody::RiskSignals(hash)
                | StateAttachmentBody::ReviewSignatures(hash)
                | StateAttachmentBody::Discussions(hash)
                | StateAttachmentBody::StructuredConflicts(hash) => {
                    excluded_hashes.insert(hash);
                }
                StateAttachmentBody::SemanticIndex(root) => {
                    collect_semantic_hashes(store, root, &mut excluded_hashes)?;
                }
                StateAttachmentBody::Signature(_) => {}
            }
        }
    }

    Ok((excluded_states, excluded_hashes))
}

fn collect_tree_hashes(
    store: &impl ObjectStore,
    tree_hash: ContentHash,
    excluded: &mut HashSet<ContentHash>,
) -> Result<()> {
    if !excluded.insert(tree_hash) {
        return Ok(());
    }

    let tree = match store.get_tree(&tree_hash)? {
        Some(tree) => tree,
        None => return Ok(()),
    };

    for entry in tree.entries() {
        match entry.target() {
            TreeEntryTarget::Blob { hash, .. } | TreeEntryTarget::Symlink { hash } => {
                excluded.insert(*hash);
            }
            TreeEntryTarget::Tree { hash } => {
                collect_tree_hashes(store, *hash, excluded)?;
            }
            TreeEntryTarget::Gitlink { .. } => {}
            // Native child-spool edge: its target lives in a separate spool
            // object graph, not this store, so it is not walked here.
            TreeEntryTarget::Spoollink { .. } => {}
        }
    }

    Ok(())
}

pub fn is_ancestor(
    store: &impl ObjectStore,
    ancestor: StateId,
    descendant: StateId,
) -> Result<bool> {
    if ancestor == descendant {
        return Ok(true);
    }

    let mut seen: HashSet<StateId> = HashSet::new();
    let mut queue: VecDeque<StateId> = VecDeque::new();
    queue.push_back(descendant);

    while let Some(id) = queue.pop_front() {
        if !seen.insert(id) {
            continue;
        }
        let state = match store.get_state(&id)? {
            Some(s) => s,
            None => return Ok(false),
        };
        for parent in state.parents {
            if parent == ancestor {
                return Ok(true);
            }
            queue.push_back(parent);
        }
    }

    Ok(false)
}

#[cfg(test)]
mod tests {
    use std::{
        collections::HashSet,
        sync::atomic::{AtomicUsize, Ordering},
    };

    use chrono::Utc;
    use objects::{
        object::{
            Action, ActionId, Attribution, Blob, ContentHash, Discussion, DiscussionResolution,
            DiscussionTurn, DiscussionsBlob, Principal, Redaction, State, StateAttachment,
            StateAttachmentBody, StateId, StateVisibility, SymbolAnchor, Tree, TreeEntry,
            VisibilityTier,
        },
        store::{ObjectStore, Result as StoreResult},
    };
    use repo::Repository;
    use sley::ObjectId as GitObjectId;
    use tempfile::TempDir;

    use super::{
        ObjectId, ObjectInfo, ObjectType, PlannedObject, StateClosureOptions,
        enumerate_state_closure_plan_with_options,
        enumerate_state_closure_transfer_from_boundaries,
        enumerate_state_closure_transfer_with_options, enumerate_state_closure_with_options,
        missing_blobs_in_tree,
    };

    fn pairs_from_full(objects: &[ObjectInfo]) -> HashSet<(ObjectId, ObjectType)> {
        objects
            .iter()
            .map(|info| (info.id.clone(), info.obj_type))
            .collect()
    }

    fn pairs_from_plan(objects: &[PlannedObject]) -> HashSet<(ObjectId, ObjectType)> {
        objects
            .iter()
            .map(|info| (info.id.clone(), info.obj_type))
            .collect()
    }

    fn object_info_fingerprint(
        objects: &[ObjectInfo],
    ) -> Vec<(ObjectId, ObjectType, u64, Option<ContentHash>)> {
        objects
            .iter()
            .map(|info| (info.id.clone(), info.obj_type, info.size, info.delta_base))
            .collect()
    }

    fn assert_plan_parity(
        repo: &Repository,
        state_id: StateId,
        options: StateClosureOptions,
    ) -> HashSet<(ObjectId, ObjectType)> {
        let full =
            enumerate_state_closure_with_options(repo.store(), state_id, options.clone()).unwrap();
        let plan =
            enumerate_state_closure_plan_with_options(repo.store(), state_id, options).unwrap();

        let full_pairs = pairs_from_full(&full);
        let plan_pairs = pairs_from_plan(&plan);
        assert_eq!(full_pairs, plan_pairs);
        full_pairs
    }

    fn assert_contains_object(
        objects: &HashSet<(ObjectId, ObjectType)>,
        id: ObjectId,
        obj_type: ObjectType,
    ) {
        assert!(
            objects.contains(&(id.clone(), obj_type)),
            "expected closure to contain {id:?} as {obj_type:?}: {objects:?}"
        );
    }

    struct CountingStore<'a, S> {
        inner: &'a S,
        state_reads: AtomicUsize,
    }

    impl<'a, S> CountingStore<'a, S> {
        fn new(inner: &'a S) -> Self {
            Self {
                inner,
                state_reads: AtomicUsize::new(0),
            }
        }

        fn state_reads(&self) -> usize {
            self.state_reads.load(Ordering::SeqCst)
        }
    }

    impl<S: ObjectStore> ObjectStore for CountingStore<'_, S> {
        fn get_blob(&self, hash: &ContentHash) -> StoreResult<Option<Blob>> {
            self.inner.get_blob(hash)
        }

        fn put_blob(&self, blob: &Blob) -> StoreResult<ContentHash> {
            self.inner.put_blob(blob)
        }

        fn has_blob(&self, hash: &ContentHash) -> StoreResult<bool> {
            self.inner.has_blob(hash)
        }

        fn get_tree(&self, hash: &ContentHash) -> StoreResult<Option<Tree>> {
            self.inner.get_tree(hash)
        }

        fn put_tree(&self, tree: &Tree) -> StoreResult<ContentHash> {
            self.inner.put_tree(tree)
        }

        fn has_tree(&self, hash: &ContentHash) -> StoreResult<bool> {
            self.inner.has_tree(hash)
        }

        fn get_state(&self, id: &StateId) -> StoreResult<Option<State>> {
            self.state_reads.fetch_add(1, Ordering::SeqCst);
            self.inner.get_state(id)
        }

        fn put_state(&self, state: &State) -> StoreResult<()> {
            self.inner.put_state(state)
        }

        fn has_state(&self, id: &StateId) -> StoreResult<bool> {
            self.inner.has_state(id)
        }

        fn list_states(&self) -> StoreResult<Vec<StateId>> {
            self.inner.list_states()
        }

        fn get_action(&self, id: &ActionId) -> StoreResult<Option<Action>> {
            self.inner.get_action(id)
        }

        fn put_action(&self, action: &mut Action) -> StoreResult<ActionId> {
            self.inner.put_action(action)
        }

        fn list_actions(&self) -> StoreResult<Vec<ActionId>> {
            self.inner.list_actions()
        }

        fn list_blobs(&self) -> StoreResult<Vec<ContentHash>> {
            self.inner.list_blobs()
        }

        fn list_trees(&self) -> StoreResult<Vec<ContentHash>> {
            self.inner.list_trees()
        }
    }

    fn test_attribution() -> Attribution {
        Attribution::human(Principal::new("Graph Tester", "graph@example.com"))
    }

    #[test]
    fn lean_closure_planner_matches_object_info_ids_and_types() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        std::fs::create_dir_all(temp.path().join("src")).unwrap();
        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
        std::fs::write(temp.path().join("src/lib.rs"), "pub fn hi() {}\n").unwrap();
        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();

        let full = enumerate_state_closure_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();
        let lean = enumerate_state_closure_plan_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();

        let full_pairs = full
            .into_iter()
            .map(|info| (info.id, info.obj_type))
            .collect::<std::collections::HashSet<_>>();
        let lean_pairs = lean
            .into_iter()
            .map(|info| (info.id, info.obj_type))
            .collect::<std::collections::HashSet<_>>();

        assert_eq!(full_pairs, lean_pairs);
        assert!(
            full_pairs
                .iter()
                .any(|(id, _)| matches!(id, ObjectId::StateId(_)))
        );
    }

    #[test]
    fn transfer_boundary_stops_at_server_head_without_walking_its_history() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        let path = temp.path().join("story.txt");

        std::fs::write(&path, "base\n").unwrap();
        let base = repo.snapshot(Some("base".to_string()), None).unwrap();
        std::fs::write(&path, "middle\n").unwrap();
        let middle = repo.snapshot(Some("middle".to_string()), None).unwrap();
        std::fs::write(&path, "tip\n").unwrap();
        let tip = repo.snapshot(Some("tip".to_string()), None).unwrap();

        let counting = CountingStore::new(repo.store());
        let transfer = enumerate_state_closure_transfer_from_boundaries(
            &counting,
            tip.state_id,
            &[middle.state_id],
            512,
        )
        .unwrap();
        let states = transfer
            .planned_objects
            .iter()
            .filter_map(|object| match object.id {
                ObjectId::StateId(state) if object.obj_type == ObjectType::State => Some(state),
                _ => None,
            })
            .collect::<Vec<_>>();

        assert_eq!(states, vec![tip.state_id]);
        assert!(!states.contains(&middle.state_id));
        assert!(!states.contains(&base.state_id));
        assert_eq!(
            counting.state_reads(),
            1,
            "the advertised server boundary must stop the walk before reading old states"
        );
    }

    #[test]
    fn transfer_projection_matches_full_and_plan_on_mixed_state_closure_fixture() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();

        let excluded_blob = repo
            .store()
            .put_blob(&Blob::from("excluded"))
            .expect("put excluded blob");
        let excluded_tree_hash = repo
            .store()
            .put_tree(&Tree::from_entries(vec![
                TreeEntry::file("excluded.txt", excluded_blob, false).unwrap(),
            ]))
            .expect("put excluded tree");
        let excluded_parent = State::new(excluded_tree_hash, Vec::new(), test_attribution());
        repo.store()
            .put_state(&excluded_parent)
            .expect("put excluded parent");

        let redacted_blob = repo
            .store()
            .put_blob(&Blob::from("secret"))
            .expect("put redacted blob");
        let nested_blob = repo
            .store()
            .put_blob(&Blob::from("nested"))
            .expect("put nested blob");
        let symlink_blob = repo
            .store()
            .put_blob(&Blob::from("target"))
            .expect("put symlink blob");
        let context_blob = repo
            .store()
            .put_blob(&Blob::from("context"))
            .expect("put context blob");
        let provenance_blob = repo
            .store()
            .put_blob(&Blob::from("provenance"))
            .expect("put provenance blob");
        let risk_blob = repo
            .store()
            .put_blob(&Blob::from("risk"))
            .expect("put risk blob");
        let review_blob = repo
            .store()
            .put_blob(&Blob::from("review"))
            .expect("put review blob");
        let discussions_blob = repo
            .store()
            .put_blob(&Blob::from("discussion"))
            .expect("put discussion blob");
        let conflicts_blob = repo
            .store()
            .put_blob(&Blob::from("conflicts"))
            .expect("put conflicts blob");

        let nested_tree_hash = repo
            .store()
            .put_tree(&Tree::from_entries(vec![
                TreeEntry::file("nested.txt", nested_blob, false).unwrap(),
                TreeEntry::symlink("latest", symlink_blob).unwrap(),
            ]))
            .expect("put nested tree");
        let context_tree_hash = repo
            .store()
            .put_tree(&Tree::from_entries(vec![
                TreeEntry::file("context.txt", context_blob, false).unwrap(),
            ]))
            .expect("put context tree");
        let provenance_tree_hash = repo
            .store()
            .put_tree(&Tree::from_entries(vec![
                TreeEntry::file("lineage.txt", provenance_blob, false).unwrap(),
            ]))
            .expect("put provenance tree");
        let gitlink_target: GitObjectId = "0303030303030303030303030303030303030303"
            .parse()
            .expect("git oid");
        let root_tree_hash = repo
            .store()
            .put_tree(&Tree::from_entries(vec![
                TreeEntry::file("secret.txt", redacted_blob, false).unwrap(),
                TreeEntry::directory("nested", nested_tree_hash).unwrap(),
                TreeEntry::gitlink("vendor", gitlink_target).unwrap(),
            ]))
            .expect("put root tree");
        let state = State::new(
            root_tree_hash,
            vec![excluded_parent.state_id],
            test_attribution(),
        )
        .with_provenance(provenance_tree_hash);
        repo.store().put_state(&state).expect("put state");
        for body in [
            StateAttachmentBody::Context(context_tree_hash),
            StateAttachmentBody::RiskSignals(risk_blob),
            StateAttachmentBody::ReviewSignatures(review_blob),
            StateAttachmentBody::Discussions(discussions_blob),
            StateAttachmentBody::StructuredConflicts(conflicts_blob),
        ] {
            repo.put_state_attachment(&StateAttachment {
                state_id: state.id(),
                body,
                attribution: state.attribution.clone(),
                created_at: Utc::now(),
                supersedes: None,
            })
            .unwrap();
        }

        repo.put_redaction(Redaction {
            redacted_blob,
            state: state.state_id,
            path: "secret.txt".to_string(),
            reason: "test leak".to_string(),
            redactor: Principal::new("Tester", "tester@example.test"),
            redacted_at: Utc::now(),
            signature: None,
            purged_at: None,
            supersedes: None,
        })
        .expect("put redaction");
        repo.put_state_visibility(StateVisibility {
            state: state.state_id,
            tier: VisibilityTier::Restricted {
                scope_label: "security".to_string(),
            },
            embargo_until: None,
            declarer: Principal::new("Tester", "tester@example.test"),
            declared_at: Utc::now(),
            signature: None,
            supersedes: None,
        })
        .expect("put visibility");

        let options = StateClosureOptions {
            depth: None,
            exclude_states: vec![excluded_parent.state_id],
        };
        let transfer = enumerate_state_closure_transfer_with_options(
            repo.store(),
            state.state_id,
            options.clone(),
            512,
        )
        .expect("transfer projection");

        let full =
            enumerate_state_closure_with_options(repo.store(), state.state_id, options.clone())
                .expect("full closure");
        let plan = enumerate_state_closure_plan_with_options(repo.store(), state.state_id, options)
            .expect("plan closure");
        assert_eq!(
            transfer
                .full_objects
                .as_deref()
                .map(object_info_fingerprint),
            Some(object_info_fingerprint(&full))
        );
        assert_eq!(transfer.planned_objects, plan);

        let full_pairs = pairs_from_full(&full);
        assert_eq!(full_pairs, pairs_from_plan(&plan));
        assert_contains_object(
            &full_pairs,
            ObjectId::StateId(state.state_id),
            ObjectType::State,
        );
        assert_contains_object(
            &full_pairs,
            ObjectId::StateId(state.state_id),
            ObjectType::StateVisibility,
        );
        assert_contains_object(&full_pairs, ObjectId::Hash(redacted_blob), ObjectType::Blob);
        assert_contains_object(
            &full_pairs,
            ObjectId::Hash(redacted_blob),
            ObjectType::Redaction,
        );
        for hash in [
            root_tree_hash,
            nested_tree_hash,
            context_tree_hash,
            provenance_tree_hash,
        ] {
            assert_contains_object(&full_pairs, ObjectId::Hash(hash), ObjectType::Tree);
        }
        for hash in [
            nested_blob,
            symlink_blob,
            context_blob,
            provenance_blob,
            risk_blob,
            review_blob,
            discussions_blob,
            conflicts_blob,
        ] {
            assert_contains_object(&full_pairs, ObjectId::Hash(hash), ObjectType::Blob);
        }
        assert!(!full_pairs.contains(&(
            ObjectId::StateId(excluded_parent.state_id),
            ObjectType::State
        )));
        assert!(!full_pairs.contains(&(ObjectId::Hash(excluded_tree_hash), ObjectType::Tree)));
        assert!(!full_pairs.contains(&(ObjectId::Hash(excluded_blob), ObjectType::Blob)));
    }

    #[test]
    fn transfer_projection_reads_root_state_once_on_small_transfer() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        let blob = repo
            .store()
            .put_blob(&Blob::from("hello\n"))
            .expect("put blob");
        let tree_hash = repo
            .store()
            .put_tree(&Tree::from_entries(vec![
                TreeEntry::file("README.md", blob, false).unwrap(),
            ]))
            .expect("put tree");
        let state = State::new(tree_hash, Vec::new(), test_attribution());
        repo.store().put_state(&state).expect("put state");
        let store = CountingStore::new(repo.store());

        let transfer = enumerate_state_closure_transfer_with_options(
            &store,
            state.state_id,
            StateClosureOptions::default(),
            512,
        )
        .expect("transfer projection");

        assert!(
            !transfer.planned_objects.is_empty(),
            "lean projection should be available"
        );
        assert!(transfer.full_objects.is_some());
        assert_eq!(
            store.state_reads(),
            1,
            "small transfer projection must not read the root state through a second closure walk"
        );
    }

    #[test]
    fn transfer_projection_drops_full_descriptors_after_threshold() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();

        let transfer = enumerate_state_closure_transfer_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
            0,
        )
        .expect("transfer projection");

        assert!(
            !transfer.planned_objects.is_empty(),
            "lean projection should still be available over the threshold"
        );
        assert!(transfer.full_objects.is_none());
    }

    #[test]
    fn depth_and_exclude_options_match_between_full_and_plan() {
        use std::collections::BTreeMap;

        use objects::object::{BindingDelta, SemanticIndexRoot, SemanticTreeNode};

        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        let path = temp.path().join("story.txt");

        std::fs::write(&path, "base\n").unwrap();
        let base = repo.snapshot(Some("base".to_string()), None).unwrap();
        std::fs::write(&path, "middle\n").unwrap();
        let middle = repo.snapshot(Some("middle".to_string()), None).unwrap();
        std::fs::write(&path, "tip\n").unwrap();
        let tip = repo.snapshot(Some("tip".to_string()), None).unwrap();

        let (semantic_tree, semantic_digest) = SemanticTreeNode::new(Vec::new());
        let semantic_tree_hash = repo
            .store()
            .put_blob(&Blob::new(semantic_tree.encode().unwrap()))
            .unwrap();
        let attach_delta = |state: StateId, parent: Option<ContentHash>| {
            let delta = BindingDelta::new(parent, Vec::new());
            let delta_hash = repo
                .store()
                .put_blob(&Blob::new(delta.encode().unwrap()))
                .unwrap();
            let root =
                SemanticIndexRoot::new(1, BTreeMap::new(), semantic_tree_hash, semantic_digest)
                    .with_binding_delta(delta_hash, 1);
            let root_hash = repo
                .store()
                .put_blob(&Blob::new(root.encode().unwrap()))
                .unwrap();
            repo.put_state_attachment(&StateAttachment {
                state_id: state,
                body: StateAttachmentBody::SemanticIndex(root_hash),
                attribution: test_attribution(),
                created_at: Utc::now(),
                supersedes: None,
            })
            .unwrap();
            delta_hash
        };
        let base_delta = attach_delta(base.state_id, None);
        let middle_delta = attach_delta(middle.state_id, Some(base_delta));
        let tip_delta = attach_delta(tip.state_id, Some(middle_delta));

        let depth_zero = assert_plan_parity(
            &repo,
            tip.state_id,
            StateClosureOptions {
                depth: Some(0),
                exclude_states: Vec::new(),
            },
        );
        assert!(depth_zero.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
        assert!(!depth_zero.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
        assert!(!depth_zero.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
        assert!(depth_zero.contains(&(ObjectId::Hash(tip_delta), ObjectType::Blob)));
        assert!(!depth_zero.contains(&(ObjectId::Hash(middle_delta), ObjectType::Blob)));
        assert!(!depth_zero.contains(&(ObjectId::Hash(base_delta), ObjectType::Blob)));

        let depth_one = assert_plan_parity(
            &repo,
            tip.state_id,
            StateClosureOptions {
                depth: Some(1),
                exclude_states: Vec::new(),
            },
        );
        assert!(depth_one.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
        assert!(depth_one.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
        assert!(!depth_one.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
        assert!(depth_one.contains(&(ObjectId::Hash(tip_delta), ObjectType::Blob)));
        assert!(depth_one.contains(&(ObjectId::Hash(middle_delta), ObjectType::Blob)));
        assert!(!depth_one.contains(&(ObjectId::Hash(base_delta), ObjectType::Blob)));

        let exclude_middle = assert_plan_parity(
            &repo,
            tip.state_id,
            StateClosureOptions {
                depth: None,
                exclude_states: vec![middle.state_id],
            },
        );
        assert!(exclude_middle.contains(&(ObjectId::StateId(tip.state_id), ObjectType::State)));
        assert!(!exclude_middle.contains(&(ObjectId::StateId(middle.state_id), ObjectType::State)));
        assert!(!exclude_middle.contains(&(ObjectId::StateId(base.state_id), ObjectType::State)));
    }

    #[test]
    fn shared_tree_and_blob_references_are_emitted_once() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();

        let shared_blob = Blob::from("shared contents\n");
        let shared_blob_hash = repo.store().put_blob(&shared_blob).unwrap();
        let shared_tree = Tree::from_entries(vec![
            TreeEntry::file("shared.txt", shared_blob_hash, false).unwrap(),
        ]);
        let shared_tree_hash = repo.store().put_tree(&shared_tree).unwrap();
        let root = Tree::from_entries(vec![
            TreeEntry::directory("left", shared_tree_hash).unwrap(),
            TreeEntry::directory("right", shared_tree_hash).unwrap(),
        ]);
        let root_hash = repo.store().put_tree(&root).unwrap();
        let state = State::new(root_hash, Vec::new(), test_attribution());
        repo.store().put_state(&state).unwrap();

        let full = enumerate_state_closure_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();
        let plan = enumerate_state_closure_plan_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();

        assert_eq!(
            pairs_from_full(&full),
            pairs_from_plan(&plan),
            "full and lean closure enumerators must dedup the same objects"
        );

        assert_eq!(
            full.iter()
                .filter(|info| info.id == ObjectId::Hash(root_hash)
                    && info.obj_type == ObjectType::Tree)
                .count(),
            1
        );
        assert_eq!(
            full.iter()
                .filter(|info| info.id == ObjectId::Hash(shared_tree_hash)
                    && info.obj_type == ObjectType::Tree)
                .count(),
            1
        );
        assert_eq!(
            full.iter()
                .filter(|info| info.id == ObjectId::Hash(shared_blob_hash)
                    && info.obj_type == ObjectType::Blob)
                .count(),
            1
        );
    }

    #[test]
    fn state_closure_skips_gitlink_targets() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        let target: GitObjectId = "0303030303030303030303030303030303030303"
            .parse()
            .expect("git oid");
        let root = Tree::from_entries(vec![
            TreeEntry::gitlink("vendor", target).expect("gitlink entry"),
        ]);
        let root_hash = repo.store().put_tree(&root).unwrap();
        let state = State::new(root_hash, Vec::new(), test_attribution());
        repo.store().put_state(&state).unwrap();

        let full = enumerate_state_closure_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();
        let plan = enumerate_state_closure_plan_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();

        assert_eq!(pairs_from_full(&full), pairs_from_plan(&plan));
        assert!(
            !full.iter().any(|info| info.obj_type == ObjectType::Blob),
            "gitlinks carry foreign Git commit ids, not Heddle blob dependencies: {full:?}"
        );
        assert!(full.iter().any(|info| {
            info.id == ObjectId::Hash(root_hash) && info.obj_type == ObjectType::Tree
        }));
    }

    #[test]
    fn missing_blobs_in_tree_skips_gitlinks_and_walks_nested_side_paths() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        let present_blob = repo
            .store()
            .put_blob(&Blob::from("already local"))
            .expect("put present blob");
        let missing_nested = ContentHash::from_bytes([7; 32]);
        let missing_symlink = ContentHash::from_bytes([8; 32]);
        let nested_tree = Tree::from_entries(vec![
            TreeEntry::file("remote.txt", missing_nested, false).unwrap(),
            TreeEntry::symlink("remote-link", missing_symlink).unwrap(),
        ]);
        let nested_tree_hash = repo
            .store()
            .put_tree(&nested_tree)
            .expect("put nested tree");
        let gitlink_target: GitObjectId = "0404040404040404040404040404040404040404"
            .parse()
            .expect("git oid");
        let root = Tree::from_entries(vec![
            TreeEntry::file("local.txt", present_blob, false).unwrap(),
            TreeEntry::directory("nested", nested_tree_hash).unwrap(),
            TreeEntry::gitlink("vendor", gitlink_target).unwrap(),
        ]);
        let root_hash = repo.store().put_tree(&root).expect("put root tree");

        let missing = missing_blobs_in_tree(repo.store(), root_hash).expect("missing blobs");

        assert_eq!(
            missing.into_iter().collect::<HashSet<_>>(),
            HashSet::from([missing_nested, missing_symlink])
        );
    }

    /// Once a redaction is declared for a blob in a snapshot, the
    /// state closure must include an `ObjectType::Redaction` entry
    /// keyed on that blob's hash — that's the wire-side signal the
    /// receiver replays.
    #[test]
    fn enumerate_state_closure_emits_redaction_for_redacted_blob() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        std::fs::write(temp.path().join("secret.toml"), "api_token = \"x\"\n").unwrap();
        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();

        // Find the blob hash for secret.toml by walking the snapshot's tree.
        let tree = repo
            .store()
            .get_tree(&state.tree)
            .unwrap()
            .expect("tree present");
        let blob_hash = tree
            .iter()
            .find(|e| e.name() == "secret.toml")
            .expect("entry present")
            .blob_hash()
            .expect("secret.toml is a blob");

        let redaction = Redaction {
            redacted_blob: blob_hash,
            state: state.state_id,
            path: "secret.toml".to_string(),
            reason: "test leak".to_string(),
            redactor: Principal {
                name: "Tester".into(),
                email: "tester@heddle.sh".into(),
            },
            redacted_at: Utc::now(),
            signature: None,
            purged_at: None,
            supersedes: None,
        };
        repo.put_redaction(redaction).unwrap();

        let full = enumerate_state_closure_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();
        let plan = enumerate_state_closure_plan_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();

        assert!(
            full.iter()
                .any(|info| info.obj_type == ObjectType::Redaction
                    && info.id == ObjectId::Hash(blob_hash)),
            "full closure must include a Redaction entry for the redacted blob"
        );
        assert!(
            plan.iter()
                .any(|p| p.obj_type == ObjectType::Redaction && p.id == ObjectId::Hash(blob_hash)),
            "plan closure must include a Redaction entry for the redacted blob"
        );
    }

    #[test]
    fn enumerate_state_closure_emits_state_visibility_for_visible_state() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();

        repo.put_state_visibility(StateVisibility {
            state: state.state_id,
            tier: VisibilityTier::Restricted {
                scope_label: "security-embargo".into(),
            },
            embargo_until: None,
            declarer: Principal {
                name: "Tester".into(),
                email: "tester@heddle.sh".into(),
            },
            declared_at: Utc::now(),
            signature: None,
            supersedes: None,
        })
        .unwrap();

        let full = enumerate_state_closure_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();
        let plan = enumerate_state_closure_plan_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();

        assert!(
            full.iter()
                .any(|info| info.obj_type == ObjectType::StateVisibility
                    && info.id == ObjectId::StateId(state.state_id)),
            "full closure must include a StateVisibility entry for the visible state"
        );
        assert!(
            plan.iter()
                .any(|p| p.obj_type == ObjectType::StateVisibility
                    && p.id == ObjectId::StateId(state.state_id)),
            "plan closure must include a StateVisibility entry for the visible state"
        );
    }

    #[test]
    fn enumerate_state_closure_emits_state_metadata_blobs() {
        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();

        let principal = Principal::new("Tester", "tester@example.test");
        let discussion_bytes = DiscussionsBlob::new(vec![Discussion {
            id: "disc-1".to_string(),
            anchor: SymbolAnchor::new("src/lib.rs", "answer"),
            opened_against_state: state.state_id,
            opened_at: 1_782_400_000,
            thread_ref: None,
            turns: vec![DiscussionTurn {
                author: principal,
                body: "Should this sync?".to_string(),
                posted_at: 1_782_400_000,
                references: Vec::new(),
            }],
            resolution: DiscussionResolution::Open,
            body_changed_since_open: false,
            orphaned: false,
            visibility: VisibilityTier::default(),
            resolved_annotation_id: None,
        }])
        .encode()
        .expect("encode discussions");
        let discussion_hash = repo
            .store()
            .put_blob(&Blob::new(discussion_bytes))
            .expect("put discussions blob");
        let risk_hash = repo
            .store()
            .put_blob(&Blob::from_slice(b"risk signals"))
            .expect("put risk blob");
        let review_hash = repo
            .store()
            .put_blob(&Blob::from_slice(b"review signatures"))
            .expect("put review blob");
        let conflicts_hash = repo
            .store()
            .put_blob(&Blob::from_slice(b"structured conflicts"))
            .expect("put conflicts blob");
        for body in [
            StateAttachmentBody::RiskSignals(risk_hash),
            StateAttachmentBody::ReviewSignatures(review_hash),
            StateAttachmentBody::Discussions(discussion_hash),
            StateAttachmentBody::StructuredConflicts(conflicts_hash),
        ] {
            repo.put_state_attachment(&StateAttachment {
                state_id: state.id(),
                body,
                attribution: state.attribution.clone(),
                created_at: Utc::now(),
                supersedes: None,
            })
            .unwrap();
        }

        let full = enumerate_state_closure_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();
        let plan = enumerate_state_closure_plan_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();

        for metadata_hash in [risk_hash, review_hash, discussion_hash, conflicts_hash] {
            assert!(
                full.iter().any(|info| info.obj_type == ObjectType::Blob
                    && info.id == ObjectId::Hash(metadata_hash)),
                "full closure must include state metadata blob {metadata_hash}"
            );
            assert!(
                plan.iter().any(
                    |p| p.obj_type == ObjectType::Blob && p.id == ObjectId::Hash(metadata_hash)
                ),
                "plan closure must include state metadata blob {metadata_hash}"
            );
        }
    }

    /// The push/pull packability split routes `StateAttachment` off the push
    /// pack (weft#549 forgery seal) while leaving pull carriage and every other
    /// type untouched.
    #[test]
    fn packable_predicates_split_state_attachment_by_direction() {
        // Sidecar records are never packable in either direction.
        for sidecar in [
            ObjectType::Redaction,
            ObjectType::StateVisibility,
            ObjectType::KeyBinding,
        ] {
            assert!(!sidecar.packable_for_push(), "{sidecar:?} push");
            assert!(!sidecar.packable_for_pull(), "{sidecar:?} pull");
        }
        // Content-addressed objects ride the pack in both directions.
        for packable in [
            ObjectType::Blob,
            ObjectType::Tree,
            ObjectType::State,
            ObjectType::Action,
        ] {
            assert!(packable.packable_for_push(), "{packable:?} push");
            assert!(packable.packable_for_pull(), "{packable:?} pull");
        }
        // The attachment record: excluded from the push pack, kept on pull.
        assert!(!ObjectType::StateAttachment.packable_for_push());
        assert!(ObjectType::StateAttachment.packable_for_pull());
    }

    /// A pushed state's semantic-index attachment RECORD must be excluded from
    /// the push pack (it rides the sidecar lane) while its semantic-index
    /// content blobs still ride the pack in both directions, and the same
    /// record stays packable server->client on pull.
    #[test]
    fn semantic_index_attachment_excluded_from_push_pack_but_kept_for_pull() {
        use std::collections::BTreeMap;

        use objects::object::{
            BindingDelta, FileBindingDelta, SemanticIndexRoot, SemanticTreeNode,
        };

        let temp = TempDir::new().unwrap();
        let repo = Repository::init_default(temp.path()).unwrap();
        std::fs::write(temp.path().join("README.md"), "hello\n").unwrap();
        let state = repo.snapshot(Some("seed".to_string()), None).unwrap();

        // Minimal valid semantic-index fixture: an empty tree node under a root.
        let (node, node_digest) = SemanticTreeNode::new(Vec::new());
        let node_hash = repo
            .store()
            .put_blob(&Blob::new(node.encode().unwrap()))
            .expect("put semantic tree node");
        let base_delta = BindingDelta::new(
            None,
            vec![FileBindingDelta::new(
                "unreachable-parent.rs",
                None,
                Vec::new(),
            )],
        );
        let base_delta_hash = repo
            .store()
            .put_blob(&Blob::new(base_delta.encode().unwrap()))
            .expect("put base binding delta");
        let delta = BindingDelta::new(Some(base_delta_hash), Vec::new());
        let delta_hash = repo
            .store()
            .put_blob(&Blob::new(delta.encode().unwrap()))
            .expect("put binding delta");
        let root = SemanticIndexRoot::new(1, BTreeMap::new(), node_hash, node_digest)
            .with_binding_delta(delta_hash, 1);
        let root_hash = repo
            .store()
            .put_blob(&Blob::new(root.encode().unwrap()))
            .expect("put semantic index root");
        repo.put_state_attachment(&StateAttachment {
            state_id: state.state_id,
            body: StateAttachmentBody::SemanticIndex(root_hash),
            attribution: test_attribution(),
            created_at: Utc::now(),
            supersedes: None,
        })
        .unwrap();

        let plan = enumerate_state_closure_plan_with_options(
            repo.store(),
            state.state_id,
            StateClosureOptions::default(),
        )
        .unwrap();

        // Every StateAttachment record (at least the semantic index authored
        // above) is push-excluded and pull-included.
        let attachments: Vec<_> = plan
            .iter()
            .filter(|p| p.obj_type == ObjectType::StateAttachment)
            .collect();
        assert!(
            !attachments.is_empty(),
            "closure must contain the semantic-index attachment record"
        );
        for attachment in &attachments {
            assert!(matches!(attachment.id, ObjectId::StateAttachment { .. }));
            // Push: excluded from the pack (sidecar lane). Pull: kept in pack.
            assert!(
                !attachment.obj_type.packable_for_push(),
                "attachment record must be excluded from the push pack"
            );
            assert!(
                attachment.obj_type.packable_for_pull(),
                "attachment record must stay in the pull pack"
            );
        }

        // The semantic-index CONTENT blobs are ordinary content-addressed
        // objects and still ride the pack in both directions — only the
        // attachment record is sidecar'd on push.
        for content in [root_hash, node_hash, delta_hash] {
            let obj = plan
                .iter()
                .find(|p| p.id == ObjectId::Hash(content))
                .unwrap_or_else(|| panic!("semantic content blob {content} in closure"));
            assert_eq!(obj.obj_type, ObjectType::Blob);
            assert!(obj.obj_type.packable_for_push());
            assert!(obj.obj_type.packable_for_pull());
        }
        assert!(
            !plan
                .iter()
                .any(|object| object.id == ObjectId::Hash(base_delta_hash)),
            "a binding delta belonging to an unreachable parent state must not ride this state's closure"
        );

        // Partitioning the plan by push-packability puts the record on the
        // sidecar side and never in the pack side.
        let (push_pack, push_sidecar): (Vec<_>, Vec<_>) =
            plan.iter().partition(|p| p.obj_type.packable_for_push());
        assert!(
            push_sidecar
                .iter()
                .any(|p| p.obj_type == ObjectType::StateAttachment),
            "attachment record routed to the push sidecar partition"
        );
        assert!(
            !push_pack
                .iter()
                .any(|p| p.obj_type == ObjectType::StateAttachment),
            "attachment record must not be in the push pack partition"
        );
    }
}