onetaskgraph_core/engine/

copy.rs

//! The copy verb: one item out of one source and into another, by the rules that make a
//! second copy an update rather than a duplicate.
//!
//! Correspondence lives on the items and never in a table. A copied item carries
//! [`GlobalId::ORIGIN_KEY`], whose value is the qualified id it was copied from, and the
//! item it was copied from carries [`MetadataKey::COPIES_KEY`], an object naming its
//! counterpart at each destination it was copied into. The rules below read exactly those —
//! so nothing here is written down outside the plugin that owns the item, and the invariant
//! this engine is built around is untouched.
//!
//! 0. **Follow the link.** An item whose `onetaskgraph.copies` names an item at the
//!    destination has that item read by id, and when that item's own origin names this one
//!    back it is the target: one read, and no scan. A link naming nothing there any more is
//!    refused as `stale-link` unless `--recreate` is given, for the reason a stale origin
//!    is; a link naming an item that no longer names this one back — somebody re-pointed it
//!    — is ignored, and the rules below decide.
//! 1. **Follow the origin.** An item already carrying an origin whose source half is the
//!    destination names the destination item *directly*, and the copy updates it. This is
//!    the half that makes an edit's copy-back an update: the local file came from the
//!    remote item and knows which one.
//! 2. **Search by origin.** Otherwise the destination is scanned, one page at a time, for
//!    an item whose origin is the id being copied. Found, the copy updates it; not found,
//!    the copy creates one carrying that origin.
//!
//! Which rule found the item decides what the copy records there. A copy that got its
//! target from rule 1 is a copy-back — the destination is the *original*, and the item
//! being copied is the one that came from it — so the destination keeps the origin it
//! already holds, holding none included. Every other copy records the id it was copied
//! from. See [`recorded`] for what stamping a copy-back's own id there costs.
//!
//! The same distinction decides the link. Once the whole copy has landed, every item whose
//! target was found by any rule but rule 1 — or was created — has its link for this
//! destination recorded or refreshed through its source's narrow metadata write, and that
//! write joins the undo journal like every destination write. Rule 1 records nothing: that
//! correspondence is already written down, on the destination item.
//!
//! A destination write is at the user's explicit request, names its destination, goes
//! through that source's own write interface into that source's own store, and is never
//! read back to answer a query. That is what makes it a write and not a cache.

use std::collections::{BTreeMap, BTreeSet};

use onetaskgraph_plugin_api::{
    AssetPayload, AssetUploads, AssetWrite, Capabilities, Cursor, DependencyEdge,
    DependencyEndpoint, DependencyKind, Direction, Document, DocumentQuery, ItemKind, ItemWrite,
    Location, MetadataKey, MetadataMatch, Metering, NativeId, Page, PageRequest, Project,
    ProjectQuery, Repository, SourceError, SourceName, StatusCategory, Task, TaskQuery, TaskRef,
    TextFields, TextQuery,
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::GlobalId;
use crate::config::Placement;
use crate::resolve::ResolvedSource;
use crate::template::{Sha256Digest, TemplateProvenance, body_digest};

use super::assets;
use super::delivery::{Delivered, targets};
use super::fetch::{fits, unrepeated};
use super::local::ProjectSelector;
use super::narrow::holds_priority;
use super::{
    DocumentFilters, DocumentRequest, Engine, EngineError, Filters, LeftBehind, Paging, Qualified,
    TaskRequest,
};

/// A request to copy work into one configured destination.
#[derive(Debug, Clone)]
pub struct CopyRequest {
    /// The qualified items to copy, in the order they were named.
    pub items: CopyItems,
    /// What those ids name, and what comes with them.
    pub scope: CopyScope,
    /// The configured source to copy into — a source name, never a qualified id.
    pub destination: SourceName,
    /// How to re-establish a correspondence the two origin rules cannot find.
    pub match_by: Option<MatchBy>,
    /// Whether an origin naming nothing at the destination falls through to the search
    /// rule instead of refusing.
    pub recreate: bool,
    /// Whether the caller asserts the destination holds no counterpart of any item named,
    /// so each is created there without the correspondence lookup — no origin search of the
    /// destination at all.
    ///
    /// Sound for a caller that has just asked the destination itself: the origin query is
    /// an index that lags a write, so repeating it here would be behind by exactly as much as
    /// the caller's own was and could find nothing the caller's did not. Refused beside
    /// [`match_by`](Self::match_by) and [`recreate`](Self::recreate), which are ways of
    /// looking, and for an item that itself records a counterpart at the destination — a
    /// link or an origin naming it — which is a carrier the caller's assertion overlooked.
    // llmlint: ignore-block[invalid_states_unrepresentable] One enum of the ways a copy finds its target would make `create` beside `match_by` or `recreate` unrepresentable, but only by replacing `match_by` and `recreate`, which every Rust caller of this request sets by name and the command line maps flag for flag — rewriting what each existing caller writes. `create` is one more field beside them, as `recreate` and `dry_run` were, and the one combination it makes possible that means nothing is refused by `Engine::copy`, as `EngineError::CreateWith`, before anything is read.
    pub create: bool,
    // llmlint: ignore-end[invalid_states_unrepresentable]
    /// Whether to perform every read and no write.
    pub dry_run: bool,
}

/// A way of looking for a counterpart that [`CopyRequest::create`] says not to take, named as
/// the command line spells it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CopyLookup {
    /// [`CopyRequest::match_by`], `--match-by`.
    MatchBy,
    /// [`CopyRequest::recreate`], `--recreate`.
    Recreate,
}

impl std::fmt::Display for CopyLookup {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(match self {
            Self::MatchBy => "--match-by",
            Self::Recreate => "--recreate",
        })
    }
}

/// The items one copy names: at least one, because a copy naming none is not a copy.
///
/// A newtype rather than a bare `Vec`, for the reason [`Repository`] is one: the empty
/// list is not a copy of nothing, it is a caller mistake, and a type that can hold it
/// leaves every reader to decide what it means — a report with no entries, an error, a
/// silent success. None of those is better than not being able to say it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CopyItems(Vec<GlobalId>);

impl CopyItems {
    /// The items a caller named, or `None` when they named none.
    #[must_use]
    pub fn new(items: Vec<GlobalId>) -> Option<Self> {
        (!items.is_empty()).then_some(Self(items))
    }

    /// The items, in the order they were named.
    #[must_use]
    pub fn as_slice(&self) -> &[GlobalId] {
        &self.0
    }
}

/// What the ids a copy names are, and what travels with them.
///
/// One value rather than a kind beside a flag, because three of the four combinations
/// those two would make are real and the fourth — tasks, with the tasks of each also
/// copied — means nothing. The member list is a variant for the same reason: it narrows a
/// project copy that carries its tasks, and it has nothing to say to the other three.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CopyScope {
    /// The ids name tasks, and only those tasks are copied.
    Tasks,
    /// The ids name projects.
    Projects {
        /// Whether the tasks in each project are copied too.
        tasks: bool,
    },
    /// The ids name projects, and of the tasks in them exactly these are copied.
    ///
    /// A member the list does not name is not read at the destination, not compared, not
    /// written and not reported, and no walk for what the copy left behind runs. A named
    /// member with no counterpart there is still created: the list narrows which members
    /// are read and written, never which outcomes are possible.
    ///
    /// An edge from a copied item to a member the list does not name resolves to the
    /// destination id that member's own [`GlobalId::ORIGIN_KEY`] records at the source,
    /// without reading the destination for it — or, for a copy into a source with `routes`,
    /// the id its own `onetaskgraph.copies` link records in any source the copy reaches.
    /// When that member records neither, the copy is refused before anything is written.
    Members(CopyItems),
    /// The ids name documents, and only those documents are copied.
    ///
    /// Nothing travels with a document: it takes part in no dependency graph, and it holds
    /// nothing of its own the way a project holds tasks.
    Documents,
}

/// The caller-named escape for a correspondence neither origin rule can find.
///
/// A person editing Markdown who deletes or corrupts the origin key leaves an item rule 1
/// cannot use and rule 2 cannot find, and the next copy would create a second item. This
/// is how that is re-established without hand-editing ids.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MatchBy {
    /// Match the item whose title is the same.
    Title,
    /// Match the item whose value at this metadata key is the same.
    Metadata(String),
}

impl MatchBy {
    /// The spelling a caller types, `title` or any metadata key.
    #[must_use]
    pub fn parse(key: &str) -> Self {
        if key == "title" {
            Self::Title
        } else {
            Self::Metadata(key.to_owned())
        }
    }
}

/// What a copy did, one entry per item.
///
/// The same per-item outcomes reach every consumer: the machine-readable output renders
/// this, the rendered output renders this, and a Rust caller is handed it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct CopyReport {
    /// One entry per item the copy considered, in the order it considered them.
    pub items: Vec<CopyOutcome>,
    // Three flat scalars rather than one nested object, and a `!skip_serializing_if` beside
    // every skip: both are load-bearing for the generated SDKs rather than matters of
    // taste, and AGENTS.md's note on what a copied document's references are pointed at is
    // where that reasoning lives.
    /// Reference occurrences the copy rewrote to the destination's own location for the
    /// record they name.
    ///
    /// A silent bound is indistinguishable from a bug, so the copy says what it did to the
    /// references the documents it carried hold. This and the two below are totals over the
    /// whole invocation rather than figures per document, and all three default to zero, so
    /// a consumer written against the output before they existed is unaffected.
    ///
    /// **What these figures do not claim.** The referent set is a document's own project,
    /// so a reference to a record in a *different* project is never recognised at all and
    /// cannot appear in [`Self::references_unresolved`] either. These are the references
    /// the copy recognised; they are not a census of every reference a document holds.
    /// Noticing an out-of-scope reference would need exactly the unbounded destination walk
    /// this design refuses.
    #[serde(default, skip_serializing_if = "nothing_to_report")]
    #[schemars(!skip_serializing_if)]
    pub references_rewritten: u64,
    /// Reference occurrences the copy recognised and left byte-for-byte as they were,
    /// because the correspondence could not be established.
    #[serde(default, skip_serializing_if = "nothing_to_report")]
    #[schemars(!skip_serializing_if)]
    pub references_unresolved: u64,
    /// How many of [`Self::references_unresolved`] were left alone because the
    /// correspondence was **ambiguous** rather than merely absent. A sub-count, never
    /// larger than it.
    ///
    /// Split out because the two mean different things to a reader. A reference with no
    /// counterpart is ordinary and expected under the bound above — the design working. An
    /// ambiguous one says the destination holds duplicate records for one work item, or the
    /// source reports one location for two records, and re-running the copy will never
    /// clear it.
    #[serde(default, skip_serializing_if = "nothing_to_report")]
    #[schemars(!skip_serializing_if)]
    // llmlint: ignore[invalid_states_unrepresentable] JSON Schema cannot express an
    // inequality between two numbers, so a private constructor here would hold this in one
    // consumer of three while both SDKs' generated models went on admitting it. What holds
    // it is `substitute`: `Resolution` has no variant that counts an occurrence ambiguous
    // without counting it unresolved.
    pub references_ambiguous: u64,
    /// `delivers` entries the copy rewrote to the destination's own id for a member of the
    /// copied set, over the whole invocation. Every other entry arrives qualified and is not
    /// counted. Left out when zero, as the three figures above are.
    #[serde(default, skip_serializing_if = "nothing_to_report")]
    #[schemars(!skip_serializing_if)]
    pub delivers_rewritten: u64,
    /// One entry per delivered task the copy kept in step with a task it landed, after the
    /// whole copy was complete — see `task status set`, which reports the same entries. Left
    /// out when there were none.
    ///
    /// A failed entry does not undo the copy: the tasks it landed stay landed, and the
    /// command exits `4`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    #[schemars(!skip_serializing_if)]
    pub delivered: Vec<Delivered>,
    /// What this copy spent, summed over the sources in the command that meter their own
    /// requests — and absent, never zero, when none of them does.
    ///
    /// Omitted from the wire when absent, like the three figures above, so a consumer
    /// written against the output before it existed reads the same document.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub spent: Option<Spent>,
}

/// Whether one of [`CopyReport`]'s reference figures has anything to say.
///
/// A copy that recognised no reference reports that by leaving the figure out rather than
/// by writing a nought, so the machine output of a task or project copy is exactly what it
/// was before these figures existed. The human rendering says it in words either way,
/// because a reader there needs to be told the copy looked.
fn nothing_to_report(figure: &u64) -> bool {
    *figure == 0
}

/// What one command spent, summed over the sources in it that meter their own requests.
///
/// **Source-owned.** Every figure is what a source said it sent and spent while the command
/// ran, read through [`TaskSource::metering`](onetaskgraph_plugin_api::TaskSource::metering)
/// before the command and again after it. The engine adds the differences up by name and
/// interprets none of them, which is why the budget and unit names are open vocabulary.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct Spent {
    /// How many HTTP requests those sources sent for this command.
    pub requests: u64,
    /// What those requests spent, one entry per budget and unit, ordered by budget name.
    pub budgets: Vec<BudgetSpent>,
}

/// What one command spent against one budget.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct BudgetSpent {
    /// The budget, as the source names it — `graphql`, `rest`.
    // llmlint: ignore[invalid_states_unrepresentable] The contract this report lands states `budget` and `unit` as strings in an open vocabulary the engine interprets none of, and both SDKs are generated from this type's schema; the one value such a vocabulary can refuse, the empty name, never reaches here, because `difference` below treats a source reporting one as not metering.
    pub budget: String,
    /// What it is metered in — `points`, `requests`.
    // llmlint: ignore[invalid_states_unrepresentable] As `budget` above.
    pub unit: String,
    /// How much was spent against it, in that unit.
    pub amount: u64,
    /// Whether any part of `amount` was modelled by a source rather than reported by its
    /// backend or counted, which makes `amount` a lower bound on what the backend charged
    /// rather than a measurement of it.
    pub lower_bound: bool,
}

/// One document's reference figures, before they are folded into the invocation's.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct Counted {
    /// Occurrences rewritten.
    rewritten: u64,
    /// Occurrences recognised and left alone.
    unresolved: u64,
    /// How many of those were ambiguous.
    ambiguous: u64,
}

impl Counted {
    /// Fold one document's figures into the invocation's.
    fn add(&mut self, other: Self) {
        self.rewritten += other.rewritten;
        self.unresolved += other.unresolved;
        self.ambiguous += other.ambiguous;
    }
}

/// What happened to one item.
///
/// `action` and `destination` are one value rather than two fields side by side: an
/// updated item without a destination id, or an orphan without one, are states this type
/// must not be able to say — the id *is* what those outcomes are about. The one outcome
/// that legitimately has none is a dry run that would create, because nothing was
/// created and there is no id to report.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct CopyOutcome {
    /// The qualified id the item was read from.
    pub source: GlobalId,
    /// What happened to it, and where.
    #[serde(flatten)]
    pub action: CopyAction,
    /// Which source a routed copy placed it in, and the index of the route entry that
    /// matched — `null` there when none did.
    ///
    /// Present for every item a copy into a source declaring `routes` landed, a dry run's
    /// included, so a would-be create still says where it would go; absent for a copy into
    /// a source with none, whose output is what it always was, and for an orphan, which was
    /// not placed at all.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub placed: Option<Placement>,
}

/// Which rule found the destination counterpart an item was updated at, or found already
/// reading as it does.
///
/// Tried in this order, and the first that answers is the one reported: the link the item
/// records, its own origin, a search of the destination for an item recording it as its
/// origin, and the caller's `--match-by`. When none answers, the item is created and says
/// so as [`NoCounterpart`] instead.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum CopyVia {
    /// The item's `onetaskgraph.copies` entry for the destination named an item whose own
    /// origin names this one back. Found by one read by id, and no scan.
    Link,
    /// The item's own `onetaskgraph.origin` named the destination item: a copy-back.
    Origin,
    /// A search of the destination for an item recording this one as its origin.
    Scan,
    /// A search of the destination by the caller's `--match-by`.
    Match,
}

/// What a created item reports as its `via`: no rule found a counterpart.
///
/// One word, and a type of its own rather than a fifth [`CopyVia`], so an item updated at a
/// counterpart cannot say none was found and a created one cannot name a rule that found
/// one.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum NoCounterpart {
    /// No counterpart was found, so one was created — or, in a dry run, would have been.
    #[default]
    Created,
}

/// What a copy did to the `onetaskgraph.copies` entry the copied item holds for the
/// destination.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum CopyLink {
    /// The entry was written, because the item held none for this destination or held one
    /// naming another item.
    Recorded,
    /// Nothing was written: the entry already named this destination item, or the item was
    /// found by its own origin, where the correspondence is already recorded on the other
    /// side.
    Unchanged,
    /// The item's source cannot hold the entry — it has no write side, or it refuses the
    /// narrow metadata write — so the copy landed without recording it.
    Unrecorded,
}

impl CopyOutcome {
    /// The qualified id this outcome landed on, when it landed on one.
    #[must_use]
    pub fn destination(&self) -> Option<&GlobalId> {
        self.action.destination()
    }
}

/// The four things a copy can do to one item, and the id each of them is about.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "action", rename_all = "kebab-case")]
pub enum CopyAction {
    // llmlint: ignore[names_match_behavior] `created` is Contract D's serialized action
    // for both a completed create and a dry run that would create; the optional destination
    // distinguishes those cases, and renaming this public variant would break Rust callers.
    /// The destination held no counterpart, so one was created.
    Created {
        /// The id it was created under, or `null` for a dry run that would have created
        /// one — there is no id, because nothing was.
        destination: Option<GlobalId>,
        /// That no rule found a counterpart. The one word it can be, so a report written
        /// before there was a `via` reads as saying it.
        #[serde(default)]
        via: NoCounterpart,
        /// What the copy did to the link the copied item records for this destination at
        /// `onetaskgraph.copies`; absent for a dry run, which writes nothing.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        // llmlint: ignore[invalid_states_unrepresentable] Whether a copy is a dry run is a
        // property of the request, which the report does not carry, and the contract this
        // lands — shared with the follow-up tooling built against it — says a dry run leaves
        // `link` out rather than naming a sixth word for it. What holds it is `Engine::link`:
        // the one writer of this field, run for every item a copy that writes has landed and
        // for none of a dry run's, which `a_dry_run_says_which_rule_would_answer_and_records_no_link`
        // and every `link` assertion in tests/copy_link.rs drive.
        link: Option<CopyLink>,
    },
    /// The destination held a counterpart and it now reads as the source does.
    Updated {
        /// The item that was updated.
        destination: GlobalId,
        /// Which rule found it.
        via: CopyVia,
        /// What the copy did to the copied item's link; absent for a dry run.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        // llmlint: ignore[invalid_states_unrepresentable] As `link` on `Created` above.
        link: Option<CopyLink>,
    },
    /// The destination held a counterpart that already read that way; nothing was written.
    Unchanged {
        /// The item that already said it.
        destination: GlobalId,
        /// Which rule found it.
        via: CopyVia,
        /// What the copy did to the copied item's link; absent for a dry run.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        // llmlint: ignore[invalid_states_unrepresentable] As `link` on `Created` above.
        link: Option<CopyLink>,
    },
    /// The destination holds a counterpart the source no longer does. A copy never
    /// deletes, so it was left exactly as it is — and no rule was asked where it goes, and
    /// no link was touched, because it was not copied.
    Orphaned {
        /// The item that was left alone.
        destination: GlobalId,
    },
}

impl CopyAction {
    /// The qualified id this action is about, when there is one.
    #[must_use]
    pub fn destination(&self) -> Option<&GlobalId> {
        match self {
            Self::Created { destination, .. } => destination.as_ref(),
            Self::Updated { destination, .. }
            | Self::Unchanged { destination, .. }
            | Self::Orphaned { destination } => Some(destination),
        }
    }

    /// Which rule found the counterpart, or `None` when none did — the item was created —
    /// and for an orphan, which no rule was asked about.
    #[must_use]
    pub fn found_by(&self) -> Option<CopyVia> {
        match self {
            Self::Updated { via, .. } | Self::Unchanged { via, .. } => Some(*via),
            Self::Created { .. } | Self::Orphaned { .. } => None,
        }
    }

    /// What became of the copied item's link, or `None` for a dry run and for an orphan.
    #[must_use]
    pub fn link(&self) -> Option<CopyLink> {
        match self {
            Self::Created { link, .. }
            | Self::Updated { link, .. }
            | Self::Unchanged { link, .. } => *link,
            Self::Orphaned { .. } => None,
        }
    }

    /// The link slot of an item this copy landed, `None` for an orphan.
    fn link_slot(&mut self) -> Option<&mut Option<CopyLink>> {
        match self {
            Self::Created { link, .. }
            | Self::Updated { link, .. }
            | Self::Unchanged { link, .. } => Some(link),
            Self::Orphaned { .. } => None,
        }
    }

    /// The word this action serializes as, taken from its own `Serialize`.
    ///
    /// Read back off the wire form rather than written out again in a `match`, for the
    /// reason `render::wire` gives: a second spelling of `unchanged` would be a second
    /// place for it to drift from the one a caller reads.
    #[must_use]
    pub fn name(&self) -> String {
        serde_json::to_value(self).expect("a contract enum serialises")["action"]
            .as_str()
            .expect("an internally tagged enum carries its tag")
            .to_owned()
    }
}

/// What one item points at, resolved as far as the copy has got: its forward edges and the
/// tasks it delivers.
struct Pointing<'a> {
    /// Its forward edges, `None` where the far end is a member not landed yet.
    edges: &'a [Option<DependencyEdge>],
    /// The tasks it delivers that can already be named at the destination.
    delivers: &'a [TaskRef],
}

/// Where one item is going at the destination.
enum Target {
    /// Update the destination item with this id, reached by the rule named.
    Update {
        /// The destination item this copy updates.
        id: NativeId,
        /// Which rule found it.
        found: Found,
    },
    /// Create one.
    Create,
}

/// Which of the rules above found the destination item a copy is updating.
///
/// Every one is the same instruction — update that item — and rule 1 is a different answer
/// about the origin, which is why the distinction is carried this far rather than dropped
/// where it is made. See [`recorded`]. It is also what the report's [`CopyVia`] says and
/// what decides whether the copied item's link is written.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Found {
    /// The item's own link named it, and it names the item back.
    Link,
    /// Rule 1: the item being copied already named it, so this copy is a copy-back.
    Origin,
    /// Rule 2: the destination was searched for an item recording this one as its origin.
    Scan,
    /// The caller's matching escape.
    Match,
}

impl Found {
    /// The word the report says it in.
    fn via(self) -> CopyVia {
        match self {
            Self::Link => CopyVia::Link,
            Self::Origin => CopyVia::Origin,
            Self::Scan => CopyVia::Scan,
            Self::Match => CopyVia::Match,
        }
    }
}

impl Target {
    /// Which rule found this target, or `None` for one this copy creates.
    fn found(&self) -> Option<Found> {
        match self {
            Self::Update { found, .. } => Some(*found),
            Self::Create => None,
        }
    }
}

/// What a scan of the destination is looking for.
enum Wanted {
    /// An item recording this qualified id as its origin.
    Origin(String),
    /// An item whose title is this.
    Title(String),
    /// An item holding this value at this metadata key.
    Metadata(String, Value),
}

impl Wanted {
    /// Whether one destination item is the one being looked for.
    fn found(&self, title: &str, metadata: &BTreeMap<String, Value>) -> bool {
        match self {
            Self::Origin(id) => {
                metadata.get(GlobalId::ORIGIN_KEY) == Some(&Value::String(id.clone()))
            }
            Self::Title(wanted) => title == wanted,
            Self::Metadata(key, value) => metadata.get(key) == Some(value),
        }
    }

    /// The task query that asks `capabilities`' source for this item rather than for all of
    /// them, when that source applies the predicate itself.
    ///
    /// Only ever a narrowing the source declared it applies, so a source that cannot is
    /// asked for everything exactly as before; and never the last word, because
    /// [`found`](Self::found) still confirms each row. A text search is wider than an equal
    /// title, and a source may answer a match it cannot yet see as absent — which is the
    /// same answer the whole-store walk gave for an item its own listing was behind on.
    ///
    /// A title or a value with no letter or digit is never pushed down: a source whose search
    /// indexes words cannot answer it and may refuse it, and the walk it replaces answered it
    /// already, so it is asked for everything as it always was.
    fn task_query(&self, capabilities: &Capabilities) -> TaskQuery {
        match self {
            Self::Origin(id) if capabilities.filter_by_origin.is_native() => TaskQuery {
                origin: Some(id.clone()),
                ..TaskQuery::default()
            },
            Self::Title(title) if capabilities.search_title.is_native() && has_words(title) => {
                TaskQuery {
                    text: Some(TextQuery {
                        terms: title.clone(),
                        fields: TextFields::Title,
                    }),
                    ..TaskQuery::default()
                }
            }
            Self::Metadata(key, Value::String(value))
                if capabilities.filter_by_metadata.is_native() && has_words(value) =>
            {
                MetadataMatch::new(key.clone(), Vec::new(), value.clone())
                    .map(|wanted| TaskQuery {
                        metadata: vec![wanted],
                        ..TaskQuery::default()
                    })
                    .unwrap_or_default()
            }
            _ => TaskQuery::default(),
        }
    }
}

/// Whether `terms` holds a letter or a digit — a word a search that indexes words can find.
fn has_words(terms: &str) -> bool {
    terms.chars().any(char::is_alphanumeric)
}

/// What the destination held before this copy touched one item.
///
/// Read once, in [`Engine::land`], and used three times over: to decide whether the write
/// would change anything, to repair the item's edges once the rest of the copy has landed,
/// and — if the copy cannot finish — to put the item back exactly as it was.
#[derive(Clone)]
struct Prior {
    /// The item as the destination held it.
    item: Item,
    /// Its forward edges there.
    edges: Vec<DependencyEdge>,
    /// The image assets it held there, with the bytes of each its own record does not
    /// already say the destination serves — `None` when it held none and no write of this
    /// copy carried any, so putting it back writes it exactly as it did before assets.
    assets: Option<Vec<AssetPayload>>,
    /// What its `onetaskgraph.assets` says the destination serves, checked where it was read.
    recorded: Option<AssetUploads>,
}

/// One item that landed with an edge whose far end was not written yet.
///
/// Held until every item of the whole copy has landed, because the far end may be in
/// another project of the same command: a copy of two projects at once is one copied set,
/// not two, and an edge across them is remapped rather than written as a foreign id.
struct Deferred {
    /// The item, as it was read and resolved.
    item: Planned,
    /// The destination project it was filed under.
    filed: Option<NativeId>,
    /// Where it landed.
    destination: NativeId,
    /// What the destination held there before, when it held anything.
    prior: Option<Prior>,
}

/// What one item's undo has to do to put the destination back.
///
/// Each entry names the source it was written to, because a routed copy writes into more
/// than one, and it is undone in every one of them.
enum Undo {
    /// The copy created it, so undoing means removing it.
    Created {
        /// The source it was created in.
        at: SourceName,
        /// Which write interface removes it.
        kind: Level,
        /// The destination id it was created under.
        id: NativeId,
    },
    /// The copy overwrote something, so undoing means writing that something back.
    ///
    /// No `kind` beside the id, unlike the variant above: what was there says which of the
    /// two write interfaces takes it back, and a second spelling of that could disagree
    /// with it.
    Updated {
        /// The source it was overwritten in.
        at: SourceName,
        /// The destination id that was overwritten.
        id: NativeId,
        /// What was there before.
        prior: Prior,
    },
}

impl Undo {
    /// The source this entry was written to.
    fn at(&self) -> &SourceName {
        match self {
            Self::Created { at, .. } | Self::Updated { at, .. } => at,
        }
    }

    /// The destination id this entry is about.
    fn id(&self) -> &NativeId {
        match self {
            Self::Created { id, .. } | Self::Updated { id, .. } => id,
        }
    }

    /// Which of the destination's three write interfaces this entry belongs to.
    ///
    /// An id alone does not identify a destination item: nothing stops a destination
    /// numbering its tasks and its projects in one namespace, and a local-Markdown store
    /// filing `alpha.md` under both is the ordinary case rather than the contrived one.
    /// So this pairs with `id` wherever one entry has to be told from another.
    fn kind(&self) -> Level {
        match self {
            Self::Created { kind, .. } => *kind,
            // Read off what was there, for the reason the variant carries no `kind` of
            // its own: two spellings of one fact can disagree, and this one cannot.
            Self::Updated { prior, .. } => prior.item.level(),
        }
    }
}

/// Refuse a write of `item` whose status `destination` has no name for, before the write: a
/// document carries no status, so it asks nothing.
async fn check_status(
    destination: &ResolvedSource,
    item: &Item,
    target: Option<&NativeId>,
) -> Result<(), EngineError> {
    let (kind, category) = match item {
        Item::Task(task) => (ItemKind::Task, task.status.category),
        Item::Project(project) => (ItemKind::Project, project.status.category),
        Item::Document(_) => return Ok(()),
    };
    destination
        .source()
        .check_status_write(kind, category, target)
        .await
        .map_err(|error| refused(destination, error))
}

/// Everything one copy has written, in the order it wrote it, so a copy that cannot finish
/// can undo its own writes.
///
/// This is not state the engine keeps: it lives for the length of one `copy` call and is
/// dropped with it, so the invariant that nothing of a user's work is written down outside
/// the plugin that owns it is untouched.
#[derive(Default)]
struct Journal {
    /// One entry per destination item this copy first touched, in that order.
    entries: Vec<Undo>,
    /// One entry per copied item whose `onetaskgraph.copies` link this copy wrote, in that
    /// order.
    ///
    /// Kept apart from `entries` because these are the *sources'* items rather than the
    /// destination's, and an id alone does not say which store it is in: a copy inside one
    /// source can write a link on an item and land on another item with the same kind.
    links: Vec<Unlink>,
}

/// How to take back one link this copy wrote on a copied item.
struct Unlink {
    /// The item, at its own source.
    item: GlobalId,
    /// Which of that source's interfaces holds it.
    level: Level,
    /// What it held at `onetaskgraph.copies` before, or `None` when it held nothing there.
    before: Option<Value>,
}

impl Journal {
    /// Record what has to happen to put one destination item back.
    ///
    /// The *first* entry for an id is the one that matters and later ones are dropped: an
    /// item written twice — once as it lands, once when its edges are repaired — was only
    /// ever one thing before this copy started, and that is what undoing it restores.
    fn record(&mut self, entry: Undo) {
        if self.entries.iter().any(|held| {
            held.at() == entry.at() && held.kind() == entry.kind() && held.id() == entry.id()
        }) {
            return;
        }
        self.entries.push(entry);
    }
}

/// What one copy invocation carries from the item it lands to the next.
///
/// Like the [`Journal`] beside it, this lives for the length of one `copy` call and is
/// dropped with it: nothing here is state the engine keeps, and nothing in it is read back
/// to answer a later query.
#[derive(Default)]
struct Running {
    /// Every item an edge of this command resolves to a destination id rather than writing as
    /// it was read: the ids named, what travels with them, and — for a member copy — each
    /// member the copy does not carry whose own recorded origin already names its
    /// destination item. That last kind is resolvable without being copied.
    resolvable: Vec<GlobalId>,
    /// The destination item each of those corresponds to, as soon as it is known — whether
    /// this command found it, landed it, or read it off a member's recorded origin.
    ///
    /// Keyed by the qualified id's own rendering, which is what a recorded origin holds
    /// anyway — making `GlobalId` orderable for a local map would put an ordering on a
    /// contract type for a reason no caller of it has.
    ///
    /// Qualified, because a routed copy lands items in more than one source, and whether an
    /// edge to one is written bare or qualified depends on where both of its ends landed.
    counterparts: BTreeMap<String, GlobalId>,
    /// The items held back for [`Engine::repair`], across the whole request.
    deferred: Vec<Deferred>,
    /// The reference figures, one total for the whole invocation rather than one per
    /// document.
    references: Counted,
    /// The destination project each source project corresponds to, once this command has
    /// looked for it — so a second task filed under the same project does not walk the
    /// destination for it again.
    ///
    /// Keyed by the source project and the source the item lands in, because a routed task
    /// lands in its home's member project rather than in the home.
    filings: BTreeMap<(String, SourceName), Option<NativeId>>,
    /// Every home project this copy has learned the member projects of, by the home's
    /// qualified id: which members it holds, and whether this copy added one.
    homes: BTreeMap<String, Home>,
    /// `delivers` entries naming a member of the copied set, over the whole invocation.
    delivers_rewritten: u64,
    /// Every task this copy landed, for the relation rule once the whole copy is complete.
    landed: Vec<LandedTask>,
    /// Every item this copy landed, in the order it landed them, for the link each records
    /// once the whole copy has landed.
    linking: Vec<Linking>,
}

/// The member project a routed write was filed under, and what finding it wrote.
pub(super) struct Filed {
    /// The member project, or `None` when the home is not there to have one.
    pub(super) project: Option<NativeId>,
    /// The member created and the home's list rewritten, if either was.
    journal: Journal,
}

/// One home project, and the member projects it names at `onetaskgraph.members`.
struct Home {
    /// The home, where it landed.
    id: GlobalId,
    /// Its title and status, which a member created for it takes.
    title: String,
    /// See `title`.
    status: onetaskgraph_plugin_api::Status,
    /// The members it names, at most one per source.
    members: Vec<GlobalId>,
    /// Whether this copy added or replaced one, so the home's own list is written once the
    /// copy has landed.
    grew: bool,
}

impl Home {
    /// A home as its source holds it, with the members it already names.
    ///
    /// # Errors
    ///
    /// The home's own source refusing, as malformed, a member list [`members_of`] cannot
    /// read: a copy filing into a plan it cannot read whole would add a second member beside
    /// one it failed to see.
    fn of(id: GlobalId, held: &Project) -> Result<Self, EngineError> {
        let members =
            members_of(&id, &held.metadata).map_err(|message| EngineError::SourceRefused {
                name: id.source.to_string(),
                error: SourceError::Malformed { message },
            })?;
        Ok(Self {
            id,
            title: held.title.clone(),
            status: held.status.clone(),
            members,
            grew: false,
        })
    }
}

/// The member projects the home `home` records at [`MetadataKey::MEMBERS_KEY`]; none when it
/// records the key not at all.
///
/// # Errors
///
/// Why, when the key holds anything but a list of qualified ids, each in a different source
/// from the home and from the others — the one shape the routed write that keeps it writes.
pub(crate) fn members_of(
    home: &GlobalId,
    metadata: &BTreeMap<String, Value>,
) -> Result<Vec<GlobalId>, String> {
    let Some(value) = metadata.get(MetadataKey::MEMBERS_KEY) else {
        return Ok(Vec::new());
    };
    let refuse = |why: String| {
        format!(
            "{home} records {key} as {value}, which {why}; a home's member list is a list of \
             qualified project ids, at most one per source and none in the home's own",
            key = MetadataKey::MEMBERS_KEY
        )
    };
    let Value::Array(entries) = value else {
        return Err(refuse("is not a list".to_owned()));
    };
    let mut members: Vec<GlobalId> = Vec::new();
    for entry in entries {
        let member = entry
            .as_str()
            .and_then(|entry| entry.parse::<GlobalId>().ok())
            .ok_or_else(|| refuse(format!("holds {entry}, which is not a qualified id")))?;
        if member.source == home.source || members.iter().any(|kept| kept.source == member.source) {
            return Err(refuse(format!(
                "names {member}, a second project in {}",
                member.source
            )));
        }
        members.push(member);
    }
    Ok(members)
}

/// One item a copy landed, and what settling its `onetaskgraph.copies` link needs.
struct Linking {
    /// The item, at its own source.
    item: GlobalId,
    /// Which of that source's interfaces holds it.
    level: Level,
    /// Which rule found where it landed, or `None` when this copy created it.
    found: Option<Found>,
    /// Where it landed.
    destination: GlobalId,
    /// What it held at `onetaskgraph.copies` when it was read, if anything.
    held: Option<Value>,
}

/// One task a copy landed, and what the relation rule reads of it once the copy is complete.
struct LandedTask {
    /// Where it landed.
    destination: GlobalId,
    /// The source it was read from, which a bare `delivers` entry names a task of.
    origin: SourceName,
    /// Its `delivers`, as its source reported them.
    delivers: Vec<TaskRef>,
    /// The `delivers` the destination held there before this copy, qualified.
    before: Vec<GlobalId>,
    /// Its status category, as the copy wrote it.
    category: StatusCategory,
}

/// A task a copy landed, with the tasks it delivers now and delivered before, qualified.
struct Deliverer {
    destination: GlobalId,
    category: StatusCategory,
    now: Vec<GlobalId>,
    before: Vec<GlobalId>,
}

/// One item, read and resolved, on its way into the destination.
struct Planned {
    /// Where it came from.
    source: GlobalId,
    /// The item as its source reported it.
    item: Item,
    /// Its forward edges, as its source reported them.
    edges: Vec<DependencyEdge>,
    /// Where it is going.
    target: Target,
    /// What the destination held at that target, read once where the target was found.
    ///
    /// The read that decided an item exists is the read of what it holds, so the two are
    /// one round trip rather than two against a hosted destination.
    held: Option<Prior>,
    /// The source it lands in: the copy's destination, or the source a route sent it to.
    to: SourceName,
    /// What the report says placed it there, for a copy into a source with routes.
    placed: Option<Placement>,
    /// The image assets its content references, with their bytes as its source holds them,
    /// in the order the content first references them; none for a project.
    assets: Vec<AssetPayload>,
}

/// One item read out of its source, before anything decides where it lands.
struct Read {
    /// Where it came from.
    source: GlobalId,
    /// The item as its source reported it.
    item: Item,
}

/// A task, a project or a document, so the copy path is written once.
#[derive(Clone)]
enum Item {
    /// A task.
    Task(Box<Task>),
    /// A project.
    Project(Box<Project>),
    /// A document.
    Document(Box<Document>),
}

impl Item {
    fn id(&self) -> &NativeId {
        match self {
            Self::Task(task) => &task.id,
            Self::Project(project) => &project.id,
            Self::Document(document) => &document.id,
        }
    }

    fn level(&self) -> Level {
        match self {
            Self::Task(_) => Level::Task,
            Self::Project(_) => Level::Project,
            Self::Document(_) => Level::Document,
        }
    }
}

/// Which of a destination's three read-and-write interfaces one item belongs to.
///
/// Deliberately not [`ItemKind`]: that enum names what a *dependency endpoint* points at,
/// and the contract gives it no document variant because nothing may point at a document.
/// This one names which pair of methods reads and writes an item, which is a different
/// question with a third answer.
///
/// Ordered so it can key a map of what a destination holds, per interface: an id alone
/// does not identify a destination item, for the reason [`Undo::kind`] records.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(super) enum Level {
    /// `get_task`, `write_task`, `delete_task`.
    Task,
    /// `get_project`, `write_project`, `delete_project`.
    Project,
    /// `get_document`, `write_document`, `delete_document`.
    Document,
}

/// One record filed under a document's own project, and the location string its source
/// reports for it.
///
/// A reference is a **literal occurrence, in a document's content, of the exact location
/// string the source reports for a related record** — the `String` inside
/// [`Location::Path`] or [`Location::Url`]. Both ends of a rewrite come from the plugins'
/// own reported [`Location`]: nothing here composes an address out of a name, an id or a
/// root, because a source that reports a canonical absolute path and a source that reports
/// an issue link are the two things the contract lets this ask about.
#[derive(Clone)]
struct Referent {
    /// Its qualified id at the source.
    id: GlobalId,
    /// The origin it records in its own metadata, when it records one.
    origin: Option<GlobalId>,
    /// Which of the destination's three interfaces its counterpart would be read from.
    level: Level,
    /// The non-empty location string its source reports for it.
    location: String,
}

impl Referent {
    /// The two keys a destination record's own recorded origin is matched against.
    ///
    /// **A destination record is this referent's counterpart when its
    /// [`GlobalId::ORIGIN_KEY`] equals either this referent's own qualified source id, or
    /// the origin this referent itself records.** In plainer terms: when the destination
    /// record was copied directly from this referent, or directly from the one predecessor
    /// this referent itself records.
    ///
    /// That reach is exactly one recorded hop of ancestry on each side, and nothing more:
    ///
    /// - **A single hop resolves.** The destination record came straight from the referent.
    /// - **A one-level fan-out resolves.** A record copied from one store into two, with
    ///   the document arriving by one route and naming records that arrived by the other,
    ///   so both sides trace to one common predecessor. That is what the second key buys,
    ///   and it is the only thing it buys.
    /// - **A chain of two or more hops does not resolve**, on either side, and that is
    ///   permanent rather than pending: only one origin is ever recorded and every hop
    ///   overwrites it. [`carried`] removes the key from the incoming metadata outright and
    ///   [`recorded`] writes the id at the *immediate* source on every path but the
    ///   copy-back, so A → B → C leaves C keyed by B and A's id gone.
    ///
    /// The second key costs no read — the referent's metadata is already in hand. Chasing
    /// the chain further would need the intermediate stores configured and reachable, which
    /// would put a third party's availability inside a copy; a durable lineage id would
    /// identify only records written after it landed, so it would resolve nothing already
    /// on a destination.
    fn keys(&self) -> Vec<String> {
        let mut keys = vec![self.id.to_string()];
        if let Some(origin) = &self.origin {
            let recorded = origin.to_string();
            if recorded != keys[0] {
                keys.push(recorded);
            }
        }
        keys
    }
}

/// What every whole-reference occurrence of one location string becomes.
///
/// Three variants rather than a rewrite beside a flag, because the two ways of leaving an
/// occurrence alone are what the two figures a copy reports are *about*: one is the design
/// working and the other says the destination or the source holds something a re-run will
/// never fix. A bool would let a reader of this type read them as the same outcome.
enum Resolution {
    /// The destination's own location string for the counterpart.
    Rewrite(String),
    /// The destination holds no counterpart, or holds one it reports no location for, so
    /// the occurrence is left byte-for-byte. Ordinary and expected under the bound this
    /// design works to.
    NoCounterpart,
    /// The correspondence could not be established **confidently** — more than one
    /// destination record matches the referent's two keys, or two referents report this one
    /// location string. The occurrence is left byte-for-byte, no record is chosen, and a
    /// re-run will never clear it.
    Ambiguous,
}

/// Every destination record that records an origin, read **once per copy invocation**.
///
/// Several documents in one project is the ordinary case, and a walk per document would
/// multiply reads against a rate limiter for no gain — so this is built once, for the
/// levels the invocation's own documents really name, and every document and every
/// referent of that invocation is answered out of it. A copy whose documents hold no
/// candidate reference builds none at all.
///
/// **This is a stricter discipline than [`Engine::scan`], deliberately.** That lookup takes
/// the first hit and stops, and every consumer of the copy already depends on it doing so;
/// it chooses the copy's own target, where a caller named the item. This one edits the
/// content of somebody's document, where a wrong answer is silent corruption of prose a
/// person will act on — so where more than one record matches, it chooses none. The two
/// lookups answer different questions and are meant to disagree on a destination holding
/// duplicates.
#[derive(Default)]
struct Counterparts {
    /// The records at one interface recording one origin.
    by_origin: BTreeMap<(Level, String), Vec<Held>>,
}

/// One record a destination walk found: where it is there, and the location string that
/// destination reports for it — `None` when it reports none.
type Held = (NativeId, Option<String>);

impl Counterparts {
    /// Record one destination item, when it records an origin at all.
    fn note(
        &mut self,
        level: Level,
        id: &NativeId,
        location: Option<&Location>,
        metadata: &BTreeMap<String, Value>,
    ) {
        let Some(origin) = origin_of(metadata) else {
            return;
        };
        self.by_origin
            .entry((level, origin.to_string()))
            .or_default()
            .push((id.clone(), located(location)));
    }

    /// What one referent's occurrences become, by the two-key rule.
    ///
    /// Where the correspondence cannot be established **confidently**, the text is left
    /// exactly as it is and no record is chosen — see the note on this type.
    fn resolve(&self, referent: &Referent) -> Resolution {
        let mut candidates: Vec<&Held> = Vec::new();
        for key in referent.keys() {
            for record in self
                .by_origin
                .get(&(referent.level, key))
                .into_iter()
                .flatten()
            {
                // One destination record matching both keys is one record, not two.
                if !candidates.iter().any(|held| held.0 == record.0) {
                    candidates.push(record);
                }
            }
        }
        match candidates.as_slice() {
            [] => Resolution::NoCounterpart,
            // A counterpart the destination reports no location for names nowhere a reader
            // could go, so the source's own string is left standing rather than removed.
            [(_, location)] => location
                .clone()
                .map_or(Resolution::NoCounterpart, Resolution::Rewrite),
            _ => Resolution::Ambiguous,
        }
    }
}

impl Engine {
    /// Copy every item a request names into one configured destination.
    ///
    /// This is the whole of the verb, and the command line drives exactly this: a copy a
    /// Rust caller makes and a copy typed at a shell are the same call, so the two cannot
    /// answer the same copy differently.
    ///
    /// # Errors
    ///
    /// Returns [`EngineError`] when the destination is not configured, cannot be built,
    /// cannot be written, or — for a document copy — declares it has no documents; when an
    /// id names nothing; when an origin names an item the
    /// destination no longer holds and `--recreate` was not given; and when the
    /// destination refuses the write — including a field or a metadata key it cannot
    /// carry, which it names rather than dropping.
    pub async fn copy(&self, request: &CopyRequest) -> Result<CopyReport, EngineError> {
        // Before anything is read: `--create` says there is nothing to look for, and the other
        // two say how to look.
        if request.create {
            for (given, flag) in [
                (request.match_by.is_some(), CopyLookup::MatchBy),
                (request.recreate, CopyLookup::Recreate),
            ] {
                if given {
                    return Err(EngineError::CreateWith { flag });
                }
            }
        }
        let destination = self.writable(&request.destination)?;
        // Before anything is read, and from the declaration rather than from a failed
        // write: a destination that says it has no documents has nowhere to put one.
        if request.scope == CopyScope::Documents {
            documentary(destination)?;
        }
        // The sources this command names, each read once before and once after, so what it
        // spent is the difference between two of each one's own running totals.
        let mut metered = vec![destination];
        for routed in self.routes.reachable(&request.destination) {
            if let Some(source) = self.ready().find(|source| source.name() == &routed)
                && !metered.iter().any(|held| held.name() == source.name())
            {
                metered.push(source);
            }
        }
        for id in request.items.as_slice() {
            if let Some(source) = self.ready().find(|source| source.name() == &id.source)
                && !metered.iter().any(|held| held.name() == source.name())
            {
                metered.push(source);
            }
        }
        let before = readings(&metered).await;
        let mut journal = Journal::default();
        match self.copy_all(request, &mut journal).await {
            Ok((mut report, deliverers)) => {
                // After the whole copy is complete and outside the journal: what the relation
                // rule writes is the delivered tasks' own, and a failure there is reported
                // against that task rather than undoing a copy that landed.
                for deliverer in &deliverers {
                    report.delivered.extend(
                        self.deliver(
                            &deliverer.destination,
                            deliverer.category,
                            &deliverer.now,
                            &deliverer.before,
                        )
                        .await,
                    );
                }
                report.spent = spent_between(&before, &readings(&metered).await);
                Ok(report)
            }
            Err(error) => Err(self.undo(journal, error).await),
        }
    }

    /// The copy itself, with everything it writes recorded so a failure can be undone.
    ///
    /// The ids named together are **one** copied set, and that is what makes an edge
    /// between any two of them a real edge at the destination: a copy of two projects at
    /// once knows that a task in the first depends on a task in the second, and a task
    /// knows that the project it belongs to is being created beside it. Copying them one
    /// at a time could not, and wrote the far end as the id it had at its *source* — a
    /// dangling reference to somewhere the destination has never heard of.
    async fn copy_all(
        &self,
        request: &CopyRequest,
        journal: &mut Journal,
    ) -> Result<(CopyReport, Vec<Deliverer>), EngineError> {
        let mut running = Running::default();
        // The whole copied set, established before anything is written. For a project
        // copy that means reading every named project's membership first: the set is the
        // whole request rather than one project of it.
        let items = match &request.scope {
            CopyScope::Tasks | CopyScope::Documents => {
                let kind = if request.scope == CopyScope::Documents {
                    Level::Document
                } else {
                    Level::Task
                };
                running
                    .resolvable
                    .extend(request.items.as_slice().iter().cloned());
                let mut planned = Vec::new();
                for id in request.items.as_slice() {
                    planned.push(self.plan(request, kind, id, &mut running).await?);
                }
                self.copy_items(request, planned, None, &mut running, journal)
                    .await?
            }
            CopyScope::Projects { tasks } => {
                let mut projects = Vec::new();
                for id in request.items.as_slice() {
                    let members = if *tasks {
                        self.project_members(id).await?
                    } else {
                        Vec::new()
                    };
                    running.resolvable.push(id.clone());
                    running.resolvable.extend(members.iter().cloned());
                    projects.push((id.clone(), members));
                }
                self.copy_projects(request, &projects, &[], &mut running, journal)
                    .await?
            }
            CopyScope::Members(named) => {
                let (projects, unrecorded) =
                    self.named_members(request, named, &mut running).await?;
                self.copy_projects(request, &projects, &unrecorded, &mut running, journal)
                    .await?
            }
        };
        let references = running.references;
        let delivers_rewritten = running.delivers_rewritten;
        // Every member's destination id is known once the first pass has landed it, so the
        // tasks each landed task delivers are settled here rather than after the repair.
        let deliverers: Vec<Deliverer> = running
            .landed
            .iter()
            .map(|task| Deliverer {
                destination: task.destination.clone(),
                category: task.category,
                now: targets(
                    &resolved_entries(&mapped_delivers(
                        &task.delivers,
                        &task.origin,
                        &task.destination.source,
                        &running.resolvable,
                        &running.counterparts,
                    )),
                    &task.destination.source,
                ),
                before: task.before.clone(),
            })
            .collect();
        let linking = std::mem::take(&mut running.linking);
        let homes = std::mem::take(&mut running.homes);
        self.repair(request, running, journal).await?;
        if !request.dry_run {
            self.record_members(homes, journal).await?;
        }
        let mut items = items;
        self.link(linking, &mut items, journal).await?;
        Ok((
            CopyReport {
                items,
                references_rewritten: references.rewritten,
                references_unresolved: references.unresolved,
                references_ambiguous: references.ambiguous,
                delivers_rewritten,
                // Filled in by `copy`, once the copy is complete.
                delivered: Vec::new(),
                // Filled in by `copy`, which is the one place both readings are taken.
                spent: None,
            },
            deliverers,
        ))
    }

    /// Write every deferred item again, now that every destination id is known.
    ///
    /// This is the second half of the two passes an edge between two items of one copy
    /// needs: the far end's destination id does not exist until it has been created, so
    /// the item that points at it lands first without that edge and is completed here.
    /// It runs once for the whole request rather than once per project, because a far end
    /// may be in a project this copy has not reached yet.
    async fn repair(
        &self,
        request: &CopyRequest,
        running: Running,
        journal: &mut Journal,
    ) -> Result<(), EngineError> {
        if request.dry_run {
            return Ok(());
        }
        let Running {
            resolvable,
            counterparts,
            deferred,
            ..
        } = running;
        for entry in deferred {
            let destination = self.writable(&entry.item.to)?;
            let edges = mapped_edges(
                &entry.item.edges,
                &entry.item.source.source,
                &entry.item.to,
                &resolvable,
                &counterparts,
            );
            let delivers = delivers_of(&entry.item, &entry.item.to, &resolvable, &counterparts);
            self.write(
                destination,
                &entry.item,
                Some(entry.destination),
                entry.filed,
                &resolved(&edges),
                &resolved_entries(&delivers),
                entry.prior,
                journal,
            )
            .await?;
        }
        Ok(())
    }

    /// Put the destination back the way this copy found it, then report why it failed.
    ///
    /// Undone in reverse, and an item this copy created is removed rather than restored —
    /// the entry recording what it looked like a moment after creation is not a state
    /// anybody asked for. When the destination cannot take one of them back, the refusal
    /// says so and names what is still there, because a user told "the copy failed" about
    /// a destination that is not as they left it will copy again over a tree nobody
    /// described.
    async fn undo(&self, journal: Journal, error: EngineError) -> EngineError {
        let created: Vec<(&SourceName, Level, &NativeId)> = journal
            .entries
            .iter()
            .filter_map(|entry| match entry {
                Undo::Created { at, kind, id } => Some((at, *kind, id)),
                Undo::Updated { .. } => None,
            })
            .collect();
        // The ids and the refusal are one value rather than two, because they are one
        // fact: an item is only left behind because the destination refused to take it
        // back, so the first refusal carries the first id and neither half can be
        // recorded without the other.
        let mut unrestored: Option<(LeftBehind, SourceError)> = None;
        // The links first, because they were written last: once the whole copy had landed.
        for entry in journal.links.iter().rev() {
            if let Err(problem) = self.unlink(entry).await {
                match &mut unrestored {
                    Some((left_behind, _)) => left_behind.push(entry.item.clone()),
                    None => unrestored = Some((LeftBehind::new(entry.item.clone()), problem)),
                }
            }
        }
        for entry in journal.entries.iter().rev() {
            let outcome = match self.writable(entry.at()) {
                Err(failed) => Err(SourceError::Refused {
                    message: failed.to_string(),
                }),
                Ok(destination) => match entry {
                    Undo::Created { kind, id, .. } => remove(destination, *kind, id).await,
                    Undo::Updated { at, id, prior }
                        if !created.contains(&(at, prior.item.level(), id)) =>
                    {
                        restore(destination, id, prior).await
                    }
                    Undo::Updated { .. } => Ok(()),
                },
            };
            if let Err(problem) = outcome {
                let id = GlobalId::new(entry.at().clone(), entry.id().clone());
                match &mut unrestored {
                    Some((left_behind, _)) => left_behind.push(id),
                    None => unrestored = Some((LeftBehind::new(id), problem)),
                }
            }
        }
        match unrestored {
            None => error,
            Some((left_behind, refusal)) => EngineError::CopyNotUndone {
                error: Box::new(error),
                left_behind,
                refusal,
            },
        }
    }

    /// Record, on every item this copy landed, where it landed — once the whole copy has
    /// landed, so a copy that fails before this point has written nothing on any source.
    ///
    /// Each outcome is told what happened to its item's link. The items landed are exactly
    /// the outcomes that are not orphans, in the order they landed; a dry run records
    /// nothing, so this is never asked about one.
    async fn link(
        &self,
        linking: Vec<Linking>,
        items: &mut [CopyOutcome],
        journal: &mut Journal,
    ) -> Result<(), EngineError> {
        let mut landed = items
            .iter_mut()
            .filter(|outcome| !matches!(outcome.action, CopyAction::Orphaned { .. }));
        for entry in linking {
            let slot = landed
                .next()
                .filter(|outcome| outcome.source == entry.item)
                .and_then(|outcome| outcome.action.link_slot())
                .expect("every item a copy landed is reported, in the order it landed");
            *slot = Some(self.linked(entry, journal).await?);
        }
        Ok(())
    }

    /// Write one landed item's `onetaskgraph.copies` entry for the destination, when it
    /// does not already say this, and say what was done.
    ///
    /// An item found by its own origin records nothing: that correspondence is already
    /// recorded, on the destination item. A source that cannot hold the entry — it has no
    /// write side, or refuses the narrow metadata write — leaves the copy as it landed and
    /// is reported so; any other failure fails the copy, which then undoes every write it
    /// made, this one's journal entry included.
    async fn linked(&self, entry: Linking, journal: &mut Journal) -> Result<CopyLink, EngineError> {
        // Keyed by the source the item really landed in, which a routed copy decides per item.
        let landed_in = entry.destination.source.clone();
        if entry.found == Some(Found::Origin) {
            return Ok(CopyLink::Unchanged);
        }
        let wanted = Value::String(entry.destination.to_string());
        // Only the entries that are links are kept. Anything else under the key — a value
        // that is not an object, or an entry a person spelled wrong — is not a link this or
        // any later copy can read, so it is replaced by what is, exactly as an entry naming
        // another item is rewritten.
        let mut links = well_formed(entry.held.as_ref());
        if links.get(landed_in.as_str()) == Some(&wanted) {
            return Ok(CopyLink::Unchanged);
        }
        let source = self.readable(&entry.item.source)?;
        if !source.source().writes().is_supported() {
            return Ok(CopyLink::Unrecorded);
        }
        links.insert(landed_in.to_string(), wanted);
        // Journalled before the write, for the reason a destination write is: a write that
        // stopped part way is only put back by what this journal holds.
        journal.links.push(Unlink {
            item: entry.item.clone(),
            level: entry.level,
            before: entry.held,
        });
        match set_link(
            source,
            entry.level,
            &entry.item.native,
            &Value::Object(links),
        )
        .await
        {
            Ok(true) => Ok(CopyLink::Recorded),
            // Nothing was written: the source no longer holds the item, or said it cannot
            // hold the key.
            Ok(false) | Err(SourceError::Refused { .. }) => {
                journal.links.pop();
                Ok(CopyLink::Unrecorded)
            }
            // §4.18 once let a plugin refuse any key in this namespace as malformed, and one
            // written then still may. Its answer says nothing about whether it wrote, so the
            // journal entry stays and a later failure puts the item back regardless.
            Err(SourceError::Malformed { .. }) => Ok(CopyLink::Unrecorded),
            Err(error) => Err(refused(source, error)),
        }
    }

    /// Put one copied item's `onetaskgraph.copies` back to what it held before this copy.
    ///
    /// A key it held is written back through the same narrow write. A key it did not hold
    /// cannot be removed through that write, so the item is read as it now is and written
    /// back whole, without the key, through the source's own write interface — the same
    /// way an overwritten destination item is restored.
    async fn unlink(&self, entry: &Unlink) -> Result<(), SourceError> {
        let as_source_error = |error: EngineError| match error {
            EngineError::SourceRefused { error, .. } => error,
            other => SourceError::Refused {
                message: other.to_string(),
            },
        };
        let source = self.readable(&entry.item.source).map_err(as_source_error)?;
        if let Some(before) = &entry.before {
            return set_link(source, entry.level, &entry.item.native, before)
                .await
                .map(|_| ());
        }
        let Some(mut held) = self
            .prior(source, entry.level, &entry.item.native)
            .await
            .map_err(as_source_error)?
        else {
            return Ok(());
        };
        metadata_of(&mut held.item).remove(MetadataKey::COPIES_KEY);
        restore(source, &entry.item.native, &held).await
    }

    /// The member project `home` names in `to`, created and recorded on the home when it
    /// names none — for a write that is not a copy, such as `task create`, routed away from
    /// the source its project is in. `None` when the home itself is not there.
    ///
    /// What it wrote is handed back, so a write that then fails takes it back through
    /// [`Engine::unfile`] and leaves no member nobody asked for.
    pub(super) async fn member_project(
        &self,
        home: &GlobalId,
        to: &SourceName,
    ) -> Result<Filed, EngineError> {
        let mut running = Running::default();
        let mut journal = Journal::default();
        let mut filed = self
            .member_in(false, home, to, &mut running, &mut journal)
            .await;
        if filed.is_ok() {
            let homes = std::mem::take(&mut running.homes);
            if let Err(error) = self.record_members(homes, &mut journal).await {
                filed = Err(error);
            }
        }
        match filed {
            Ok(project) => Ok(Filed { project, journal }),
            Err(error) => Err(self.undo(journal, error).await),
        }
    }

    /// Take back what [`Engine::member_project`] wrote, because the write it filed for failed
    /// with `error`.
    pub(super) async fn unfile(&self, filed: Filed, error: EngineError) -> EngineError {
        self.undo(filed.journal, error).await
    }

    /// The destination source, once it is established it exists and can be written.
    fn writable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
        let name = self.known(name)?;
        if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
            return Err(EngineError::DestinationUnavailable {
                name: name.to_string(),
                error: unavailable.error().clone(),
            });
        }
        let source = self
            .ready()
            .find(|source| source.name() == &name)
            .ok_or(EngineError::NoSources)?;
        if !source.source().writes().is_supported() {
            return Err(EngineError::NotWritable {
                name: name.to_string(),
                kind: source.kind().to_owned(),
            });
        }
        Ok(source)
    }

    /// Copy each project and what travels with it: every task in it, the members a member
    /// copy names, or nothing at all.
    ///
    /// Every item of every project is read and its target decided **before any of them is
    /// written, and each exactly once.** That is what lets a member copy refuse an edge it
    /// cannot resolve while the destination is still as it was found, and what stops a
    /// whole copy resolving one target twice — once to learn the ids the project's own
    /// edges name, and again to land it — which against a hosted destination is a read per
    /// item spent on an answer the command already had.
    async fn copy_projects(
        &self,
        request: &CopyRequest,
        projects: &[(GlobalId, Vec<GlobalId>)],
        unrecorded: &[GlobalId],
        running: &mut Running,
        journal: &mut Journal,
    ) -> Result<Vec<CopyOutcome>, EngineError> {
        let mut plans = Vec::new();
        for (id, members) in projects {
            if self.routes.routes(&request.destination) {
                // Every task is read before the project is aimed, because where its home goes
                // is decided by where they do.
                let project = self.read(Level::Project, id).await?;
                let mut reads = Vec::new();
                for member in members {
                    reads.push(self.read(Level::Task, member).await?);
                }
                let home = self.home_placement(request, &project, &reads).await?;
                let project = self.aim(request, project, home).await?;
                let mut tasks = Vec::new();
                for read in reads {
                    let placement = match &read.item {
                        Item::Task(task) => {
                            self.routes.place(&request.destination, &task.repositories)
                        }
                        Item::Project(_) | Item::Document(_) => {
                            self.placed_at(request, &request.destination)
                        }
                    };
                    tasks.push(self.aim(request, read, placement).await?);
                }
                plans.push((project, tasks));
            } else {
                let project = self.plan(request, Level::Project, id, running).await?;
                let mut tasks = Vec::new();
                for member in members {
                    tasks.push(self.plan(request, Level::Task, member, running).await?);
                }
                plans.push((project, tasks));
            }
        }
        for (project, tasks) in &plans {
            for item in std::iter::once(project).chain(tasks) {
                unrecorded_far_end(item, unrecorded)?;
            }
        }
        let mut outcomes = Vec::new();
        for (project, tasks) in plans {
            outcomes.extend(
                self.copy_project(request, project, tasks, running, journal)
                    .await?,
            );
        }
        Ok(outcomes)
    }

    /// Where a routed project copy lands the project: its **home**.
    ///
    /// A home that already exists, in any source the copy can reach, stays where it is —
    /// found by the project's own link or origin. Otherwise the home follows its tasks only
    /// when every one copied routes to one source, and so does the project's own
    /// repositories when it names any; any other project's home is the copy's destination.
    /// With no task copied, the project's own repositories alone decide.
    async fn home_placement(
        &self,
        request: &CopyRequest,
        project: &Read,
        tasks: &[Read],
    ) -> Result<Placement, EngineError> {
        let Item::Project(held) = &project.item else {
            return Ok(self.placed_at(request, &request.destination));
        };
        let mut placements: Vec<Placement> = tasks
            .iter()
            .filter_map(|read| match &read.item {
                Item::Task(task) => {
                    Some(self.routes.place(&request.destination, &task.repositories))
                }
                Item::Project(_) | Item::Document(_) => None,
            })
            .collect();
        if !held.repositories.is_empty() {
            placements.push(self.routes.place(&request.destination, &held.repositories));
        }
        let rule = match placements.first() {
            Some(first)
                if placements
                    .iter()
                    .all(|placement| placement.destination == first.destination) =>
            {
                first.clone()
            }
            _ => self.placed_at(request, &request.destination),
        };
        for elsewhere in self.routes.reachable(&request.destination) {
            if elsewhere == rule.destination {
                continue;
            }
            let there = self.writable(&elsewhere)?;
            let named = link_of(&held.metadata, &elsewhere)
                .or_else(|| origin_of(&held.metadata).filter(|origin| origin.source == elsewhere));
            let Some(named) = named else {
                continue;
            };
            let Some(found) = self.prior(there, Level::Project, &named.native).await? else {
                continue;
            };
            let names_back = origin_of(described(&found.item).1).as_ref() == Some(&project.source);
            if names_back || origin_of(&held.metadata).as_ref() == Some(&named) {
                return Ok(self.placed_at(request, &elsewhere));
            }
        }
        Ok(rule)
    }

    /// Land one planned project, then the tasks planned beside it.
    async fn copy_project(
        &self,
        request: &CopyRequest,
        project: Planned,
        tasks: Vec<Planned>,
        running: &mut Running,
        journal: &mut Journal,
    ) -> Result<Vec<CopyOutcome>, EngineError> {
        let (carries_tasks, walks_orphans) = match request.scope {
            CopyScope::Projects { tasks } => (tasks, tasks),
            CopyScope::Members(_) => (true, false),
            CopyScope::Tasks | CopyScope::Documents => (false, false),
        };
        let id = project.source.clone();
        let members: Vec<GlobalId> = tasks.iter().map(|task| task.source.clone()).collect();
        // The project lands with every edge it can already resolve — its members' targets
        // are known from their plans — so a project that has not changed is not written at
        // all. What it *reports* is read the way it always has been: against the ids known
        // before its own members' were, unless every edge resolves and nothing differs.
        // Reading it any other way would move the word a repeat copy reports for a project
        // whose only difference is an edge, which is not this change's to move.
        let mut before_members = running.counterparts.clone();
        if let Target::Update { id: target, .. } = &project.target {
            before_members.insert(
                id.to_string(),
                GlobalId::new(project.to.clone(), target.clone()),
            );
        }
        for item in std::iter::once(&project).chain(&tasks) {
            if let Target::Update { id: target, .. } = &item.target {
                running.counterparts.insert(
                    item.source.to_string(),
                    GlobalId::new(item.to.clone(), target.clone()),
                );
            }
        }
        let home_source = project.to.clone();
        let unchanged = match &project.target {
            Target::Update { id: target, .. } => {
                let settled = mapped_edges(
                    &project.edges,
                    &id.source,
                    &home_source,
                    &running.resolvable,
                    &running.counterparts,
                );
                let first = mapped_edges(
                    &project.edges,
                    &id.source,
                    &home_source,
                    &running.resolvable,
                    &before_members,
                );
                let held = project.held.as_ref();
                let unchanged_with = |edges: &[Option<DependencyEdge>]| {
                    !changes(
                        held,
                        &project,
                        target,
                        &None,
                        &resolved(edges),
                        &[],
                        &home_source,
                    )
                };
                Some(
                    (!settled.iter().any(Option::is_none) && unchanged_with(&settled))
                        || unchanged_with(&first),
                )
            }
            Target::Create => None,
        };
        // A project this copy creates holds nothing it did not file itself, so nothing filed
        // under it can be an orphan and the destination is not walked for one.
        let created = matches!(project.target, Target::Create);
        let mut outcomes = self
            .copy_items(request, vec![project], None, running, journal)
            .await?;
        if let (Some(unchanged), Some(landed), Some(via)) = (
            unchanged,
            outcomes[0].destination().cloned(),
            outcomes[0].action.found_by(),
        ) {
            let link = outcomes[0].action.link();
            outcomes[0].action = if unchanged {
                CopyAction::Unchanged {
                    destination: landed,
                    via,
                    link,
                }
            } else {
                CopyAction::Updated {
                    destination: landed,
                    via,
                    link,
                }
            };
        }
        if !carries_tasks {
            return Ok(outcomes);
        }
        // `None` when a dry run would have created the project: nothing was written, so
        // there is no destination project id to file the tasks under. Every task is still
        // read and still reported, because that is what a dry run is for.
        let filed = outcomes.first().and_then(CopyOutcome::destination).cloned();
        let filing = match &filed {
            None => None,
            Some(home) => {
                let mut filing = Filing::from([(home.source.clone(), home.native.clone())]);
                // A task routed to another source than its home's is filed under the home's
                // member project there, found or created before any of those tasks lands.
                let mut elsewhere: Vec<SourceName> = Vec::new();
                for task in &tasks {
                    if task.to != home.source && !elsewhere.contains(&task.to) {
                        elsewhere.push(task.to.clone());
                    }
                }
                for to in elsewhere {
                    if let Some(member) = self
                        .member_in(request.dry_run, home, &to, running, journal)
                        .await?
                    {
                        filing.insert(to, member);
                    }
                }
                Some(filing)
            }
        };
        outcomes.extend(
            self.copy_items(request, tasks, filing.as_ref(), running, journal)
                .await?,
        );
        // A member copy was told which members it carries, so it cannot tell a member it
        // left out from one the source no longer holds, and does not walk for either.
        if walks_orphans
            && !created
            && let Some(filed) = filed
        {
            let destination = self.writable(&filed.source)?;
            outcomes.extend(
                self.orphans(destination, &id, &filed.native, &members)
                    .await?,
            );
            // A routed plan's tasks are filed under its members too, so each member is walked
            // in its own source for what the source plan no longer holds.
            if self.routes.routes(&request.destination) {
                for member in self.members_named(&filed, running).await? {
                    let there = self.writable(&member.source)?;
                    outcomes.extend(self.orphans(there, &id, &member.native, &members).await?);
                }
            }
        }
        Ok(outcomes)
    }

    /// The member projects `home` names: the list this copy already holds for it, grown
    /// members included, or else the one the home itself records.
    async fn members_named(
        &self,
        home: &GlobalId,
        running: &Running,
    ) -> Result<Vec<GlobalId>, EngineError> {
        if let Some(held) = running.homes.get(&home.to_string()) {
            return Ok(held.members.clone());
        }
        let at = self.writable(&home.source)?;
        let Some(held) = at
            .source()
            .get_project(&home.native)
            .await
            .map_err(|error| refused(at, error))?
        else {
            return Ok(Vec::new());
        };
        members_of(home, &held.metadata).map_err(|message| EngineError::SourceRefused {
            name: home.source.to_string(),
            error: SourceError::Malformed { message },
        })
    }

    /// The members a member copy names, per project and in the order they were named, and
    /// every member it does not name that records no destination id.
    ///
    /// Settled before anything is read at the destination. A named id that is a member of
    /// none of the projects is refused here, naming it. An unnamed member whose recorded
    /// origin names the destination joins the copied set with that id already known, so an
    /// edge to it resolves exactly as an edge to a copied item does, and costs no read. An
    /// unnamed member recording none is returned, so an edge to it is refused before
    /// anything is written — see [`unrecorded_far_end`].
    async fn named_members(
        &self,
        request: &CopyRequest,
        named: &CopyItems,
        running: &mut Running,
    ) -> Result<(Vec<(GlobalId, Vec<GlobalId>)>, Vec<GlobalId>), EngineError> {
        let projects = request.items.as_slice();
        // A routed copy can land a member in any source it reaches, so an origin naming any
        // of them is a destination id already known.
        let reachable = self.routes.reachable(&request.destination);
        let mut carried = Vec::new();
        let mut unrecorded = Vec::new();
        for project in projects {
            let held = self.project_member_tasks(project).await?;
            let mut members: Vec<GlobalId> = Vec::new();
            for id in named.as_slice() {
                if held.iter().any(|task| &task.id == id) && !members.contains(id) {
                    members.push(id.clone());
                }
            }
            for task in held {
                if members.contains(&task.id) {
                    continue;
                }
                // A routed copy also reads the member's own link, because a plan authored
                // where it is copied from records where each task landed rather than an
                // origin, and its tasks may have landed in any source the copy reaches.
                let linked = || {
                    reachable
                        .iter()
                        .find_map(|there| link_of(&task.item.metadata, there))
                        .filter(|_| self.routes.routes(&request.destination))
                };
                match origin_of(&task.item.metadata)
                    .filter(|origin| reachable.contains(&origin.source))
                    .or_else(linked)
                {
                    Some(origin) => {
                        running.counterparts.insert(task.id.to_string(), origin);
                        running.resolvable.push(task.id);
                    }
                    None => unrecorded.push(task.id),
                }
            }
            running.resolvable.push(project.clone());
            running.resolvable.extend(members.iter().cloned());
            carried.push((project.clone(), members));
        }
        if let Some(stray) = named
            .as_slice()
            .iter()
            .find(|id| !carried.iter().any(|(_, members)| members.contains(id)))
        {
            return Err(EngineError::NotAMember {
                id: stray.clone(),
                projects: projects.to_vec(),
            });
        }
        Ok((carried, unrecorded))
    }

    /// Every task the source holds in `project`, by qualified id.
    async fn project_members(&self, project: &GlobalId) -> Result<Vec<GlobalId>, EngineError> {
        Ok(self
            .project_member_tasks(project)
            .await?
            .into_iter()
            .map(|task| task.id)
            .collect())
    }

    /// Every task the source holds in `project`, as the source reported it.
    ///
    /// The ids alone are what a copy files under a project; the whole task is what a
    /// document's references need, because the location a reference names is a field of it.
    async fn project_member_tasks(
        &self,
        project: &GlobalId,
    ) -> Result<Vec<Qualified<Task>>, EngineError> {
        let mut request = TaskRequest {
            sources: vec![project.source.clone()],
            filters: Filters::default(),
            project: ProjectSelector::Qualified(project.clone()),
            priorities: Vec::new(),
            commented_since: None,
            metadata: Vec::new(),
            origin: None,
            include_members: false,
            paging: Paging {
                limit: PROJECT_PAGE,
                token: None,
            },
        };
        let mut members = Vec::new();
        // Pages by this engine's own token rather than by a source cursor, and the
        // asymmetry with the three walks below is deliberate. A source answering two
        // cursors with each other advances on every page, so `unrepeated` under the list
        // verb never fires; the cycle shows only as a token handed back unchanged, which
        // is what this loop pages by. Point it at the source and a project copy spins.
        //
        // No page bound here for the same reason: `Engine::tasks` merges under the budget
        // it was asked, so nothing longer than `PROJECT_PAGE` can arrive. `fits` in
        // `fetch::walk` refuses the page a source can really overrun, its own, while
        // these very members are read.
        let misbehaved = |error| EngineError::SourceRefused {
            name: project.source.to_string(),
            error,
        };
        loop {
            let asked = request.paging.token.clone();
            let response = self.tasks(&request).await?;
            if let Some(failure) = response.errors.first() {
                return Err(EngineError::SourceRefused {
                    name: failure.source.to_string(),
                    error: failure.error.clone(),
                });
            }
            unrepeated(
                response.next.as_ref(),
                asked.as_ref(),
                "the tasks of a project were being read for a copy",
            )
            .map_err(misbehaved)?;
            members.extend(response.items);
            match response.next {
                Some(token) => request.paging.token = Some(token),
                None => return Ok(members),
            }
        }
    }

    /// Every document the source holds in `project`, as the source reported it.
    ///
    /// Paged by this engine's own token for the reason the member walk above is, and the
    /// note there says why.
    async fn project_documents(
        &self,
        project: &GlobalId,
    ) -> Result<Vec<Qualified<Document>>, EngineError> {
        let mut request = DocumentRequest {
            sources: vec![project.source.clone()],
            filters: DocumentFilters::default(),
            project: ProjectSelector::Qualified(project.clone()),
            paging: Paging {
                limit: PROJECT_PAGE,
                token: None,
            },
        };
        let mut held = Vec::new();
        let misbehaved = |error| EngineError::SourceRefused {
            name: project.source.to_string(),
            error,
        };
        loop {
            let asked = request.paging.token.clone();
            let response = self.documents(&request).await?;
            if let Some(failure) = response.errors.first() {
                return Err(EngineError::SourceRefused {
                    name: failure.source.to_string(),
                    error: failure.error.clone(),
                });
            }
            unrepeated(
                response.next.as_ref(),
                asked.as_ref(),
                "the documents of a project were being read for a copy",
            )
            .map_err(misbehaved)?;
            held.extend(response.items);
            match response.next {
                Some(token) => request.paging.token = Some(token),
                None => return Ok(held),
            }
        }
    }

    /// Point every reference the documents of this copy hold at the destination's own
    /// records, and say how many it could not.
    ///
    /// A document copied out of a local Markdown store used to arrive naming absolute paths
    /// under one checkout on one machine, dead for the only reader the copy exists for,
    /// while the destination held its own record for every one of them the whole time. This
    /// is what closes that, and it is deliberately **not** a Markdown-link parser: the
    /// artifact that motivated it holds bare absolute paths inside backticks in a table
    /// cell, which `[text](target)` matching would have left exactly as it found them.
    ///
    /// The correspondence is re-established from what the *destination* records at
    /// [`GlobalId::ORIGIN_KEY`], not from the mapping this copy holds. That mapping is not
    /// available when it is needed: `project copy` and `document copy` are separate verbs,
    /// one invocation carries one [`CopyScope`], and a project copy carries no documents —
    /// so by the time the document is copied, the tasks were written by a process that has
    /// exited. Reading the destination is also what makes a document copied on its own
    /// work, which a same-run mapping never could.
    ///
    // llmlint: ignore-block[contracts_have_one_source_or_a_drift_gate] This comment is required to state the rule; it is held by the journeys `a_rendering_whose_references_a_copy_rewrites_records_the_digest_of_what_it_was_given`, `a_copy_carries_provenance_it_cannot_vouch_for_verbatim` and `a_rendering_whose_references_a_copy_leaves_alone_carries_its_provenance_unchanged` in `crates/onetaskgraph-e2e/tests/e2e/rendered.rs`, and the entry's fields are `TemplateProvenance`'s own.
    /// Tasks and projects are not touched. A document's content is rewritten, and so is
    /// one thing beside it: when this substitutes at least one reference into a rendering
    /// that still hashes to the `body_digest` its [`TemplateProvenance`] records, the copy
    /// records as `body_digest` the digest of the rendering as this rewrote its references,
    /// carrying the entry's other three fields; any other entry, or none, is carried
    /// verbatim. The answers that rendering was made from, and the `answers_digest` naming
    /// them, still hold the pre-copy locations, so the copy is not a fresh rendering of them
    /// and a regenerate from them reproduces the pre-copy content.
    // llmlint: ignore-end[contracts_have_one_source_or_a_drift_gate]
    async fn rewrite_references(
        &self,
        planned: &mut [Planned],
        counts: &mut Counted,
    ) -> Result<(), EngineError> {
        // Read at the source, once per project rather than once per document: several
        // documents of one project is the ordinary case.
        let mut by_project: BTreeMap<String, Vec<Referent>> = BTreeMap::new();
        let mut named: Vec<Vec<Referent>> = Vec::new();
        for item in planned.iter() {
            named.push(self.named_referents(item, &mut by_project).await?);
        }
        // The destination is walked only for a copy that really names something, and only
        // for the interfaces those referents are read from — in each source a document of
        // this call lands in, which a routed copy decides per document.
        let mut levels: BTreeMap<SourceName, Vec<Level>> = BTreeMap::new();
        for (item, referents) in planned.iter().zip(&named) {
            for referent in referents {
                let wanted = levels.entry(item.to.clone()).or_default();
                if !wanted.contains(&referent.level) {
                    wanted.push(referent.level);
                }
            }
        }
        if levels.is_empty() {
            return Ok(());
        }
        let mut walked: BTreeMap<SourceName, Counterparts> = BTreeMap::new();
        for (to, levels) in levels {
            let destination = self.writable(&to)?;
            walked.insert(to, self.counterparts(destination, &levels).await?);
        }
        for (item, referents) in planned.iter_mut().zip(named) {
            let Some(counterparts) = walked.get(&item.to) else {
                continue;
            };
            let Item::Document(document) = &mut item.item else {
                continue;
            };
            let Some(content) = &document.content else {
                continue;
            };
            let (rewritten, made) = substitute(content, &table_for(&referents, counterparts));
            if made.rewritten > 0
                && let Some(provenance) = restamped(&document.metadata, content, &rewritten)
            {
                document
                    .metadata
                    .insert(TemplateProvenance::KEY.to_owned(), provenance.to_value());
            }
            document.content = Some(rewritten);
            counts.add(made);
        }
        Ok(())
    }

    /// The records of one document's own project whose location string its content really
    /// holds, as a whole reference.
    ///
    /// A document with no content, or with no project at the source, names nothing: the
    /// referent set is the document's own project, its tasks and its other documents, and
    /// there is no such set without a project.
    async fn named_referents(
        &self,
        item: &Planned,
        by_project: &mut BTreeMap<String, Vec<Referent>>,
    ) -> Result<Vec<Referent>, EngineError> {
        let Item::Document(document) = &item.item else {
            return Ok(Vec::new());
        };
        let (Some(content), Some(project)) =
            (document.content.as_deref(), document.project.as_ref())
        else {
            return Ok(Vec::new());
        };
        if content.is_empty() {
            return Ok(Vec::new());
        }
        let project = GlobalId::new(item.source.source.clone(), project.clone());
        let key = project.to_string();
        if !by_project.contains_key(&key) {
            let read = self.referents(&project).await?;
            by_project.insert(key.clone(), read);
        }
        Ok(by_project[&key]
            .iter()
            // A document does not name itself: the referent set is every *other* record
            // filed under the project. Told by the interface as well as the id, because an
            // id alone does not identify a record — a folder of Markdown filing `A.md`
            // under both `tasks/` and `documents/` is the ordinary case rather than the
            // contrived one, and excluding by id alone would drop that task from the set.
            .filter(|referent| referent.level != Level::Document || referent.id != item.source)
            .filter(|referent| holds(content, &referent.location))
            .cloned()
            .collect())
    }

    /// Every record filed under one project at its own source, with the location string
    /// that source reports for it.
    ///
    /// The project record itself, every task filed under it, and every document filed under
    /// it. All three are reads this engine already knows how to make.
    async fn referents(&self, project: &GlobalId) -> Result<Vec<Referent>, EngineError> {
        let source = self.readable(&project.source)?;
        let mut referents = Vec::new();
        if let Some(held) = source
            .source()
            .get_project(&project.native)
            .await
            .map_err(|error| refused(source, error))?
        {
            note(
                &mut referents,
                project.clone(),
                Level::Project,
                held.location.as_ref(),
                &held.metadata,
            );
        }
        for task in self.project_member_tasks(project).await? {
            note(
                &mut referents,
                task.id,
                Level::Task,
                task.item.location.as_ref(),
                &task.item.metadata,
            );
        }
        for document in self.project_documents(project).await? {
            note(
                &mut referents,
                document.id,
                Level::Document,
                document.item.location.as_ref(),
                &document.item.metadata,
            );
        }
        Ok(referents)
    }

    /// Walk the destination once for every record it holds at the levels named, one page
    /// at a time.
    ///
    /// One page is *read* at a time, and what is kept from each is three fields of each
    /// record — its id, the origin it records and the location the destination reports —
    /// never the page. That is more than [`Engine::scan`] keeps, and deliberately: the whole
    /// point of this walk is that one pass answers every document and every referent of the
    /// invocation, so what it learns has to outlive the page it learned it from. Nothing is
    /// written down and the index is dropped with the call.
    ///
    /// The other difference from `scan` is the *answer*: this one keeps every match, so a
    /// destination holding two records for one work item is reported as ambiguous rather
    /// than resolved to the first.
    async fn counterparts(
        &self,
        destination: &ResolvedSource,
        levels: &[Level],
    ) -> Result<Counterparts, EngineError> {
        let mut found = Counterparts::default();
        for level in levels {
            // Every cursor this level has already been sent. `unrepeated` below catches a
            // source that hands back the cursor it was just given, and its own note says
            // why it catches no more than that: a source cycling through two cursors
            // advances on every page, and seeing it needs memory the walks that share that
            // helper do not keep. This walk does keep memory — it is building an index that
            // outlives each page — so here the memory exists and the cycle is caught. The
            // walks one level up catch the same defect as a page token handed back
            // unchanged; this one pages by the source's own cursor and has no level above
            // it, so nothing else would.
            let mut asked_before: BTreeSet<String> = BTreeSet::new();
            let mut cursor: Option<Cursor> = None;
            loop {
                if let Some(next) = &cursor
                    && !asked_before.insert(next.0.clone())
                {
                    return Err(refused(
                        destination,
                        SourceError::Malformed {
                            message: "the source returned a cursor it had already been \
                                      given while the destination was being walked for the \
                                      records a document's references name, so the walk \
                                      would never end"
                                .to_owned(),
                        },
                    ));
                }
                let asked = cursor.clone();
                let request = request_for(destination, cursor);
                let next = match level {
                    Level::Task => {
                        let page = destination
                            .source()
                            .query_tasks(&TaskQuery::default(), &request)
                            .await
                            .map_err(|error| refused(destination, error))?;
                        fits(page.items.len(), request.limit)
                            .map_err(|error| refused(destination, error))?;
                        for task in &page.items {
                            found.note(*level, &task.id, task.location.as_ref(), &task.metadata);
                        }
                        page.next
                    }
                    Level::Project => {
                        let page = destination
                            .source()
                            .query_projects(&ProjectQuery::default(), &request)
                            .await
                            .map_err(|error| refused(destination, error))?;
                        fits(page.items.len(), request.limit)
                            .map_err(|error| refused(destination, error))?;
                        for project in &page.items {
                            found.note(
                                *level,
                                &project.id,
                                project.location.as_ref(),
                                &project.metadata,
                            );
                        }
                        page.next
                    }
                    Level::Document => {
                        let page = destination
                            .source()
                            .query_documents(&DocumentQuery::default(), &request)
                            .await
                            .map_err(|error| refused(destination, error))?;
                        fits(page.items.len(), request.limit)
                            .map_err(|error| refused(destination, error))?;
                        for document in &page.items {
                            found.note(
                                *level,
                                &document.id,
                                document.location.as_ref(),
                                &document.metadata,
                            );
                        }
                        page.next
                    }
                };
                unrepeated(
                    next.as_ref(),
                    asked.as_ref(),
                    "the destination was being walked for the records a document's \
                     references name",
                )
                .map_err(|error| refused(destination, error))?;
                match next {
                    Some(next) => cursor = Some(next),
                    None => break,
                }
            }
        }
        Ok(found)
    }

    /// Destination tasks filed under the copied project whose origin the source no longer
    /// holds.
    ///
    /// A copy never deletes, so each is left exactly as it is and reported.
    async fn orphans(
        &self,
        destination: &ResolvedSource,
        project: &GlobalId,
        at_destination: &NativeId,
        copied: &[GlobalId],
    ) -> Result<Vec<CopyOutcome>, EngineError> {
        let mut orphans = Vec::new();
        let mut cursor: Option<Cursor> = None;
        loop {
            let asked = cursor.clone();
            let request = request_for(destination, cursor);
            let page: Page<Task> = destination
                .source()
                .query_tasks(&TaskQuery::default(), &request)
                .await
                .map_err(|error| refused(destination, error))?;
            fits(page.items.len(), request.limit).map_err(|error| refused(destination, error))?;
            for task in &page.items {
                if task.project.as_ref() != Some(at_destination) {
                    continue;
                }
                let Some(origin) = origin_of(&task.metadata) else {
                    continue;
                };
                if origin.source != project.source || copied.contains(&origin) {
                    continue;
                }
                orphans.push(CopyOutcome {
                    source: origin,
                    action: CopyAction::Orphaned {
                        destination: GlobalId::new(destination.name().clone(), task.id.clone()),
                    },
                    placed: None,
                });
            }
            unrepeated(
                page.next.as_ref(),
                asked.as_ref(),
                "the destination was being read for items the copy left behind",
            )
            .map_err(|error| refused(destination, error))?;
            match page.next {
                Some(next) => cursor = Some(next),
                None => return Ok(orphans),
            }
        }
    }

    /// Resolve and write every item planned, holding back the ones whose edges are not
    /// resolvable yet.
    ///
    /// An edge between two items of one copy can point at a member whose destination id
    /// does not exist until it has been created, so the item that points at it lands
    /// without that edge and is handed to `deferred`. [`Engine::repair`] finishes it once
    /// the *whole* request has landed — not once this call has, because the far end may
    /// be in another project of the same command.
    async fn copy_items(
        &self,
        request: &CopyRequest,
        mut planned: Vec<Planned>,
        project: Option<&Filing>,
        running: &mut Running,
        journal: &mut Journal,
    ) -> Result<Vec<CopyOutcome>, EngineError> {
        // Only a document's own content names other records, and only once every document
        // of this call has been read: the destination is walked once for all of them, and
        // the content the rest of this call lands is the rewritten one — which is what
        // makes a repeat copy of an already-rewritten document report `unchanged`.
        if planned
            .iter()
            .any(|item| item.item.level() == Level::Document)
        {
            self.rewrite_references(&mut planned, &mut running.references)
                .await?;
        }

        for item in &planned {
            if let Target::Update { id, .. } = &item.target {
                running.counterparts.insert(
                    item.source.to_string(),
                    GlobalId::new(item.to.clone(), id.clone()),
                );
            }
            if let Item::Task(task) = &item.item {
                running.delivers_rewritten +=
                    members_named(&task.delivers, &item.source.source, &running.resolvable);
            }
        }

        // Resolved once per item, and used by both passes: the repair pass writes the
        // same item again, and re-deriving this there could file it somewhere else.
        let mut filed = Vec::new();
        for item in &planned {
            filed.push(match (project, &item.item) {
                (_, Item::Project(_)) => None,
                (Some(filing), _) => filing.get(&item.to).cloned(),
                (None, _) => self.filed(request, item, None, running, journal).await?,
            });
        }

        let mut outcomes = Vec::new();
        let mut unresolved = Vec::new();
        let mut priors = Vec::new();
        for (index, item) in planned.iter().enumerate() {
            let destination = self.writable(&item.to)?;
            let edges = mapped_edges(
                &item.edges,
                &item.source.source,
                &item.to,
                &running.resolvable,
                &running.counterparts,
            );
            let delivers = delivers_of(item, &item.to, &running.resolvable, &running.counterparts);
            if edges.iter().any(Option::is_none) || delivers.iter().any(Option::is_none) {
                unresolved.push(index);
            }
            let resolvable_delivers = resolved_entries(&delivers);
            let (outcome, prior) = self
                .land(
                    destination,
                    request,
                    item,
                    filed[index].clone(),
                    Pointing {
                        edges: &edges,
                        delivers: &resolvable_delivers,
                    },
                    journal,
                )
                .await?;
            if let Some(id) = outcome.destination() {
                running
                    .counterparts
                    .insert(item.source.to_string(), id.clone());
                if !request.dry_run {
                    running.linking.push(Linking {
                        item: item.source.clone(),
                        level: item.item.level(),
                        found: item.target.found(),
                        destination: id.clone(),
                        held: described(&item.item)
                            .1
                            .get(MetadataKey::COPIES_KEY)
                            .cloned(),
                    });
                }
            }
            if !request.dry_run
                && let (Item::Task(task), Some(landed)) = (&item.item, outcome.destination())
            {
                running.landed.push(LandedTask {
                    destination: landed.clone(),
                    origin: item.source.source.clone(),
                    delivers: task.delivers.clone(),
                    before: match prior.as_ref().map(|prior| &prior.item) {
                        Some(Item::Task(held)) => targets(&held.delivers, destination.name()),
                        _ => Vec::new(),
                    },
                    category: task.status.category,
                });
            }
            outcomes.push(outcome);
            priors.push(prior);
        }

        if !request.dry_run {
            for (index, item) in planned.into_iter().enumerate() {
                if !unresolved.contains(&index) {
                    continue;
                }
                // Every item a copy that is not a dry run lands has a destination id: the
                // one outcome without one is a dry run that would have created, and this
                // block does not run for a dry run.
                let id = outcomes[index]
                    .destination()
                    .expect("a copy that writes lands every item it planned")
                    .clone();
                running.deferred.push(Deferred {
                    item,
                    filed: filed[index].clone(),
                    destination: id.native,
                    prior: priors[index].clone(),
                });
            }
        }
        Ok(outcomes)
    }

    /// Read one item, place it, read its forward edges, and decide where it is going.
    ///
    /// A task is placed by its own repositories; a document goes with its project's home.
    async fn plan(
        &self,
        request: &CopyRequest,
        kind: Level,
        id: &GlobalId,
        running: &mut Running,
    ) -> Result<Planned, EngineError> {
        let read = self.read(kind, id).await?;
        let placement = match &read.item {
            Item::Task(task) => self.routes.place(&request.destination, &task.repositories),
            Item::Document(document) => {
                self.document_placement(request, &read.source, document, running)
                    .await?
            }
            Item::Project(project) => self
                .routes
                .place(&request.destination, &project.repositories),
        };
        self.aim(request, read, placement).await
    }

    /// Read one item out of its own source.
    async fn read(&self, kind: Level, id: &GlobalId) -> Result<Read, EngineError> {
        let source = self.readable(&id.source)?;
        if kind == Level::Document {
            documentary(source)?;
        }
        let item = match kind {
            Level::Task => source
                .source()
                .get_task(&id.native)
                .await
                .map_err(|error| refused(source, error))?
                .map(|task| Item::Task(Box::new(task))),
            Level::Project => source
                .source()
                .get_project(&id.native)
                .await
                .map_err(|error| refused(source, error))?
                .map(|project| Item::Project(Box::new(project))),
            Level::Document => source
                .source()
                .get_document(&id.native)
                .await
                .map_err(|error| refused(source, error))?
                .map(|document| Item::Document(Box::new(document))),
        }
        .ok_or_else(|| EngineError::NoSuchItem { id: id.to_string() })?;
        Ok(Read {
            source: id.clone(),
            item,
        })
    }

    /// Decide where one read item is going, once it is known which source it lands in.
    async fn aim(
        &self,
        request: &CopyRequest,
        read: Read,
        placement: Placement,
    ) -> Result<Planned, EngineError> {
        let Read { source: id, item } = read;
        let destination = self.writable(&placement.destination)?;
        if item.level() == Level::Document {
            documentary(destination)?;
        }
        // Before the destination is read, and so before it is written: a destination that
        // holds no priority has nowhere to put one, and every item of a copy is planned before
        // any of them lands. A task carrying `none` passes and writes exactly as it always did.
        if let Item::Task(task) = &item {
            holds_priority(destination, &id.to_string(), task.priority)?;
        }
        let routed = self.routes.routes(&request.destination);
        // A task or a document whose counterpart already sits somewhere a route could have
        // sent it, other than where it routes now, was moved by a repository changed after
        // the copy. That is a person's decision rather than a silent move, so it is refused
        // before anything is written. A project's home is exempt: it stays where it landed.
        if routed && item.level() != Level::Project {
            self.misrouted(request, &id, &item, &placement.destination)
                .await?;
        }
        let source = self.readable(&id.source)?;
        // Read, and refused, before the destination is read and before anything is written:
        // an asset its source does not hold, and a destination that stores none.
        let assets = carried_assets(source, &id, &item).await?;
        if let Some(first) = assets.first() {
            assets::stores(destination, &id.to_string(), &first.name)?;
        }
        let edges = forward_edges(source, &id.native, item.level()).await?;
        let (target, held) = self.target(destination, request, &id, &item).await?;
        Ok(Planned {
            source: id,
            item,
            edges,
            target,
            held,
            to: placement.destination.clone(),
            placed: routed.then_some(placement),
            assets,
        })
    }

    /// Refuse an item whose existing counterpart, by its link or its origin, sits in a source
    /// its copy could reach other than `to`, the one it routes to now.
    async fn misrouted(
        &self,
        request: &CopyRequest,
        id: &GlobalId,
        item: &Item,
        to: &SourceName,
    ) -> Result<(), EngineError> {
        let metadata = described(item).1;
        for elsewhere in self.routes.reachable(&request.destination) {
            if &elsewhere == to {
                continue;
            }
            let there = self.writable(&elsewhere)?;
            let linked = match link_of(metadata, &elsewhere) {
                Some(link) => self
                    .prior(there, item.level(), &link.native)
                    .await?
                    .filter(|held| origin_of(described(&held.item).1).as_ref() == Some(id))
                    .map(|_| link),
                None => None,
            };
            let originated = match origin_of(metadata).filter(|origin| origin.source == elsewhere) {
                Some(origin) => self
                    .prior(there, item.level(), &origin.native)
                    .await?
                    .map(|_| origin),
                None => None,
            };
            if let Some(counterpart) = linked.or(originated) {
                return Err(EngineError::Misrouted {
                    item: id.clone(),
                    counterpart,
                    route: to.clone(),
                });
            }
        }
        Ok(())
    }

    /// Where a document lands: with its project's home, wherever that is; by its own
    /// repositories when its project has none yet, or it is in no project.
    async fn document_placement(
        &self,
        request: &CopyRequest,
        id: &GlobalId,
        document: &Document,
        running: &mut Running,
    ) -> Result<Placement, EngineError> {
        if self.routes.routes(&request.destination)
            && let Some(project) = &document.project
        {
            let qualified = GlobalId::new(id.source.clone(), project.clone());
            if let Some(home) = self.find_home(request, &qualified, running).await? {
                return Ok(self.placed_at(request, &home.source));
            }
        }
        Ok(self
            .routes
            .place(&request.destination, &document.repositories))
    }

    /// What the report says placed an item in `at`: the first route entry sending there, or
    /// none for the copy's own destination.
    fn placed_at(&self, request: &CopyRequest, at: &SourceName) -> Placement {
        Placement {
            destination: at.clone(),
            route: self
                .routes
                .of(&request.destination)
                .iter()
                .position(|route| route.to() == at)
                .map(|index| u32::try_from(index).unwrap_or(u32::MAX)),
        }
    }

    /// Which destination item this one corresponds to, by the link the item records, the
    /// two origin rules and the caller's escape, and what the destination holds there.
    async fn target(
        &self,
        destination: &ResolvedSource,
        request: &CopyRequest,
        id: &GlobalId,
        item: &Item,
    ) -> Result<(Target, Option<Prior>), EngineError> {
        let (title, metadata) = described(item);
        // A caller asserting there is nothing to find is believed, and nothing is looked for —
        // unless the item itself says otherwise, by naming a counterpart at the destination.
        if request.create {
            let carrier = link_of(metadata, destination.name()).or_else(|| {
                origin_of(metadata).filter(|origin| &origin.source == destination.name())
            });
            if let Some(carrier) = carrier {
                return Err(EngineError::CreateCarried {
                    item: id.clone(),
                    carrier,
                });
            }
            return Ok((Target::Create, None));
        }
        // The link first, and it is believed only when the item it names still names this
        // one back: a person who re-pointed that item's origin has said it is not this
        // one's counterpart any more, so the link is ignored and rewritten to whatever the
        // rules below find.
        if let Some(link) = link_of(metadata, destination.name()) {
            match self.prior(destination, item.level(), &link.native).await? {
                Some(held) if origin_of(described(&held.item).1).as_ref() == Some(id) => {
                    return Ok((
                        Target::Update {
                            id: link.native,
                            found: Found::Link,
                        },
                        Some(held),
                    ));
                }
                Some(_) => {}
                None if request.recreate => {}
                None => {
                    return Err(EngineError::StaleLink {
                        item: id.to_string(),
                        link: link.to_string(),
                    });
                }
            }
        }
        if let Some(origin) = origin_of(metadata)
            && &origin.source == destination.name()
        {
            // The read that says the origin still names something is the read of what it
            // holds, so rule 1 costs one round trip rather than two.
            if let Some(held) = self
                .prior(destination, item.level(), &origin.native)
                .await?
            {
                return Ok((
                    Target::Update {
                        id: origin.native,
                        found: Found::Origin,
                    },
                    Some(held),
                ));
            }
            if !request.recreate {
                return Err(EngineError::StaleOrigin {
                    item: id.to_string(),
                    origin: origin.to_string(),
                });
            }
        }
        if let Some(found) = self
            .scan(destination, item.level(), &Wanted::Origin(id.to_string()))
            .await?
        {
            let held = self.prior(destination, item.level(), &found).await?;
            return Ok((
                Target::Update {
                    id: found,
                    found: Found::Scan,
                },
                held,
            ));
        }
        let wanted = match &request.match_by {
            Some(MatchBy::Title) => Some(Wanted::Title(title.to_owned())),
            Some(MatchBy::Metadata(key)) => metadata
                .get(key)
                .map(|value| Wanted::Metadata(key.clone(), value.clone())),
            None => None,
        };
        if let Some(wanted) = wanted
            && let Some(found) = self.scan(destination, item.level(), &wanted).await?
        {
            let held = self.prior(destination, item.level(), &found).await?;
            return Ok((
                Target::Update {
                    id: found,
                    found: Found::Match,
                },
                held,
            ));
        }
        Ok((Target::Create, None))
    }

    /// Walk the destination one page at a time, looking for `wanted`.
    ///
    /// One page is held at a time and nothing is written down, which is the same bound
    /// every other compensation in this engine works under.
    async fn scan(
        &self,
        destination: &ResolvedSource,
        kind: Level,
        wanted: &Wanted,
    ) -> Result<Option<NativeId>, EngineError> {
        let mut cursor: Option<Cursor> = None;
        let tasks = wanted.task_query(&destination.source().capabilities());
        loop {
            let asked = cursor.clone();
            let request = request_for(destination, cursor);
            let next = match kind {
                Level::Task => {
                    let page = destination
                        .source()
                        .query_tasks(&tasks, &request)
                        .await
                        .map_err(|error| refused(destination, error))?;
                    fits(page.items.len(), request.limit)
                        .map_err(|error| refused(destination, error))?;
                    for task in &page.items {
                        if wanted.found(&task.title, &task.metadata) {
                            return Ok(Some(task.id.clone()));
                        }
                    }
                    page.next
                }
                Level::Project => {
                    let page = destination
                        .source()
                        .query_projects(&ProjectQuery::default(), &request)
                        .await
                        .map_err(|error| refused(destination, error))?;
                    fits(page.items.len(), request.limit)
                        .map_err(|error| refused(destination, error))?;
                    for project in &page.items {
                        if wanted.found(&project.title, &project.metadata) {
                            return Ok(Some(project.id.clone()));
                        }
                    }
                    page.next
                }
                Level::Document => {
                    let page = destination
                        .source()
                        .query_documents(&DocumentQuery::default(), &request)
                        .await
                        .map_err(|error| refused(destination, error))?;
                    fits(page.items.len(), request.limit)
                        .map_err(|error| refused(destination, error))?;
                    for document in &page.items {
                        if wanted.found(&document.title, &document.metadata) {
                            return Ok(Some(document.id.clone()));
                        }
                    }
                    page.next
                }
            };
            unrepeated(
                next.as_ref(),
                asked.as_ref(),
                "the destination was being scanned for the item to update",
            )
            .map_err(|error| refused(destination, error))?;
            match next {
                Some(next) => cursor = Some(next),
                None => return Ok(None),
            }
        }
    }

    /// Write one planned item, or say what a dry run would have done.
    ///
    /// Answers with what the destination held there beforehand as well, which is what
    /// makes an item written twice restorable to what it was rather than to what this
    /// copy's first pass left.
    async fn land(
        &self,
        destination: &ResolvedSource,
        request: &CopyRequest,
        item: &Planned,
        project: Option<NativeId>,
        pointing: Pointing<'_>,
        journal: &mut Journal,
    ) -> Result<(CopyOutcome, Option<Prior>), EngineError> {
        let Pointing { edges, delivers } = pointing;
        let (target, found) = match &item.target {
            Target::Update { id, found } => (Some(id.clone()), Some(*found)),
            Target::Create => (None, None),
        };
        // Every link is settled once the whole copy has landed, by `Engine::link`, so none
        // is known here.
        let reached = |destination: Option<GlobalId>, changed: bool| match (found, destination) {
            (Some(via), Some(destination)) if changed => CopyAction::Updated {
                destination,
                via: via.via(),
                link: None,
            },
            (Some(via), Some(destination)) => CopyAction::Unchanged {
                destination,
                via: via.via(),
                link: None,
            },
            (_, destination) => CopyAction::Created {
                destination,
                via: NoCounterpart::Created,
                link: None,
            },
        };
        // The one read of the destination item, made where its target was found, used to
        // decide whether the write changes anything and — if the copy cannot finish — to
        // put that item back.
        let prior = item.held.clone();
        let edges = resolved(edges);
        let qualified = |native: NativeId| GlobalId::new(destination.name().clone(), native);
        if let Some(id) = &target
            && !changes(
                prior.as_ref(),
                item,
                id,
                &project,
                &edges,
                delivers,
                destination.name(),
            )
        {
            return Ok((
                CopyOutcome {
                    source: item.source.clone(),
                    action: reached(Some(qualified(id.clone())), false),
                    placed: item.placed.clone(),
                },
                prior,
            ));
        }
        if request.dry_run {
            return Ok((
                CopyOutcome {
                    source: item.source.clone(),
                    // Null only for a create here: nothing was created, so there is no id to
                    // report.
                    action: reached(target.map(qualified), true),
                    placed: item.placed.clone(),
                },
                prior,
            ));
        }
        let written = qualified(
            self.write(
                destination,
                item,
                target,
                project,
                &edges,
                delivers,
                prior.clone(),
                journal,
            )
            .await?,
        );
        Ok((
            CopyOutcome {
                source: item.source.clone(),
                action: reached(Some(written), true),
                placed: item.placed.clone(),
            },
            prior,
        ))
    }

    /// Which destination project this item is filed under, when it is filed at all.
    ///
    /// A task copied as part of a project copy is filed under that project's counterpart,
    /// which the copy has just established. A task copied on its own has to find it.
    async fn filed(
        &self,
        request: &CopyRequest,
        item: &Planned,
        project: Option<NativeId>,
        running: &mut Running,
        journal: &mut Journal,
    ) -> Result<Option<NativeId>, EngineError> {
        match (&item.item, project) {
            (Item::Task(task), None) => {
                self.counterpart(request, item, task.project.as_ref(), running, journal)
                    .await
            }
            (Item::Document(document), None) => {
                self.counterpart(request, item, document.project.as_ref(), running, journal)
                    .await
            }
            (Item::Task(_) | Item::Document(_), filed) => Ok(filed),
            (Item::Project(_), _) => Ok(None),
        }
    }

    /// The destination project this task's own project corresponds to, when there is one.
    ///
    /// A task copied on its own keeps its source's project id when the destination holds
    /// no counterpart: the field is opaque to this engine, and dropping it would lose
    /// what the source said.
    ///
    /// Looked for once per source project and landing source per command. A project this
    /// command itself copied is already known, and one an earlier task of this command was
    /// filed under was already looked for; walking the destination again for either would
    /// spend a scan per task on an answer the command holds. A project whose own link names
    /// its counterpart there is found by that link — see [`Engine::linked_project`] — and
    /// the destination is walked only when it does not.
    ///
    /// A routed item whose project's home landed in another source than it does is filed
    /// under that home's member project in its own source — see [`Engine::member_in`].
    async fn counterpart(
        &self,
        request: &CopyRequest,
        item: &Planned,
        project: Option<&NativeId>,
        running: &mut Running,
        journal: &mut Journal,
    ) -> Result<Option<NativeId>, EngineError> {
        let Some(project) = project else {
            return Ok(None);
        };
        let qualified = GlobalId::new(item.source.source.clone(), project.clone());
        let key = (qualified.to_string(), item.to.clone());
        if let Some(looked) = running.filings.get(&key) {
            return Ok(looked.clone());
        }
        let filed = match self.find_home(request, &qualified, running).await? {
            Some(home) if home.source == item.to => Some(home.native),
            Some(home) => {
                self.member_in(request.dry_run, &home, &item.to, running, journal)
                    .await?
            }
            None => Some(project.clone()),
        };
        running.filings.insert(key, filed.clone());
        Ok(filed)
    }

    /// The counterpart a source project has in some source this copy can reach — its home —
    /// when it has one: the one this command landed, the one its own link names, or the one a
    /// walk of each reachable source finds recording it as its origin.
    async fn find_home(
        &self,
        request: &CopyRequest,
        project: &GlobalId,
        running: &Running,
    ) -> Result<Option<GlobalId>, EngineError> {
        if let Some(landed) = running.counterparts.get(&project.to_string()) {
            return Ok(Some(landed.clone()));
        }
        if let Some(linked) = self.linked_project(request, project).await? {
            return Ok(Some(linked));
        }
        for reachable in self.routes.reachable(&request.destination) {
            let there = self.writable(&reachable)?;
            if let Some(found) = self
                .scan(there, Level::Project, &Wanted::Origin(project.to_string()))
                .await?
            {
                return Ok(Some(GlobalId::new(reachable, found)));
            }
        }
        Ok(None)
    }

    /// The member project `home` names in `to`, creating it when it names none there.
    ///
    /// Found through the home's `onetaskgraph.members`, read once per command. A member it
    /// names that `to` no longer holds is replaced rather than refused: the member exists
    /// only to file the home's routed tasks, and nothing a person wrote lives on it. A dry
    /// run creates nothing, so a member it would create files its tasks under nothing.
    async fn member_in(
        &self,
        dry_run: bool,
        home: &GlobalId,
        to: &SourceName,
        running: &mut Running,
        journal: &mut Journal,
    ) -> Result<Option<NativeId>, EngineError> {
        let key = home.to_string();
        if !running.homes.contains_key(&key) {
            let at = self.writable(&home.source)?;
            let Some(held) = at
                .source()
                .get_project(&home.native)
                .await
                .map_err(|error| refused(at, error))?
            else {
                return Ok(None);
            };
            running
                .homes
                .insert(key.clone(), Home::of(home.clone(), &held)?);
        }
        let named = running.homes[&key]
            .members
            .iter()
            .find(|member| &member.source == to)
            .cloned();
        let there = self.writable(to)?;
        if let Some(member) = named
            && there
                .source()
                .get_project(&member.native)
                .await
                .map_err(|error| refused(there, error))?
                .is_some()
        {
            return Ok(Some(member.native));
        }
        if dry_run {
            return Ok(None);
        }
        let held = &running.homes[&key];
        let member = Project {
            id: home.native.clone(),
            title: held.title.clone(),
            content: None,
            status: held.status.clone(),
            labels: Vec::new(),
            url: None,
            location: None,
            created_at: None,
            updated_at: None,
            metadata: BTreeMap::from([(
                MetadataKey::MEMBER_OF_KEY.to_owned(),
                Value::String(home.to_string()),
            )]),
            repositories: Vec::new(),
        };
        let created = there
            .source()
            .write_project(&ItemWrite {
                target: None,
                item: member,
                depends_on: Vec::new(),
            })
            .await
            .map_err(|error| refused(there, error))?;
        journal.record(Undo::Created {
            at: to.clone(),
            kind: Level::Project,
            id: created.clone(),
        });
        let entry = running
            .homes
            .get_mut(&key)
            .expect("the home was recorded above");
        entry.members.retain(|member| &member.source != to);
        entry
            .members
            .push(GlobalId::new(to.clone(), created.clone()));
        entry.grew = true;
        Ok(Some(created))
    }

    /// Write each home's grown `onetaskgraph.members` through its own project write, once the
    /// whole copy has landed.
    ///
    /// The home is read as it now stands and written back with that one entry changed, and
    /// the read is what the journal restores when the copy cannot finish — unless the copy
    /// already holds an earlier entry for it, which is what it was before this copy began.
    async fn record_members(
        &self,
        homes: BTreeMap<String, Home>,
        journal: &mut Journal,
    ) -> Result<(), EngineError> {
        for home in homes.into_values().filter(|home| home.grew) {
            let at = self.writable(&home.id.source)?;
            let Some(prior) = self.prior(at, Level::Project, &home.id.native).await? else {
                continue;
            };
            let Item::Project(mut project) = prior.item.clone() else {
                continue;
            };
            project.metadata.insert(
                MetadataKey::MEMBERS_KEY.to_owned(),
                Value::Array(
                    home.members
                        .iter()
                        .map(|member| Value::String(member.to_string()))
                        .collect(),
                ),
            );
            let edges = prior.edges.clone();
            at.source()
                .check_status_write(
                    ItemKind::Project,
                    project.status.category,
                    Some(&home.id.native),
                )
                .await
                .map_err(|error| refused(at, error))?;
            journal.record(Undo::Updated {
                at: home.id.source.clone(),
                id: home.id.native.clone(),
                prior,
            });
            at.source()
                .write_project(&ItemWrite {
                    target: Some(home.id.native.clone()),
                    item: *project,
                    depends_on: edges,
                })
                .await
                .map_err(|error| refused(at, error))?;
        }
        Ok(())
    }

    /// The home project a source project's own `onetaskgraph.copies` names in a source this
    /// copy can reach, when the project it names there still records that source project as
    /// its origin.
    ///
    /// The filing lookup's half of following a link: one read of the project at its source
    /// and one of its counterpart by id, in place of a walk of the destination's projects.
    /// A link that names nothing there, or names a project somebody re-pointed, is not a
    /// refusal here as it is for a copy's own target — this only decides where an item is
    /// filed — so the walk answers instead, exactly as it did before there were links.
    async fn linked_project(
        &self,
        request: &CopyRequest,
        project: &GlobalId,
    ) -> Result<Option<GlobalId>, EngineError> {
        let source = self.readable(&project.source)?;
        if !source.source().capabilities().projects.is_native() {
            return Ok(None);
        }
        let Some(held) = source
            .source()
            .get_project(&project.native)
            .await
            .map_err(|error| refused(source, error))?
        else {
            return Ok(None);
        };
        for reachable in self.routes.reachable(&request.destination) {
            let Some(link) = link_of(&held.metadata, &reachable) else {
                continue;
            };
            let there = self.writable(&reachable)?;
            let named = there
                .source()
                .get_project(&link.native)
                .await
                .map_err(|error| refused(there, error))?;
            if named.is_some_and(|there| origin_of(&there.metadata).as_ref() == Some(project)) {
                return Ok(Some(link));
            }
        }
        Ok(None)
    }

    /// What the destination holds at one id, item and forward edges together.
    ///
    /// One read for both purposes it serves — deciding whether a write changes anything,
    /// and putting the item back if the copy cannot finish — because a second read of the
    /// same item is a second round trip against a hosted destination for nothing.
    async fn prior(
        &self,
        destination: &ResolvedSource,
        kind: Level,
        id: &NativeId,
    ) -> Result<Option<Prior>, EngineError> {
        let held = match kind {
            Level::Task => destination
                .source()
                .get_task(id)
                .await
                .map_err(|error| refused(destination, error))?
                .map(|task| Item::Task(Box::new(task))),
            Level::Project => destination
                .source()
                .get_project(id)
                .await
                .map_err(|error| refused(destination, error))?
                .map(|project| Item::Project(Box::new(project))),
            Level::Document => destination
                .source()
                .get_document(id)
                .await
                .map_err(|error| refused(destination, error))?
                .map(|document| Item::Document(Box::new(document))),
        };
        let Some(item) = held else {
            return Ok(None);
        };
        let edges = forward_edges(destination, id, kind).await?;
        let recorded =
            assets::recorded(described(&item).1).map_err(|error| refused(destination, error))?;
        let assets = held_assets(destination, kind, id, recorded.as_ref()).await?;
        Ok(Some(Prior {
            item,
            edges,
            assets,
            recorded,
        }))
    }

    /// Hand one item to the destination's own write interface, recording how to take it
    /// back.
    // llmlint: ignore[suppressions_justified] A write is the item, where it is going, what
    // it is filed under, its edges, what was there before and the journal that records how
    // to put it back. Each is a distinct decision made by a different part of the copy, and
    // grouping them would only move the argument list to a constructor.
    #[allow(clippy::too_many_arguments)]
    async fn write(
        &self,
        destination: &ResolvedSource,
        item: &Planned,
        target: Option<NativeId>,
        project: Option<NativeId>,
        edges: &[DependencyEdge],
        delivers: &[TaskRef],
        prior: Option<Prior>,
        journal: &mut Journal,
    ) -> Result<NativeId, EngineError> {
        let created_kind = item.item.level();
        let suggested = target
            .clone()
            .unwrap_or_else(|| created_id(&item.item, project.as_ref()));
        // Settled before the journal takes `prior`, and from that same read: what the
        // destination holds at the origin key is what a copy-back leaves there, and what it
        // holds as `delivered_by` is what the item keeps.
        let origin = recorded(item, prior.as_ref());
        let landing = outgoing(item, suggested, project, &origin, delivers, prior.as_ref());
        // Asked before the journal records anything or the destination is sent anything: a
        // status the destination has no name for refuses this write while there is nothing to
        // put back, where refused inside the write it would leave the journal restoring an
        // item the write never touched.
        check_status(destination, &landing, target.as_ref()).await?;
        // Recorded *before* the write rather than after it. A destination's own write is
        // several calls — `docs/plugin-protocol.md` §4.9 — and one of them failing leaves
        // the ones before it applied. No source can put those back, because only this
        // journal holds what was there; recorded after a successful write, an update that
        // stopped part way was the one way a copy could end and leave the destination
        // altered. A restore of an item the write never reached rewrites what is already
        // there, which costs one mutation and is what "either complete or it never
        // happened" is worth.
        // A write carries assets when the item's content references any, and when the item it
        // overwrites holds any: what lands then holds exactly the assets this copy carries.
        let held_assets = prior.as_ref().and_then(|prior| prior.assets.clone());
        let carrying = !item.assets.is_empty() || held_assets.is_some();
        let recorded = prior.as_ref().and_then(|prior| prior.recorded.clone());
        if let (Some(id), Some(mut prior)) = (target.clone(), prior) {
            if carrying && prior.assets.is_none() {
                // Putting this item back then has to take away the assets this write adds.
                prior.assets = Some(Vec::new());
            }
            journal.record(Undo::Updated {
                at: destination.name().clone(),
                id,
                prior,
            });
        }
        let landed = match landing {
            Item::Task(task) if carrying => destination
                .source()
                .write_task_with_assets(
                    &ItemWrite {
                        target: target.clone(),
                        item: *task,
                        depends_on: edges.to_vec(),
                    },
                    None,
                    &assets::write_of(item.assets.clone(), recorded.clone()),
                )
                .await
                .map(|written| written.id)
                .map_err(|error| refused(destination, error))?,
            Item::Document(document) if carrying => destination
                .source()
                .write_document_with_assets(
                    &ItemWrite {
                        target: target.clone(),
                        item: *document,
                        depends_on: Vec::new(),
                    },
                    None,
                    &assets::write_of(item.assets.clone(), recorded.clone()),
                )
                .await
                .map(|written| written.id)
                .map_err(|error| refused(destination, error))?,
            Item::Task(task) => destination
                .source()
                .write_task(&ItemWrite {
                    target: target.clone(),
                    item: *task,
                    depends_on: edges.to_vec(),
                })
                .await
                .map_err(|error| refused(destination, error))?,
            Item::Project(project) => destination
                .source()
                .write_project(&ItemWrite {
                    target: target.clone(),
                    item: *project,
                    depends_on: edges.to_vec(),
                })
                .await
                .map_err(|error| refused(destination, error))?,
            // No edges, and that is the contract: a document takes part in no dependency
            // graph, so there is nothing here for `depends_on` to carry.
            Item::Document(document) => destination
                .source()
                .write_document(&ItemWrite {
                    target: target.clone(),
                    item: *document,
                    depends_on: Vec::new(),
                })
                .await
                .map_err(|error| refused(destination, error))?,
        };
        // A created item can only be journalled here: its id is what the write answers
        // with. A create that fails leaves nothing behind — §4.9 makes taking the item
        // back the source's own duty, because a write that refused must not leave an item
        // nobody asked for.
        if target.is_none() {
            journal.record(Undo::Created {
                at: destination.name().clone(),
                kind: created_kind,
                id: landed.clone(),
            });
        }
        Ok(landed)
    }

    /// A configured source that built, for reading an item out of.
    fn readable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
        let name = self.known(name)?;
        if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
            return Err(EngineError::SourceRefused {
                name: name.to_string(),
                error: unavailable.error().clone(),
            });
        }
        self.ready()
            .find(|source| source.name() == &name)
            .ok_or(EngineError::NoSources)
    }
}

/// How many tasks of a project are read at once while walking it.
const PROJECT_PAGE: std::num::NonZeroU32 = std::num::NonZeroU32::new(50).expect("50 is not zero");

/// Each source's own running totals, in the order the sources were given.
///
/// A reading a source could not take is read as that source not metering: what a command
/// spent is a report about the work, and a failed reading must not become a failure of the
/// work itself. That holds when only one of a source's two readings failed as well, since
/// a difference needs both ends.
pub(super) async fn readings(sources: &[&ResolvedSource]) -> Vec<Option<Metering>> {
    let mut read = Vec::with_capacity(sources.len());
    for source in sources {
        read.push(source.source().metering().await.ok().flatten());
    }
    read
}

/// What the sources spent between two readings of them, or `None` when none of them meters.
///
/// A source that does not meter contributes nothing and does not make the total zero: a
/// command whose sources all declined reports that it cannot say, not that it spent nothing.
/// A budget is keyed by its name and unit together, so two sources naming one budget add up
/// and two units of one budget stay apart.
///
/// A source's two readings are a plugin's word, so they are held to [`Metering`]'s contract
/// before either is believed, and a pair that breaks it is that source not metering — see
/// [`difference`].
pub(super) fn spent_between(
    before: &[Option<Metering>],
    after: &[Option<Metering>],
) -> Option<Spent> {
    let mut metered = false;
    let mut requests = 0_u64;
    let mut budgets: BTreeMap<(String, String), (u64, u64)> = BTreeMap::new();
    for (before, after) in before.iter().zip(after) {
        let (Some(before), Some(after)) = (before, after) else {
            continue;
        };
        let Some((sent, spent)) = difference(before, after) else {
            continue;
        };
        metered = true;
        requests = requests.saturating_add(sent);
        for (key, measured, modelled) in spent {
            let total = budgets.entry(key).or_default();
            total.0 = total.0.saturating_add(measured).saturating_add(modelled);
            total.1 = total.1.saturating_add(modelled);
        }
    }
    metered.then(|| Spent {
        requests,
        budgets: budgets
            .into_iter()
            .map(|((budget, unit), (amount, modelled))| BudgetSpent {
                budget,
                unit,
                amount,
                lower_bound: modelled > 0,
            })
            .collect(),
    })
}

/// One budget's name and unit, and the measured and modelled amounts spent against it.
type BudgetDifference = ((String, String), u64, u64);

/// What one source sent and spent between two of its readings, or `None` when the pair
/// cannot be a running total's.
///
/// A running total never falls, never drops a budget it has named, and names every budget
/// it keeps, once; a pair that breaks any of that is a source that reset its figures or
/// reported something else, and a difference taken over it would be a number that measures
/// nothing. Such a source is reported as not metering rather than as having spent a clamped
/// zero.
fn difference(before: &Metering, after: &Metering) -> Option<(u64, Vec<BudgetDifference>)> {
    let sent = after.requests.checked_sub(before.requests)?;
    for reading in [before, after] {
        let mut named = BTreeSet::new();
        for budget in &reading.budgets {
            if budget.budget.is_empty()
                || budget.unit.is_empty()
                || !named.insert((&budget.budget, &budget.unit))
            {
                return None;
            }
        }
    }
    let held = |from: &Metering, budget: &onetaskgraph_plugin_api::Metered| {
        from.budgets
            .iter()
            .find(|held| held.budget == budget.budget && held.unit == budget.unit)
            .cloned()
    };
    if before
        .budgets
        .iter()
        .any(|budget| held(after, budget).is_none())
    {
        return None;
    }
    let mut spent = Vec::with_capacity(after.budgets.len());
    for budget in &after.budgets {
        let earlier = held(before, budget);
        let measured = budget
            .measured
            .checked_sub(earlier.as_ref().map_or(0, |held| held.measured))?;
        let modelled = budget
            .modelled
            .checked_sub(earlier.as_ref().map_or(0, |held| held.modelled))?;
        spent.push((
            (budget.budget.clone(), budget.unit.clone()),
            measured,
            modelled,
        ));
    }
    Some((sent, spent))
}

/// One page request against `source`, at the largest page it will serve.
fn request_for(source: &ResolvedSource, cursor: Option<Cursor>) -> PageRequest {
    PageRequest {
        cursor,
        limit: source.source().capabilities().max_page_size.max(1),
    }
}

/// Whether writing this item would change what the destination already holds.
///
/// A free function over the state already read rather than a method that reads it again:
/// the same answer is wanted where the item is landed and where a repeat copy of a project
/// decides whether it settled, and a second read there is a second round trip for nothing.
fn changes(
    held: Option<&Prior>,
    item: &Planned,
    target: &NativeId,
    project: &Option<NativeId>,
    edges: &[DependencyEdge],
    delivers: &[TaskRef],
    destination: &SourceName,
) -> bool {
    let Some(held) = held else {
        return true;
    };
    let mut outgoing = outgoing(
        item,
        target.clone(),
        project.clone(),
        &recorded(item, Some(held)),
        delivers,
        Some(held),
    );
    if (!item.assets.is_empty() || held.assets.is_some())
        && !lands_as_held(item, held, &mut outgoing)
    {
        return true;
    }
    !same(&held.item, &outgoing, destination) || !same_edges(&held.edges, edges)
}

/// Whether `held` already holds exactly the assets `item` carries, by name and digest — and,
/// when it does, `outgoing` made what the destination would store of it.
///
/// A destination that serves its assets at a URL says what it holds in its own record,
/// `onetaskgraph.assets`, and would rewrite the references to the URLs recorded there; one
/// that keeps them beside the record says so in the assets it lists, and stores the references
/// as written.
fn lands_as_held(item: &Planned, held: &Prior, outgoing: &mut Item) -> bool {
    let Some(recorded) = &held.recorded else {
        let holds: Vec<onetaskgraph_plugin_api::Asset> = held
            .assets
            .iter()
            .flatten()
            .map(|payload| onetaskgraph_plugin_api::Asset {
                name: payload.name.clone(),
                sha256: payload.sha256.clone(),
                content_type: payload.content_type,
                path: None,
            })
            .collect();
        return assets::same_set(&holds, &item.assets);
    };
    if recorded.0.len() != item.assets.len() {
        return false;
    }
    let mut served = AssetUploads::default();
    for payload in &item.assets {
        let Some(url) = recorded.reusable(&payload.name, &payload.sha256) else {
            return false;
        };
        served.0.insert(
            payload.name.clone(),
            onetaskgraph_plugin_api::AssetUpload {
                sha256: payload.sha256.clone(),
                url: url.to_owned(),
            },
        );
    }
    let (content, metadata) = match outgoing {
        Item::Task(task) => (&mut task.content, &mut task.metadata),
        Item::Document(document) => (&mut document.content, &mut document.metadata),
        Item::Project(_) => return true,
    };
    if let Some(text) = content {
        *text = onetaskgraph_plugin_api::serve_asset_references(text, metadata, &served);
    }
    true
}

/// The image assets `item`'s content references, each with its bytes as `source` holds them,
/// in the order the content first references them — none for a project.
///
/// # Errors
///
/// [`EngineError::AssetNotHeld`] naming the record and the asset for a reference `source` holds
/// no asset for, and [`EngineError::SourceRefused`] when it cannot answer.
async fn carried_assets(
    source: &ResolvedSource,
    id: &GlobalId,
    item: &Item,
) -> Result<Vec<AssetPayload>, EngineError> {
    let (content, owner) = match item {
        Item::Task(task) => (task.content.as_deref(), assets::Owner::Task),
        Item::Document(document) => (document.content.as_deref(), assets::Owner::Document),
        Item::Project(_) => return Ok(Vec::new()),
    };
    let referenced = onetaskgraph_plugin_api::asset_references(content.unwrap_or_default());
    if referenced.is_empty() {
        return Ok(Vec::new());
    }
    let listed = match owner {
        assets::Owner::Document => source.source().document_assets(&id.native).await,
        assets::Owner::Task => source.source().task_assets(&id.native).await,
    }
    .map_err(|error| refused(source, error))?;
    let mut carried = Vec::with_capacity(referenced.len());
    for name in referenced {
        let not_held = || EngineError::AssetNotHeld {
            record: id.to_string(),
            asset: name.to_string(),
        };
        if !listed.iter().any(|asset| asset.name == name) {
            return Err(not_held());
        }
        let bytes = match owner {
            assets::Owner::Document => source.source().document_asset(&id.native, &name).await,
            assets::Owner::Task => source.source().task_asset(&id.native, &name).await,
        }
        .map_err(|error| refused(source, error))?
        .ok_or_else(not_held)?;
        carried.push(AssetPayload::of(name, bytes));
    }
    Ok(carried)
}

/// The image assets the item `id` of kind `kind` holds at `destination`, as [`Prior::assets`]
/// keeps them: `None` for a destination that stores none, for a project, and for an item that
/// holds none.
async fn held_assets(
    destination: &ResolvedSource,
    kind: Level,
    id: &NativeId,
    recorded: Option<&AssetUploads>,
) -> Result<Option<Vec<AssetPayload>>, EngineError> {
    let owner = match kind {
        Level::Task => assets::Owner::Task,
        Level::Document => assets::Owner::Document,
        Level::Project => return Ok(None),
    };
    if !destination.source().capabilities().assets.is_native() {
        return Ok(None);
    }
    let listed = match owner {
        assets::Owner::Document => destination.source().document_assets(id).await,
        assets::Owner::Task => destination.source().task_assets(id).await,
    }
    .map_err(|error| refused(destination, error))?;
    if listed.is_empty() {
        // A destination that lists none may still serve some — one reached over the stdio
        // protocol lists nothing — and its own record of what it uploaded is then what it holds,
        // every one of them reusable by its digest without reading a byte back.
        return Ok(recorded
            .filter(|recorded| !recorded.0.is_empty())
            .map(|recorded| {
                recorded
                    .0
                    .iter()
                    .map(|(name, upload)| AssetPayload {
                        name: name.clone(),
                        sha256: upload.sha256.clone(),
                        content_type: name.content_type(),
                        bytes: None,
                    })
                    .collect()
            }));
    }
    let mut held = Vec::with_capacity(listed.len());
    for asset in listed {
        // Bytes the destination's own record says it serves are not read back: putting them
        // back reuses what it serves.
        let bytes = if recorded
            .is_some_and(|recorded| recorded.reusable(&asset.name, &asset.sha256).is_some())
        {
            None
        } else {
            match owner {
                assets::Owner::Document => {
                    destination.source().document_asset(id, &asset.name).await
                }
                assets::Owner::Task => destination.source().task_asset(id, &asset.name).await,
            }
            .map_err(|error| refused(destination, error))?
        };
        held.push(AssetPayload {
            name: asset.name,
            sha256: asset.sha256,
            content_type: asset.content_type,
            bytes,
        });
    }
    Ok(Some(held))
}

/// Remove one item this copy created, through the destination's own write interface.
async fn remove(
    destination: &ResolvedSource,
    kind: Level,
    id: &NativeId,
) -> Result<(), SourceError> {
    match kind {
        Level::Task => destination.source().delete_task(id).await,
        Level::Project => destination.source().delete_project(id).await,
        Level::Document => destination.source().delete_document(id).await,
    }
}

/// Write one item back exactly as the destination held it before this copy.
async fn restore(
    destination: &ResolvedSource,
    id: &NativeId,
    prior: &Prior,
) -> Result<(), SourceError> {
    if let Some(held) = &prior.assets {
        let assets = AssetWrite {
            assets: held.clone(),
            recorded_assets: prior.recorded.clone(),
        };
        match &prior.item {
            Item::Task(task) => {
                return destination
                    .source()
                    .write_task_with_assets(
                        &ItemWrite {
                            target: Some(id.clone()),
                            item: (**task).clone(),
                            depends_on: prior.edges.clone(),
                        },
                        None,
                        &assets,
                    )
                    .await
                    .map(|_| ());
            }
            Item::Document(document) => {
                return destination
                    .source()
                    .write_document_with_assets(
                        &ItemWrite {
                            target: Some(id.clone()),
                            item: (**document).clone(),
                            depends_on: Vec::new(),
                        },
                        None,
                        &assets,
                    )
                    .await
                    .map(|_| ());
            }
            Item::Project(_) => {}
        }
    }
    match &prior.item {
        Item::Task(task) => destination
            .source()
            .write_task(&ItemWrite {
                target: Some(id.clone()),
                item: (**task).clone(),
                depends_on: prior.edges.clone(),
            })
            .await
            .map(|_| ()),
        Item::Project(project) => destination
            .source()
            .write_project(&ItemWrite {
                target: Some(id.clone()),
                item: (**project).clone(),
                depends_on: prior.edges.clone(),
            })
            .await
            .map(|_| ()),
        Item::Document(document) => destination
            .source()
            .write_document(&ItemWrite {
                target: Some(id.clone()),
                item: (**document).clone(),
                depends_on: Vec::new(),
            })
            .await
            .map(|_| ()),
    }
}

/// Hold `value` at `onetaskgraph.copies` on one item, through its source's narrow metadata
/// write — `false` when the source holds no such item.
async fn set_link(
    source: &ResolvedSource,
    level: Level,
    id: &NativeId,
    value: &Value,
) -> Result<bool, SourceError> {
    let key = MetadataKey::copies();
    Ok(match level {
        Level::Task => source
            .source()
            .set_task_metadata(id, &key, value)
            .await?
            .is_some(),
        Level::Project => source
            .source()
            .set_project_metadata(id, &key, value)
            .await?
            .is_some(),
        Level::Document => source
            .source()
            .set_document_metadata(id, &key, value)
            .await?
            .is_some(),
    })
}

/// The metadata of either kind of item, to edit.
fn metadata_of(item: &mut Item) -> &mut BTreeMap<String, Value> {
    match item {
        Item::Task(task) => &mut task.metadata,
        Item::Project(project) => &mut project.metadata,
        Item::Document(document) => &mut document.metadata,
    }
}

/// Refuse a document copy addressed to a source that declares it has none.
///
/// Read off the declaration rather than by asking, which is what "not asked" means: the
/// engine learned at the handshake that this source holds no documents, so it refuses
/// naming the source and its plugin instead of sending a read that would be refused there.
/// Applied at both ends of a copy — a source with no documents holds nothing to copy out,
/// and a destination with none has nowhere to put one.
fn documentary(source: &ResolvedSource) -> Result<(), EngineError> {
    if source.source().capabilities().documents.is_native() {
        return Ok(());
    }
    Err(EngineError::NoDocuments {
        name: source.name().to_string(),
        kind: source.kind().to_owned(),
    })
}

/// One source failing while a copy was mid-flight.
fn refused(source: &ResolvedSource, error: SourceError) -> EngineError {
    EngineError::SourceRefused {
        name: source.name().to_string(),
        error,
    }
}

/// Every forward edge at one item, walked to exhaustion one page at a time.
pub(super) async fn forward_edges(
    source: &ResolvedSource,
    id: &NativeId,
    kind: Level,
) -> Result<Vec<DependencyEdge>, EngineError> {
    // A document has no edges to walk, and asking for them would mean asking a source for
    // a graph the contract says nothing may point into.
    if kind == Level::Document {
        return Ok(Vec::new());
    }
    let mut edges = Vec::new();
    let mut cursor: Option<Cursor> = None;
    loop {
        let asked = cursor.clone();
        let request = request_for(source, cursor);
        let page = match kind {
            Level::Task | Level::Document => {
                source
                    .source()
                    .task_dependencies(id, Direction::DependsOn, &request)
                    .await
            }
            Level::Project => {
                source
                    .source()
                    .project_dependencies(id, Direction::DependsOn, &request)
                    .await
            }
        }
        .map_err(|error| refused(source, error))?;
        fits(page.items.len(), request.limit).map_err(|error| refused(source, error))?;
        edges.extend(page.items);
        unrepeated(
            page.next.as_ref(),
            asked.as_ref(),
            "an item's dependencies were being read for a copy",
        )
        .map_err(|error| refused(source, error))?;
        match page.next {
            Some(next) => cursor = Some(next),
            None => return Ok(edges),
        }
    }
}

/// The location string one record reports, when it reports a usable one.
///
/// Either variant's own `String`, and `None` for a record the source gave no location for
/// or gave an empty string for: there is nothing to look for in a document's content and
/// nothing to point a reader at.
fn located(location: Option<&Location>) -> Option<String> {
    let (Location::Path(held) | Location::Url(held)) = location?;
    (!held.is_empty()).then(|| held.clone())
}

/// Record one candidate referent, when its source said where it is.
fn note(
    into: &mut Vec<Referent>,
    id: GlobalId,
    level: Level,
    location: Option<&Location>,
    metadata: &BTreeMap<String, Value>,
) {
    if let Some(location) = located(location) {
        into.push(Referent {
            id,
            origin: origin_of(metadata),
            level,
            location,
        });
    }
}

/// What every location string this document names becomes, longest first.
///
/// Longest first because a shorter location may start where a longer one does — a
/// project's directory and a task's file under it — and the longer of the two is the
/// record that occurrence names.
fn table_for(referents: &[Referent], counterparts: &Counterparts) -> Vec<(String, Resolution)> {
    let mut table: Vec<(String, Resolution)> = Vec::new();
    for referent in referents {
        // Two referents reporting one location string: an occurrence of it cannot be
        // attributed to either, and a rewrite would be *confidently wrong* rather than
        // merely unhelpful. So neither is chosen and both occurrences are counted.
        if let Some(held) = table
            .iter_mut()
            .find(|(location, _)| location == &referent.location)
        {
            held.1 = Resolution::Ambiguous;
            continue;
        }
        table.push((referent.location.clone(), counterparts.resolve(referent)));
    }
    table.sort_by_key(|(location, _)| std::cmp::Reverse(location.len()));
    table
}

/// Whether `content` holds `location` at least once, stopped on both sides.
fn holds(content: &str, location: &str) -> bool {
    (0..content.len()).any(|at| delimited_at(content, at, location))
}

/// Whether `location` occurs at `at` **stopped on both sides** — by
/// [`stops_a_location`], or by the end of the content — rather than as part of a longer
/// location-like string.
///
/// A location string occurring inside a longer one is a different string naming a
/// different record: `/…/tasks/p/t.md` must not be rewritten inside `/…/tasks/p/t.md.bak`,
/// `https://example.invalid/1` must not be rewritten inside `https://example.invalid/12`,
/// and a project's location that is a directory prefix of a task's must not be rewritten
/// inside that task's. What deciding it this way costs is stated on [`stops_a_location`].
fn delimited_at(content: &str, at: usize, location: &str) -> bool {
    if !content.is_char_boundary(at) || !content[at..].starts_with(location) {
        return false;
    }
    let before = content[..at].chars().next_back();
    let after = content[at + location.len()..].chars().next();
    stops_a_location(before) && stops_a_location(after)
}

/// Whether a character cannot continue a path or a link, so a location string beside one
/// ends there.
///
/// Stated as what *stops* a location rather than as what one may contain, because the
/// second list is unbounded — a path may hold very nearly any byte, and a URL more. Every
/// character not named here continues, which is what leaves the three cases above alone;
/// the end of the content counts as a stop. The set is what the artifact this exists for
/// really wraps a bare path in — a backtick in a table cell — plus the delimiters prose
/// and Markdown put next to one.
///
/// **Sentence punctuation is deliberately absent, and that is a stated cost rather than an
/// oversight.** `.`, `!`, `?` and `:` each equally *continue* a real location — `/…/t.md`
/// and `/…/t.md.bak` are two files, `…/1` and `…/1?q=2` two pages — so admitting them as
/// stops would rewrite one record's location into another's. The price is that a location
/// written bare at the end of a sentence is not recognised at all: its text is left
/// byte-for-byte and it is counted in neither figure, exactly as a reference to another
/// project's record is. That is the direction to be wrong in, because this edits the
/// content of somebody's document, where a confidently wrong rewrite is worse than one
/// that never happens.
fn stops_a_location(character: Option<char>) -> bool {
    match character {
        None => true,
        Some(character) => character.is_whitespace() || "`\"'()[]{}<>|,;".contains(character),
    }
}

/// One document's content with every whole reference rewritten, and what that took.
///
/// A location string with no confident counterpart is left **byte-for-byte** as it was
/// rather than removed or guessed at, and so is every character of the content that is not
/// a rewritten reference. Nothing is added and nothing is reformatted.
fn substitute(content: &str, table: &[(String, Resolution)]) -> (String, Counted) {
    let mut written = String::with_capacity(content.len());
    let mut counts = Counted::default();
    let mut at = 0;
    while at < content.len() {
        if let Some((location, resolution)) = table
            .iter()
            .find(|(location, _)| delimited_at(content, at, location))
        {
            match resolution {
                Resolution::Rewrite(there) => {
                    written.push_str(there);
                    counts.rewritten += 1;
                }
                // Both left-alone outcomes count as unresolved on the branch that counts
                // them, which is what really holds `ambiguous` at or below `unresolved`.
                Resolution::NoCounterpart => {
                    written.push_str(location);
                    counts.unresolved += 1;
                }
                Resolution::Ambiguous => {
                    written.push_str(location);
                    counts.unresolved += 1;
                    counts.ambiguous += 1;
                }
            }
            at += location.len();
            continue;
        }
        let character = content[at..]
            .chars()
            .next()
            .expect("a character at a boundary this walk only ever lands on");
        written.push(character);
        at += character.len_utf8();
    }
    (written, counts)
}

/// The entry [`Engine::rewrite_references`] records for a document it rewrote from `read`
/// to `written`, or `None` to carry what the document records verbatim.
///
/// `None` unless `read` is still the rendering its entry vouches for, because only then is
/// the one change between it and `written` this copy's own: a hand edit is never laundered.
fn restamped(
    metadata: &BTreeMap<String, Value>,
    read: &str,
    written: &str,
) -> Option<TemplateProvenance> {
    let mut provenance = TemplateProvenance::read(metadata).ok().flatten()?;
    if provenance.body_digest.as_str() != body_digest(read) {
        return None;
    }
    provenance.body_digest =
        Sha256Digest::parse(body_digest(written)).expect("`body_digest` spells a digest");
    Some(provenance)
}

/// The origin one item records, when it records a usable one.
fn origin_of(metadata: &BTreeMap<String, Value>) -> Option<GlobalId> {
    metadata
        .get(GlobalId::ORIGIN_KEY)?
        .as_str()?
        .parse::<GlobalId>()
        .ok()
}

/// The destination item one item's `onetaskgraph.copies` entry names at `destination`, when
/// it holds a usable one.
///
/// An entry that is not a qualified id, or is one naming another source than the key it is
/// filed under, is no link at all: the copy goes on by the rules that follow and rewrites
/// it to what they find.
fn link_of(metadata: &BTreeMap<String, Value>, destination: &SourceName) -> Option<GlobalId> {
    let linked = metadata
        .get(MetadataKey::COPIES_KEY)?
        .as_object()?
        .get(destination.as_str())?
        .as_str()?
        .parse::<GlobalId>()
        .ok()?;
    (&linked.source == destination).then_some(linked)
}

/// The entries of a `onetaskgraph.copies` value that are links: a source name, and a
/// qualified id of that same source.
fn well_formed(value: Option<&Value>) -> serde_json::Map<String, Value> {
    let Some(Value::Object(held)) = value else {
        return serde_json::Map::new();
    };
    held.iter()
        .filter(|(destination, linked)| {
            linked
                .as_str()
                .and_then(|linked| linked.parse::<GlobalId>().ok())
                .is_some_and(|linked| linked.source.as_str() == destination.as_str())
        })
        .map(|(destination, linked)| (destination.clone(), linked.clone()))
        .collect()
}

/// Why a value is not one a `onetaskgraph.copies` entry may hold, or nothing when it is: a
/// JSON object mapping each destination source name to a qualified id of that source.
///
/// What the stdio boundary holds an incoming value of that key to, so a plugin is never
/// handed a link it could not answer a later copy with.
pub(crate) fn malformed_links(value: &Value) -> Option<String> {
    let Value::Object(held) = value else {
        return Some(format!(
            "{} holds an object of destination source names to qualified ids, not {value}",
            MetadataKey::COPIES_KEY
        ));
    };
    let wrong = held.len() - well_formed(Some(value)).len();
    (wrong > 0).then(|| {
        format!(
            "{} holds an object of destination source names to qualified ids of that source, \
             and {wrong} of its entries is not one: {value}",
            MetadataKey::COPIES_KEY
        )
    })
}

/// The title and metadata of either kind of item.
fn described(item: &Item) -> (&str, &BTreeMap<String, Value>) {
    match item {
        Item::Task(task) => (&task.title, &task.metadata),
        Item::Project(project) => (&project.title, &project.metadata),
        Item::Document(document) => (&document.title, &document.metadata),
    }
}

/// The item as the destination should hold it.
///
/// `url`, `location`, `created_at` and `updated_at` are the destination's own and are
/// never written — where the *source* holds an item says nothing about where the
/// destination does, which is why a copied document does not arrive claiming the path or
/// the link its source reported. The two reserved keys this product encodes typed fields
/// under are removed, because those fields travel as themselves — leaving the encoding
/// beside them would have the destination hold one thing twice, and disagree with itself
/// the moment one changed.
///
/// A task's `delivers` is `delivers`, already resolved against the destination, and its
/// `delivered_by` is the one the destination holds — never the source's, and empty for an
/// item this copy creates: that list is the store's to keep, at the destination. The
/// `onetaskgraph.copies` link is kept on the same terms: it says where the *destination*
/// item was copied to, which is the destination's to hold, never the source's to send.
fn outgoing(
    item: &Planned,
    id: NativeId,
    project: Option<NativeId>,
    origin: &Origin,
    delivers: &[TaskRef],
    held: Option<&Prior>,
) -> Item {
    let own = held.map(|held| described(&held.item).1);
    let carried = |metadata: &BTreeMap<String, Value>| carried(metadata, origin, own);
    match &item.item {
        Item::Task(task) => Item::Task(Box::new(Task {
            id,
            url: None,
            location: None,
            created_at: None,
            updated_at: None,
            project,
            metadata: carried(&task.metadata),
            delivers: delivers.to_vec(),
            delivered_by: match held.map(|held| &held.item) {
                Some(Item::Task(held)) => held.delivered_by.clone(),
                _ => Vec::new(),
            },
            ..(**task).clone()
        })),
        Item::Project(project) => Item::Project(Box::new(Project {
            id,
            url: None,
            location: None,
            created_at: None,
            updated_at: None,
            metadata: carried(&project.metadata),
            ..(**project).clone()
        })),
        Item::Document(document) => Item::Document(Box::new(Document {
            id,
            url: None,
            location: None,
            created_at: None,
            updated_at: None,
            project,
            metadata: carried(&document.metadata),
            ..(**document).clone()
        })),
    }
}

/// The id a created item is offered to the destination under.
///
/// A task whose id is `<its project's id>/<rest>` is offered as `<rest>` under the
/// destination project it is filed in, so a destination whose ids are paths files it in that
/// project rather than the source's. Any other id, and every update, is left as it is. The
/// `a_project_copied_*` journeys in `crates/onetaskgraph-e2e/tests/e2e/copy.rs` hold the Markdown
/// plugin's id shape to this rule.
fn created_id(item: &Item, filed: Option<&NativeId>) -> NativeId {
    if let (Item::Task(task), Some(filed)) = (item, filed)
        && let Some(own) = &task.project
        && let Some(rest) = task
            .id
            .as_str()
            .strip_prefix(own.as_str())
            .and_then(|rest| rest.strip_prefix('/'))
        && !rest.is_empty()
    {
        return NativeId(format!("{}/{rest}", filed.as_str()));
    }
    item.id().clone()
}

/// The metadata a copy carries: the caller's own keys untouched, the origin settled, and the
/// destination's own store-kept keys — its `onetaskgraph.copies` link, and the
/// `onetaskgraph.members` or `onetaskgraph.member_of` that ties it to its member projects or
/// its home — kept as `own`, the destination item, holds them.
///
/// The key is removed before it is settled rather than overwritten, because the item being
/// copied carries an origin of its own and [`Origin::Keeps`] must not let it through. Its
/// link is removed for the same reason: it names where *that* item was copied to, and on
/// the destination it would name the destination item's own counterpart as itself. Its
/// member keys name projects tied to *it*, which say nothing about the destination's.
fn carried(
    metadata: &BTreeMap<String, Value>,
    origin: &Origin,
    own: Option<&BTreeMap<String, Value>>,
) -> BTreeMap<String, Value> {
    let mut carried = metadata.clone();
    carried.remove(Repository::METADATA_KEY);
    carried.remove(DependencyEdge::RECORDED_KEY);
    carried.remove(TaskRef::DELIVERS_KEY);
    carried.remove(TaskRef::DELIVERED_BY_KEY);
    carried.remove(GlobalId::ORIGIN_KEY);
    for kept in [
        MetadataKey::COPIES_KEY,
        MetadataKey::MEMBERS_KEY,
        MetadataKey::MEMBER_OF_KEY,
        MetadataKey::ASSETS_KEY,
    ] {
        carried.remove(kept);
        if let Some(held) = own.and_then(|own| own.get(kept)) {
            carried.insert(kept.to_owned(), held.clone());
        }
    }
    let held = match origin {
        Origin::Records(id) => Some(Value::String(id.to_string())),
        Origin::Keeps(held) => held.clone(),
    };
    if let Some(held) = held {
        carried.insert(GlobalId::ORIGIN_KEY.to_owned(), held);
    }
    carried
}

/// What one landed item records at [`GlobalId::ORIGIN_KEY`].
enum Origin {
    /// The qualified id this item was copied from, as the id type rather than as its
    /// spelling: the key holds a [`GlobalId`] and nothing else may be recorded there.
    Records(GlobalId),
    /// Whatever the destination already holds there — `None` when it holds nothing, which
    /// is written as the key being absent rather than as a null.
    ///
    /// A [`Value`] and not a [`GlobalId`], because this variant does not interpret what it
    /// carries: it is the destination's own metadata entry, held for the length of one
    /// write and put back exactly as it was read. Parsing it would turn a value a
    /// destination holds and this engine cannot read into a value this engine deletes,
    /// which is the opposite of what keeping it means.
    Keeps(Option<Value>),
}

/// Which of the two a copy of this item does.
///
/// A copy that reached its target by rule 1 is a copy-back: the item being copied names
/// the destination item, so the destination is the *original* and the id being copied
/// belongs to the copy that came out of it. Recording that id there would overwrite the
/// original's own provenance — and with it the correspondence every later copy from the
/// source it was authored in depends on. That copy would then match nothing and create a
/// second item beside the one it meant to update, which is the whole failure: nothing is
/// reported, and whoever reads that board now has two. So a copy-back leaves the
/// destination's origin exactly as the destination holds it, absent included, and every
/// other copy records the id it was copied from.
fn recorded(item: &Planned, held: Option<&Prior>) -> Origin {
    if let Target::Update {
        found: Found::Origin,
        ..
    } = &item.target
    {
        return Origin::Keeps(
            held.and_then(|held| described(&held.item).1.get(GlobalId::ORIGIN_KEY).cloned()),
        );
    }
    Origin::Records(item.source.clone())
}

/// Whether the destination already reads exactly as this copy would leave it.
///
/// The destination's own `url` and timestamps are excluded because a copy never writes
/// them, so a difference there is not one this copy would close.
fn same(held: &Item, outgoing: &Item, destination: &SourceName) -> bool {
    match (held, outgoing) {
        (Item::Task(held), Item::Task(outgoing)) => {
            // Qualified before they are compared, so `T-1` and `folder:T-1` at the destination
            // `folder` are the one entry they are.
            targets(&held.delivers, destination) == targets(&outgoing.delivers, destination)
                && targets(&held.delivered_by, destination)
                    == targets(&outgoing.delivered_by, destination)
                && held.title == outgoing.title
                && held.content == outgoing.content
                && held.status == outgoing.status
                && held.priority == outgoing.priority
                && held.labels == outgoing.labels
                && held.project == outgoing.project
                && held.metadata == outgoing.metadata
                && held.repositories == outgoing.repositories
        }
        (Item::Project(held), Item::Project(outgoing)) => {
            held.title == outgoing.title
                && held.content == outgoing.content
                && held.status == outgoing.status
                && held.labels == outgoing.labels
                && held.metadata == outgoing.metadata
                && held.repositories == outgoing.repositories
        }
        // No status, because a document has none; no edges, because it is in no graph.
        (Item::Document(held), Item::Document(outgoing)) => {
            held.title == outgoing.title
                && held.content == outgoing.content
                && held.labels == outgoing.labels
                && held.project == outgoing.project
                && held.metadata == outgoing.metadata
                && held.repositories == outgoing.repositories
        }
        _ => false,
    }
}

/// Whether the destination's forward edges already say what this copy would write.
fn same_edges(held: &[DependencyEdge], outgoing: &[DependencyEdge]) -> bool {
    let ends = |edges: &[DependencyEdge]| {
        let mut ends: Vec<(String, ItemKind, DependencyKind)> = edges
            .iter()
            .map(|edge| (edge.to.id().to_owned(), edge.to.kind, edge.kind))
            .collect();
        ends.sort_by(|left, right| left.0.cmp(&right.0));
        ends
    };
    ends(held) == ends(outgoing)
}

/// Each read edge as the destination should record it, or `None` when its far end is a
/// member of this copy whose destination id is not known yet.
///
/// `destination` is the source the near end lands in. A far end this copy landed in another
/// source — a routed copy's other half — is written qualified, which every source records
/// as a cross-source edge on the near item.
fn mapped_edges(
    edges: &[DependencyEdge],
    origin: &SourceName,
    destination: &SourceName,
    copied: &[GlobalId],
    written: &BTreeMap<String, GlobalId>,
) -> Vec<Option<DependencyEdge>> {
    edges
        .iter()
        .map(|edge| {
            let far = GlobalId::new(origin.clone(), NativeId(edge.to.id().to_owned()));
            let id = if let Some(native) = names(&edge.to, destination) {
                // A far end already qualified to the destination's own source is that
                // source's own item, so it is written the way that source names its own:
                // unqualified. Leaving it qualified would have the destination hold an
                // edge into itself written as if it left, which is the one spelling the
                // reserved key exists to keep for edges that really do.
                Some(native)
            } else if !edge.to.is_qualified() && copied.contains(&far) {
                // A member of this copy, inside one source included: a copy there lands
                // beside the item it was read from, so the edge names the copy rather than
                // the original the copy's own project does not hold.
                written
                    .get(&far.to_string())
                    .map(|landed| spelled_from(landed, destination))
            } else if edge.to.is_qualified() || origin == destination {
                // Already naming a source of its own, or a copy inside one source where
                // the far end's own id is the destination's id.
                Some(edge.to.id().to_owned())
            } else {
                Some(far.to_string())
            }?;
            DependencyEndpoint::new(id, edge.to.kind)
                .ok()
                .map(|to| DependencyEdge {
                    from: edge.from.clone(),
                    to,
                    kind: edge.kind,
                })
        })
        .collect()
}

/// Each `delivers` entry of one planned task as the destination should hold it, or `None`
/// where it names a member of this copy whose destination id is not known yet. Empty for
/// anything but a task.
fn delivers_of(
    item: &Planned,
    destination: &SourceName,
    copied: &[GlobalId],
    written: &BTreeMap<String, GlobalId>,
) -> Vec<Option<TaskRef>> {
    match &item.item {
        Item::Task(task) => mapped_delivers(
            &task.delivers,
            &item.source.source,
            destination,
            copied,
            written,
        ),
        Item::Project(_) | Item::Document(_) => Vec::new(),
    }
}

/// Each `delivers` entry as the destination should hold it, or `None` when it names a member
/// of this copy whose destination id is not known yet.
///
/// An entry naming a member of the copied set becomes that member's own id at the
/// destination — bare, because the member is the destination's, unless the id holds a colon
/// a bare entry would be misread by. Every other entry is carried through qualified, so a
/// bare one read at `origin` goes on naming a task of `origin`.
fn mapped_delivers(
    entries: &[TaskRef],
    origin: &SourceName,
    destination: &SourceName,
    copied: &[GlobalId],
    written: &BTreeMap<String, GlobalId>,
) -> Vec<Option<TaskRef>> {
    entries
        .iter()
        .map(|entry| {
            let qualified = entry.in_source(origin);
            let Ok(far) = qualified.as_str().parse::<GlobalId>() else {
                return Some(qualified);
            };
            if !copied.contains(&far) {
                return Some(qualified);
            }
            let landed = written.get(&far.to_string())?;
            let native = &landed.native;
            Some(
                if &landed.source != destination || native.as_str().contains(':') {
                    TaskRef::qualified(&landed.source, native)
                } else {
                    TaskRef::new(native.as_str())
                        .unwrap_or_else(|_| TaskRef::qualified(destination, native))
                },
            )
        })
        .collect()
}

/// How many of one task's `delivers` entries name a member of the copied set.
fn members_named(entries: &[TaskRef], origin: &SourceName, copied: &[GlobalId]) -> u64 {
    let named = entries
        .iter()
        .filter(|entry| {
            entry
                .in_source(origin)
                .as_str()
                .parse::<GlobalId>()
                .is_ok_and(|far| copied.contains(&far))
        })
        .count();
    u64::try_from(named).unwrap_or(u64::MAX)
}

/// The entries that could be resolved, which is every one of them on the second pass.
fn resolved_entries(entries: &[Option<TaskRef>]) -> Vec<TaskRef> {
    entries.iter().flatten().cloned().collect()
}

/// How an item landed at `landed` is named from one landed in `near`: by its native id in
/// the same source, qualified from any other.
fn spelled_from(landed: &GlobalId, near: &SourceName) -> String {
    if &landed.source == near {
        landed.native.0.clone()
    } else {
        landed.to_string()
    }
}

/// The native id a qualified endpoint names at `destination`, when it names one there.
fn names(endpoint: &DependencyEndpoint, destination: &SourceName) -> Option<String> {
    if !endpoint.is_qualified() {
        return None;
    }
    let id: GlobalId = endpoint.id().parse().ok()?;
    (&id.source == destination).then_some(id.native.0)
}

/// Where a project copy files the tasks it carries: in each source a task lands in, the
/// home there or the home's member project.
type Filing = BTreeMap<SourceName, NativeId>;

/// The edges that could be resolved, which is every one of them on the second pass.
fn resolved(edges: &[Option<DependencyEdge>]) -> Vec<DependencyEdge> {
    edges.iter().flatten().cloned().collect()
}

/// Refuse an item of a member copy whose edge names a member that copy does not carry
/// and whose destination id nothing records.
///
/// `unrecorded` is empty for every other copy, which is what makes this a no-op there. An
/// edge already naming a source of its own, and every edge of a copy inside one source,
/// is written as it was read by [`mapped_edges`], so neither can need a recorded origin.
fn unrecorded_far_end(item: &Planned, unrecorded: &[GlobalId]) -> Result<(), EngineError> {
    if item.source.source == item.to {
        return Ok(());
    }
    for edge in &item.edges {
        if edge.to.is_qualified() {
            continue;
        }
        let far = GlobalId::new(
            item.source.source.clone(),
            NativeId(edge.to.id().to_owned()),
        );
        if unrecorded.contains(&far) {
            return Err(EngineError::UnrecordedMember {
                item: item.source.clone(),
                member: far,
                destination: item.to.clone(),
            });
        }
    }
    Ok(())
}