faculties 0.21.0

//! Canonical root descriptors for faculty collections.
//!
//! A root collection used to be anchored by an opaque minted scope id. It
//! discriminated roots correctly and told a reader nothing: the id lived as a
//! hex constant in one faculty's source, so "which collection is this?" was
//! answerable only by someone holding the code. A root is now a self-describing
//! fragment containing its name, representation, and immutable READ and WRITE
//! admission policies. The fragment's content handle is the collection
//! identity.
//!
//! The scope ids have not gone anywhere — they remain each schema's stable
//! identifier and the key this table is read by, because the migration that
//! re-seats existing data has to speak both languages at once.

use std::ffi::OsString;

use anybytes::View;
use anyhow::{anyhow, bail, Context};
use ed25519_dalek::VerifyingKey;

use triblespace::core::blob::encodings::simplearchive::SimpleArchive;
use triblespace::core::blob::encodings::utf8string::UTF8String;
use triblespace::core::blob::{Blob, TryFromBlob};
use triblespace::core::collection::{
    descriptor, generation, records::CollectionHandle, AdmissionPolicy, Collection,
    CollectionPolicy, CollectionRead, CollectionRecordSelector, CollectionRegistrationError,
    CollectionStoreExt,
};
use triblespace::core::id::Id;
use triblespace::core::inline::Inline;
use triblespace::core::repo::{
    BlobStoreGet, BlobStoreList, BlobStorePut, CapabilityProofRead, SnapshotSource, StoreSnapshot,
};
use triblespace::core::trible::TribleSet;

use crate::schemas::{
    atlas, blockdag, body, code, cognition, compass, config, decide, discord, embeddings, files,
    habit, headspace, mail, memory, message, orient, planner, posture, relations, status,
    swarm_health, teams, voice, web, wiki,
};
use crate::secrets::DEFAULT_SCOPE_ID as SECRETS_SCOPE_ID;

/// Every root collection this build writes: the scope that used to anchor it,
/// and the name it is known by.
///
/// A faculty that is missing here cannot be opened at all, which is the point:
/// a nameless collection is one the pile cannot describe, and shipping one
/// silently is how the old scope model stayed opaque for so long. Every
/// collection in this table deliberately uses a direct READ and WRITE policy
/// rooted at the pile's durable signer. Sharing a collection later means
/// creating or migrating to a descriptor whose policy says so; it is not an
/// ambient property hidden in this table.
///
/// The ones worth naming individually:
///
/// - `memory` and `memory-comb` are the journal, first-person and personal.
/// - `compass` and `wiki` are the two JP has floated sharing. Neither becomes
///   public here: his own design for compass is a collection that shares its
///   goals but not the personal notes attached to them, which is *two*
///   collections, not one made public. A public sibling is the shape, with its
///   own explicit admission policy.
pub fn table() -> Vec<(Id, &'static str)> {
    vec![
        (atlas::DEFAULT_SCOPE_ID, "atlas"),
        (blockdag::DEFAULT_SCOPE_ID, "blockdag"),
        (body::DEFAULT_SCOPE_ID, "body"),
        (code::DEFAULT_SCOPE_ID, "code"),
        (cognition::DEFAULT_SCOPE_ID, "cognition"),
        (compass::DEFAULT_SCOPE_ID, "compass"),
        (config::DEFAULT_SCOPE_ID, config::COLLECTION_NAME),
        (decide::DEFAULT_SCOPE_ID, "decide"),
        (discord::DEFAULT_SCOPE_ID, "discord"),
        (embeddings::DEFAULT_SCOPE_ID, "embeddings"),
        (files::DEFAULT_SCOPE_ID, "files"),
        (habit::DEFAULT_SCOPE_ID, "habit"),
        (headspace::DEFAULT_SCOPE_ID, "headspace"),
        (mail::DEFAULT_SCOPE_ID, "mail"),
        (memory::DEFAULT_SCOPE_ID, "memory-journal"),
        (memory::DEFAULT_COMB_SCOPE_ID, "memory-comb"),
        (message::DEFAULT_SCOPE_ID, "message"),
        (orient::DEFAULT_SCOPE_ID, "orient"),
        (planner::DEFAULT_SCOPE_ID, "planner"),
        (posture::DEFAULT_POLICY_SCOPE_ID, "posture-policy"),
        (posture::DEFAULT_SCAN_SCOPE_ID, "posture-scan"),
        (relations::DEFAULT_SCOPE_ID, "relations"),
        (SECRETS_SCOPE_ID, "secrets"),
        (status::DEFAULT_SCOPE_ID, "status"),
        (
            swarm_health::DEFAULT_SCOPE_ID,
            swarm_health::COLLECTION_NAME,
        ),
        (teams::DEFAULT_SCOPE_ID, "teams"),
        (voice::COLLECTION_SCOPE_ID, "voice"),
        (web::DEFAULT_SCOPE_ID, "web"),
        (wiki::DEFAULT_SCOPE_ID, "wiki"),
    ]
}

/// The name for one scope, or `None` if this build does not know it.
pub fn name_for(scope: Id) -> Option<&'static str> {
    table()
        .into_iter()
        .find(|(candidate, _)| *candidate == scope)
        .map(|(_, name)| name)
}

/// The name for one scope, or a panic naming the scope that is missing.
///
/// Every collection this build opens is one it wrote the table entry for, so an
/// absence is a bug in this crate rather than anything a pile can cause. It is
/// loud because the alternative — inventing a name — would root real data at a
/// collection nothing else can find.
pub fn require_name(scope: Id) -> &'static str {
    name_for(scope).unwrap_or_else(|| {
        panic!(
            "no collection name for scope {scope:X}; add it to \
             faculties::collection_names::table"
        )
    })
}

/// The private policy deliberately shared by every current faculty root.
pub fn private_policy(authority: VerifyingKey) -> CollectionPolicy {
    CollectionPolicy::new(
        AdmissionPolicy::direct(authority),
        AdmissionPolicy::direct(authority),
    )
}

/// Prefix for exact descriptor overrides understood by every faculty.
///
/// The suffix is the canonical collection name, uppercased with `-` replaced
/// by `_`: `wiki` is `TRIBLESPACE_COLLECTION_WIKI`, while `memory-journal` is
/// `TRIBLESPACE_COLLECTION_MEMORY_JOURNAL`. Keeping one variable per name lets
/// a process which reads several faculty collections select each one
/// independently instead of applying one ambient collection identity to all
/// of them.
pub const COLLECTION_OVERRIDE_PREFIX: &str = "TRIBLESPACE_COLLECTION_";

/// Deterministic environment-variable name for one faculty collection.
pub fn override_env_name(scope: Id) -> String {
    let name = require_name(scope);
    let mut variable = String::with_capacity(COLLECTION_OVERRIDE_PREFIX.len() + name.len());
    variable.push_str(COLLECTION_OVERRIDE_PREFIX);
    variable.extend(name.bytes().map(|byte| match byte {
        b'a'..=b'z' => char::from(byte - b'a' + b'A'),
        b'A'..=b'Z' | b'0'..=b'9' => char::from(byte),
        b'-' => '_',
        _ => panic!("collection name {name:?} cannot form an environment variable"),
    }));
    variable
}

fn parse_override(variable: &str, raw: OsString) -> anyhow::Result<CollectionHandle> {
    let raw = raw
        .into_string()
        .map_err(|_| anyhow!("{variable} is not valid UTF-8"))?;
    let raw = raw.trim();
    let raw = raw.strip_prefix("blake3:").unwrap_or(raw);
    if raw.len() != 64 {
        bail!("{variable} must be one exact 64-digit hexadecimal collection descriptor handle");
    }
    let mut bytes = [0_u8; 32];
    hex::decode_to_slice(raw, &mut bytes)
        .with_context(|| format!("{variable} is not a hexadecimal collection descriptor handle"))?;
    Ok(Inline::new(bytes))
}

/// Exact descriptor selected for `scope` by the environment, if it supplies one.
///
/// Invalid values fail loudly. Falling back to a signer-private descriptor in
/// that case would silently fork a shared collection into a different identity.
///
/// This is the ENVIRONMENT half only. The other half -- what this pile's own
/// `config` collection resolves the name to -- needs a snapshot, and so lives
/// in [`open_configured`], which already has one. Keeping this function
/// storage-free is what lets callers that hold no pile still ask the question.
pub fn configured_handle(scope: Id) -> anyhow::Result<Option<CollectionHandle>> {
    let variable = override_env_name(scope);
    match std::env::var_os(&variable) {
        Some(raw) => Ok(Some(parse_override(&variable, raw)?)),
        None => Ok(None),
    }
}

/// Open the operator-selected exact descriptor, or construct the ordinary
/// signer-private faculty descriptor when no override is present.
///
/// The override path is non-registering: its canonical descriptor must already
/// be resident and carry the name assigned to this faculty scope. Local
/// publication is intentionally unconditional; WRITE admission decides which
/// commits enter an admitted snapshot, and later evidence may activate an
/// earlier offline commit.
pub fn open_configured<S>(
    storage: &mut S,
    scope: Id,
    authority: VerifyingKey,
) -> anyhow::Result<Collection<SimpleArchive>>
where
    S: CollectionStoreExt + SnapshotSource,
    <S as SnapshotSource>::Snapshot: BlobStoreGet + CollectionRead,
{
    let Some(handle) = configured_handle(scope)? else {
        let collection =
            open(storage, scope, authority).context("register signer-private descriptor")?;
        // A host with no configured handle must not quietly start a new
        // generation beside ones the other hosts already write to. The private
        // descriptor is only content until something commits to it, so
        // registering it costs nothing; using it here would.
        //
        // A private generation that already holds commits is not being
        // started: it is this host's own collection, reopened. Siblings with
        // content of their own do not change that -- `cat` of another host's
        // pile brings in exactly such siblings, and the merge must not stop
        // this host from opening what it already wrote.
        let snapshot = storage
            .snapshot()
            .context("freeze store to look for other generations of this name")?;
        if let Some(report) = generation::named_generations(&snapshot, collection.handle())
            .map_err(|error| anyhow!("look for other generations: {error}"))?
        {
            if report.selected().commits() == 0 && report.strands_records() {
                let variable = override_env_name(scope);
                let siblings: Vec<String> = report
                    .siblings()
                    .iter()
                    .filter(|sibling| sibling.commits() > 0)
                    .map(|sibling| {
                        format!(
                            "blake3:{} ({} commit(s))",
                            hex::encode(sibling.handle().raw),
                            sibling.commits()
                        )
                    })
                    .collect();
                bail!(
                    "{} is not configured on this host and this pile already holds {} \
                     generation(s) of {:?} with content: {}. Set {variable} to the one \
                     this host should use instead of starting another.",
                    variable,
                    siblings.len(),
                    require_name(scope),
                    siblings.join(", ")
                );
            }
        }
        return Ok(collection);
    };

    let snapshot = storage
        .snapshot()
        .context("freeze store while opening configured collection descriptor")?;
    let collection = open_exact_in(&snapshot, scope, handle)?;
    if let Some(warning) = empty_beside_content(&snapshot, scope, handle) {
        eprintln!("warning: {warning}");
    }
    Ok(collection)
}

/// Open a configured descriptor at a synchronous foreground I/O boundary.
/// Only exact descriptor/name bytes may be acquired; no records are selected
/// again after acquisition. Call outside `Runtime::block_on`.
pub fn open_configured_acquiring<S>(
    storage: &mut S,
    scope: Id,
    authority: VerifyingKey,
    runtime: &std::sync::Arc<tokio::runtime::Runtime>,
) -> anyhow::Result<Collection<SimpleArchive>>
where
    S: CollectionStoreExt + SnapshotSource,
    S::Snapshot:
        BlobStoreGet + CollectionRead + triblespace::core::repo::async_store::AsyncBlobStoreGet,
{
    let Some(handle) = configured_handle(scope)? else {
        return open_configured(storage, scope, authority);
    };
    let snapshot = storage.snapshot().context("freeze configured collection")?;
    let reader = crate::storage::AcquiringReader::new(snapshot.clone(), runtime.clone());
    let collection = open_exact_in(&reader, scope, handle)?;
    if let Some(warning) = empty_beside_content(&snapshot, scope, handle) {
        eprintln!("warning: {warning}");
    }
    Ok(collection)
}

/// READ admission over frozen proofs with exact descriptor, name, and
/// capability-definition acquisition. Unavailable or inapplicable evidence
/// still grants nothing; obtaining bytes does not discover new proofs.
pub fn open_configured_read_acquiring<S>(
    storage: &mut S,
    scope: Id,
    subject: VerifyingKey,
    runtime: &std::sync::Arc<tokio::runtime::Runtime>,
) -> anyhow::Result<Collection<SimpleArchive>>
where
    S: CollectionStoreExt + SnapshotSource,
    S::Snapshot: BlobStoreList
        + CapabilityProofRead
        + triblespace::core::repo::async_store::AsyncBlobStoreGet,
{
    let Some(handle) = configured_handle(scope)? else {
        return open(storage, scope, subject).context("register signer-private descriptor");
    };
    let snapshot = storage
        .snapshot()
        .context("freeze configured collection READ evidence")?;
    let reader = crate::storage::AcquiringReader::new(snapshot, runtime.clone());
    open_exact_read_in(&reader, scope, subject, handle)
}

/// The silent failure a re-mint produces, made audible: the configured
/// generation holds nothing while other generations of the same name hold
/// records. Not a refusal, because an empty new generation beside dead ones
/// is also what a deliberate cutover looks like the moment before its drain,
/// and for some names (orient) the old content is never meant to be carried.
/// Costs one indexed probe on a healthy host; the whole-store detector runs
/// only when the configured generation is empty.
fn empty_beside_content<S>(snapshot: &S, scope: Id, handle: CollectionHandle) -> Option<String>
where
    S: CollectionRead + BlobStoreGet,
{
    let selected = std::collections::BTreeSet::from([CollectionRecordSelector::Collection(handle)]);
    if !snapshot
        .select_records(&selected)
        .map(|records| records.is_empty())
        .unwrap_or(false)
    {
        return None;
    }
    let report = generation::named_generations(snapshot, handle).ok()??;
    if !report.strands_records() {
        return None;
    }
    let holding = report
        .siblings()
        .iter()
        .filter(|sibling| sibling.commits() > 0)
        .count();
    Some(format!(
        "{} names an empty generation of {:?} (blake3:{}) while {} other generation(s) in this \
         pile hold {} record(s); if this host was meant to read them, drain them into this \
         generation (trible pile collection adopt --into blake3:{} --siblings) or configure \
         the generation that holds them",
        override_env_name(scope),
        require_name(scope),
        hex::encode(handle.raw),
        holding,
        report.stranded_records(),
        hex::encode(handle.raw),
    ))
}

/// Open the operator-selected exact descriptor for a reader, or construct the
/// ordinary signer-private descriptor when no override is present.
///
/// Unlike [`open_configured`], an exact override requires READ rather than
/// WRITE admission. This is the appropriate boundary for consumers of a
/// shared collection which never publish to it. Admission uses the descriptor
/// and proof evidence in the supplied frozen snapshot.
/// Refuse a COMMAND whose record would be published but never admitted.
///
/// [`open_configured`] says why publication itself stays unconditional, and
/// that stays true: a library may publish an offline COMMIT that later
/// evidence activates. What must not stay silent is a command a person typed.
/// An unadmitted COMMIT is appended and then invisible — every read goes
/// through a maintained projection that carries admitted support only — so the
/// command prints an id, exits zero, and changes nothing anyone can observe.
/// That is how a wrong signing key ran for eight hours without a single
/// symptom.
///
/// On the ordinary path this can never fire: the descriptor's WRITE authority
/// IS the signer's own key, so admission is self-satisfied. It exists for the
/// override path, where `TRIBLESPACE_COLLECTION_<FACULTY>` names an exact
/// descriptor whose authority may be somebody else's key — which is precisely
/// the configuration that produced the incident.
///
/// `faculty` names the collection in the message and `reader_hint` names a
/// command whose output would silently not change, because "your write went
/// nowhere" is only actionable if the reader knows where to look.
pub fn require_command_write_admission<S>(
    store: &mut S,
    collection: Collection<SimpleArchive>,
    signer: &ed25519_dalek::SigningKey,
    faculty: &str,
    reader_hint: &str,
) -> anyhow::Result<()>
where
    S: SnapshotSource,
    S::Snapshot: StoreSnapshot + BlobStoreGet + CapabilityProofRead,
{
    let snapshot = store
        .snapshot()
        .map_err(|error| anyhow!("freeze {faculty} publication authority: {error}"))?;
    require_command_write_in(&snapshot, collection, signer, faculty, reader_hint)
}

/// The command WRITE guard with exact-byte acquisition and frozen proof evidence.
/// Like the other acquiring entry points, call only at the synchronous boundary.
pub fn require_command_write_admission_acquiring<S>(
    store: &mut S,
    collection: Collection<SimpleArchive>,
    signer: &ed25519_dalek::SigningKey,
    faculty: &str,
    reader_hint: &str,
    runtime: &std::sync::Arc<tokio::runtime::Runtime>,
) -> anyhow::Result<()>
where
    S: SnapshotSource,
    S::Snapshot: CapabilityProofRead + triblespace::core::repo::async_store::AsyncBlobStoreGet,
{
    let snapshot = store.snapshot().context("freeze publication authority")?;
    let reader = crate::storage::AcquiringReader::new(snapshot, runtime.clone());
    require_command_write_in(&reader, collection, signer, faculty, reader_hint)
}

fn require_command_write_in<S>(
    snapshot: &S,
    collection: Collection<SimpleArchive>,
    signer: &ed25519_dalek::SigningKey,
    faculty: &str,
    reader_hint: &str,
) -> anyhow::Result<()>
where
    S: StoreSnapshot + BlobStoreGet + CapabilityProofRead,
{
    let admitted = collection
        .writer_is_admitted_acquiring(snapshot, signer.verifying_key())
        .map_err(|error| anyhow!("check {faculty} collection WRITE admission: {error}"))?;
    if !admitted {
        bail!(
            "key {} is not admitted to write the {faculty} collection {}. The record would be \
             appended as a raw ledger entry that never enters an admitted snapshot, so no \
             reader — `{reader_hint}` included — would ever see it. Grant that key WRITE on the \
             collection, or run with an admitted key.",
            hex::encode_upper(signer.verifying_key().to_bytes()),
            hex::encode_upper(collection.handle().raw),
        );
    }
    Ok(())
}

pub fn open_configured_read<S>(
    storage: &mut S,
    scope: Id,
    subject: VerifyingKey,
) -> anyhow::Result<Collection<SimpleArchive>>
where
    S: CollectionStoreExt + SnapshotSource,
    <S as SnapshotSource>::Snapshot: BlobStoreGet + BlobStoreList + CapabilityProofRead,
{
    let Some(handle) = configured_handle(scope)? else {
        return open(storage, scope, subject).context("register signer-private descriptor");
    };

    let snapshot = storage
        .snapshot()
        .context("freeze store while opening configured collection descriptor")?;
    open_exact_read_in(&snapshot, scope, subject, handle)
}

/// Open and validate one exact faculty descriptor in an existing snapshot.
///
/// This is the coherent publication-boundary form used by callers which
/// already froze a pile prefix. It validates only the descriptor's type and
/// faculty name. Local publication does not require present WRITE admission.
pub fn open_exact_in<S>(
    snapshot: &S,
    scope: Id,
    handle: CollectionHandle,
) -> anyhow::Result<Collection<SimpleArchive>>
where
    S: BlobStoreGet,
{
    open_exact_descriptor_in(snapshot, scope, handle)
}

/// Open and validate one exact faculty descriptor for a READ-only consumer
/// using the snapshot's frozen descriptor and proof evidence.
pub fn open_exact_read_in<S>(
    snapshot: &S,
    scope: Id,
    subject: VerifyingKey,
    handle: CollectionHandle,
) -> anyhow::Result<Collection<SimpleArchive>>
where
    S: StoreSnapshot + BlobStoreGet + BlobStoreList + CapabilityProofRead,
{
    let collection = open_exact_descriptor_in(snapshot, scope, handle)?;
    let expected = require_name(scope);
    if !collection
        .reader_is_admitted_acquiring(snapshot, subject)
        .context("check configured collection READ admission")?
    {
        bail!(
            "durable signer {} is not admitted to READ configured collection {:?}",
            hex::encode(subject.to_bytes()),
            expected,
        );
    }
    Ok(collection)
}

fn open_exact_descriptor_in<S>(
    snapshot: &S,
    scope: Id,
    handle: CollectionHandle,
) -> anyhow::Result<Collection<SimpleArchive>>
where
    S: BlobStoreGet,
{
    let collection = Collection::open(snapshot, handle).with_context(|| {
        format!(
            "open exact {} descriptor from {}",
            require_name(scope),
            override_env_name(scope)
        )
    })?;
    let blob: Blob<SimpleArchive> = snapshot
        .get(handle)
        .context("read configured collection descriptor while checking its name")?;
    let facts = TribleSet::try_from_blob(blob)
        .context("decode configured collection descriptor while checking its name")?;
    let name_handle = descriptor::name(&facts)
        .context("decode configured collection name")?
        .ok_or_else(|| anyhow!("configured faculty collection is derived and has no root name"))?;
    let name: View<str> = snapshot
        .get::<View<str>, UTF8String>(name_handle)
        .context("read configured collection name")?;
    let expected = require_name(scope);
    if &*name != expected {
        bail!(
            "{} names collection {:?}, not expected faculty collection {:?}",
            override_env_name(scope),
            &*name,
            expected,
        );
    }
    Ok(collection)
}

/// Register one faculty root and return its typed descriptor handle.
///
/// Registration is idempotent and owns the descriptor's complete attachment
/// closure. Later publication and snapshots take only the returned handle;
/// the store remains owned by its caller.
pub fn open<S>(
    storage: &mut S,
    scope: Id,
    authority: VerifyingKey,
) -> Result<Collection<SimpleArchive>, CollectionRegistrationError<<S as BlobStorePut>::PutError>>
where
    S: CollectionStoreExt,
{
    storage.collection(require_name(scope), private_policy(authority))
}

#[cfg(test)]
mod tests {
    use super::*;

    type MemorySnapshot = <MemoryRepo as SnapshotSource>::Snapshot;

    /// Exact-byte provider for startup tests. Its proof view never advances
    /// when a requested blob is supplied from the later source observation.
    #[derive(Clone)]
    struct StartupSnapshot {
        frozen: MemorySnapshot,
        source: MemorySnapshot,
        requested: std::sync::Arc<std::sync::Mutex<Vec<[u8; 32]>>>,
    }

    impl StoreSnapshot for StartupSnapshot {}

    impl CapabilityProofRead for StartupSnapshot {
        type ProofsError = <MemorySnapshot as CapabilityProofRead>::ProofsError;
        type ProofIter<'a> = <MemorySnapshot as CapabilityProofRead>::ProofIter<'a>;

        fn proofs(&self) -> Result<Self::ProofIter<'_>, Self::ProofsError> {
            self.frozen.proofs()
        }
    }

    impl BlobStoreList for StartupSnapshot {
        type Err = <MemorySnapshot as BlobStoreList>::Err;
        type Iter<'a> = <MemorySnapshot as BlobStoreList>::Iter<'a>;

        fn blobs(&self) -> Self::Iter<'_> {
            self.frozen.blobs()
        }
    }

    impl triblespace::core::repo::async_store::AsyncBlobStoreGet for StartupSnapshot {
        type GetError<E: std::error::Error + Send + Sync + 'static> =
            <MemorySnapshot as BlobStoreGet>::GetError<E>;

        fn get<T, E>(
            &self,
            handle: Inline<triblespace::core::inline::encodings::hash::Handle<E>>,
        ) -> impl std::future::Future<Output = Result<T, Self::GetError<T::Error>>> + Send
        where
            E: triblespace::core::blob::BlobEncoding + 'static,
            T: TryFromBlob<E>,
            triblespace::core::inline::encodings::hash::Handle<E>:
                triblespace::core::inline::InlineEncoding,
        {
            let raw = handle.raw;
            async move {
                let handle =
                    Inline::<triblespace::core::inline::encodings::hash::Handle<E>>::new(raw);
                if self.frozen.contains_blob(handle).unwrap() {
                    return self.frozen.get(handle);
                }
                self.requested.lock().unwrap().push(raw);
                self.source.get(handle)
            }
        }
    }

    fn startup_reader(
        source: &mut MemoryRepo,
        missing: Option<[u8; 32]>,
        include_proofs: bool,
    ) -> (
        crate::storage::AcquiringReader<StartupSnapshot>,
        StartupSnapshot,
    ) {
        use anybytes::Bytes;
        use triblespace::core::blob::encodings::UnknownBlob;
        let source = source.snapshot().unwrap();
        let mut local = MemoryRepo::default();
        for info in source.blobs() {
            let handle = info.unwrap().handle;
            if Some(handle.raw) != missing {
                let bytes: Bytes = source.get(handle).unwrap();
                local.put::<UnknownBlob, _>(bytes).unwrap();
            }
        }
        if include_proofs {
            for proof in source.proofs().unwrap() {
                local.insert_proof(proof.unwrap()).unwrap();
            }
        }
        let snapshot = StartupSnapshot {
            frozen: local.snapshot().unwrap(),
            source,
            requested: Default::default(),
        };
        let reader = crate::storage::AcquiringReader::new(
            snapshot.clone(),
            std::sync::Arc::new(crate::storage::runtime().unwrap()),
        );
        (reader, snapshot)
    }

    #[test]
    fn startup_acquires_missing_descriptor_and_name_without_relaxing_validation() {
        let owner = SigningKey::from_bytes(&[0x65; 32]);
        let mut source = MemoryRepo::default();
        let collection = open(&mut source, wiki::DEFAULT_SCOPE_ID, owner.verifying_key()).unwrap();
        let snapshot = source.snapshot().unwrap();
        let descriptor: Blob<SimpleArchive> = snapshot.get(collection.handle()).unwrap();
        let name = descriptor::name(&TribleSet::try_from_blob(descriptor).unwrap())
            .unwrap()
            .unwrap();
        for missing in [collection.handle().raw, name.raw] {
            let (reader, evidence) = startup_reader(&mut source, Some(missing), true);
            assert!(open_exact_in(
                &evidence.frozen,
                wiki::DEFAULT_SCOPE_ID,
                collection.handle()
            )
            .is_err());
            assert_eq!(
                open_exact_in(&reader, wiki::DEFAULT_SCOPE_ID, collection.handle()).unwrap(),
                collection
            );
            let requested = evidence.requested.lock().unwrap();
            assert!(!requested.is_empty());
            assert!(requested.iter().all(|handle| *handle == missing));
            drop(requested);
            assert!(
                !reader
                    .contains_blob(Inline::<
                        triblespace::core::inline::encodings::hash::Handle<
                            triblespace::core::blob::encodings::UnknownBlob,
                        >,
                    >::new(missing))
                    .unwrap(),
                "acquisition must not advance frozen residency"
            );
        }
        let (reader, evidence) = startup_reader(&mut source, None, true);
        assert!(open_exact_in(&reader, relations::DEFAULT_SCOPE_ID, collection.handle()).is_err());
        assert!(evidence.requested.lock().unwrap().is_empty());
    }

    #[test]
    fn startup_acquires_admission_definitions_but_never_later_proofs() {
        let owner = SigningKey::from_bytes(&[0x66; 32]);
        let subject = SigningKey::from_bytes(&[0x67; 32]);
        let outsider = SigningKey::from_bytes(&[0x68; 32]);
        let mut source = MemoryRepo::default();
        let collection = open(&mut source, wiki::DEFAULT_SCOPE_ID, owner.verifying_key()).unwrap();
        grant_collection_read(
            &mut source,
            collection.handle(),
            &owner,
            subject.verifying_key(),
        )
        .unwrap();
        let definition = triblespace::core::collection::read_capability();
        let (reader, evidence) = startup_reader(&mut source, Some(definition.raw), true);
        assert!(open_exact_read_in(
            &evidence.frozen,
            wiki::DEFAULT_SCOPE_ID,
            subject.verifying_key(),
            collection.handle()
        )
        .is_err());
        assert_eq!(
            open_exact_read_in(
                &reader,
                wiki::DEFAULT_SCOPE_ID,
                subject.verifying_key(),
                collection.handle()
            )
            .unwrap(),
            collection
        );
        assert!(evidence.requested.lock().unwrap().contains(&definition.raw));

        let (resident, evidence) = startup_reader(&mut source, None, true);
        assert!(open_exact_read_in(
            &resident,
            wiki::DEFAULT_SCOPE_ID,
            outsider.verifying_key(),
            collection.handle()
        )
        .is_err());
        assert!(evidence.requested.lock().unwrap().is_empty());

        let (before_grant, _) = startup_reader(&mut source, Some(definition.raw), false);
        assert!(
            open_exact_read_in(
                &before_grant,
                wiki::DEFAULT_SCOPE_ID,
                subject.verifying_key(),
                collection.handle()
            )
            .is_err(),
            "provider's newer proof must not enter frozen admission"
        );
        assert!(before_grant.proofs().unwrap().next().is_none());

        let (_, mut offline) = startup_reader(&mut source, Some(definition.raw), true);
        offline.source = offline.frozen.clone();
        let offline = crate::storage::AcquiringReader::new(
            offline,
            std::sync::Arc::new(crate::storage::runtime().unwrap()),
        );
        assert!(
            open_exact_read_in(
                &offline,
                wiki::DEFAULT_SCOPE_ID,
                subject.verifying_key(),
                collection.handle()
            )
            .is_err(),
            "an unavailable definition cannot grant READ"
        );
    }

    /// A host with no configured handle may mint the private descriptor on an
    /// empty pile, but not beside a generation of the same name that already
    /// holds content: that is how a fourth generation would start by accident.
    #[test]
    fn open_configured_refuses_to_start_a_generation_beside_one_with_content() {
        use triblespace::core::collection::{CollectionCommit, CollectionRecord, CollectionStore};

        let scope = decide::DEFAULT_SCOPE_ID;
        let variable = override_env_name(scope);
        assert!(
            std::env::var_os(&variable).is_none(),
            "{variable} must be unset for this test"
        );
        let mut store = MemoryRepo::default();
        let mac = SigningKey::from_bytes(&[71; 32]);
        let sky = SigningKey::from_bytes(&[72; 32]);

        // An empty pile: the private descriptor is the only generation.
        let first = open_configured(&mut store, scope, mac.verifying_key())
            .expect("the first generation on an empty pile");
        // Still fine while nobody has committed anything anywhere.
        open_configured(&mut store, scope, sky.verifying_key())
            .expect("a second descriptor is only content until something commits");

        store
            .insert(CollectionRecord::Commit(CollectionCommit::sign(
                &mac,
                first.handle(),
                Inline::new([1; 32]),
                Inline::new([2; 32]),
            )))
            .unwrap();
        let error = open_configured(&mut store, scope, sky.verifying_key())
            .expect_err("a generation with content exists; refuse to start another")
            .to_string();
        assert!(error.contains(&variable), "{error}");
        assert!(error.contains(&hex::encode(first.handle().raw)), "{error}");
        // The same host's own generation, reopened, is not "another".
        open_configured(&mut store, scope, mac.verifying_key())
            .expect("reopening the generation that holds the content");

        // Nor is it once a sibling holds content of its own, which is what
        // `cat` of another host's pile produces: each host keeps opening the
        // generation it already wrote to.
        let theirs = open(&mut store, scope, sky.verifying_key()).unwrap();
        store
            .insert(CollectionRecord::Commit(CollectionCommit::sign(
                &sky,
                theirs.handle(),
                Inline::new([3; 32]),
                Inline::new([4; 32]),
            )))
            .unwrap();
        open_configured(&mut store, scope, mac.verifying_key())
            .expect("a sibling with content does not stop a host reopening its own");
        open_configured(&mut store, scope, sky.verifying_key())
            .expect("each host with content reopens its own generation");
        // A host with nothing of its own is still starting one, and refused.
        let fresh = SigningKey::from_bytes(&[70; 32]);
        open_configured(&mut store, scope, fresh.verifying_key())
            .expect_err("an empty generation beside content is still not started");
    }

    /// A configured generation that reads empty while a same-named sibling
    /// holds records is the re-mint failure that ran silently four times;
    /// it is now said out loud, and only then.
    #[test]
    fn an_empty_configured_generation_beside_content_is_named_not_silent() {
        use triblespace::core::collection::{CollectionCommit, CollectionRecord, CollectionStore};

        let scope = decide::DEFAULT_SCOPE_ID;
        let mut store = MemoryRepo::default();
        let mac = SigningKey::from_bytes(&[73; 32]);
        let sky = SigningKey::from_bytes(&[74; 32]);
        let old = open(&mut store, scope, mac.verifying_key()).unwrap();
        let new = open(&mut store, scope, sky.verifying_key()).unwrap();

        // Both empty: nothing to say.
        let snapshot = store.snapshot().unwrap();
        assert_eq!(empty_beside_content(&snapshot, scope, new.handle()), None);

        store
            .insert(CollectionRecord::Commit(CollectionCommit::sign(
                &mac,
                old.handle(),
                Inline::new([1; 32]),
                Inline::new([2; 32]),
            )))
            .unwrap();
        let snapshot = store.snapshot().unwrap();
        // The generation holding the content is fine to configure.
        assert_eq!(empty_beside_content(&snapshot, scope, old.handle()), None);
        // The empty one beside it is named, with the remedy.
        let warning = empty_beside_content(&snapshot, scope, new.handle())
            .expect("an empty generation beside content is said out loud");
        assert!(
            warning.contains(&hex::encode(new.handle().raw)),
            "{warning}"
        );
        assert!(warning.contains("--siblings"), "{warning}");
        assert!(warning.contains(&override_env_name(scope)), "{warning}");
    }

    /// The guard refuses an unadmitted writer and says enough to fix it.
    ///
    /// Also pins the half that must NOT change: publication itself stays
    /// unconditional, so the same key can still append through the library
    /// path. The guard is on the command, not on the commit.
    #[test]
    fn the_command_guard_refuses_an_unadmitted_writer_and_names_the_remedy() {
        let mut store = MemoryRepo::default();
        let owner = SigningKey::from_bytes(&[61; 32]);
        let outsider = SigningKey::from_bytes(&[62; 32]);
        let collection = open(&mut store, decide::DEFAULT_SCOPE_ID, owner.verifying_key())
            .expect("register the signer-private descriptor");

        require_command_write_admission(&mut store, collection, &owner, "Decide", "decide show")
            .expect("the descriptor's own authority is admitted");

        let error = require_command_write_admission(
            &mut store,
            collection,
            &outsider,
            "Decide",
            "decide show",
        )
        .expect_err("an unadmitted writer must not get a silent success");
        let message = format!("{error:#}");
        assert!(message.contains("not admitted to write"), "{message}");
        assert!(
            message.contains(&hex::encode_upper(outsider.verifying_key().to_bytes())),
            "the refusal names the key to grant: {message}"
        );
        assert!(
            message.contains(&hex::encode_upper(collection.handle().raw)),
            "the refusal names the collection to grant it on: {message}"
        );
        assert!(
            message.contains("decide show"),
            "the refusal names a reader that would silently not change: {message}"
        );

        // The library path is deliberately untouched.
        let fragment = entity! { metadata::description: "raw outsider publication".to_owned() };
        store
            .commit(collection, &outsider, fragment)
            .expect("publication stays unconditional");
    }

    use std::collections::BTreeSet;

    use ed25519_dalek::SigningKey;
    use triblespace::core::capability::{CapabilityProof, CapabilityResource};
    use triblespace::core::collection::grant_collection_read;
    use triblespace::core::metadata;
    use triblespace::core::repo::memoryrepo::MemoryRepo;
    use triblespace::core::repo::{CapabilityProofStore, SnapshotSource};
    use triblespace::core::trible::TribleSet;
    use triblespace::macros::entity;

    #[test]
    fn every_name_is_nonempty_and_no_two_scopes_share_one() {
        let mut names = BTreeSet::new();
        let mut scopes = BTreeSet::new();
        let mut variables = BTreeSet::new();
        for (scope, name) in table() {
            assert!(!name.is_empty());
            assert!(names.insert(name), "two scopes both claim the name {name}");
            assert!(scopes.insert(scope), "scope {scope:X} appears twice");
            assert!(
                variables.insert(override_env_name(scope)),
                "two collections normalize to one override variable"
            );
        }
    }

    #[test]
    fn a_scope_with_no_name_is_loud_rather_than_invented() {
        assert!(name_for(Id::new([0x5a; 16]).unwrap()).is_none());
    }

    #[test]
    fn root_policy_is_identity_and_snapshot_admission() {
        let local = SigningKey::from_bytes(&[0x31; 32]);
        let foreign = SigningKey::from_bytes(&[0x73; 32]);
        let scope = wiki::DEFAULT_SCOPE_ID;
        let evidence = entity! { _ @ metadata::tag: &scope };
        let expected = evidence.facts().clone();
        let mut store = MemoryRepo::default();
        let collection = open(&mut store, scope, local.verifying_key()).unwrap();
        store
            .commit(collection, &foreign, evidence.clone())
            .unwrap();
        let store_snapshot = store.snapshot().unwrap();
        let facts = collection.read::<TribleSet, _>(&store_snapshot).unwrap();
        assert!(facts.is_empty());

        store.commit(collection, &local, evidence).unwrap();
        let store_snapshot = store.snapshot().unwrap();
        let facts = collection.read::<TribleSet, _>(&store_snapshot).unwrap();
        assert!(expected.difference(&facts).is_empty());
    }

    #[test]
    fn override_names_and_handles_are_exact() {
        assert_eq!(
            override_env_name(memory::DEFAULT_SCOPE_ID),
            "TRIBLESPACE_COLLECTION_MEMORY_JOURNAL"
        );
        let variable = override_env_name(wiki::DEFAULT_SCOPE_ID);
        let raw = "ab".repeat(32);
        assert_eq!(
            parse_override(&variable, OsString::from(&raw)).unwrap().raw,
            [0xab; 32]
        );
        assert_eq!(
            parse_override(&variable, OsString::from(format!("blake3:{raw}")))
                .unwrap()
                .raw,
            [0xab; 32]
        );
        assert!(parse_override(&variable, OsString::from("ab")).is_err());
        assert!(parse_override(&variable, OsString::from("zz".repeat(32))).is_err());
    }

    #[test]
    fn exact_publication_open_requires_the_expected_name_not_current_write_admission() {
        let operator = SigningKey::from_bytes(&[0x41; 32]);
        let tenant = SigningKey::from_bytes(&[0x52; 32]);
        let mut store = MemoryRepo::default();
        let shared = store
            .collection("wiki", private_policy(operator.verifying_key()))
            .unwrap();
        let snapshot = store.snapshot().unwrap();
        let opened = open_exact_in(&snapshot, wiki::DEFAULT_SCOPE_ID, shared.handle()).unwrap();
        assert_eq!(opened, shared);
        let private = open(&mut store, wiki::DEFAULT_SCOPE_ID, tenant.verifying_key()).unwrap();

        assert_ne!(opened, private);

        let wrong_name = store
            .collection("relations", private_policy(tenant.verifying_key()))
            .unwrap();
        let snapshot = store.snapshot().unwrap();
        let error =
            open_exact_in(&snapshot, wiki::DEFAULT_SCOPE_ID, wrong_name.handle()).unwrap_err();
        assert!(error
            .to_string()
            .contains("not expected faculty collection"));
    }

    #[test]
    fn exact_read_open_requires_current_read_admission() {
        let operator = SigningKey::from_bytes(&[0x61; 32]);
        let reader = SigningKey::from_bytes(&[0x62; 32]);
        let mut store = MemoryRepo::default();
        let shared = store
            .collection("wiki", private_policy(operator.verifying_key()))
            .unwrap();

        let snapshot = store.snapshot().unwrap();
        let error = open_exact_read_in(
            &snapshot,
            wiki::DEFAULT_SCOPE_ID,
            reader.verifying_key(),
            shared.handle(),
        )
        .unwrap_err();
        assert!(error.to_string().contains("is not admitted to READ"));
        drop(snapshot);

        grant_collection_read(
            &mut store,
            shared.handle(),
            &operator,
            reader.verifying_key(),
        )
        .unwrap();
        let snapshot = store.snapshot().unwrap();
        assert_eq!(
            open_exact_read_in(
                &snapshot,
                wiki::DEFAULT_SCOPE_ID,
                reader.verifying_key(),
                shared.handle(),
            )
            .unwrap(),
            shared
        );
    }

    #[test]
    fn exact_read_open_reuses_unchanged_snapshot_evidence() {
        let operator = SigningKey::from_bytes(&[0x63; 32]);
        let reader = SigningKey::from_bytes(&[0x64; 32]);
        let mut store = MemoryRepo::default();
        let shared = store
            .collection("wiki", private_policy(operator.verifying_key()))
            .unwrap();
        store
            .insert_proof(CapabilityProof::new(
                CapabilityResource::from(shared.handle()),
                &operator,
                triblespace::core::collection::read_capability(),
                reader.verifying_key(),
            ))
            .unwrap();
        let valid = store.snapshot().unwrap();
        let later = store.snapshot().unwrap();
        assert!(later.changes_since(&valid).is_empty());
        assert!(valid.changes_since(&later).is_empty());
        assert!(open_exact_read_in(
            &later,
            wiki::DEFAULT_SCOPE_ID,
            reader.verifying_key(),
            shared.handle(),
        )
        .is_ok());
        assert_eq!(
            open_exact_read_in(
                &valid.clone(),
                wiki::DEFAULT_SCOPE_ID,
                reader.verifying_key(),
                shared.handle(),
            )
            .unwrap(),
            shared
        );
    }
}