armdb 0.14.2

use std::any::Any;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;

use armour_core::{
    CollectionKind, CollectionNode, CompositeRef, FieldRef, GetRefs, GetType, KeyShape, OnMissing,
    PathSeg, RefVisitor, RelationEdge, RelationKind, RowPart, SchemaGraph, StorageClass, Typ,
};

use super::access::SchemaCollection;
use super::plan::{IdFlush, IdMap, MatchPlan, VisitRefsFn, build_match_plan, scope_of_path};
use super::validate::RefDesc;
use crate::CollectionMeta;
use crate::counter::{CounterSink, CounterTargets};
use crate::error::DbResult;
use crate::hook::NoHook;
use crate::key::Key;
use crate::projection::{IndexTargets, ProjectableIndex, Projected, ProjectionSink};
use crate::queue::{QueueMode, QueueSpec};
use crate::unique::{ClaimSink, UNIQUE_OWNER_KEY_CAP, UniqueTargets, collection_tag};
use crate::{EmitSink, Events, Queue, QueueTargets};

type ScanAnyFn = Box<dyn Fn(&mut dyn FnMut(&[u8], &dyn Any, &dyn Any)) + Send + Sync>;
type UndecodableRowsFn = Box<dyn Fn() -> u64 + Send + Sync>;
type ContainsAnyFn = Box<dyn Fn(&dyn Any) -> Option<bool> + Send + Sync>;
type WithValueAnyFn = Box<dyn Fn(&dyn Any, &mut dyn FnMut(&dyn Any)) -> Option<bool> + Send + Sync>;
type ValueEqBytesFn = Box<dyn Fn(&[u8], &[u8]) -> DbResult<bool> + Send + Sync>;
/// `(src K any, src V any, source projection budget, out rows per target index)`.
///
/// The budget is the source collection's already-resolved
/// `min(hook budget, Config::max_projected_*)`, read off the open handle at
/// registration — so a declaration cannot evaluate a contract writes never used.
/// `None` (a non-projecting family) falls back to the crate defaults.
type IndexProjectFn = Box<
    dyn Fn(
            &dyn Any,
            &dyn Any,
            Option<(usize, usize)>,
            &mut Vec<(usize, Vec<u8>, Vec<u8>)>,
        ) -> DbResult<()>
        + Send
        + Sync,
>;
pub(super) type ValueToBytesFn = Arc<dyn Fn(&dyn Any) -> Option<Vec<u8>> + Send + Sync>;

pub(super) struct IndexTargetDecl {
    pub name: &'static str,
    pub index: Arc<dyn ProjectableIndex>,
    pub value_eq: ValueEqBytesFn,
    pub value_to_bytes: ValueToBytesFn,
}
type DenormProjectFn =
    Box<dyn Fn(&dyn Any, &dyn Any) -> Option<(Vec<u8>, Box<dyn Any>)> + Send + Sync>;
type DenormConsistentFn = Box<dyn Fn(&dyn Any, &dyn Any) -> Option<bool> + Send + Sync>;
/// Erased counter projection: `(src K any, src V any, sink)`. Unlike
/// [`IndexProjectFn`] the sink (budget + declared targets) is constructed by
/// the caller — mirrors the typed [`Projected::and_counters`] closure shape,
/// just with `K`/`T` erased to `dyn Any`.
type CounterProjectErasedFn =
    Box<dyn Fn(&dyn Any, &dyn Any, &mut CounterSink<'_>) -> DbResult<()> + Send + Sync>;
/// Erased claim projection: `(src K any, src V any, sink)` — the T10 twin of
/// [`CounterProjectErasedFn`], same erasure shape as
/// [`Projected::and_unique`](crate::projection::Projected::and_unique)'s typed
/// closure.
type ClaimProjectErasedFn =
    Box<dyn Fn(&dyn Any, &dyn Any, &mut ClaimSink<'_>) -> DbResult<()> + Send + Sync>;
type ContainsFn = Box<dyn Fn(&[u64]) -> Option<bool> + Send + Sync>;

pub(in crate::schema) struct Home {
    pub(super) name: &'static str,
    pub(super) shape: &'static [&'static str],
    pub(super) contains: ContainsFn,
}

pub(super) struct HomePending {
    shape: &'static [&'static str],
    contains: ContainsFn,
}

fn sorted_shape(shape: &[&'static str]) -> Box<[&'static str]> {
    let mut v: Vec<_> = shape.to_vec();
    v.sort_unstable();
    v.into_boxed_slice()
}

/// Registry consistency error reported by [`SchemaRegistry::finish`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
pub enum RegistryBuildError {
    #[error("collection `{0}` registered twice")]
    DuplicateCollection(&'static str),
    /// Two different sources declared a counter for the same `(collection,
    /// field)`. With one source, the sum of a full scan *is* the absolute
    /// value — exactly what the open-time recount relies on. With several,
    /// each scan only yields one addend, and writing it would erase the
    /// others.
    #[error("collection `{collection}` field `{field}` has more than one declared counter source")]
    DuplicateCounterSource {
        collection: &'static str,
        field: &'static str,
    },
}

/// Display label for an FK edge in the graph/report (not extraction — closures do that).
#[derive(Debug, Clone)]
pub struct FkLabel(pub String);

impl FkLabel {
    pub fn field(name: &str) -> Self {
        Self(name.to_string())
    }
}

pub(super) struct CollectionEntry {
    pub name: &'static str,
    pub kind: CollectionKind,
    pub typ: Typ,
    pub key_typ: Typ,
    pub self_shape: Option<&'static [&'static str]>,
    pub storage: StorageClass,
    pub auto_refs: Vec<(RowPart, FieldRef)>,
    pub composites: Vec<(RowPart, CompositeRef)>,
    pub match_plan: MatchPlan,
    /// Scan rows; callback receives key bytes and id maps for key/value halves.
    pub visit_refs: VisitRefsFn,
    /// Scan with erased types; callback: (key_bytes, &K as Any, &V as Any).
    pub scan_any: ScanAnyFn,
    /// Rows a `scan` silently skips (undecodable values) — reported as a finding
    /// so a corrupt collection cannot pass validation as clean.
    pub undecodable_rows: UndecodableRowsFn,
    /// The budget this collection's writes project under, read off the open
    /// handle. A declaration sourced here evaluates the same contract.
    pub projection_limits: Option<(usize, usize)>,
    /// None — wrong key type (declaration error), Some(b) — presence.
    pub contains_any: ContainsAnyFn,
    /// Some(false) — no record, Some(true) — record existed and `f` was called.
    pub with_value_any: WithValueAnyFn,
    /// Taken in `finish` to populate the homes map.
    pub entity_home: Option<HomePending>,
}

/// Result of an explicit FK key-builder.
pub(super) enum FkBuild {
    /// No reference (e.g. `Option::None`) — valid.
    NoRef,
    Key(Box<dyn Any>),
    TypeMismatch,
}

type FkBuildFn = Box<dyn Fn(&dyn Any, &dyn Any) -> FkBuild + Send + Sync>;

pub(super) enum Decl {
    Index {
        source: &'static str,
        targets: Vec<IndexTargetDecl>,
        project: IndexProjectFn,
    },
    Fk {
        from: &'static str,
        to: &'static str,
        label: String,
        on_missing: OnMissing,
        /// (src K any, src V any) -> target key; inner `None` = no reference (valid).
        build: FkBuildFn,
    },
    Denorm {
        source: &'static str,
        denorm: &'static str,
        project: DenormProjectFn,
        consistent: DenormConsistentFn,
    },
    Counter {
        source: &'static str,
        targets: CounterTargets,
        project: CounterProjectErasedFn,
    },
    /// A source's declared unique-namespace claims (T10 Task 3). Reports
    /// through [`BuiltRegistry::graph`] / [`BuiltRegistry::unique_decl_sources`],
    /// and `project` is consumed at data-validation time by
    /// [`super::expected::build_unique`] — the same erased-closure shape
    /// `expected::build_counters` uses for `Decl::Counter.project`.
    Unique {
        source: &'static str,
        targets: UniqueTargets,
        project: ClaimProjectErasedFn,
    },
    /// Graph-only declaration. Queue state is intentionally excluded from
    /// expected-state construction, validation, and repair.
    Stream {
        source: &'static str,
        targets: QueueTargets,
    },
}

#[derive(Default)]
pub struct SchemaRegistry {
    pub(super) entries: Vec<CollectionEntry>,
    pub(super) decls: Vec<Decl>,
    pub(super) external: Vec<(&'static str, CollectionKind)>,
    projection_sources: HashSet<&'static str>,
    projection_targets: HashSet<&'static str>,
    queue_names: HashSet<&'static str>,
    stream_sources: HashSet<&'static str>,
    complete_indices: HashSet<&'static str>,
    /// One entry per `(target collection, field, source collection)` declared
    /// by [`SchemaRegistry::counted`] — checked for a single source per
    /// `(target, field)` in [`SchemaRegistry::finish`].
    counter_declarations: Vec<(&'static str, &'static str, &'static str)>,
    /// `(namespace, owner tag) -> source name` already registered into that
    /// namespace by [`SchemaRegistry::unique_index`]. Checked eagerly, per
    /// call, rather than in [`SchemaRegistry::finish`]: `finish()` runs after
    /// every source is already open, by which point a colliding claim could
    /// already be live — see [`SchemaRegistry::register_owner_tag`].
    owner_tags: HashMap<(&'static str, u64), &'static str>,
}

/// Registry after a successful consistency check.
pub struct BuiltRegistry {
    pub(super) reg: SchemaRegistry,
    /// Sorted brand multiset -> entity homes for that key shape.
    pub(super) homes: HashMap<Box<[&'static str]>, Vec<Home>>,
    pub(super) complete_indices: HashSet<&'static str>,
}

impl SchemaRegistry {
    pub fn new() -> Self {
        Self::default()
    }

    /// Register a collection with an open handle.
    pub fn register<S: SchemaCollection>(&mut self, handle: &Arc<S>, kind: CollectionKind) {
        self.register_inner(handle, kind, None);
    }

    /// Entity whose key shape becomes an auto-FK home.
    pub fn register_entity<S: SchemaCollection>(&mut self, handle: &Arc<S>)
    where
        S::K: KeyShape,
    {
        let h = handle.clone();
        let shape = S::K::SHAPE;
        if shape.iter().any(|b| b.is_empty()) {
            self.register_inner(handle, CollectionKind::Entity, None);
            return;
        }
        let contains = Box::new(move |ids: &[u64]| -> Option<bool> {
            let key = S::K::from_ids(ids)?;
            Some(h.contains_key(&key))
        });
        self.register_inner(
            handle,
            CollectionKind::Entity,
            Some(HomePending { shape, contains }),
        );
    }

    /// Graph node without a handle (no validation; clears UnclassifiedCollection).
    pub fn register_external(&mut self, name: &'static str, kind: CollectionKind) {
        self.external.push((name, kind));
    }

    /// Register a queue's physical storage for schema graph introspection.
    ///
    /// Queues deliberately have no [`SchemaCollection`] access handle: their
    /// delivery state is neither validated nor repaired by the schema layer.
    pub fn register_queue<N>(&mut self, _queue: &Arc<Queue<N>>)
    where
        N: QueueSpec,
    {
        self.register_queue_name(N::NAME);
        if N::Mode::HAS_PAYLOAD_ROW {
            self.register_queue_name(N::PAYLOAD_NAME);
        }
    }

    fn register_queue_name(&mut self, name: &'static str) {
        assert!(
            !self.projection_sources.contains(name) && !self.projection_targets.contains(name),
            "queue `{name}` participates in a schema projection (cascade forbidden)"
        );
        self.queue_names.insert(name);
        self.register_external(name, CollectionKind::Queue);
    }

    fn register_inner<S: SchemaCollection>(
        &mut self,
        handle: &Arc<S>,
        kind: CollectionKind,
        home: Option<HomePending>,
    ) {
        let mut auto_refs: Vec<(RowPart, FieldRef)> = S::K::refs()
            .into_iter()
            .map(|r| (RowPart::Key, r))
            .collect();
        auto_refs.extend(S::V::refs().into_iter().map(|r| (RowPart::Value, r)));

        let mut composites: Vec<(RowPart, CompositeRef)> = S::K::composites()
            .into_iter()
            .map(|c| (RowPart::Key, c))
            .collect();
        composites.extend(S::V::composites().into_iter().map(|c| (RowPart::Value, c)));

        let h = handle.clone();
        #[allow(clippy::type_complexity)]
        let visit_refs = Box::new(move |f: &mut dyn FnMut(&[u8], &IdMap, &IdMap, IdFlush)| {
            type ElemFlush<'a> = dyn FnMut(&IdMap, IdFlush) + 'a;
            struct Visitor<'a> {
                stack: Vec<PathSeg>,
                ids: &'a mut IdMap,
                /// When set, leaving `Elem` flushes planned edges for that instance.
                flush: Option<&'a mut ElemFlush<'a>>,
            }
            impl RefVisitor for Visitor<'_> {
                fn enter(&mut self, seg: PathSeg) {
                    self.stack.push(seg);
                }
                fn leave(&mut self) {
                    if self.stack.last() == Some(&PathSeg::Elem)
                        && let Some(flush) = self.flush.as_mut()
                    {
                        let scope = self.stack.clone();
                        flush(self.ids, IdFlush::Scope(scope.clone()));
                        self.ids.retain(|(_, _, full), _| {
                            full.len() < scope.len() || full[..scope.len()] != scope[..]
                        });
                    }
                    self.stack.pop();
                }
                fn id(&mut self, seg: Option<PathSeg>, brand: &'static str, id: u64) {
                    let mut full_path = self.stack.clone();
                    // `None` — the visited value *is* the id (`Fuid`/`Id64` at a
                    // root), whose static `refs()` path is the stack alone.
                    if let Some(seg) = seg {
                        full_path.push(seg);
                    }
                    let scope = scope_of_path(&full_path);
                    self.ids.insert((scope, brand, full_path), id);
                }
            }
            h.scan(&mut |k, v| {
                let kb = k.as_bytes();
                let mut key_ids = IdMap::new();
                k.visit_refs(&mut Visitor {
                    stack: Vec::new(),
                    ids: &mut key_ids,
                    flush: None,
                });
                let mut value_ids = IdMap::new();
                {
                    let mut on_flush = |ids: &IdMap, flush: IdFlush| {
                        f(kb, &key_ids, ids, flush);
                    };
                    v.visit_refs(&mut Visitor {
                        stack: Vec::new(),
                        ids: &mut value_ids,
                        flush: Some(&mut on_flush),
                    });
                }
                f(kb, &key_ids, &value_ids, IdFlush::Final);
            });
        }) as VisitRefsFn;

        let h = handle.clone();
        let scan_any: ScanAnyFn = Box::new(move |f| {
            h.scan(&mut |k, v| f(k.as_bytes(), k as &dyn Any, v as &dyn Any));
        });

        let h = handle.clone();
        let undecodable_rows: UndecodableRowsFn = Box::new(move || h.undecodable_rows());

        let h = handle.clone();
        let contains_any =
            Box::new(move |key: &dyn Any| key.downcast_ref::<S::K>().map(|k| h.contains_key(k)))
                as ContainsAnyFn;

        let h = handle.clone();
        let with_value_any = Box::new(move |key: &dyn Any, f: &mut dyn FnMut(&dyn Any)| {
            key.downcast_ref::<S::K>()
                .map(|k| h.with_value(k, &mut |v| f(v as &dyn Any)))
        }) as WithValueAnyFn;

        let self_shape = home.as_ref().map(|h| h.shape);

        self.entries.push(CollectionEntry {
            name: S::V::NAME,
            kind,
            typ: <S::V as GetType>::TYPE,
            key_typ: <S::K as GetType>::TYPE,
            self_shape,
            storage: S::STORAGE,
            auto_refs,
            composites,
            match_plan: MatchPlan {
                ambiguities: Vec::new(),
                edges: Vec::new(),
                unvalidated_brands: HashSet::new(),
            },
            visit_refs,
            scan_any,
            undecodable_rows,
            projection_limits: handle.effective_projection_limits(),
            contains_any,
            with_value_any,
            entity_home: home,
        });
    }

    /// Consistency check: unique collection names; assemble homes map.
    /// Declarations are validated as they are added in later tasks.
    pub fn finish(mut self) -> Result<BuiltRegistry, RegistryBuildError> {
        let mut names: HashSet<&'static str> = HashSet::new();
        for e in &self.entries {
            if !names.insert(e.name) {
                return Err(RegistryBuildError::DuplicateCollection(e.name));
            }
        }
        // One source per (target, field): a second source declared for the
        // same field would only ever contribute an addend, not the absolute
        // value the open-time recount needs.
        let mut counter_owner: HashMap<(&'static str, &'static str), &'static str> = HashMap::new();
        for &(collection, field, source) in &self.counter_declarations {
            let owner = *counter_owner.entry((collection, field)).or_insert(source);
            if owner != source {
                return Err(RegistryBuildError::DuplicateCounterSource { collection, field });
            }
        }
        let mut homes: HashMap<Box<[&'static str]>, Vec<Home>> = HashMap::new();
        for e in &mut self.entries {
            if let Some(pending) = e.entity_home.take() {
                let key = sorted_shape(pending.shape);
                homes.entry(key).or_default().push(Home {
                    name: e.name,
                    shape: pending.shape,
                    contains: pending.contains,
                });
            }
        }
        for e in &mut self.entries {
            e.match_plan =
                build_match_plan(e.name, e.self_shape, &e.auto_refs, &e.composites, &homes);
        }
        let complete_indices = std::mem::take(&mut self.complete_indices);
        Ok(BuiltRegistry {
            reg: self,
            homes,
            complete_indices,
        })
    }

    /// Every producer of this index is declared in this registry.
    pub fn complete_index<I: SchemaCollection>(&mut self, _index: &Arc<I>) {
        self.complete_indices.insert(I::V::NAME);
    }

    /// Every producer of this unique namespace is declared in this registry
    /// (T10 Task 10) — the namespace twin of [`Self::complete_index`], and
    /// deliberately reusing its exact gate: same `complete_indices` set, same
    /// meaning, same consumers (`Finding::UniqueOrphan`'s `complete` field
    /// here; `RepairOptions::delete_orphans`'s partner check in
    /// `schema::unique_repair::repair_namespaces`).
    ///
    /// A separate entry point rather than an overload of `complete_index`
    /// itself: that method's bound is `I: SchemaCollection`, which requires
    /// `Collection` (a public write path) — `UniqueIndex` has none of that by
    /// design (see `unique/index.rs`'s module doc), so it cannot satisfy the
    /// bound `complete_index` already has. The set the two write into is the
    /// same one, so a namespace and an index sharing a `NAME` would collide —
    /// not a real risk in practice, since a namespace is opened via
    /// `Db::open_unique_index` and every other family through its own
    /// `open_*`, and `Db` itself already refuses two collections sharing one
    /// name (`reserve_name`).
    pub fn complete_namespace<N>(&mut self, _ns: &Arc<crate::unique::UniqueIndex<N>>)
    where
        N: crate::unique::UniqueNamespace,
        N::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
    {
        self.complete_indices.insert(N::NAME);
    }

    fn reject_index_decl(&self, source: &'static str, targets: &IndexTargets) {
        assert!(
            !targets.is_empty(),
            "secondary_index: IndexTargets must declare at least one target"
        );
        for t in targets.erased_targets() {
            assert!(
                !self.queue_names.contains(t.name),
                "secondary_index: queue `{}` cannot be a projection target",
                t.name
            );
            assert_ne!(
                t.name, source,
                "secondary_index: target `{0}` must not equal source `{source}`",
                t.name
            );
            assert!(
                !self.projection_sources.contains(t.name),
                "secondary_index: target `{0}` is already a projecting source (cascade forbidden)",
                t.name
            );
        }
        assert!(
            !self.queue_names.contains(source),
            "secondary_index: queue `{source}` cannot be a projection source"
        );
        assert!(
            !self.projection_targets.contains(source),
            "secondary_index: source `{source}` is already a projection target (cascade forbidden)"
        );
    }

    /// Same rejection as [`Self::reject_index_decl`], for a counter
    /// declaration: shares `projection_sources`/`projection_targets` with the
    /// index half, so an index target cannot become a counter source (or vice
    /// versa) any more than an index target can become another index source.
    /// A collection that is both a counter target and a projecting source is
    /// rejected at commit time by `DbError::TxTargetIsAlsoASource` and also
    /// breaks the open-time recount (it would clear its own dirty flag and
    /// strand the source upstream of it) — catching it here is where the
    /// error is cheapest to understand.
    fn reject_counter_decl(&self, source: &'static str, targets: &CounterTargets) {
        assert!(
            !targets.is_empty(),
            "counted: CounterTargets must declare at least one target"
        );
        for t in targets.all() {
            let name = t.collection_name();
            assert!(
                !self.queue_names.contains(name),
                "counted: queue `{name}` cannot be a projection target"
            );
            assert_ne!(
                name, source,
                "counted: target `{name}` must not equal source `{source}`",
            );
            assert!(
                !self.projection_sources.contains(name),
                "counted: target `{name}` is already a projecting source (cascade forbidden)",
            );
        }
        assert!(
            !self.queue_names.contains(source),
            "counted: queue `{source}` cannot be a projection source"
        );
        assert!(
            !self.projection_targets.contains(source),
            "counted: source `{source}` is already a projection target (cascade forbidden)"
        );
    }

    /// Same rejection as [`Self::reject_index_decl`] / [`Self::reject_counter_decl`],
    /// for a unique-namespace claim declaration (T10): shares
    /// `projection_sources`/`projection_targets` with the index and counter
    /// halves, so a namespace cannot become a projecting/counting source (or
    /// vice versa) any more than an index target can.
    fn reject_unique_decl(&self, source: &'static str, targets: &UniqueTargets) {
        assert!(
            !targets.is_empty(),
            "unique_index: UniqueTargets must declare at least one target"
        );
        for t in targets.all() {
            let name = t.collection_name();
            assert_ne!(
                name, source,
                "unique_index: namespace `{name}` must not equal source `{source}`",
            );
            assert!(
                !self.projection_sources.contains(name),
                "unique_index: namespace `{name}` is already a projecting source (cascade forbidden)",
            );
        }
        assert!(
            !self.projection_targets.contains(source),
            "unique_index: source `{source}` is already a projection target (cascade forbidden)"
        );
    }

    /// Record that `source` claims into `namespace` under `tag`
    /// (`collection_tag(source)`). Panics when `tag` is already registered
    /// for that namespace under a **different** source name — an FNV-1a
    /// collision between two logically distinct sources sharing one
    /// namespace, which would make one source's owner rows indistinguishable
    /// from the other's (see [`crate::unique::UniqueOwner::belongs_to`]).
    /// Idempotent for the same `(namespace, source)` pair, so declaring the
    /// same source into the same namespace twice (e.g. two `unique_index`
    /// calls sharing a target) is not itself a collision.
    ///
    /// Runs eagerly, from [`Self::unique_index`] — not from [`Self::finish`],
    /// which only runs after every source is already open, by which point a
    /// colliding claim could already be live.
    fn register_owner_tag(&mut self, namespace: &'static str, source: &'static str, tag: u64) {
        match self.owner_tags.entry((namespace, tag)) {
            std::collections::hash_map::Entry::Occupied(e) => {
                assert_eq!(
                    *e.get(),
                    source,
                    "unique_index: namespace `{namespace}` owner tag collision — sources \
                     `{}` and `{source}` hash to the same collection tag",
                    e.get()
                );
            }
            std::collections::hash_map::Entry::Vacant(e) => {
                e.insert(source);
            }
        }
    }

    fn reject_stream_decl(&self, source: &'static str, targets: &QueueTargets) {
        assert!(
            !targets.is_empty(),
            "change_stream: QueueTargets must declare at least one target"
        );
        assert!(
            !self.stream_sources.contains(source),
            "change_stream: source `{source}` already declares a stream"
        );
        assert!(
            !self.queue_names.contains(source),
            "change_stream: queue `{source}` cannot be a stream source"
        );
        assert!(
            !self.projection_targets.contains(source),
            "change_stream: source `{source}` is already a projection target (cascade forbidden)"
        );
        for name in targets.names() {
            assert!(
                self.queue_names.contains(name),
                "change_stream: queue `{name}` is not registered"
            );
            assert!(
                !self.projection_sources.contains(name),
                "change_stream: queue `{name}` is already a projection source (cascade forbidden)"
            );
        }
    }

    /// Declare a secondary index: returns a [`Projected`] carrier for opening the source.
    ///
    /// Several sources may feed one or more index collections (one declaration per
    /// source): the validator's orphan check runs against the union of expected keys
    /// from all declared sources of that index.
    pub fn secondary_index<SV, P>(
        &mut self,
        targets: IndexTargets,
        project: P,
    ) -> Projected<SV::SelfId, SV, NoHook>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, &SV, &mut ProjectionSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    {
        self.reject_index_decl(SV::NAME, &targets);

        let target_decls: Vec<IndexTargetDecl> = targets
            .erased_targets()
            .iter()
            .map(|t| {
                let index = Arc::clone(&t.inner);
                IndexTargetDecl {
                    name: t.name,
                    index: Arc::clone(&t.inner),
                    value_eq: Box::new(move |a, b| index.values_equal(a, b)),
                    value_to_bytes: Arc::clone(&t.value_to_bytes),
                }
            })
            .collect();

        self.projection_sources.insert(SV::NAME);
        for t in &target_decls {
            self.projection_targets.insert(t.name);
        }

        let project_fn = Arc::new(project);
        let project_erased = Arc::clone(&project_fn);
        let sink_targets = targets.clone();

        self.decls.push(Decl::Index {
            source: SV::NAME,
            targets: target_decls,
            project: Box::new(move |k, v, limits, out| {
                let k = k.downcast_ref::<SV::SelfId>().ok_or_else(|| {
                    crate::error::DbError::Internal("index decl: key type mismatch")
                })?;
                let v = v.downcast_ref::<SV>().ok_or_else(|| {
                    crate::error::DbError::Internal("index decl: value type mismatch")
                })?;
                // Resolved off the open source collection, so this is the exact
                // budget its writes project under — not a reconstruction of it.
                let (max_rows, max_bytes) = limits.unwrap_or((
                    crate::projection::DEFAULT_MAX_PROJECTED_ROWS,
                    crate::projection::DEFAULT_MAX_PROJECTED_BYTES,
                ));
                let mut sink = ProjectionSink::new(&sink_targets, max_rows, max_bytes);
                project_erased(k, v, &mut sink)?;
                for row in sink.rows() {
                    let Some(idx) = sink_targets
                        .erased_targets()
                        .iter()
                        .position(|t| t.name == row.target)
                    else {
                        return Err(crate::error::DbError::UndeclaredIndexTarget {
                            index: row.target,
                        });
                    };
                    out.push((idx, row.key.clone(), row.value.clone()));
                }
                Ok(())
            }),
        });

        Projected::new(targets, move |k, v, sink| project_fn(k, v, sink))
    }

    /// Declare a transactional change stream from one source to queue targets.
    /// The declaration exists solely for schema graph introspection; stream
    /// events cannot be reconstructed from current queue state.
    pub fn change_stream<SV, P>(
        &mut self,
        targets: QueueTargets,
        project: P,
    ) -> Projected<SV::SelfId, SV, NoHook, Events<SV::SelfId, SV>>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, Option<&SV>, Option<&SV>, &mut EmitSink<'_>) -> DbResult<()>
            + Send
            + Sync
            + 'static,
    {
        self.change_stream_on(
            Projected::new(IndexTargets::new(), |_, _, _| Ok(())),
            targets,
            project,
        )
    }

    /// Add a stream half to an existing schema-declared projection carrier.
    pub fn change_stream_on<SV, H, P>(
        &mut self,
        base: Projected<SV::SelfId, SV, H>,
        targets: QueueTargets,
        project: P,
    ) -> Projected<SV::SelfId, SV, H, Events<SV::SelfId, SV>>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, Option<&SV>, Option<&SV>, &mut EmitSink<'_>) -> DbResult<()>
            + Send
            + Sync
            + 'static,
    {
        assert!(
            SV::TTL.is_none(),
            "change_stream: source `{}` cannot declare TTL",
            SV::NAME
        );
        self.reject_stream_decl(SV::NAME, &targets);
        self.stream_sources.insert(SV::NAME);
        self.projection_sources.insert(SV::NAME);
        for name in targets.names() {
            self.projection_targets.insert(name);
        }
        self.decls.push(Decl::Stream {
            source: SV::NAME,
            targets: targets.clone(),
        });
        base.and_events(targets, project)
    }

    /// Explicit FK: typed key-builder (covers unbranded fields and composite targets).
    pub fn fk<SV, TV, F>(&mut self, on_missing: OnMissing, label: FkLabel, build: F)
    where
        SV: CollectionMeta + 'static,
        SV::SelfId: Key + 'static,
        TV: CollectionMeta + 'static,
        TV::SelfId: Key + 'static,
        F: Fn(&SV::SelfId, &SV) -> Option<TV::SelfId> + Send + Sync + 'static,
    {
        self.decls.push(Decl::Fk {
            from: SV::NAME,
            to: TV::NAME,
            label: label.0,
            on_missing,
            build: Box::new(move |k, v| {
                let (Some(k), Some(v)) = (k.downcast_ref::<SV::SelfId>(), v.downcast_ref::<SV>())
                else {
                    return FkBuild::TypeMismatch;
                };
                match build(k, v) {
                    Some(tk) => FkBuild::Key(Box::new(tk)),
                    None => FkBuild::NoRef,
                }
            }),
        });
    }

    /// Denormalization: key projection + consistency predicate.
    pub fn denormalized<SV, DV, PF, EQ>(&mut self, project: PF, consistent: EQ)
    where
        SV: CollectionMeta + 'static,
        SV::SelfId: Key + 'static,
        DV: CollectionMeta + 'static,
        DV::SelfId: Key + 'static,
        PF: Fn(&SV::SelfId, &SV) -> DV::SelfId + Send + Sync + 'static,
        EQ: Fn(&SV, &DV) -> bool + Send + Sync + 'static,
    {
        self.decls.push(Decl::Denorm {
            source: SV::NAME,
            denorm: DV::NAME,
            project: Box::new(move |k, v| {
                let k = k.downcast_ref::<SV::SelfId>()?;
                let v = v.downcast_ref::<SV>()?;
                let dk = project(k, v);
                Some((dk.as_bytes().to_vec(), Box::new(dk) as Box<dyn Any>))
            }),
            consistent: Box::new(move |s, d| {
                Some(consistent(s.downcast_ref::<SV>()?, d.downcast_ref::<DV>()?))
            }),
        });
    }

    /// Declare an engine-maintained counter: returns a [`Projected`] carrier
    /// for opening the source, exactly as [`Self::secondary_index`] does for
    /// indexes — under the hood this is a no-op index projection with the
    /// counter projection layered on top via [`Projected::and_counters`].
    ///
    /// `targets` names the counter target(s) `project` may contribute to (via
    /// [`CounterSink::add`] / [`CounterSink::add_one_of`]) — see
    /// [`CounterTargets::with_counter`]. Several *different* `(target,
    /// field)` pairs may each be fed by their own source, but a single
    /// `(target, field)` may have only one declared source in a registry:
    /// [`SchemaRegistry::finish`] returns
    /// [`RegistryBuildError::DuplicateCounterSource`] otherwise, and a
    /// cascade (a counter target that is also a projecting source, or vice
    /// versa) panics at declaration time — see [`Self::reject_counter_decl`].
    pub fn counted<SV, P>(
        &mut self,
        targets: CounterTargets,
        project: P,
    ) -> Projected<SV::SelfId, SV, NoHook>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, &SV, &mut CounterSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    {
        self.counted_on(
            Projected::new(IndexTargets::new(), |_, _, _| Ok(())),
            targets,
            project,
        )
    }

    /// Same declaration as [`Self::counted`], attaching the counter half onto
    /// an existing carrier (`base`) instead of a fresh one — composes with
    /// whatever halves `base` already carries (e.g. a real index declared via
    /// [`Self::secondary_index`], or a claim half from
    /// [`Self::unique_index_on`]). [`Self::counted`] is implemented in terms
    /// of this method over a fresh carrier, so there is one code path for the
    /// counter half, not two that could drift apart — the same reasoning that
    /// makes [`Self::unique_index_on`] the single source of truth for the
    /// claim half.
    pub fn counted_on<SV, P>(
        &mut self,
        base: Projected<SV::SelfId, SV, NoHook>,
        targets: CounterTargets,
        project: P,
    ) -> Projected<SV::SelfId, SV, NoHook>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, &SV, &mut CounterSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    {
        self.reject_counter_decl(SV::NAME, &targets);

        for t in targets.all() {
            self.counter_declarations
                .push((t.collection_name(), t.field(), SV::NAME));
        }

        self.projection_sources.insert(SV::NAME);
        for t in targets.all() {
            self.projection_targets.insert(t.collection_name());
        }

        let project_fn = Arc::new(project);
        let project_erased = Arc::clone(&project_fn);

        self.decls.push(Decl::Counter {
            source: SV::NAME,
            targets: targets.clone(),
            project: Box::new(move |k, v, sink| {
                let k = k.downcast_ref::<SV::SelfId>().ok_or_else(|| {
                    crate::error::DbError::Internal("counter decl: key type mismatch")
                })?;
                let v = v.downcast_ref::<SV>().ok_or_else(|| {
                    crate::error::DbError::Internal("counter decl: value type mismatch")
                })?;
                project_erased(k, v, sink)
            }),
        });

        base.and_counters(targets, move |k, v, sink| project_fn(k, v, sink))
    }

    /// Declare a source's claims into one or more unique namespaces (T10 Task
    /// 3): returns a [`Projected`] carrier for opening the source, exactly as
    /// [`Self::counted`] does — under the hood this is a no-op index
    /// projection with the claim projection layered on top via
    /// [`Projected::and_unique`].
    ///
    /// Three checks run at declaration time, before the source is ever
    /// opened (see [`Self::register_owner_tag`] for why the last one cannot
    /// wait for [`Self::finish`]):
    /// 1. no cascades — a namespace may not be a projecting/counting source,
    ///    a source may not be its own namespace (see [`Self::reject_unique_decl`]);
    /// 2. the source's own key must fit a namespace owner row
    ///    (`UNIQUE_OWNER_KEY_CAP` bytes);
    /// 3. no owner-tag collision between this source and any other source
    ///    already declared into the same namespace.
    ///
    /// Uniqueness itself is enforced inside `db.tx()`, by `resolve_claims`
    /// (`txn::projection_commit`) — this method only records the declaration
    /// (registry composition, the topology checks above, and the data it
    /// hands to `schema::validate`/`schema::repair`).
    pub fn unique_index<SV, P>(
        &mut self,
        targets: UniqueTargets,
        project: P,
    ) -> Projected<SV::SelfId, SV, NoHook>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, &SV, &mut ClaimSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    {
        self.unique_index_on(
            Projected::new(IndexTargets::new(), |_, _, _| Ok(())),
            targets,
            project,
        )
    }

    /// Same declaration as [`Self::unique_index`], attaching the claim half
    /// onto an existing carrier (`base`) instead of a fresh one — composes
    /// with whatever halves `base` already carries (e.g. a real index from
    /// [`Self::secondary_index`], or a counter from [`Self::counted_on`]).
    /// [`Self::unique_index`] is implemented in terms of this method over a
    /// fresh carrier, so there is one code path for the claim half, not two:
    /// a source declaring both a registry-tracked counter *and* a
    /// registry-tracked claim on one collection previously had to attach one
    /// half through its registry method and the other raw (via
    /// [`Projected::and_counters`] / [`Projected::and_unique`] directly),
    /// which left the registry's `Decl` closure and the carrier's actual
    /// write-path closure free to diverge — see the review that added these
    /// `_on` forms for the concrete hazard.
    pub fn unique_index_on<SV, P>(
        &mut self,
        base: Projected<SV::SelfId, SV, NoHook>,
        targets: UniqueTargets,
        project: P,
    ) -> Projected<SV::SelfId, SV, NoHook>
    where
        SV: CollectionMeta + Send + Sync + 'static,
        SV::SelfId: Key + 'static,
        P: Fn(&SV::SelfId, &SV, &mut ClaimSink<'_>) -> DbResult<()> + Send + Sync + 'static,
    {
        self.reject_unique_decl(SV::NAME, &targets);

        const {
            assert!(
                size_of::<SV::SelfId>() <= UNIQUE_OWNER_KEY_CAP,
                "a claim source's key does not fit a unique namespace owner (56 bytes)"
            );
        }
        // T10 Task 5 review (F2): a claiming source must not declare a TTL.
        // A sweep runs outside any transaction — the same "outside db.tx()"
        // window Task 5's non-transactional guard exists to close for
        // ordinary writes — so an expiring row could vanish without
        // releasing its claim, permanently locking the name with no owner.
        // Same shape as T9's "a counter target with TTL is unsupported". A
        // runtime `assert!` (not `const { assert!(..) }`) to match every
        // other `reject_*_decl` check in this file — and so the rejection
        // can be pinned with `#[should_panic]` like the others, rather than
        // a compile-fail test this crate has no harness for.
        assert!(
            SV::TTL.is_none(),
            "unique_index: a claiming source must not declare a TTL — a sweep runs outside any \
             transaction, so an expiring row could not release its claim atomically"
        );

        for t in targets.all() {
            self.register_owner_tag(t.collection_name(), SV::NAME, collection_tag(SV::NAME));
        }

        self.projection_sources.insert(SV::NAME);
        for t in targets.all() {
            self.projection_targets.insert(t.collection_name());
        }

        let project_fn = Arc::new(project);
        let project_erased = Arc::clone(&project_fn);

        self.decls.push(Decl::Unique {
            source: SV::NAME,
            targets: targets.clone(),
            project: Box::new(move |k, v, sink| {
                let k = k.downcast_ref::<SV::SelfId>().ok_or_else(|| {
                    crate::error::DbError::Internal("unique decl: key type mismatch")
                })?;
                let v = v.downcast_ref::<SV>().ok_or_else(|| {
                    crate::error::DbError::Internal("unique decl: value type mismatch")
                })?;
                project_erased(k, v, sink)
            }),
        });

        base.and_unique(targets, move |k, v, sink| project_fn(k, v, sink))
    }
}

impl BuiltRegistry {
    pub(super) fn homes_for_shape(&self, shape: &[&'static str]) -> Option<&[Home]> {
        self.homes.get(&sorted_shape(shape)).map(Vec::as_slice)
    }

    /// Returns whether every producer of `name` was declared complete.
    ///
    /// The same bit controls destructive orphan repair for secondary indexes and
    /// persistent unique namespaces.
    pub fn is_complete_index(&self, name: &str) -> bool {
        self.complete_indices.contains(name)
    }

    pub fn graph(&self) -> SchemaGraph {
        let mut g = SchemaGraph::default();
        for e in &self.reg.entries {
            g.collections.push(CollectionNode {
                name: e.name.to_string(),
                kind: e.kind,
                ty: e.typ,
                key_ty: Some(e.key_typ),
                self_shape: e
                    .self_shape
                    .map(|s| s.iter().map(|b| (*b).to_string()).collect()),
                storage: Some(e.storage),
            });
        }
        for (name, kind) in &self.reg.external {
            g.collections.push(CollectionNode {
                name: name.to_string(),
                kind: *kind,
                ty: Typ::Custom("external", &[]),
                key_ty: None,
                self_shape: None,
                storage: None,
            });
        }
        for e in &self.reg.entries {
            for edge in &e.match_plan.edges {
                for (home_name, home_shape) in edge.home_names.iter().zip(edge.home_shapes.iter()) {
                    let home_shape_refs: Vec<&str> = home_shape.to_vec();
                    g.relations.push(RelationEdge {
                        from: e.name.to_string(),
                        to: home_name.to_string(),
                        kind: RelationKind::Fk {
                            field_path: super::plan::reference_path_for_graph(
                                &home_shape_refs,
                                &edge.slots,
                            ),
                            brand: home_shape.join("+"),
                            on_missing: OnMissing::Error,
                            validated: true,
                        },
                    });
                }
            }
        }
        for d in &self.reg.decls {
            match d {
                Decl::Index {
                    source, targets, ..
                } => {
                    for t in targets {
                        g.relations.push(RelationEdge {
                            from: source.to_string(),
                            to: t.name.to_string(),
                            kind: RelationKind::Index,
                        });
                    }
                }
                Decl::Fk {
                    from,
                    to,
                    label,
                    on_missing,
                    ..
                } => g.relations.push(RelationEdge {
                    from: from.to_string(),
                    to: to.to_string(),
                    kind: RelationKind::Fk {
                        field_path: vec![],
                        brand: label.clone(),
                        on_missing: *on_missing,
                        validated: true,
                    },
                }),
                Decl::Denorm { source, denorm, .. } => g.relations.push(RelationEdge {
                    from: source.to_string(),
                    to: denorm.to_string(),
                    kind: RelationKind::Denorm,
                }),
                Decl::Counter {
                    source, targets, ..
                } => {
                    for t in targets.all() {
                        g.relations.push(RelationEdge {
                            from: t.collection_name().to_string(),
                            to: source.to_string(),
                            kind: RelationKind::Counter {
                                field: t.field().to_string(),
                            },
                        });
                    }
                }
                Decl::Unique {
                    source, targets, ..
                } => {
                    for t in targets.all() {
                        g.relations.push(RelationEdge {
                            from: source.to_string(),
                            to: t.collection_name().to_string(),
                            kind: RelationKind::Unique,
                        });
                    }
                }
                Decl::Stream { source, targets } => {
                    for target in targets.names() {
                        g.relations.push(RelationEdge {
                            from: source.to_string(),
                            to: target.to_string(),
                            kind: RelationKind::Stream,
                        });
                    }
                }
            }
        }
        g
    }

    /// Source collection names with a declared unique-namespace claim (one
    /// per [`SchemaRegistry::unique_index`] / [`SchemaRegistry::unique_index_on`]
    /// call) — introspection/test surface; data-level findings
    /// (`UniqueMissing`/`UniqueConflict`/`UniqueOrphan`) are `schema::validate`'s
    /// job, driven by [`super::expected::build_unique`], not this accessor's.
    /// `#[doc(hidden)]`: not part of the documented surface of a published
    /// crate.
    ///
    /// Named `_sources`, not `_declarations`, because [`SchemaRegistry`]
    /// already has a private field `counter_declarations` holding a different
    /// shape (`(target, field, source)` triples) — reusing that name for a
    /// public accessor of a different shape would mean two things called
    /// "counter_declarations" in the same type.
    #[doc(hidden)]
    pub fn unique_decl_sources(&self) -> Vec<&'static str> {
        self.reg
            .decls
            .iter()
            .filter_map(|d| match d {
                Decl::Unique { source, .. } => Some(*source),
                _ => None,
            })
            .collect()
    }

    /// Source collection names with a declared counter (one per
    /// [`SchemaRegistry::counted`] / [`SchemaRegistry::counted_on`] call) —
    /// introspection/test surface, mirrors [`Self::unique_decl_sources`].
    /// `#[doc(hidden)]`: not part of the documented surface of a published
    /// crate.
    #[doc(hidden)]
    pub fn counted_decl_sources(&self) -> Vec<&'static str> {
        self.reg
            .decls
            .iter()
            .filter_map(|d| match d {
                Decl::Counter { source, .. } => Some(*source),
                _ => None,
            })
            .collect()
    }
}

fn find<'a>(entries: &'a [CollectionEntry], name: &str) -> Option<&'a CollectionEntry> {
    entries.iter().find(|e| e.name == name)
}

pub(crate) fn scan_target_actual<'a>(
    reg: &'a BuiltRegistry,
    cache: &'a mut HashMap<&'static str, HashMap<Vec<u8>, Vec<u8>>>,
    name: &'static str,
    encode: &ValueToBytesFn,
) -> &'a HashMap<Vec<u8>, Vec<u8>> {
    cache.entry(name).or_insert_with(|| {
        let Some(entry) = reg.reg.entries.iter().find(|e| e.name == name) else {
            return HashMap::new();
        };
        let mut map = HashMap::new();
        (entry.scan_any)(&mut |kb, _k, v| {
            if let Some(vb) = encode(v) {
                map.insert(kb.to_vec(), vb);
            }
        });
        map
    })
}

/// Resolve a namespace owner's opaque collection tag back to a source name,
/// for diagnostics — the validator's own twin of
/// `txn::projection_commit::render_owner` ([`crate::error::DbError::UniqueViolation`]'s
/// renderer), except this one can actually do the resolution (`render_owner`
/// cannot: the commit path has no registry in scope). Checks every declared
/// `Decl::Unique` source first, then falls back to *any* registered entry
/// whose name hashes to this tag — a source can be a registered
/// `CollectionEntry` (via `register`/`register_entity`) without (or no longer)
/// having a `Decl::Unique` of its own, e.g. a claim declaration that was
/// dropped from the registry after the fact; that source's rows would
/// otherwise render as an opaque tag even though its name is known.
fn resolve_unique_owner_name(reg: &BuiltRegistry, tag: u64) -> Option<String> {
    reg.reg
        .decls
        .iter()
        .find_map(|d| match d {
            Decl::Unique { source, .. } if crate::unique::collection_tag(source) == tag => {
                Some((*source).to_string())
            }
            _ => None,
        })
        .or_else(|| {
            reg.reg
                .entries
                .iter()
                .find(|e| crate::unique::collection_tag(e.name) == tag)
                .map(|e| e.name.to_string())
        })
}

/// Render a namespace owner as one string, for [`super::Finding::UniqueOrphan`].
fn describe_unique_owner(reg: &BuiltRegistry, owner: &crate::unique::UniqueOwner) -> String {
    let tag = owner.collection_id();
    let key_hex = super::validate::hex(owner.key_bytes());
    match resolve_unique_owner_name(reg, tag) {
        Some(name) => format!("{name} key={key_hex}"),
        None => format!("collection_tag={tag:#x} key={key_hex}"),
    }
}

/// Render a namespace owner as `(source-or-tag, hex key)`, for
/// [`super::Finding::UniqueConflict`]'s `owners` — the same shape
/// `expected::UniqueClaim`'s own `(source, source_key)` pair already uses, so
/// a conflict between a declared claimant and a foreign row's *actual* owner
/// (found via [`crate::unique::resolve::decide`], not a second declared
/// claimant) reads the same way a two-claimant conflict does.
pub(super) fn unique_owner_pair(
    reg: &BuiltRegistry,
    owner: &crate::unique::UniqueOwner,
) -> (String, String) {
    let tag = owner.collection_id();
    let name =
        resolve_unique_owner_name(reg, tag).unwrap_or_else(|| format!("collection_tag={tag:#x}"));
    (name, super::validate::hex(owner.key_bytes()))
}

pub(super) fn run_decls(reg: &BuiltRegistry, report: &mut super::validate::ValidationReport) {
    use super::expected::{self, BuildIssue};
    use super::validate::{Finding, hex};
    use std::collections::HashMap;

    let mut actual_by_target: HashMap<&'static str, HashMap<Vec<u8>, Vec<u8>>> = HashMap::new();

    // Structural checks first — they need no data, and a declaration naming an
    // unregistered collection is a usage bug, not index/counter drift.
    for d in &reg.reg.decls {
        if let Decl::Index {
            source, targets, ..
        } = d
        {
            if reg.reg.entries.iter().all(|e| e.name != *source) {
                report.findings.push(Finding::RegistryError {
                    message: format!("index decl: source `{source}` not registered"),
                });
            }
            for t in targets {
                if reg.reg.entries.iter().all(|e| e.name != t.name) {
                    report.findings.push(Finding::RegistryError {
                        message: format!("index decl: target `{}` not registered", t.name),
                    });
                }
            }
        }
        // An unregistered counter source is not just a missing diagnostic —
        // without a `CollectionEntry` `build_counters` silently skips the
        // whole declaration, so its targets would otherwise look like they
        // expect zero. Reported here; the per-declaration loop below also
        // skips comparing/repairing that declaration's targets so a "not
        // registered" source can never be read as "confirmed empty".
        if let Decl::Counter { source, .. } = d
            && reg.reg.entries.iter().all(|e| e.name != *source)
        {
            report.findings.push(Finding::RegistryError {
                message: format!("counter decl: source `{source}` not registered"),
            });
        }
        // Same reasoning as the counter check above: an unregistered claim
        // source makes `expected::build_unique` silently skip the whole
        // declaration, which must not read as "confirmed no claims".
        if let Decl::Unique {
            source, targets, ..
        } = d
        {
            if reg.reg.entries.iter().all(|e| e.name != *source) {
                report.findings.push(Finding::RegistryError {
                    message: format!("unique decl: source `{source}` not registered"),
                });
            }
            // T10 review F7: the index pass above checks that each source and
            // each *index* target is registered; nothing checked that a claim
            // declaration's *namespace* is registered too. A namespace has no
            // handle of its own (`UniqueIndex` is not a `SchemaCollection`), so
            // its only registration path is `reg.register_external(namespace,
            // CollectionKind::UniqueIndex)` — forgetting that call, or
            // registering it under the wrong kind, previously reported a clean
            // registry while `BuiltRegistry::graph()` still emitted a `Unique`
            // edge whose target node is missing or misclassified.
            for t in targets.all() {
                let namespace = t.collection_name();
                let kind = reg
                    .reg
                    .entries
                    .iter()
                    .find(|e| e.name == namespace)
                    .map(|e| e.kind)
                    .or_else(|| {
                        reg.reg
                            .external
                            .iter()
                            .find(|(name, _)| *name == namespace)
                            .map(|(_, kind)| *kind)
                    });
                match kind {
                    None => report.findings.push(Finding::RegistryError {
                        message: format!("unique decl: namespace `{namespace}` not registered"),
                    }),
                    Some(k) if k != CollectionKind::UniqueIndex => {
                        report.findings.push(Finding::RegistryError {
                            message: format!(
                                "unique decl: namespace `{namespace}` registered as {k:?}, \
                                 expected CollectionKind::UniqueIndex"
                            ),
                        });
                    }
                    Some(_) => {}
                }
            }
        }
    }

    // The expected set, shared with `repair` so the two cannot disagree about
    // what an index should hold. The validator collects every issue rather than
    // stopping, so the callback always returns `Ok` and `build` cannot fail.
    let (expected, _skipped) = expected::build(reg, &mut |issue| {
        match issue {
            BuildIssue::Collision { index, key, owners } => {
                report.findings.push(Finding::IndexKeyCollision {
                    index: index.to_string(),
                    key: hex(&key),
                    owners: owners.to_vec(),
                });
            }
            BuildIssue::Projection { source, error } => {
                report.findings.push(Finding::RegistryError {
                    message: format!("index decl `{source}`: projection failed: {error}"),
                });
            }
        }
        Ok(())
    })
    .unwrap_or_default();

    let all_targets = expected::declared_targets(reg);

    // Compare per expected *key*, not per emitted row: a projection may emit one
    // key twice, and comparing the first value against a target that correctly
    // holds the second reports a mismatch on an index `repair` considers clean.
    for target_name in &all_targets {
        let Some(rows) = expected.rows(target_name) else {
            continue;
        };
        let Some(t) = expected::target_decl(reg, target_name) else {
            continue;
        };
        let actual = scan_target_actual(reg, &mut actual_by_target, t.name, &t.value_to_bytes);
        let mut found: Vec<(String, Finding)> = Vec::new();
        for (key, (value, owner)) in rows {
            let key_hex = hex(key);
            match actual.get(key) {
                None => found.push((
                    key_hex.clone(),
                    Finding::IndexMissing {
                        source: owner.0.clone(),
                        index: t.name.to_string(),
                        key: key_hex,
                    },
                )),
                Some(actual_bytes) => match (t.value_eq)(value, actual_bytes) {
                    Ok(false) => found.push((
                        key_hex.clone(),
                        Finding::IndexValueMismatch {
                            index: t.name.to_string(),
                            key: key_hex,
                        },
                    )),
                    Ok(true) => {}
                    Err(e) => found.push((
                        key_hex,
                        Finding::RegistryError {
                            message: format!("index decl `{}`: value compare failed: {e}", t.name),
                        },
                    )),
                },
            }
        }
        // The expected set is a hash map; sort so a report diffs cleanly.
        found.sort_by(|a, b| a.0.cmp(&b.0));
        report.findings.extend(found.into_iter().map(|(_, f)| f));
    }

    // Orphans: index rows no declared source claims.
    for target_name in &all_targets {
        let Some(tdecl) = expected::target_decl(reg, target_name) else {
            continue;
        };
        let actual = scan_target_actual(
            reg,
            &mut actual_by_target,
            target_name,
            &tdecl.value_to_bytes,
        );
        let complete = reg.complete_indices.contains(target_name);
        let mut orphans: Vec<String> = actual
            .keys()
            .filter(|kb| !expected.contains(target_name, kb))
            .map(|kb| hex(kb))
            .collect();
        orphans.sort();
        for key in orphans {
            report.findings.push(Finding::IndexOrphan {
                index: target_name.to_string(),
                key,
                complete,
            });
        }
    }

    // Counter half of the shared expected-value pass (see
    // `expected::build_counters`'s rustdoc for why a hard failure here — a
    // projection error or sum overflow — is fatal to the counter section). A
    // build failure is reported once and the whole counter section is
    // skipped below (`counter_build_failed`) rather than compared against an
    // empty map, which would otherwise report every stored counter row as a
    // spurious `CounterMismatch { expected: 0 }` on top of the real error.
    let (counter_expected, counter_build_failed) = match expected::build_counters(reg) {
        Ok(m) => (m, false),
        Err(e) => {
            report.findings.push(Finding::RegistryError {
                message: format!("counter decl: expected-value build failed: {e}"),
            });
            (HashMap::new(), true)
        }
    };

    // Unique-namespace half (T10 Task 9): the expected set is shared across
    // every declaration feeding one namespace (see `expected::build_unique`'s
    // doc), so — like the index/counter passes above — this runs once per
    // namespace, not once per `Decl::Unique`; the per-decl loop below leaves
    // `Decl::Unique` empty, mirroring `Decl::Index { .. } => {}`.
    //
    // T10 review F6: mirrors `counter_build_failed` above. A build failure
    // (a projection closure error or a budget overrun) is reported once here;
    // the missing/conflict/orphan comparison pass below must then be *skipped*,
    // not run against the substituted empty map — comparing against "expects
    // nothing" would additionally report every stored permanent claim as a
    // phantom `UniqueOrphan` on top of the real `RegistryError`.
    let (unique_expected, unique_build_failed) = match expected::build_unique(reg) {
        Ok((x, _skipped)) => (x, false),
        Err(e) => {
            report.findings.push(Finding::RegistryError {
                message: format!("unique decl: expected-claim build failed: {e}"),
            });
            (HashMap::new(), true)
        }
    };

    // Scan every declared namespace once, reporting `UniqueRowCorrupt`
    // directly and keeping every well-formed row for the per-key pass below.
    // A corrupt key is excluded from that pass entirely (tracked in
    // `corrupt_by_ns`) — already reported, and `UniqueRow::owner()`/
    // `key_bytes()` cannot be trusted on one (T10 Task 9 review, C1: this is
    // exactly the panic `is_well_formed` exists to prevent, so a corrupt row
    // must never reach `decide` or `describe_unique_owner` below).
    //
    // `corrupt_findings` is collected across *every* namespace before it is
    // sorted once, by `(namespace, key)` — not sorted per-namespace inside
    // the loop below and extended as each namespace is visited. `unique_handles`
    // is a `HashMap`, so its iteration order (i.e. namespace order) is
    // nondeterministic; sorting only within one namespace at a time (as an
    // earlier version of this pass did — T10 Task 9 review, M1, carried to
    // Task 10) left cross-namespace order exactly as nondeterministic as the
    // map itself, even though every finding was individually sorted. The
    // missing/conflict/orphan pass just below was fixed for exactly this
    // reason; this is the other half of that same fix.
    let unique_handles = expected::declared_unique_handles(reg);
    let mut stored_by_ns: HashMap<&'static str, HashMap<Vec<u8>, crate::UniqueRow>> =
        HashMap::new();
    let mut corrupt_by_ns: HashMap<&'static str, std::collections::HashSet<Vec<u8>>> =
        HashMap::new();
    let mut corrupt_findings: Vec<((&'static str, String), Finding)> = Vec::new();
    // Same nondeterminism-across-a-`HashMap`-iteration hazard as
    // `corrupt_findings` above (review round M2, the sibling of the
    // `UniqueRowCorrupt` fix): a scan failure is collected here and sorted by
    // namespace once, rather than pushed straight into `report.findings`
    // from inside the loop, so two namespaces failing to scan in one run
    // report in a stable order.
    let mut scan_error_findings: Vec<(&'static str, Finding)> = Vec::new();
    for (ns_name, handle) in &unique_handles {
        let stored = stored_by_ns.entry(ns_name).or_default();
        let corrupt = corrupt_by_ns.entry(ns_name).or_default();
        let scan_result = handle.scan_rows(&mut |key_bytes, row| {
            if !row.is_well_formed() {
                corrupt.insert(key_bytes.to_vec());
                corrupt_findings.push((
                    (*ns_name, hex(key_bytes)),
                    Finding::UniqueRowCorrupt {
                        namespace: ns_name.to_string(),
                        key: hex(key_bytes),
                    },
                ));
                return Ok(());
            }
            stored.insert(key_bytes.to_vec(), row);
            Ok(())
        });
        if let Err(e) = scan_result {
            scan_error_findings.push((
                ns_name,
                Finding::RegistryError {
                    message: format!("unique decl `{ns_name}`: namespace scan failed: {e}"),
                },
            ));
        }
    }
    corrupt_findings.sort_by(|a, b| a.0.cmp(&b.0));
    report
        .findings
        .extend(corrupt_findings.into_iter().map(|(_, f)| f));
    scan_error_findings.sort_by_key(|(ns, _)| *ns);
    report
        .findings
        .extend(scan_error_findings.into_iter().map(|(_, f)| f));

    // Missing/conflict/orphan: one pass over the union of every expected
    // `(namespace, key)` and every physically stored one (minus the corrupt
    // keys already reported above). A key with exactly one declared claimant
    // is decided through `unique::resolve::decide` against the *stored* row —
    // the same pure rule the write path uses (binding constraint: `decide` is
    // the shared rule) — rather than merely checking presence: presence alone
    // cannot tell "nobody claims this key" apart from "someone else's claim
    // sits on it" (T10 Task 9 review, C1). Two or more claimants for one key
    // is `UniqueConflict` regardless of what is stored (see
    // `expected::UniqueExpected`'s doc); zero claimants against a permanent
    // stored row is the "plain" `UniqueOrphan`.
    // T10 review F6: skip the comparison entirely on a build failure — a
    // `RegistryError` was already pushed above, and comparing every stored
    // permanent claim against the substituted empty map would additionally
    // report each one as a phantom `UniqueOrphan` (mirrors `counter_build_failed`
    // just above, and the same `continue` `Decl::Counter` uses in the per-decl
    // loop below).
    if !unique_build_failed {
        let mut all_pairs: std::collections::HashSet<(&'static str, Vec<u8>)> =
            unique_expected.keys().cloned().collect();
        for (ns, rows) in &stored_by_ns {
            for key in rows.keys() {
                all_pairs.insert((*ns, key.clone()));
            }
        }
        for (ns, keys) in &corrupt_by_ns {
            for key in keys {
                all_pairs.remove(&(*ns, key.clone()));
            }
        }

        let mut unique_mc: Vec<((&'static str, String), Finding)> = Vec::new();
        for (ns, key) in all_pairs {
            let key_hex = hex(&key);
            let claims = unique_expected.get(&(ns, key.clone()));
            let row = stored_by_ns.get(ns).and_then(|m| m.get(&key)).copied();

            match claims {
                None => {
                    // No declared claimant at all: a permanent row is an orphan.
                    // A reservation with no source row is its normal state.
                    if let Some(row) = row
                        && row.deadline().is_none()
                    {
                        unique_mc.push((
                            (ns, key_hex.clone()),
                            Finding::UniqueOrphan {
                                namespace: ns.to_string(),
                                key: key_hex,
                                owner: describe_unique_owner(reg, &row.owner()),
                                complete: reg.complete_indices.contains(ns),
                            },
                        ));
                    }
                }
                Some(claims) if claims.len() == 1 => {
                    let claim = &claims[0];
                    let kind = crate::txn::participant::ClaimKind::Claim;
                    match crate::unique::resolve::decide(row, &claim.owner, kind) {
                        crate::unique::resolve::ClaimDecision::NoOp => {}
                        crate::unique::resolve::ClaimDecision::Write(_) => {
                            unique_mc.push((
                                (ns, key_hex.clone()),
                                Finding::UniqueMissing {
                                    namespace: ns.to_string(),
                                    source: claim.source.to_string(),
                                    key: key_hex,
                                },
                            ));
                        }
                        crate::unique::resolve::ClaimDecision::Conflict(cur) => {
                            let (cur_source, cur_key) = unique_owner_pair(reg, &cur);
                            unique_mc.push((
                                (ns, key_hex.clone()),
                                Finding::UniqueConflict {
                                    namespace: ns.to_string(),
                                    key: key_hex,
                                    owners: vec![
                                        (claim.source.to_string(), claim.source_key.clone()),
                                        (cur_source, cur_key),
                                    ],
                                },
                            ));
                        }
                        crate::unique::resolve::ClaimDecision::Delete => {
                            // `ClaimKind::Claim` never decides `Delete` — only
                            // `Release` does (see `decide`'s own match arms).
                            // Reported, not panicked: a validator must not crash
                            // on a state that should be unreachable by
                            // construction.
                            report.findings.push(Finding::RegistryError {
                                message: format!(
                                    "unique decl `{ns}`: decide(Claim) returned Delete for key {key_hex}"
                                ),
                            });
                        }
                    }
                }
                Some(claims) => {
                    let mut owners: Vec<(String, String)> = claims
                        .iter()
                        .map(|c| (c.source.to_string(), c.source_key.clone()))
                        .collect();
                    owners.sort();
                    unique_mc.push((
                        (ns, key_hex.clone()),
                        Finding::UniqueConflict {
                            namespace: ns.to_string(),
                            key: key_hex,
                            owners,
                        },
                    ));
                }
            }
        }
        // Sorted by (namespace, key): this pass aggregates every declared
        // namespace into one vec (unlike the index/counter passes above, which
        // sort per-target), and `all_pairs` iterates a `HashSet`.
        unique_mc.sort_by(|a, b| a.0.cmp(&b.0));
        report
            .findings
            .extend(unique_mc.into_iter().map(|(_, f)| f));
    }

    for d in &reg.reg.decls {
        match d {
            Decl::Index { .. } => {}
            Decl::Fk {
                from,
                to,
                label,
                on_missing,
                build,
            } => {
                let (Some(src), Some(target)) =
                    (find(&reg.reg.entries, from), find(&reg.reg.entries, to))
                else {
                    report.findings.push(Finding::RegistryError {
                        message: format!("fk decl `{from}`->`{to}`: collection not registered"),
                    });
                    continue;
                };
                (src.scan_any)(&mut |kb, k, v| match build(k, v) {
                    FkBuild::NoRef => {}
                    FkBuild::TypeMismatch => report.findings.push(Finding::RegistryError {
                        message: format!("fk decl `{from}`->`{to}` ({label}): type mismatch"),
                    }),
                    FkBuild::Key(tk) => match (target.contains_any)(&*tk) {
                        Some(true) => {}
                        Some(false) => {
                            if *on_missing != OnMissing::AllowMissing {
                                report.findings.push(Finding::DanglingFk {
                                    collection: from.to_string(),
                                    key: hex(kb),
                                    part: RowPart::Value,
                                    path: Vec::new(),
                                    reference: RefDesc::Explicit {
                                        label: label.clone(),
                                    },
                                    homes: vec![to.to_string()],
                                    on_missing: *on_missing,
                                });
                            }
                        }
                        None => report.findings.push(Finding::RegistryError {
                            message: format!("fk decl `{to}`: key type mismatch"),
                        }),
                    },
                });
            }
            Decl::Denorm {
                source,
                denorm,
                project,
                consistent,
            } => {
                let (Some(src), Some(dn)) = (
                    find(&reg.reg.entries, source),
                    find(&reg.reg.entries, denorm),
                ) else {
                    report.findings.push(Finding::RegistryError {
                        message: format!(
                            "denorm decl `{source}`->`{denorm}`: collection not registered"
                        ),
                    });
                    continue;
                };
                (src.scan_any)(&mut |_kb, k, v| {
                    let Some((dk_bytes, dk)) = project(k, v) else {
                        report.findings.push(Finding::RegistryError {
                            message: format!(
                                "denorm decl `{source}`->`{denorm}`: type mismatch in projection"
                            ),
                        });
                        return;
                    };
                    match (dn.with_value_any)(&*dk, &mut |dv| {
                        if consistent(v, dv) == Some(false) {
                            report.findings.push(Finding::DenormMismatch {
                                source: source.to_string(),
                                denorm: denorm.to_string(),
                                key: hex(&dk_bytes),
                            });
                        }
                    }) {
                        Some(true) => {}
                        Some(false) => report.findings.push(Finding::DenormMissing {
                            source: source.to_string(),
                            denorm: denorm.to_string(),
                            key: hex(&dk_bytes),
                        }),
                        None => report.findings.push(Finding::RegistryError {
                            message: format!("denorm decl `{denorm}`: key type mismatch"),
                        }),
                    }
                });
            }
            Decl::Counter {
                source, targets, ..
            } => {
                // A build failure already produced one `RegistryError` above;
                // comparing against the fallback empty map here would only
                // add spurious noise on top of it.
                if counter_build_failed {
                    continue;
                }
                // Already reported (structural check above); an
                // unregistered source's targets must not be compared against
                // "no expected data" as if that meant "confirmed zero".
                if reg.reg.entries.iter().all(|e| e.name != *source) {
                    continue;
                }
                for t in targets.all() {
                    let drift = match expected::counter_drift(t, &counter_expected) {
                        Ok(d) => d,
                        Err(e) => {
                            report.findings.push(Finding::RegistryError {
                                message: format!(
                                    "counter decl `{}`.`{}`: reading stored rows failed: {e}",
                                    t.collection_name(),
                                    t.field()
                                ),
                            });
                            continue;
                        }
                    };
                    // The expected/existing maps are hashed; sort so a report
                    // diffs cleanly, matching the index checks above.
                    let mut found: Vec<(String, Finding)> = drift
                        .into_iter()
                        .map(|d| {
                            let key_hex = hex(&d.key);
                            (
                                key_hex.clone(),
                                Finding::CounterMismatch {
                                    collection: t.collection_name().to_string(),
                                    key: key_hex,
                                    field: t.field().to_string(),
                                    stored: d.stored,
                                    expected: d.want,
                                },
                            )
                        })
                        .collect();
                    found.sort_by(|a, b| a.0.cmp(&b.0));
                    report.findings.extend(found.into_iter().map(|(_, f)| f));
                }
            }
            // Data-level validation (`UniqueMissing`/`UniqueConflict`/
            // `UniqueOrphan`) already ran, above this per-decl loop (the
            // `unique_mc`/`all_pairs` pass over `unique_expected` and
            // `stored_by_ns`) — a namespace is fed by the union of every
            // source that declares into it, not by one `Decl::Unique` at a
            // time, so it cannot be driven from this loop the way `Counter`/
            // `Denorm`/`Fk` are. Nothing left to do here.
            Decl::Unique { .. } => {}
            Decl::Stream { .. } => {}
        }
    }
}

#[cfg(test)]
mod owner_tag_tests {
    //! T10 Task 3 review fix F4: `register_owner_tag` is a private method
    //! with no direct integration-test access and no realistic way to
    //! manufacture an actual FNV-1a collision between two collection names —
    //! a plain `mod tests` *inside* `registry.rs` is a descendant module, so
    //! it can call the private method directly with two synthetic names and
    //! a hand-picked tag (mirrors `counter.rs`'s `pub(crate) mod tests` and
    //! its `#[should_panic]` unit tests for declaration-time asserts).
    //!
    //! The critical test is `rejects_a_collision_between_two_sources`: an
    //! **empty method body** would pass every other test in this crate
    //! (including the integration test proving the *same* source can declare
    //! into the *same* namespace twice), and would only fail here — where a
    //! genuine collision must panic and an empty body would not.
    use super::SchemaRegistry;

    #[test]
    fn allows_the_same_source_registered_twice() {
        let mut reg = SchemaRegistry::new();
        reg.register_owner_tag("ns", "src_a", 42);
        // Same (namespace, tag) from the same source again — not a collision.
        reg.register_owner_tag("ns", "src_a", 42);
    }

    #[test]
    fn allows_the_same_tag_in_different_namespaces() {
        let mut reg = SchemaRegistry::new();
        reg.register_owner_tag("ns_a", "src_a", 42);
        // Same tag value, different namespace — independent keys.
        reg.register_owner_tag("ns_b", "src_b", 42);
    }

    #[test]
    #[should_panic(expected = "unique_index: namespace `ns` owner tag collision")]
    fn rejects_a_collision_between_two_sources() {
        let mut reg = SchemaRegistry::new();
        reg.register_owner_tag("ns", "src_a", 42);
        // Different source, same tag, same namespace — a genuine collision
        // that would make one source's owner rows readable as the other's.
        reg.register_owner_tag("ns", "src_b", 42);
    }
}