Huginn/crates/huginn-mind/src/docs.rs

//! The image and the batch, the two places a rule or a view may look, and
//! every derivation over them. "In force" is recursive: a document is in force
//! when no resolution that is itself in force names it, so a withdrawn
//! resolution stops closing its subject. Computed over image and batch at rule
//! time, over the image at read time, never stored, and never from
//! `stored_at`.

use cultcache_rs::CultCacheEnvelope;
use eureka_pipeline::{
    Label, OrgRepo, PipelineCampaign, PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, PipelineRun, PipelineStewardship, ResolutionOutcome,
    ResearchSource, RunOperator, Short, Slug,
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};

use crate::image::Image;
use crate::refusal::MindRefusal;
use crate::rows::root_and_local;

/// A batch document after A3 and A5: decoded, keyed, and its envelope.
pub(crate) struct Staged {
    pub(crate) kind: PipelineKind,
    pub(crate) key: String,
    pub(crate) document: PipelineDocument,
    pub(crate) envelope: CultCacheEnvelope,
}

/// The held image at one snapshot and the batch, the two places a rule or a view
/// may look. The one door of visibility: a document is there when its writing
/// receipt's ordinal is at or before `as_of`. Admission runs at the head.
pub(crate) struct Docs<'i> {
    image: &'i Image,
    as_of: u64,
    pub(crate) batch: Vec<Staged>,
}

pub(crate) fn kind_of_type(type_id: &str) -> Option<PipelineKind> {
    PipelineKind::ALL.iter().copied().find(|kind| kind.type_id() == type_id)
}

impl<'i> Docs<'i> {
    pub(crate) fn at(image: &'i Image, as_of: u64) -> Self {
        Self { image, as_of, batch: Vec::new() }
    }

    fn visible(&self, id: u32) -> bool {
        self.image.row(id).ordinal <= self.as_of
    }

    /// A4: one identity, once.
    pub(crate) fn push(&mut self, staged: Staged) -> Result<(), MindRefusal> {
        if self.batch.iter().any(|other| other.kind == staged.kind && other.key == staged.key) {
            return Err(MindRefusal::IdentityCollision { kind: staged.kind, id: staged.key });
        }
        self.batch.push(staged);
        Ok(())
    }

    pub(crate) fn in_batch(&self, kind: PipelineKind, id: &str) -> Option<&PipelineDocument> {
        self.batch.iter().find(|staged| staged.kind == kind && staged.key == id).map(|staged| &staged.document)
    }

    /// Whether the mind holds this document as its own write rather than an import.
    pub(crate) fn is_native(&self, kind: PipelineKind, id: &str) -> bool {
        self.image.id_of(kind, id).filter(|row| self.visible(*row)).is_some_and(|row| self.image.row(row).view.admission.provenance.imported_from.is_none())
    }

    pub(crate) fn in_image(&self, kind: PipelineKind, id: &str) -> Option<&PipelineDocument> {
        self.image.id_of(kind, id).filter(|row| self.visible(*row)).map(|row| &self.image.row(row).view.document)
    }

    pub(crate) fn find(&self, kind: PipelineKind, id: &str) -> Option<&PipelineDocument> {
        self.in_batch(kind, id).or_else(|| self.in_image(kind, id))
    }

    /// Whether this mind holds a native write under `root` that an import must not meet (the batch is not asked).
    pub(crate) fn root_holds_native_write(&self, root: &str) -> bool {
        self.image.root_holds_native_write(root)
    }

    pub(crate) fn of_kind(&self, kind: PipelineKind) -> impl Iterator<Item = (&str, &PipelineDocument)> {
        let batch = self.batch.iter().filter(move |staged| staged.kind == kind).map(|staged| (staged.key.as_str(), &staged.document));
        let image = self
            .image
            .of_kind(kind)
            .filter(move |(_, row)| self.visible(*row))
            .map(move |(key, row)| (key, &self.image.row(row).view.document));
        batch.chain(image)
    }

    pub(crate) fn stewardships(&self) -> impl Iterator<Item = (&str, &PipelineStewardship)> {
        self.of_kind(PipelineKind::Stewardship).filter_map(|(key, document)| match document {
            PipelineDocument::Stewardship(stewardship) => Some((key, stewardship)),
            _ => None,
        })
    }

    /// Every resolution of one subject, in image and batch: the scope a
    /// subject's sequence counts in.
    pub(crate) fn resolutions_of(&self, subject: &PipelineRef) -> impl Iterator<Item = (&str, &PipelineResolution)> {
        self.resolutions_naming(subject.kind, &subject.id.0)
    }

    /// The resolutions naming one document: the batch's first, then the held ones visible at
    /// `as_of` in resolution-key order, the order a closing resolution is found in.
    fn resolutions_naming(&self, kind: PipelineKind, id: &str) -> impl Iterator<Item = (&str, &PipelineResolution)> {
        let batch = self.batch.iter().filter_map(move |staged| match &staged.document {
            PipelineDocument::Resolution(resolution) if resolution.subject.kind == kind && resolution.subject.id.0 == id => {
                Some((staged.key.as_str(), resolution))
            }
            _ => None,
        });
        let held = self.image.closers_of(kind, id).filter(move |(_, row)| self.visible(*row)).filter_map(move |(key, row)| {
            match &self.image.row(row).view.document {
                PipelineDocument::Resolution(resolution) => Some((key, resolution)),
                _ => None,
            }
        });
        batch.chain(held)
    }

    /// Every assignment of one repo to one mind, in image and batch, in force
    /// or not: the scope a stewardship's sequence counts in.
    pub(crate) fn assignments_of(&self, mind: &Slug, repo: &OrgRepo) -> impl Iterator<Item = (&str, &PipelineStewardship)> {
        self.stewardships().filter(move |(_, stewardship)| stewardship.instance == *mind && stewardship.repo == *repo)
    }

    /// The greatest sequence among a subject's resolutions in image or batch,
    /// `0` when it has none, so the next is `latest + 1`. `own` is the
    /// document being checked, excluded by content identity so that an exact
    /// replay measures the same `latest` the first admission did.
    pub(crate) fn latest_resolution(&self, subject: &PipelineRef, own: Option<&PipelineResolution>) -> u32 {
        self.resolutions_of(subject)
            .filter(|(_, resolution)| own != Some(*resolution))
            .map(|(_, resolution)| resolution.sequence)
            .max()
            .unwrap_or(0)
    }

    /// The sequence a derivation takes: the one the record it already derived
    /// carries, when image or batch holds that record, so an exact replay
    /// re-derives the same document and A9 answers with the stored receipt;
    /// otherwise the next. A derivation is recognised by what it writes into
    /// its record: a resolution by its outcome, an assignment by its note.
    pub(crate) fn derived_resolution_sequence(
        &self,
        subject: &PipelineRef,
        derived: impl Fn(&ResolutionOutcome) -> bool,
    ) -> u32 {
        self.resolutions_of(subject)
            .find(|(_, resolution)| derived(&resolution.outcome))
            .map_or_else(|| self.latest_resolution(subject, None) + 1, |(_, resolution)| resolution.sequence)
    }

    /// The same for the assignment a hand-off derives, which notes its key.
    pub(crate) fn derived_stewardship_sequence(&self, mind: &Slug, repo: &OrgRepo, note: &str) -> u32 {
        self.stewardships()
            .find(|(_, stewardship)| stewardship.instance == *mind && stewardship.repo == *repo && stewardship.note.0 == note)
            .map_or_else(|| self.latest_stewardship(mind, repo, None) + 1, |(_, stewardship)| stewardship.sequence)
    }

    /// The same for a repo's assignments to one mind.
    pub(crate) fn latest_stewardship(&self, mind: &Slug, repo: &OrgRepo, own: Option<&PipelineStewardship>) -> u32 {
        self.assignments_of(mind, repo)
            .filter(|(_, stewardship)| own != Some(*stewardship))
            .map(|(_, stewardship)| stewardship.sequence)
            .max()
            .unwrap_or(0)
    }

    /// No resolution that is itself in force names the document. The
    /// recursion is the whole rule: a withdrawn resolution stops counting, so
    /// its subject is open again. `key(R)` is strictly longer than the id it
    /// resolves, so the chain is bounded and this terminates.
    pub(crate) fn in_force(&self, kind: PipelineKind, id: &str) -> bool {
        self.in_force_unless(kind, id, |_| false)
    }

    /// In force, ignoring a resolution the caller's own document is or
    /// derives: the ruling that answers a question is what resolves it, the
    /// hand-off that withdraws a stewardship is what resolves that, and a
    /// resolution does not resolve its own subject out from under itself.
    pub(crate) fn in_force_unless(&self, kind: PipelineKind, id: &str, own: impl Fn(&PipelineResolution) -> bool) -> bool {
        self.closing_resolution_unless(kind, id, own).is_none()
    }

    /// The in-force resolution naming the document, other than `own`, if any;
    /// exact rather than a tie-break because A8 and the Q19 cap admit at most
    /// one.
    pub(crate) fn closing_resolution_unless(
        &self,
        kind: PipelineKind,
        id: &str,
        own: impl Fn(&PipelineResolution) -> bool,
    ) -> Option<(&str, &PipelineResolution)> {
        self.resolutions_naming(kind, id).find(|&(key, resolution)| !own(resolution) && self.in_force(PipelineKind::Resolution, key))
    }

    /// The repos a campaign's cuts may name: the repo set of the highest-revision
    /// in-force target of `campaign.slug` that names one, else the campaign's own
    /// opening set. The one owner of membership (ruling campaign-repo-set-home).
    pub(crate) fn repos_in_force<'a>(&'a self, campaign: &'a PipelineCampaign) -> &'a [OrgRepo] {
        self.of_kind(PipelineKind::Target)
            .filter_map(|(key, document)| match document {
                PipelineDocument::Target(target)
                    if target.campaign == campaign.slug && !target.repos.is_empty() && self.in_force(PipelineKind::Target, key) =>
                {
                    Some(target)
                }
                _ => None,
            })
            .max_by_key(|target| target.revision)
            .map_or(campaign.repos.as_slice(), |target| target.repos.as_slice())
    }

    /// The resolution that closes the document, if any: the views' status,
    /// derived where admission derives whether there is one at all. A document
    /// with no in-force closer of its own reads closed by the Moved resolution of
    /// the campaign it is rooted at, the Moved resolution itself excepted. The
    /// fallback is the status's alone: a rule asks `in_force`, which does not
    /// have it, so a moved campaign's own closures keep their standing.
    pub(crate) fn closing_resolution(&self, kind: PipelineKind, id: &str) -> Option<(&str, &PipelineResolution)> {
        self.closing_resolution_unless(kind, id, |_| false).or_else(|| {
            let moved = self.moved_campaign(root_and_local(id).0)?;
            (kind != PipelineKind::Resolution || id != moved.0).then_some(moved)
        })
    }

    /// The Moved resolution over the campaign rooted at `root`, among the held documents visible at
    /// `as_of`: the one fact that freezes a root and closes everything under it. A campaign takes no
    /// other resolution and a Moved cannot be withdrawn (admission refuses both), so a campaign's
    /// closers are this one alone and it is always in force. The batch is not asked, so a Moved and
    /// the writes it lands beside are one receipt.
    pub(crate) fn moved_campaign(&self, root: &str) -> Option<(&'i str, &'i PipelineResolution)> {
        let image: &'i Image = self.image;
        image
            .of_kind(PipelineKind::Campaign)
            .filter(|(key, _)| root_and_local(key).0 == root)
            .flat_map(|(key, _)| image.closers_of(PipelineKind::Campaign, key))
            .filter(|(_, row)| self.visible(*row))
            .find_map(|(key, row)| match &image.row(row).view.document {
                PipelineDocument::Resolution(resolution @ PipelineResolution { outcome: ResolutionOutcome::Moved { .. }, .. }) => Some((key, resolution)),
                _ => None,
            })
    }

    /// Every in-force stewardship of `(mind, repo)`, ignoring the withdrawal
    /// `hand_off_key` derives. The stewardship row holds one in force at a
    /// time, but that exclusion can put a second back in the list: the one a
    /// replayed hand-off withdrew, beside whatever stewards the repo now.
    pub(crate) fn stewardships_of(
        &self,
        mind: &Slug,
        repo: &OrgRepo,
        hand_off_key: Option<&str>,
    ) -> Vec<(&str, &PipelineStewardship)> {
        self.assignments_of(mind, repo)
            .filter(|(key, _)| {
                self.in_force_unless(PipelineKind::Stewardship, key, |resolution| {
                    matches!(&resolution.outcome, ResolutionOutcome::Withdrawn { reason } if Some(reason.0.as_str()) == hand_off_key)
                })
            })
            .collect()
    }

    /// The in-force record, other than `base_key`, that cannot stand beside
    /// `base` on its line: the one question "may this record be in force"
    /// asks, whether `base` is being admitted (a stewardship's
    /// `AlreadyStewarded`) or put back in force by a withdrawal. A line holds
    /// one stewardship at a time, so any other stands in its way; a revision
    /// is only displaced by a later one.
    pub(crate) fn rival_in_force(&self, base_key: &str, base: &PipelineDocument) -> Option<&str> {
        let kind = base.kind();
        self.of_kind(kind)
            .find(|(key, other)| *key != base_key && rivals(base, other) && self.in_force(kind, key))
            .map(|(key, _)| key)
    }

    /// The in-force run, other than `run_key`, that holds `claim`: the one
    /// consumption fact run-is-the-grant rests on, asked by a run's opening
    /// and by the withdrawal that would put a closed run back in force.
    pub(crate) fn claim_holder(&self, run_key: &str, claim: &PipelineRef) -> Option<&str> {
        self.of_kind(PipelineKind::Run).find_map(|(other_key, document)| match document {
            PipelineDocument::Run(other)
                if other_key != run_key && other.claims.contains(claim) && self.in_force(PipelineKind::Run, other_key) =>
            {
                Some(other_key)
            }
            _ => None,
        })
    }

    /// The in-force run of hers, other than `run_key`, that holds the slot of
    /// the turn `run` is of (a Persona session and a Sleep run share the
    /// Persona slot): ruling one-live-self-run's one live run per
    /// instance and slot, asked by a run's opening and by the withdrawal that
    /// would put a closed run back in force. An operator's run neither holds
    /// the slot nor is refused for it.
    pub(crate) fn live_holder(&self, run_key: &str, run: &PipelineRun) -> Option<&str> {
        if run.operated_by == RunOperator::Operator {
            return None;
        }
        self.of_kind(PipelineKind::Run).find_map(|(other_key, document)| match document {
            PipelineDocument::Run(other)
                if other_key != run_key
                    && other.instance == run.instance
                    && other.turn.slot() == run.turn.slot()
                    && other.operated_by == RunOperator::Mind
                    && self.in_force(PipelineKind::Run, other_key) =>
            {
                Some(other_key)
            }
            _ => None,
        })
    }

    /// The stewardship of `(mind, repo)` a hand-off acts on: the one whose
    /// withdrawal already carries `hand_off_key`, when image or batch holds
    /// that withdrawal, so a replay derives the same withdrawal again and A9
    /// answers with the stored receipt; otherwise the one in force. Key order
    /// cannot pick between the two the exclusion leaves standing -- once a
    /// repo has changed hands ten times, `n11` sorts before `n2` -- so the
    /// content match picks, as it does for the sequence the other derivations
    /// take.
    pub(crate) fn stewardship_of(
        &self,
        mind: &Slug,
        repo: &OrgRepo,
        hand_off_key: Option<&str>,
    ) -> Option<(&str, &PipelineStewardship)> {
        let candidates = self.stewardships_of(mind, repo, hand_off_key);
        hand_off_key
            .and_then(|hand_off| {
                candidates.iter().find(|(key, _)| {
                    self.resolutions_naming(PipelineKind::Stewardship, key).any(|(_, resolution)| {
                        matches!(&resolution.outcome, ResolutionOutcome::Withdrawn { reason } if reason.0 == hand_off)
                    })
                })
            })
            .or_else(|| candidates.first())
            .copied()
    }
}

/// The line a record belongs to (rulings `supersession-lineage-2` and `-3`): the one place
/// lineage is decided. A record supersedes, and is reinstated over, only a record
/// of its own line. `None` is a kind the resolution matrix never lets be
/// superseded; the match names every kind, so a new kind fails to compile until
/// its line is stated.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum Lineage<'a> {
    /// A target: the campaign.
    Campaign(&'a Slug),
    /// A cut spec: the campaign and the cut.
    Cut(&'a Slug, &'a Label),
    /// A stewardship: the instance and the repo.
    Assignment(&'a Slug, &'a OrgRepo),
    /// A memory: the instance whose mind holds it, so a consolidating memory
    /// replaces only her own.
    Memories(&'a Slug),
    /// A ruling, a follow-up, a research note or an idea: operator directions and re-homed work cut
    /// across campaigns.
    Global,
}

pub(crate) fn line_of(document: &PipelineDocument) -> Option<Lineage<'_>> {
    use PipelineDocument as D;
    match document {
        D::Target(target) => Some(Lineage::Campaign(&target.campaign)),
        D::CutSpec(spec) => Some(Lineage::Cut(&spec.campaign, &spec.cut)),
        D::Stewardship(stewardship) => Some(Lineage::Assignment(&stewardship.instance, &stewardship.repo)),
        D::Memory(memory) => Some(Lineage::Memories(&memory.instance)),
        D::Ruling(_) | D::FollowUp(_) | D::Research(_) | D::Idea(_) => Some(Lineage::Global),
        D::Campaign(_)
        | D::Question(_)
        | D::CutReport(_)
        | D::Verdict(_)
        | D::Finding(_)
        | D::Resolution(_)
        | D::Instance(_)
        | D::HandOff(_)
        | D::Run(_)
        | D::Pulse(_)
        | D::Reply(_)
        | D::Thought(_) => None,
    }
}

/// Whether `other` cannot stand in force beside `base`, being on `base`'s line:
/// any other stewardship of an instance and repo, a later revision of a target
/// or cut spec. A resolution is never a base here: the cap in `resolution_rule`
/// refuses the withdrawal that would reinstate one first.
fn rivals(base: &PipelineDocument, other: &PipelineDocument) -> bool {
    use PipelineDocument as D;
    let Some(line) = line_of(base) else { return false };
    if line_of(other) != Some(line) {
        return false;
    }
    match (base, other) {
        (D::Stewardship(_), D::Stewardship(_)) => true,
        (D::Target(base), D::Target(other)) => other.revision > base.revision,
        (D::CutSpec(base), D::CutSpec(other)) => other.revision > base.revision,
        _ => false,
    }
}

/// The referring field of a citation edge, one per leaf field that names
/// another document. The wire spelling is the field's own name, so a client
/// reads an edge's role as the field it came from.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum CitationRole {
    RaisedIn,
    Answers,
    Rulings,
    Questions,
    CutSpec,
    Forks,
    CutReport,
    Findings,
    Verdict,
    Source,
    Subject,
    SupersededBy,
    ResolvedBy,
    DeferredTo,
    Documents,
    Claims,
    Consolidates,
    Serves,
    Sources,
    RestsOn,
    Toward,
    Run,
    Reads,
    DayThrough,
}

impl CitationRole {
    pub const ALL: [CitationRole; 24] = [
        Self::RaisedIn,
        Self::Answers,
        Self::Rulings,
        Self::Questions,
        Self::CutSpec,
        Self::Forks,
        Self::CutReport,
        Self::Findings,
        Self::Verdict,
        Self::Source,
        Self::Subject,
        Self::SupersededBy,
        Self::ResolvedBy,
        Self::DeferredTo,
        Self::Documents,
        Self::Claims,
        Self::Consolidates,
        Self::Serves,
        Self::Sources,
        Self::RestsOn,
        Self::Toward,
        Self::Run,
        Self::Reads,
        Self::DayThrough,
    ];

    /// The wire spelling, equal to the referring field's name.
    pub fn name(self) -> &'static str {
        match self {
            Self::RaisedIn => "raised_in",
            Self::Answers => "answers",
            Self::Rulings => "rulings",
            Self::Questions => "questions",
            Self::CutSpec => "cut_spec",
            Self::Forks => "forks",
            Self::CutReport => "cut_report",
            Self::Findings => "findings",
            Self::Verdict => "verdict",
            Self::Source => "source",
            Self::Subject => "subject",
            Self::SupersededBy => "superseded_by",
            Self::ResolvedBy => "resolved_by",
            Self::DeferredTo => "deferred_to",
            Self::Documents => "documents",
            Self::Claims => "claims",
            Self::Consolidates => "consolidates",
            Self::Serves => "serves",
            Self::Sources => "sources",
            Self::RestsOn => "rests_on",
            Self::Toward => "toward",
            Self::Run => "run",
            Self::Reads => "reads",
            Self::DayThrough => "day_through",
        }
    }

    pub fn from_name(name: &str) -> Option<Self> {
        Self::ALL.into_iter().find(|role| role.name() == name)
    }

    /// The kind whose documents carry this field.
    pub fn carrier(self) -> PipelineKind {
        use PipelineKind as K;
        match self {
            Self::RaisedIn => K::Question,
            Self::Answers => K::Ruling,
            Self::Rulings | Self::Questions => K::CutSpec,
            Self::CutSpec | Self::Forks => K::CutReport,
            Self::CutReport | Self::Findings => K::Verdict,
            Self::Verdict => K::Finding,
            Self::Source => K::FollowUp,
            Self::Subject | Self::SupersededBy | Self::ResolvedBy | Self::DeferredTo | Self::DayThrough => K::Resolution,
            Self::Documents => K::HandOff,
            Self::Claims => K::Run,
            Self::Consolidates => K::Memory,
            Self::Serves | Self::Sources => K::Research,
            Self::RestsOn | Self::Toward => K::Idea,
            Self::Run => K::Reply,
            Self::Reads => K::Thought,
        }
    }

    /// The kinds a referent of this field may be: the one kind the field is
    /// typed to, or every kind for the fields typed `PipelineRef` or a full
    /// id of any kind. "Any" is spelled as every kind, never as none.
    pub fn target_kinds(self) -> Vec<PipelineKind> {
        use PipelineKind as K;
        match self {
            Self::Answers | Self::Questions | Self::Forks => vec![K::Question],
            Self::Rulings => vec![K::Ruling],
            Self::CutSpec => vec![K::CutSpec],
            Self::CutReport => vec![K::CutReport],
            Self::Findings => vec![K::Finding],
            Self::Verdict => vec![K::Verdict],
            Self::Claims => vec![K::CutSpec, K::Finding, K::FollowUp],
            Self::Consolidates | Self::DayThrough => vec![K::Memory],
            Self::Run => vec![K::Run],
            Self::RaisedIn
            | Self::Source
            | Self::Subject
            | Self::SupersededBy
            | Self::ResolvedBy
            | Self::DeferredTo
            | Self::Documents
            | Self::Serves
            | Self::Sources
            | Self::RestsOn
            | Self::Toward
            | Self::Reads => PipelineKind::ALL.to_vec(),
        }
    }
}

/// The kind segment of a full id, for the fields typed `Short` whose referent
/// may be of any kind.
pub(crate) fn kind_of_id(id: &str) -> Option<PipelineKind> {
    let name = id.split(':').nth(1)?;
    PipelineKind::ALL.iter().copied().find(|kind| kind.name() == name)
}

fn cited(kind: PipelineKind, id: &Short) -> PipelineRef {
    PipelineRef { kind, id: id.clone() }
}

/// What a resolution's outcome names, by the field that names it.
pub(crate) fn outcome_citations(outcome: &ResolutionOutcome) -> Vec<(CitationRole, PipelineRef)> {
    match outcome {
        ResolutionOutcome::Superseded { by } => by.iter().map(|by| (CitationRole::SupersededBy, by.clone())).collect(),
        ResolutionOutcome::Answered { by } => vec![(CitationRole::ResolvedBy, by.clone())],
        ResolutionOutcome::Fixed { by: Some(by), .. } => vec![(CitationRole::ResolvedBy, by.clone())],
        ResolutionOutcome::Deferred { to } => vec![(CitationRole::DeferredTo, to.clone())],
        ResolutionOutcome::Slept { day_through, .. } => day_through.iter().map(|memory| (CitationRole::DayThrough, memory.clone())).collect(),
        ResolutionOutcome::Fixed { by: None, .. }
        | ResolutionOutcome::Recorded { .. }
        | ResolutionOutcome::Withdrawn { .. }
        | ResolutionOutcome::Moved { .. } => {
            vec![]
        }
    }
}

/// Every citation edge a document carries, derived from its leaf fields each
/// time it is asked and never stored: the one edge list. Admission's A7 reads
/// it to require the referents, and the read side's rows read it as their
/// references. `Fixed.commit`, `ForeignRef` precedents and the labels
/// (`depends_on`, `invariants`, `promise`) are not edges. A hand-off's
/// `documents` entry whose kind segment does not read cites nothing here;
/// admission refuses such an entry on the source side before it asks.
pub(crate) fn citations(document: &PipelineDocument) -> Vec<(CitationRole, PipelineRef)> {
    use CitationRole as R;
    use PipelineDocument as D;
    use PipelineKind as K;
    let mut edges = Vec::new();
    match document {
        D::Question(question) => edges.extend(question.raised_in.iter().map(|to| (R::RaisedIn, to.clone()))),
        D::Ruling(ruling) => edges.extend(ruling.answers.iter().map(|to| (R::Answers, cited(K::Question, to)))),
        D::CutSpec(spec) => {
            edges.extend(spec.rulings.iter().map(|to| (R::Rulings, cited(K::Ruling, to))));
            edges.extend(spec.questions.iter().map(|to| (R::Questions, cited(K::Question, to))));
        }
        D::CutReport(report) => {
            edges.push((R::CutSpec, cited(K::CutSpec, &report.cut_spec)));
            edges.extend(report.forks.iter().map(|to| (R::Forks, cited(K::Question, to))));
        }
        D::Verdict(verdict) => {
            edges.push((R::CutReport, cited(K::CutReport, &verdict.cut_report)));
            edges.extend(
                verdict
                    .claims
                    .iter()
                    .flat_map(|claim| claim.findings.iter())
                    .map(|to| (R::Findings, cited(K::Finding, to))),
            );
        }
        D::Finding(finding) => edges.push((R::Verdict, cited(K::Verdict, &finding.verdict))),
        D::FollowUp(follow_up) => edges.push((R::Source, follow_up.source.clone())),
        D::Resolution(resolution) => {
            edges.push((R::Subject, resolution.subject.clone()));
            edges.extend(outcome_citations(&resolution.outcome));
        }
        D::HandOff(hand_off) => edges.extend(
            hand_off.documents.iter().filter_map(|to| kind_of_id(&to.0).map(|kind| (R::Documents, cited(kind, to)))),
        ),
        D::Run(run) => edges.extend(run.claims.iter().map(|to| (R::Claims, to.clone()))),
        D::Reply(reply) => edges.push((R::Run, reply.run.clone())),
        D::Thought(thought) => edges.extend(thought.reads.iter().map(|to| (R::Reads, to.clone()))),
        D::Memory(memory) => edges.extend(memory.consolidates.iter().map(|to| (R::Consolidates, to.clone()))),
        D::Research(research) => {
            edges.extend(research.serves.iter().map(|to| (R::Serves, to.clone())));
            edges.extend(research.sources.iter().filter_map(|source| match source {
                ResearchSource::Document { reference } => Some((R::Sources, reference.clone())),
                _ => None,
            }));
        }
        D::Idea(idea) => {
            edges.extend(idea.rests_on.iter().map(|to| (R::RestsOn, to.clone())));
            edges.extend(idea.toward.iter().map(|to| (R::Toward, to.clone())));
        }
        D::Campaign(_) | D::Target(_) | D::Instance(_) | D::Stewardship(_) | D::Pulse(_) => {}
    }
    edges
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fixtures::*;
    use PipelineDocument as D;
    use PipelineKind as K;

    fn named(campaign: &str, revision: u32, repos: &[&str]) -> (String, D) {
        let D::Target(mut target) = target(revision, &[INVARIANT]) else { panic!("a target") };
        target.campaign = campaign.into();
        target.repos = repos.iter().map(|name| repo(name)).collect();
        (format!("{campaign}:target:r{revision}"), D::Target(target))
    }

    /// Membership is the highest-revision target of this campaign that is in
    /// force and names repos. The image is built by hand because admission
    /// cannot reach it: a revision above 1 supersedes its predecessor in its
    /// own batch, a target is superseded only within its campaign, and an
    /// empty revision after a repo-naming one is refused, so through admission
    /// a closed repo-naming target is always older than the in-force one.
    ///
    /// Taking the lowest revision instead of the highest changes nothing there
    /// either (one target per campaign is in force), so that mutation is
    /// equivalent and not pinned.
    #[test]
    fn repos_in_force_is_this_campaigns_in_force_naming_target() {
        let D::Campaign(this) = campaign(&[REPO]) else { panic!("a campaign") };
        let (in_force_key, in_force) = named(CAMPAIGN, 1, &[REPO, OTHER_REPO]);
        let (closed_key, closed) = named(CAMPAIGN, 2, &["GameCult/Closed"]);
        let (foreign_key, foreign) = named("other-campaign", 3, &["GameCult/Foreign"]);
        let closing = resolution(r(K::Target, &closed_key), superseded(&[r(K::Target, &in_force_key)]));
        let image = Image::of_documents(vec![
            (K::Target, in_force_key, in_force),
            (K::Target, closed_key, closed),
            (K::Target, foreign_key, foreign),
            (K::Resolution, id("resolution", "target.r2.n1"), closing),
        ]);
        let docs = Docs::at(&image, image.head);
        assert_eq!(docs.repos_in_force(&this), [repo(REPO), repo(OTHER_REPO)], "neither the closed higher revision nor another campaign's higher target decides");
    }
}