flint_lean/

manifest.rs

//! The lean subtree manifest: one JSON document, CAS-written, the
//! atomic commit point every reader trusts (checkout, sync, DR).
//!
//! Deliberately NOT `tier::manifest`: the hub's writer rebuilds the
//! whole document from its local walk and cannot merge (its Entry has
//! no author identity), which is the amputation engine the lean review
//! reproduced three ways. This writer merges three-way:
//! base = the manifest view at OUR last install, mine = this barrier's
//! walk, theirs = bucket-current. Foreign entries are preserved and
//! surfaced for the next consume; a local delete loses to a foreign
//! modify (delete/modify resolves conservative — the formal model's
//! GC-vs-merge counterexample is exactly this case).

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

use serde::{Deserialize, Serialize};

use flint_store::{
    crc64_nvme, GenerationStamps, ObjectMeta, ObjectStore, PutCondition, StoreError,
};

use super::{LeanConfig, LeanError, LeanResult};

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct LeanEntry {
    /// Object key (under `<prefix>/files/`).
    pub key: String,
    pub etag: String,
    /// Full-object CRC-64/NVME of the cited bytes, in wire (base64)
    /// form — computed by the flint client that MOVED the bytes (the
    /// publisher over what it uploaded, a consume or checkout over what
    /// it wrote), never copied from a backend header. It is the one
    /// integrity check that holds on every backend: S3 attests a
    /// checksum on the wire, Ozone attests none, and the manifest's
    /// value describes the bytes either way. Every reader verifies a
    /// fresh fetch against it before the file becomes visible.
    pub crc64_b64: String,
    pub size: u64,
    pub mode: u32,
    pub mtime_unix: i64,
    pub generation: u64,
    /// The publishing writer's lease epoch (0 = a gateway/HITL write).
    pub epoch: u64,
}

#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LeanManifest {
    pub seq: u64,
    /// path -> entry. Paths are workspace-relative, '/'-separated.
    pub entries: BTreeMap<String, LeanEntry>,
    /// This workspace is published by exactly ONE writer, so an object
    /// that has moved off its citation was moved by something that is
    /// not the publisher.
    ///
    /// It exists because a reader's default arm on a 412 is an
    /// explicit S3-wins adopt — right for a lean workspace, where an
    /// object past its citation means a human wrote newer bytes that
    /// ought to win. On a MIRROR that reading is exactly inverted: the
    /// publisher is the only party entitled to write, so "past the
    /// citation" means a stranger did, and adopting copies bytes no
    /// manifest cites into a reader's tree with no error and no
    /// conflict record. Composition drill C4 measured that against
    /// forge's legible export.
    ///
    /// Set by the installing pass from config, never inherited.
    #[serde(default)]
    pub sole_writer: bool,
    /// Which citation source installed this manifest (§2.4.1). Also
    /// stamped on the object's metadata so it is readable by HEAD.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub boundary_source: Option<String>,
    /// What each published delete RETIRED, path -> the etag the document
    /// cited before the delete, kept until the path is cited again (or
    /// `TOMBSTONE_KEEP_SEQS` generations pass). Review 2026-09-18, H1e: a
    /// writer that adopted a UI write while it was still pending, and
    /// merges after another writer cited it and then deleted it knowingly,
    /// otherwise sees only "baseline ≠ merge base, object at the key" —
    /// the same view a delete that RACED the write leaves, where modify
    /// rightly wins — and the citation repair (removed in step 5) resurrected
    /// the deleted generation. The tombstone is the one fact that tells the
    /// two apart.
    /// Empty on every document written before this field existed.
    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
    pub tombstones: BTreeMap<String, Tombstone>,
}

/// A published delete's record in the document: the generation it retired
/// and the generation of the document that published it.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Tombstone {
    pub etag: String,
    pub seq: u64,
}

/// How many document generations a tombstone outlives its delete. A writer
/// whose merge base is older than this can still re-cite what it holds —
/// the residual, named in SAFETY.md §4.14.
pub const TOMBSTONE_KEEP_SEQS: u64 = 10_000;

impl LeanManifest {
    /// COMPACT, not pretty. This document is O(entries) and it moves on
    /// every path that matters: the pre-marker checkout GET, every
    /// barrier merge GET and CAS PUT, and every gateway read verb. The
    /// indentation was ~22% of those bytes and of the parse that
    /// follows. `mc cat | jq` — the operator's and every rig's only
    /// debugging surface for the CAS commit point — is unaffected.
    pub fn to_bytes(&self) -> Vec<u8> {
        serde_json::to_vec(self).expect("manifest serializes")
    }
    pub fn parse(bytes: &[u8]) -> Result<LeanManifest, String> {
        serde_json::from_slice(bytes).map_err(|e| e.to_string())
    }
}

/// The bucket-current manifest + its document ETag (the CAS token).
/// The pointer document at `<prefix>/.flint/lean/current` — the ONLY
/// mutable metadata object under the pointer layout (design of record:
/// `docs/plans/flint-lean-manifest-pointer-design.md`).
///
/// Everything a reader needs before it decides whether to fetch entries
/// at all lives here, so the barrier's idle tick reads a few hundred
/// bytes instead of a document that runs to 264 MiB at 1M entries.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct Pointer {
    /// The manifest generation this pointer publishes. Bumped by a
    /// citation AND by a takeover rotation, which is why it is not the
    /// same number as `entries_seq`.
    pub seq: u64,
    /// The single immutable object holding every entry of this
    /// generation — the step-one layout. `None` on a CHUNKED pointer.
    ///
    /// Optional here and REQUIRED in the step-one binary's schema, and
    /// that asymmetry is the migration: a step-one reader handed a
    /// chunked pointer fails to parse it and refuses, rather than
    /// decoding a pointer it half-understands. `manifest::load` maps a
    /// missing object to `Ok(None)`, `None` means FIRST WRITE, and a
    /// barrier answers that with `If-None-Match: *` — so every layout
    /// change has to make an old reader refuse rather than conclude the
    /// project is empty. **Do not give this field a serde default in a
    /// released binary; that is what makes the refusal work.**
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub entries_key: Option<String>,
    /// The `seq` the entries object was written under. A rotation
    /// leaves this ALONE while bumping `seq`, which is exactly what
    /// lets a follower skip the entries GET after a takeover.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub entries_seq: Option<u64>,
    /// The chunk list — the step-two layout. `Some(vec![])` is a
    /// legitimately EMPTY project and is distinguishable from absence,
    /// which is why this is an Option around a Vec rather than a bare
    /// Vec with a default.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub chunks: Option<Vec<super::chunk::ChunkRef>>,
    /// Carried here as well as on the manifest because the pointer is
    /// the authority for the fields it carries, and a reader that
    /// reconstructs a manifest from it must not lose the flag that
    /// decides whether it may adopt.
    #[serde(default)]
    pub sole_writer: bool,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub boundary_source: Option<String>,
    /// The epoch that installed this pointer. Diagnostic; the fence is
    /// the CAS, not this field.
    pub epoch: u64,
    /// The content address of the document's tombstones in the chunked
    /// layout (`chunk::TombstoneBody`, at `chunk_key`), `None` when the
    /// document has none. The single-object layout carries them inline.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tombstones: Option<String>,
}

/// Where a pointer's entries actually live.
#[derive(Debug)]
pub enum Entries<'a> {
    /// One generation object (step one).
    Single { key: &'a str, seq: u64 },
    /// A chunk list (step two). Empty = an empty project.
    Chunked(&'a [super::chunk::ChunkRef]),
}

impl Pointer {
    /// Resolve the layout, refusing anything ambiguous.
    ///
    /// BOTH forms present, or NEITHER, is malformed and must fail —
    /// never pick one. A pointer carrying both would let two readers
    /// that broke the tie differently disagree about the contents of
    /// the same seq, which is the one thing a single visible object was
    /// supposed to make impossible.
    pub fn entries(&self) -> LeanResult<Entries<'_>> {
        match (&self.entries_key, &self.chunks) {
            (Some(k), None) => {
                Ok(Entries::Single { key: k, seq: self.entries_seq.unwrap_or(self.seq) })
            }
            (None, Some(c)) => Ok(Entries::Chunked(c)),
            (Some(_), Some(_)) => Err(super::LeanError::State(format!(
                "manifest pointer at seq {} names BOTH a generation object and a chunk list; \
                 refusing to guess which describes this manifest",
                self.seq
            ))),
            (None, None) => Err(super::LeanError::State(format!(
                "manifest pointer at seq {} names no entries at all — neither a generation \
                 object nor a chunk list; an empty project is an empty chunk LIST, not a \
                 missing one",
                self.seq
            ))),
        }
    }
}

/// What a writer must present to CAS. Carries the LAYOUT as well as the
/// tag, because a workspace mid-migration has a legacy manifest and no
/// pointer yet, and the two cases take different preconditions: an
/// If-Match on the object the writer read, versus an If-None-Match on a
/// pointer that must not already exist.
#[derive(Debug, Clone, PartialEq)]
pub struct ManifestHandle {
    pub etag: String,
    /// True when the document was read from the LEGACY single-object
    /// key, i.e. this write is the migration.
    pub legacy: bool,
    /// The chunk list this write is superseding, empty if the workspace
    /// is not chunked yet. Carried on the handle rather than re-read at
    /// publish time because the caller has just loaded it, and because
    /// a publish that had to GET the pointer again to find out what it
    /// already knew would pay a request per barrier for nothing.
    ///
    /// This is what makes a publish O(changed): a chunk whose address
    /// is already here is not sent.
    pub prev_chunks: Vec<super::chunk::ChunkRef>,
    /// The tombstones object the superseded pointer names, if any: retired
    /// with the chunks when the new pointer names another (M1).
    pub prev_tombstones: Option<String>,
}

/// What the legacy key is overwritten with once a workspace has moved to
/// the pointer layout. It must NOT parse as a manifest: `LeanManifest`
/// gives `seq` and `entries` no serde default, so an OLD `flint-sync`
/// reading this gets `LeanError::State` and refuses.
///
/// Deleting the legacy key instead would be a data-loss shape, not a
/// tidy-up: `manifest::load` maps a missing object to `Ok(None)`, `None`
/// means FIRST WRITE, and a barrier answers that with
/// `IfNoneMatchAny` — so an old syncer would conclude the project is
/// empty and re-seed over it.
pub const REFUSAL_DOC: &str = concat!(
    "{\"moved\":\".flint/lean/current\",",
    "\"note\":\"this workspace uses the manifest pointer layout; upgrade flint-sync\"}"
);

#[derive(Clone)]
pub struct LoadedManifest {
    pub manifest: LeanManifest,
    /// The CAS handle: the POINTER's etag under the pointer layout, the
    /// legacy object's etag before migration.
    pub etag: String,
    /// The object's own Last-Modified. The GET already carries it; a
    /// caller that wants the coherence stamp must not pay a second
    /// request for what this one threw away (review: U31).
    pub last_modified_unix: Option<u64>,
    /// `None` ⇒ this workspace is still on the legacy single-object
    /// layout and the next write migrates it.
    pub pointer: Option<Pointer>,
}

impl LoadedManifest {
    /// The handle a writer presents to CAS over this read.
    pub fn handle(&self) -> ManifestHandle {
        ManifestHandle {
            etag: self.etag.clone(),
            legacy: self.pointer.is_none(),
            prev_chunks: match self.pointer.as_ref().map(|p| p.entries()) {
                Some(Ok(Entries::Chunked(c))) => c.to_vec(),
                _ => Vec::new(),
            },
            prev_tombstones: self.pointer.as_ref().and_then(|p| p.tombstones.clone()),
        }
    }
}

/// Read the pointer alone. The barrier's idle tick wants exactly this
/// and nothing else: a few hundred bytes that answer "did anything
/// move, and were the ENTRIES part of it".
pub struct LoadedPointer {
    pub pointer: Pointer,
    pub etag: String,
    /// The pointer object's own Last-Modified — the citation's clock,
    /// which /status reports and which the GET already carries.
    pub last_modified_unix: Option<u64>,
}

pub async fn load_pointer(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
) -> LeanResult<Option<LoadedPointer>> {
    match store.get_whole(&cfg.current_key(), None).await {
        Ok((meta, bytes)) => {
            let pointer: Pointer = serde_json::from_slice(&bytes)
                .map_err(|e| super::LeanError::State(format!("manifest pointer parse: {e}")))?;
            Ok(Some(LoadedPointer { pointer, etag: meta.etag, last_modified_unix: meta.last_modified_unix }))
        }
        Err(StoreError::NotFound(_)) => Ok(None),
        Err(e) => Err(e.into()),
    }
}

/// The whole document: pointer, then the generation it names. Falls
/// back to the legacy single-object key for a workspace that has not
/// been migrated — which every existing workspace is, exactly once.
/// How many times `load` will restart onto a newer generation before
/// giving up. A restart happens only when a chunk we needed was swept
/// AND the pointer moved under us, so it is bounded by publish rate,
/// not by anything we control — three is generous for a workspace whose
/// floor is measured in seconds.
pub(crate) const LOAD_ATTEMPTS: u32 = 3;

/// What `fetch_chunks` found for one pointer.
enum Fetched {
    /// Every chunk's body, in the pointer's order and each verified against
    /// its address, and the decoded tombstones.
    Complete { bodies: Vec<bytes::Bytes>, tombstones: BTreeMap<String, Tombstone> },
    /// Something the pointer names is not in the bucket.
    Missing(Hole),
}

/// The object a load found missing.
pub(crate) enum Hole {
    Chunk { addr: String, first: String },
    Tombstones(String),
}

impl Hole {
    pub(crate) fn refusal(&self, seq: u64) -> String {
        match self {
            Hole::Chunk { addr, first } => format!(
                "manifest pointer at seq {seq} names chunk {addr} (first key {first:?}), which does \
                 not exist — refusing to serve a manifest with a hole in it as though those entries \
                 had been deleted"
            ),
            Hole::Tombstones(addr) => {
                format!("manifest pointer at seq {seq} names tombstones {addr}, which do not exist")
            }
        }
    }
}

/// The bodies of `refs` and the pointer's tombstones (latency plan lever A).
///
/// 1. The cache, where it has a body that hashes to the address asked for.
///    One that does not is dropped and fetched: the cache vouches for
///    nothing it has not just proved.
/// 2. The rest from the bucket, `chunk_fetch_parallelism` at a time — they
///    were one at a time, 220 sequential GETs at a million entries.
/// 3. Beside those, ONE LIST of the chunk prefix when anything came from
///    the cache: a body is proof of content, not of the object still being
///    there, and a chunk the live pointer names that the bucket lost must
///    still refuse the load. A LIST the credential may not make (or that
///    fails) sends the cached chunks to the bucket instead: the check is
///    never skipped, only paid for the old way.
///
/// Every fetched body is checked against its address before it is kept or
/// used, so `chunk::assemble_verified` hashes nothing twice.
async fn fetch_chunks(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    p: &Pointer,
    refs: &[super::chunk::ChunkRef],
) -> LeanResult<Fetched> {
    use futures::stream::{self, StreamExt};
    let cache = cfg.chunk_cache.as_deref();
    let mut bodies: Vec<Option<bytes::Bytes>> = vec![None; refs.len()];
    let mut tombstones: Option<BTreeMap<String, Tombstone>> = None;
    if let Some(c) = cache {
        for (slot, r) in bodies.iter_mut().zip(refs) {
            if let Some(b) = c.get(&r.addr) {
                if super::chunk::chunk_address(&b) == r.addr {
                    *slot = Some(b);
                } else {
                    c.remove(&r.addr);
                }
            }
        }
        if let Some(addr) = &p.tombstones {
            if let Some(b) = c.get(addr) {
                match super::chunk::decode_tombstones(&b, addr) {
                    Ok(t) => tombstones = Some(t),
                    Err(_) => c.remove(addr),
                }
            }
        }
    }
    let cached: Vec<usize> = (0..refs.len()).filter(|&i| bodies[i].is_some()).collect();
    let tombstones_cached = tombstones.is_some();

    // What to fetch: the misses, and on a failed LIST the hits as well.
    let fetch_round = |want_chunks: Vec<usize>, want_tombstones: bool| async move {
        let mut wants: Vec<(Option<usize>, String)> =
            want_chunks.into_iter().map(|i| (Some(i), refs[i].addr.clone())).collect();
        if want_tombstones {
            if let Some(a) = &p.tombstones {
                wants.push((None, a.clone()));
            }
        }
        stream::iter(wants.into_iter().map(|(slot, addr)| async move {
            let got = store.get_whole(&cfg.chunk_key(&addr), None).await;
            (slot, addr, got)
        }))
        .buffer_unordered(cfg.chunk_fetch_parallelism.max(1))
        .collect::<Vec<_>>()
        .await
    };
    let misses: Vec<usize> = (0..refs.len()).filter(|&i| bodies[i].is_none()).collect();
    let tombstones_missing = p.tombstones.is_some() && !tombstones_cached;
    let listing = async {
        if cached.is_empty() && !tombstones_cached {
            return None;
        }
        Some(store.list(&format!("{}/{}/chunks/", cfg.prefix, super::LEAN_DIR)).await)
    };
    let (listing, fetched) = futures::join!(listing, fetch_round(misses, tombstones_missing));
    let mut rounds = vec![fetched];
    match listing {
        None => {}
        Some(Ok(listed)) => {
            let present: std::collections::HashSet<&str> =
                listed.iter().filter_map(|o| o.key.rsplit('/').next()).collect();
            for &i in &cached {
                if !present.contains(refs[i].addr.as_str()) {
                    return Ok(Fetched::Missing(Hole::Chunk {
                        addr: refs[i].addr.clone(),
                        first: refs[i].first.clone(),
                    }));
                }
            }
            if let (true, Some(addr)) = (tombstones_cached, &p.tombstones) {
                if !present.contains(addr.as_str()) {
                    return Ok(Fetched::Missing(Hole::Tombstones(addr.clone())));
                }
            }
        }
        Some(Err(_)) => {
            // Existence by the old route: fetch what the cache served.
            rounds.push(fetch_round(cached, tombstones_cached).await);
        }
    }
    for (slot, addr, got) in rounds.into_iter().flatten() {
        let body = match got {
            Ok((_, b)) => b,
            Err(StoreError::NotFound(_)) => {
                return Ok(Fetched::Missing(match slot {
                    Some(i) => Hole::Chunk { addr, first: refs[i].first.clone() },
                    None => Hole::Tombstones(addr),
                }));
            }
            Err(e) => return Err(e.into()),
        };
        match slot {
            Some(i) => {
                // The object was fetched BY address, so a mismatch means
                // the store handed back something else — a stale read, a
                // wrong key, a truncated body.
                super::chunk::check_address_of(&body, &addr)?;
                if let Some(c) = cache {
                    c.put(&addr, &body);
                }
                bodies[i] = Some(body);
            }
            None => {
                tombstones = Some(super::chunk::decode_tombstones(&body, &addr)?);
                if let Some(c) = cache {
                    c.put(&addr, &body);
                }
            }
        }
    }
    Ok(Fetched::Complete {
        bodies: bodies.into_iter().map(|b| b.expect("every chunk is cached or fetched")).collect(),
        tombstones: tombstones.unwrap_or_default(),
    })
}

pub async fn load(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
) -> LeanResult<Option<LoadedManifest>> {
    load_from(store, cfg, None).await
}

/// [`load`], starting from a pointer read the caller ALREADY made
/// (`Some(None)`: it 404'd), so the first attempt does not GET it again —
/// the consume's cheap path reads the pointer and, when it has moved or is
/// absent, needs the document behind that same read. A restart onto a
/// newer generation reads the pointer afresh, as `load` always did.
pub async fn load_from(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    mut first: Option<Option<LoadedPointer>>,
) -> LeanResult<Option<LoadedManifest>> {
  for attempt in 0..LOAD_ATTEMPTS {
    let read = match first.take() {
        Some(already) => already,
        None => load_pointer(store, cfg).await?,
    };
    if let Some(LoadedPointer { pointer: p, etag, last_modified_unix: pointer_lm }) = read {
        // A pointer naming an object that is not there is not an empty
        // workspace — it is a broken one, and answering `None` would
        // invite a re-seed over a live project. That rule holds one
        // level deeper for chunks: a chunk list with a hole is a
        // manifest missing entries, and a missing entry reads to every
        // consumer as a file the agent deleted.
        let (mut manifest, meta_lm) = match p.entries()? {
            Entries::Single { key, .. } => {
                let (meta, bytes) = match store.get_whole(key, None).await {
                    Ok(v) => v,
                    Err(StoreError::NotFound(_)) => {
                        return Err(super::LeanError::State(format!(
                            "manifest pointer at seq {} names {key}, which does not exist — \
                             refusing to treat a broken pointer as an empty workspace",
                            p.seq
                        )))
                    }
                    Err(e) => return Err(e.into()),
                };
                let m = LeanManifest::parse(&bytes)
                    .map_err(|e| super::LeanError::State(format!("manifest parse: {e}")))?;
                (m, meta.last_modified_unix)
            }
            Entries::Chunked(refs) => {
                // The bodies come from the chunk cache where it has them and
                // from the bucket in parallel where it does not (latency
                // plan lever A, `chunkcache`). A hole FAILS rather than
                // silently shortening the manifest, however the bodies came.
                let (bodies, tombstones) = match fetch_chunks(store, cfg, &p, refs).await? {
                    Fetched::Complete { bodies, tombstones } => (bodies, tombstones),
                    Fetched::Missing(hole) => {
                        // A chunk we need is gone. Which of two very
                        // different things happened is decided by the
                        // POINTER, not by the chunk:
                        //
                        // - the pointer MOVED ⇒ we raced a sweep that
                        //   collected a generation we were still reading.
                        //   The current generation is a complete snapshot,
                        //   so start over against it: coherent, just newer.
                        //   This is what makes a reader safe for a DURATION
                        //   rather than for `Retain` publishes (chunked
                        //   design §8.2, machine-checked by
                        //   `LeanChunkGCSlowReader`).
                        // - the pointer is UNCHANGED ⇒ the live manifest
                        //   genuinely has a hole, and there is nothing newer
                        //   to restart onto. Fail, and never as `Ok(None)`,
                        //   which every caller reads as FIRST WRITE.
                        let moved = match load_pointer(store, cfg).await? {
                            Some(now) => now.etag != etag,
                            None => true,
                        };
                        if moved && attempt + 1 < LOAD_ATTEMPTS {
                            continue;
                        }
                        return Err(super::LeanError::State(hole.refusal(p.seq)));
                    }
                };
                let entries = super::chunk::assemble_verified(refs, &bodies)?;
                super::docview::remember(cfg, refs, &entries);
                // What the cache keeps from here is this document's chunks:
                // the next load needs these and whatever is new.
                if let Some(c) = cfg.chunk_cache.as_deref() {
                    let mut live: std::collections::HashSet<String> = refs.iter().map(|r| r.addr.clone()).collect();
                    live.extend(p.tombstones.clone());
                    c.retain(&live);
                }
                (
                    LeanManifest {
                        seq: p.seq,
                        entries,
                        sole_writer: p.sole_writer,
                        boundary_source: p.boundary_source.clone(),
                        tombstones,
                    },
                    None,
                )
            }
        };
        // The pointer is the authority for the fields it carries: a
        // rotation moves `seq` without rewriting the entries object, so
        // the generation's own copy is stale by construction.
        manifest.seq = p.seq;
        manifest.sole_writer = p.sole_writer;
        manifest.boundary_source = p.boundary_source.clone();
        return Ok(Some(LoadedManifest {
            manifest,
            etag,
            // The CITATION's clock is when the pointer moved, not when
            // the entries object happened to be written: a rotation
            // republishes the same entries under a new generation.
            last_modified_unix: pointer_lm.or(meta_lm),
            pointer: Some(p),
        }));
    }
    // No pointer: the pre-migration layout, which has no chunks and so
    // nothing to race a sweep over. Answer it and leave the loop.
    return match store.get_whole(&cfg.manifest_key(), None).await {
        Ok((meta, bytes)) => {
            let manifest = LeanManifest::parse(&bytes)
                .map_err(|e| super::LeanError::State(format!("manifest parse: {e}")))?;
            Ok(Some(LoadedManifest {
                manifest,
                etag: meta.etag,
                last_modified_unix: meta.last_modified_unix,
                pointer: None,
            }))
        }
        Err(StoreError::NotFound(_)) => Ok(None),
        Err(e) => Err(e.into()),
    };
  }
  // Every attempt raced a sweep. Refusing is right: the alternative is
  // to serve a manifest we know is short, and short reads as deleted.
  Err(super::LeanError::State(format!(
      "manifest load restarted {LOAD_ATTEMPTS} times without settling — a sweep is \
       collecting generations faster than this reader can finish one"
  )))
}

/// How many lost races a one-CAS edit re-reads and retries before it gives
/// up `Contended`.
pub const EDIT_CAS_ATTEMPTS: usize = 16;

/// Why a one-CAS edit did not land.
#[derive(Debug)]
pub enum EditError<E> {
    /// The edit refused the document it was shown.
    Refused(E),
    Store(LeanError),
    /// Lost `EDIT_CAS_ATTEMPTS` races in a row.
    Contended,
}

/// ONE manifest CAS for an edit by a party that holds no lease (P2): the
/// gateway's UI saves, deletes and renames, and the ingress sweep's
/// adoptions. Load the current document; let `edit` judge it and change a
/// copy — the next generation, with the installing party's fields reset as
/// `merge` resets them; keep tombstones to `TOMBSTONE_KEEP_SEQS`; CAS it in
/// with epoch 0; on a lost race do it all again against what won, so every
/// precondition is judged against the document the CAS replaces.
pub async fn commit_edit<E>(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    flush: &str,
    edit: impl Fn(Option<&LoadedManifest>, &mut LeanManifest) -> Result<(), E>,
) -> Result<(), EditError<E>> {
    commit_edit_from(store, cfg, flush, None, edit).await
}

/// [`commit_edit`], whose FIRST attempt edits `first` — a document the
/// caller loaded at the pointer it just read (the gateway's `view`, keyed by
/// the pointer's etag) — instead of loading it again. `Some(None)`: the
/// caller found no document. Safe because the CAS is If-Match on that
/// document's etag: an attempt over a document that has since moved loses,
/// and every later attempt loads the current one.
pub async fn commit_edit_from<E>(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    flush: &str,
    first: Option<Option<std::sync::Arc<LoadedManifest>>>,
    edit: impl Fn(Option<&LoadedManifest>, &mut LeanManifest) -> Result<(), E>,
) -> Result<(), EditError<E>> {
    let mut first = first;
    // The document the last attempt sent, when its CAS answered 412 or
    // 409. An SDK retries a conditional PUT whose response was lost, and
    // the retry meets the object its first attempt wrote: if the document
    // now IS the one sent, the edit committed, and it is answered so rather
    // than judged again against itself (a delete would find its path gone
    // and answer NoSuchFile; the model's GDelete acknowledges what it
    // commits, `udel`). A peer's identical edit at the same seq (the same
    // delete, raced) leaves the same document: committed either way.
    let mut sent: Option<LeanManifest> = None;
    for _ in 0..EDIT_CAS_ATTEMPTS {
        let current: Option<std::sync::Arc<LoadedManifest>> = match first.take() {
            Some(read) => read,
            None => load(store, cfg).await.map_err(EditError::Store)?.map(std::sync::Arc::new),
        };
        if let (Some(s), Some(c)) = (sent.as_ref(), current.as_ref()) {
            if same_document(s, &c.manifest) {
                return Ok(());
            }
        }
        let mut doc = current.as_ref().map(|l| l.manifest.clone()).unwrap_or_default();
        doc.seq += 1;
        doc.sole_writer = false;
        doc.boundary_source = None;
        edit(current.as_deref(), &mut doc).map_err(EditError::Refused)?;
        let seq = doc.seq;
        doc.tombstones.retain(|_, t| seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
        let expected = current.as_ref().map(|l| l.handle());
        let before = current.as_ref().map(|l| &l.manifest);
        match cas_write_retiring(store, cfg, &doc, before, expected.as_ref(), 0, flush, None).await {
            Ok(_) => return Ok(()),
            Err(LeanError::Store(StoreError::PreconditionFailed(_))) | Err(LeanError::Store(StoreError::Conflict(_))) => {
                sent = Some(doc);
                continue;
            }
            Err(e) => return Err(EditError::Store(e)),
        }
    }
    Err(EditError::Contended)
}

/// Is `b` the document `a` is — the same generation with the same
/// citations, tombstones and stamps? (`commit_edit`'s own-write check.)
fn same_document(a: &LeanManifest, b: &LeanManifest) -> bool {
    a.seq == b.seq
        && a.entries == b.entries
        && a.tombstones == b.tombstones
        && a.sole_writer == b.sole_writer
        && a.boundary_source == b.boundary_source
}

/// A one-CAS edit of a document it does not copy (the gateway's lever,
/// latency plan §16): the edit reads the document it was handed and writes
/// an overlay. `entries` answers as the edited document's map would, so an
/// edit reads as it does under `commit_edit`; `seq` is the next generation
/// and `tombstones` the document's, to change.
pub struct DocEdit<'a> {
    pub seq: u64,
    pub entries: EditEntries<'a>,
    pub tombstones: BTreeMap<String, Tombstone>,
}

/// A document's entries with an overlay on them (`DocEdit::entries`).
pub struct EditEntries<'a> {
    base: &'a BTreeMap<String, LeanEntry>,
    overlay: BTreeMap<String, Option<LeanEntry>>,
}

impl EditEntries<'_> {
    pub fn get(&self, p: &str) -> Option<&LeanEntry> {
        match self.overlay.get(p) {
            Some(v) => v.as_ref(),
            None => self.base.get(p),
        }
    }

    pub fn contains_key(&self, p: &str) -> bool {
        self.get(p).is_some()
    }

    pub fn insert(&mut self, p: String, e: LeanEntry) -> Option<LeanEntry> {
        let was = self.get(&p).cloned();
        self.overlay.insert(p, Some(e));
        was
    }

    pub fn remove(&mut self, p: &str) -> Option<LeanEntry> {
        let was = self.get(p).cloned()?;
        self.overlay.insert(p.to_string(), None);
        Some(was)
    }
}

/// How `commit_edit_paths` ended, and the document to hold next: the one it
/// installed, made from the one it was handed in place when nothing else
/// held that (no reload, no copy); or the one it was handed, unchanged,
/// when the edit refused it. `None`: load before the next edit.
pub struct EditOutcome<E> {
    pub result: Result<(), EditError<E>>,
    pub held: Option<std::sync::Arc<LoadedManifest>>,
}

/// [`commit_edit_from`] without copying the document (latency plan §16): the
/// edit writes an overlay (`DocEdit`); the commit re-chunks only from the
/// first chunk an overlay path falls in until the chunking lines up again
/// (`chunk::rechunk_with`), answers the retire set from the overlay and the
/// chunks' aliases (`docview::retired_over`), and publishes only the chunks
/// that changed. The document it installs is the one `commit_edit_from`
/// would; the shadow check (`LeanConfig::g_shadow`, and every debug build)
/// asserts the same chunk list, chunk bytes and retire set. A document that
/// is not chunked, or none, is edited whole as before, and so is every
/// document under `chunk_view = false`.
pub async fn commit_edit_paths<E>(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    flush: &str,
    first: Option<Option<std::sync::Arc<LoadedManifest>>>,
    edit: impl Fn(Option<&LoadedManifest>, &mut DocEdit<'_>) -> Result<(), E>,
) -> EditOutcome<E> {
    use std::sync::Arc;
    let mut first = first;
    // The pointer the last attempt sent, when its CAS answered 412 or 409:
    // the same own-write check as `commit_edit_from`'s, on the pointer.
    let mut sent: Option<Pointer> = None;
    for _ in 0..EDIT_CAS_ATTEMPTS {
        let current: Option<Arc<LoadedManifest>> = match first.take() {
            Some(read) => read,
            None => match load(store, cfg).await {
                Ok(l) => l.map(Arc::new),
                Err(e) => return EditOutcome { result: Err(EditError::Store(e)), held: None },
            },
        };
        let chunked: Option<Vec<super::chunk::ChunkRef>> =
            match current.as_ref().and_then(|l| l.pointer.as_ref()).map(|p| p.entries()) {
                Some(Ok(Entries::Chunked(r))) if cfg.chunked && cfg.chunk_view => Some(r.to_vec()),
                _ => None,
            };
        let refs = match chunked {
            Some(r) => r,
            None => {
                // Not chunked, nothing published, or the view turned off:
                // the whole-document edit. A refusal hands back what it read.
                let read = current.clone();
                let result = commit_edit_from(store, cfg, flush, Some(current), |cur, doc| {
                    let base = doc.clone();
                    let mut de = DocEdit {
                        seq: doc.seq,
                        entries: EditEntries { base: &base.entries, overlay: BTreeMap::new() },
                        tombstones: std::mem::take(&mut doc.tombstones),
                    };
                    edit(cur, &mut de)?;
                    doc.tombstones = de.tombstones;
                    apply_overlay(&mut doc.entries, de.entries.overlay);
                    Ok(())
                })
                .await;
                let held = match &result {
                    Err(EditError::Refused(_)) => read,
                    _ => None,
                };
                return EditOutcome { result, held };
            }
        };
        let cur = current.as_deref().expect("chunked above");
        if sent.as_ref().is_some_and(|s| cur.pointer.as_ref() == Some(s)) {
            return EditOutcome { result: Ok(()), held: None };
        }
        let base = &cur.manifest;
        let mut de = DocEdit {
            seq: base.seq + 1,
            entries: EditEntries { base: &base.entries, overlay: BTreeMap::new() },
            tombstones: base.tombstones.clone(),
        };
        if let Err(e) = edit(Some(cur), &mut de) {
            return EditOutcome { result: Err(EditError::Refused(e)), held: current };
        }
        let DocEdit { seq, entries, mut tombstones } = de;
        let overlay = entries.overlay;
        tombstones.retain(|_, t| seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
        let built = super::chunk::rechunk_with(
            &refs,
            |i| Ok(Arc::new(super::docview::chunk_entries(&base.entries, &refs, i).map(|(k, e)| (k.clone(), e.clone())).collect())),
            &overlay,
            cfg.chunk_target,
            cfg.chunk_min,
            cfg.chunk_max,
        )
        .and_then(|chunks| {
            let retired = super::docview::retired_over(cfg, &refs, &base.entries, &overlay);
            if cfg.g_shadow || cfg!(debug_assertions) {
                shadow_edit(cfg, base, &overlay, &chunks, &retired);
            }
            let tomb = match tombstones.is_empty() {
                true => None,
                false => Some(super::chunk::encode_tombstones(&tombstones)?.0),
            };
            Ok((chunks, retired, tomb))
        });
        let (chunks, retired, tomb) = match built {
            Ok(b) => b,
            Err(e) => return EditOutcome { result: Err(EditError::Store(e)), held: None },
        };
        let doc = ChunkedDoc { seq, sole_writer: false, boundary_source: None, tombstones: &tombstones };
        let expected = cur.handle();
        let sending = Pointer {
            seq,
            entries_key: None,
            entries_seq: None,
            chunks: Some(chunks.iter().map(|(r, _)| r.clone()).collect()),
            sole_writer: false,
            boundary_source: None,
            epoch: 0,
            tombstones: tomb,
        };
        match publish_chunk_list(store, cfg, &chunks, &doc, Some(&expected), &expected.prev_chunks, 0, flush).await {
            Ok((meta, chunk_refs, pointer)) => {
                debug_assert_eq!(pointer, sending, "the pointer published is not the one the own-write check compares");
                log_retired(store, cfg, seq, 0, flush, retired, Some(&chunk_refs), Some(&expected)).await;
                // The document to hold next: this one, made from what the
                // edit read, in place when nothing else holds that.
                let mut held = current.expect("chunked above");
                let l = Arc::make_mut(&mut held);
                apply_overlay(&mut l.manifest.entries, overlay);
                l.manifest.seq = seq;
                l.manifest.sole_writer = false;
                l.manifest.boundary_source = None;
                l.manifest.tombstones = tombstones;
                l.etag = meta.etag;
                // The citation's clock is this write's (a full load reads the
                // pointer's Last-Modified).
                l.last_modified_unix = meta.last_modified_unix.or_else(|| Some(super::now_unix()));
                // What the alias cache keeps is this document's chunks; the
                // new ones are scanned in the held map when first asked.
                let live: std::collections::HashSet<&str> =
                    pointer.chunks.iter().flatten().map(|r| r.addr.as_str()).collect();
                cfg.alias_cache.lock().unwrap().retain(|a, _| live.contains(a.as_str()));
                l.pointer = Some(pointer);
                return EditOutcome { result: Ok(()), held: Some(held) };
            }
            Err(LeanError::Store(StoreError::PreconditionFailed(_))) | Err(LeanError::Store(StoreError::Conflict(_))) => {
                sent = Some(sending);
                continue;
            }
            Err(e) => return EditOutcome { result: Err(EditError::Store(e)), held: None },
        }
    }
    EditOutcome { result: Err(EditError::Contended), held: None }
}

fn apply_overlay(entries: &mut BTreeMap<String, LeanEntry>, overlay: BTreeMap<String, Option<LeanEntry>>) {
    for (p, v) in overlay {
        match v {
            Some(e) => {
                entries.insert(p, e);
            }
            None => {
                entries.remove(&p);
            }
        }
    }
}

/// `commit_edit_paths`'s shadow (tests, and every debug build): the whole
/// document edited, split and diffed as `commit_edit_from` does, must give
/// the same chunk list, the same bytes for each chunk sent, and the same
/// retire set.
fn shadow_edit(
    cfg: &LeanConfig,
    base: &LeanManifest,
    overlay: &BTreeMap<String, Option<LeanEntry>>,
    chunks: &[(super::chunk::ChunkRef, Option<Vec<u8>>)],
    retired: &[String],
) {
    let mut whole = base.entries.clone();
    apply_overlay(&mut whole, overlay.clone());
    let want = super::chunk::split_with(&whole, cfg.chunk_target, cfg.chunk_min, cfg.chunk_max)
        .expect("shadow: the whole document does not split");
    let got_refs: Vec<&super::chunk::ChunkRef> = chunks.iter().map(|(r, _)| r).collect();
    let want_refs: Vec<&super::chunk::ChunkRef> = want.iter().map(|(r, _)| r).collect();
    assert_eq!(got_refs, want_refs, "edit shadow: the chunk list is not the whole document's");
    for ((r, body), (_, w)) in chunks.iter().zip(&want) {
        if let Some(b) = body {
            assert_eq!(b, w, "edit shadow: chunk {} encodes differently", r.addr);
        }
    }
    let still: std::collections::HashSet<&str> = whole.values().map(|e| e.key.as_str()).collect();
    let want_retired: Vec<String> =
        base.entries.values().map(|e| e.key.clone()).filter(|k| !still.contains(k.as_str())).collect();
    assert_eq!(retired, want_retired.as_slice(), "edit shadow: the retire set is not the whole document's");
}

/// CAS-write the manifest document. `expected`: None ⇒ first write
/// (If-None-Match:*); Some(etag) ⇒ If-Match. The caller owns 412
/// handling (merge-retry or fence — never blind re-seed).
pub async fn cas_write(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    m: &LeanManifest,
    expected: Option<&ManifestHandle>,
    epoch: u64,
    flush_uuid: &str,
) -> LeanResult<ObjectMeta> {
    cas_write_stamped(store, cfg, m, expected, epoch, flush_uuid, None).await
}

/// `cas_write` with an explicit boundary-source stamp on the manifest
/// OBJECT (§2.4.1): downstream consumers distinguish a declared-coherent
/// citation from a forced possibly-torn one from the bucket alone, for
/// the price of a HEAD rather than a GET of a manifest that can run to
/// hundreds of MiB.
pub async fn cas_write_stamped(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    m: &LeanManifest,
    expected: Option<&ManifestHandle>,
    epoch: u64,
    flush_uuid: &str,
    boundary_source: Option<&str>,
) -> LeanResult<ObjectMeta> {
    Ok(cas_write_refs(store, cfg, m, expected, epoch, flush_uuid, boundary_source).await?.0)
}

/// `cas_write_stamped`, also returning the chunk objects the new pointer
/// references (its chunks and its tombstones object) — `None` on the
/// single-object layout.
async fn cas_write_refs(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    m: &LeanManifest,
    expected: Option<&ManifestHandle>,
    epoch: u64,
    flush_uuid: &str,
    boundary_source: Option<&str>,
) -> LeanResult<(ObjectMeta, Option<Vec<String>>)> {
    // The manifest's own document must agree with its object stamp:
    // a reader that GETs and a reader that HEADs must never disagree
    // about how coherent the citation claims to be.
    // The layout switch. Every publisher routes through here — the
    // barrier, the takeover rotation and the gateway's HITL CAS — so the
    // decision lives at one site rather than three, and §6's "chunk
    // server-side on receipt" falls out rather than needing its own
    // path.
    if cfg.chunked {
        let (meta, refs) = cas_write_chunked_refs(
            store,
            cfg,
            m,
            expected,
            expected.map(|h| h.prev_chunks.as_slice()).unwrap_or(&[]),
            PublishStamps { epoch, flush_uuid, boundary_source },
        )
        .await?;
        return Ok((meta, Some(refs)));
    }
    let mut m = m.clone();
    if boundary_source.is_some() {
        m.boundary_source = boundary_source.map(|s| s.to_string());
    }
    let m = &m;
    let bytes = m.to_bytes();
    let crc = crc64_nvme(&bytes);
    let stamps = GenerationStamps {
        generation: m.seq,
        epoch,
        flush_uuid: flush_uuid.to_string(),
        // Stamp what the DOCUMENT carries, not merely what this call
        // passed. An explicit source has already been written into the
        // document above, so the two agree there either way; the case
        // this closes is a `None` caller over a document that already
        // names its clock — the gateway's HITL CAS relays a document it
        // did not author. Stamping `None` on those left the object
        // saying nothing while the document said `sentinel`, which is
        // exactly the GET/HEAD divergence the contract above forbids.
        boundary_source: m.boundary_source.clone(),
        posix: None,
    };

    // 1. The ENTRIES, to a fresh immutable key. `IfNoneMatchAny` is the
    //    whole safety story for this object: two writers that raced to
    //    the same seq cannot both land, and the loser is told so here
    //    rather than discovering it after it has already published.
    let entries_key = cfg.generation_key(m.seq, flush_uuid);
    match store
        .put_whole(&entries_key, bytes.into(), &PutCondition::IfNoneMatchAny, &stamps, crc)
        .await
    {
        Ok(_) => {}
        // The key carries this writer's flush uuid, so this is not a
        // race with another writer — it is a retry of OUR OWN write
        // whose pointer CAS we never learned the outcome of. The object
        // is byte-identical by construction; fall through to the CAS,
        // which is where the race is actually decided.
        Err(StoreError::PreconditionFailed(_)) | Err(StoreError::Conflict(_)) => {}
        Err(e) => return Err(e.into()),
    }

    // 2. The POINTER, which is what makes the publish visible and what
    //    decides the race. Everything above this line is invisible to
    //    every reader; a crash here leaves an orphan generation object
    //    that nobody reads and the sweep collects.
    let pointer = Pointer {
        seq: m.seq,
        entries_key: Some(entries_key.clone()),
        entries_seq: Some(m.seq),
        // The step-one layout. `cas_write_chunked` is what produces a
        // chunk list; this path stays byte-for-byte what it was.
        chunks: None,
        sole_writer: m.sole_writer,
        boundary_source: m.boundary_source.clone(),
        epoch,
        tombstones: None,
    };
    Ok((put_pointer(store, cfg, &pointer, expected, &stamps).await?, None))
}

/// Who is publishing, and under what coherence claim. Grouped because
/// they travel together through every publish path and are meaningless
/// apart — and because a positional list of three scalars is how an
/// epoch ends up where a flush uuid was meant to go.
pub struct PublishStamps<'a> {
    pub epoch: u64,
    pub flush_uuid: &'a str,
    pub boundary_source: Option<&'a str>,
}

/// Publish a manifest as a CHUNK LIST (chunked design §2).
///
/// The ordering is the atomicity story and is not negotiable: every
/// chunk is durable BEFORE the pointer CAS, and the pointer is the only
/// thing that ever becomes visible. A crash between the two leaves
/// orphan chunks nobody references and no reader can observe — never a
/// half-written manifest.
///
/// `prev` is the chunk list this publish is replacing. Chunks whose
/// address is already in it are skipped ENTIRELY — not uploaded and
/// re-conditioned, but never sent. That is where O(changed) actually
/// comes from: without it, a one-file publish still ships every chunk's
/// bytes over the wire and only the store's 412 saves the storage.
pub async fn cas_write_chunked(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    m: &LeanManifest,
    expected: Option<&ManifestHandle>,
    prev: &[super::chunk::ChunkRef],
    pub_stamps: PublishStamps<'_>,
) -> LeanResult<ObjectMeta> {
    Ok(cas_write_chunked_refs(store, cfg, m, expected, prev, pub_stamps).await?.0)
}

/// `cas_write_chunked`, also returning the chunk addresses the new pointer
/// references, its tombstones object included.
async fn cas_write_chunked_refs(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    m: &LeanManifest,
    expected: Option<&ManifestHandle>,
    prev: &[super::chunk::ChunkRef],
    pub_stamps: PublishStamps<'_>,
) -> LeanResult<(ObjectMeta, Vec<String>)> {
    let boundary_source = pub_stamps.boundary_source.map(|s| s.to_string()).or_else(|| m.boundary_source.clone());
    let split = super::chunk::split_with(&m.entries, cfg.chunk_target, cfg.chunk_min, cfg.chunk_max)?;
    let chunks: Vec<(super::chunk::ChunkRef, Option<Vec<u8>>)> = split.into_iter().map(|(r, b)| (r, Some(b))).collect();
    let refs: Vec<super::chunk::ChunkRef> = chunks.iter().map(|(r, _)| r.clone()).collect();
    super::docview::remember(cfg, &refs, &m.entries);
    let doc = ChunkedDoc { seq: m.seq, sole_writer: m.sole_writer, boundary_source, tombstones: &m.tombstones };
    let (meta, refs, _) =
        publish_chunk_list(store, cfg, &chunks, &doc, expected, prev, pub_stamps.epoch, pub_stamps.flush_uuid).await?;
    Ok((meta, refs))
}

/// What a chunked publish writes beside its chunk list.
pub struct ChunkedDoc<'a> {
    pub seq: u64,
    pub sole_writer: bool,
    pub boundary_source: Option<String>,
    pub tombstones: &'a BTreeMap<String, Tombstone>,
}

/// Publish a chunk list: every chunk `prev` lacks, then the tombstones
/// object, then the pointer CAS. `chunks` carries a body for each chunk it
/// encoded; one without a body is `prev`'s and is never sent. The ordering
/// is the atomicity story (`cas_write_chunked`). Returns the pointer's
/// response, the chunk addresses it references (its tombstones object
/// included), and the pointer itself.
#[allow(clippy::too_many_arguments)]
async fn publish_chunk_list(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    chunks: &[(super::chunk::ChunkRef, Option<Vec<u8>>)],
    doc: &ChunkedDoc<'_>,
    expected: Option<&ManifestHandle>,
    prev: &[super::chunk::ChunkRef],
    epoch: u64,
    flush_uuid: &str,
) -> LeanResult<(ObjectMeta, Vec<String>, Pointer)> {
    let have: std::collections::HashSet<&str> = prev.iter().map(|r| r.addr.as_str()).collect();

    let stamps = GenerationStamps {
        generation: doc.seq,
        epoch,
        flush_uuid: flush_uuid.to_string(),
        boundary_source: doc.boundary_source.clone(),
        posix: None,
    };

    for (r, body) in chunks {
        if have.contains(r.addr.as_str()) {
            continue;
        }
        let Some(body) = body else {
            return Err(LeanError::State(format!(
                "chunk {} is not in the list this publish replaces and was never encoded",
                r.addr
            )));
        };
        // UNCONDITIONAL, and that is a correctness requirement rather
        // than laziness — `LeanChunkGCAdoptSkips` in the formal model
        // fails without it.
        //
        // This chunk is not in `prev`, so nothing references it yet. It
        // may nonetheless already EXIST: a crashed publish of ours left
        // it, and content addressing means our retry produces the same
        // key. Adopting it by reference alone would make an old object
        // live without making it LOOK live, and the orphan sweep reads
        // age — so the sweep would collect it as the orphan it still
        // appears to be, between this call and our pointer CAS.
        // Rewriting refreshes the age the sweep reads. The bytes are
        // identical either way, so the write is idempotent; the cost
        // falls only on adopted orphans, which are rare by
        // construction, and chunks already in `prev` are skipped above.
        let key = cfg.chunk_key(&r.addr);
        let crc = crc64_nvme(body);
        store.put_whole(&key, body.clone().into(), &PutCondition::Unconditional, &stamps, crc).await?;
        // This writer's next load of its own document fetches nothing it
        // just wrote (latency plan lever A). Kept whether or not the CAS
        // below lands: a body is right for its address either way, and a
        // re-merge after a lost race mostly re-addresses to these.
        if let Some(c) = cfg.chunk_cache.as_deref() {
            c.put(&r.addr, body);
        }
    }

    // The tombstones, one object beside the chunks (rewritten like an
    // adopted chunk: the sweep reads age, and the bytes are identical).
    let tombstones = if doc.tombstones.is_empty() {
        None
    } else {
        let (addr, body) = super::chunk::encode_tombstones(doc.tombstones)?;
        let crc = crc64_nvme(&body);
        if let Some(c) = cfg.chunk_cache.as_deref() {
            c.put(&addr, &body);
        }
        store.put_whole(&cfg.chunk_key(&addr), body.into(), &PutCondition::Unconditional, &stamps, crc).await?;
        Some(addr)
    };
    let pointer = Pointer {
        seq: doc.seq,
        entries_key: None,
        entries_seq: None,
        chunks: Some(chunks.iter().map(|(r, _)| r.clone()).collect()),
        sole_writer: doc.sole_writer,
        boundary_source: doc.boundary_source.clone(),
        epoch,
        tombstones: tombstones.clone(),
    };
    let meta = put_pointer(store, cfg, &pointer, expected, &stamps).await?;
    let mut refs: Vec<String> = pointer.chunks.iter().flatten().map(|r| r.addr.clone()).collect();
    refs.extend(tombstones);
    Ok((meta, refs, pointer))
}

/// What a commit retired (M1): the handles its document no longer cites
/// and the chunk objects its pointer no longer references. One immutable
/// log object per commit, under `retired/`, keyed by the unix second it
/// was written — the retirement's clock. The sweeps spare everything a
/// log younger than `retire_grace_secs` names; a writer reaps the older
/// ones (`untracked.rs`).
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct RetireLog {
    pub at_unix: u64,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub handles: Vec<String>,
    /// Chunk ADDRESSES (`LeanConfig::chunk_key` names the object).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub chunks: Vec<String>,
}

pub fn retired_prefix(cfg: &LeanConfig) -> String {
    format!("{}/{}/retired/", cfg.prefix, super::LEAN_DIR)
}

/// The unix second a retire log's KEY carries (`<at:020>-...`): a listing
/// dates every log with no GET.
pub fn retire_log_at(cfg: &LeanConfig, key: &str) -> Option<u64> {
    key.strip_prefix(&retired_prefix(cfg))?.get(..20)?.parse().ok()
}

/// A commit, and the log of what it retired, written right after the CAS.
/// A crash between the two leaves those handles and chunks to the sweeps'
/// write-age rule — the shape before the retire age — and nothing worse.
/// The log is best effort for the same reason: a failure to write it costs
/// a lagging reader a re-resolve, never the commit.
#[allow(clippy::too_many_arguments)]
pub async fn cas_write_retiring(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    m: &LeanManifest,
    before: Option<&LeanManifest>,
    expected: Option<&ManifestHandle>,
    epoch: u64,
    flush_uuid: &str,
    boundary_source: Option<&str>,
) -> LeanResult<ObjectMeta> {
    let (meta, refs) = cas_write_refs(store, cfg, m, expected, epoch, flush_uuid, boundary_source).await?;
    let still: std::collections::HashSet<&str> = m.entries.values().map(|e| e.key.as_str()).collect();
    let handles: Vec<String> = before
        .map(|b| b.entries.values().map(|e| e.key.clone()).filter(|k| !still.contains(k.as_str())).collect())
        .unwrap_or_default();
    log_retired(store, cfg, m.seq, epoch, flush_uuid, handles, refs.as_deref(), expected).await;
    Ok(meta)
}

/// Lever G's commit (latency plan §12): publish the chunk list a view's
/// re-chunk made (`docview::DocView::rechunk`), and log what it retired.
/// `retired_handles` is the caller's: the keys of the document it merged
/// onto that the published one no longer cites, as `cas_write_retiring`
/// computes them over both whole documents.
#[allow(clippy::too_many_arguments)]
pub async fn cas_write_view(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    chunks: &[(super::chunk::ChunkRef, Option<Vec<u8>>)],
    doc: &ChunkedDoc<'_>,
    expected: &ManifestHandle,
    epoch: u64,
    flush_uuid: &str,
    retired_handles: Vec<String>,
) -> LeanResult<ObjectMeta> {
    let (meta, refs, _) =
        publish_chunk_list(store, cfg, chunks, doc, Some(expected), &expected.prev_chunks, epoch, flush_uuid).await?;
    log_retired(store, cfg, doc.seq, epoch, flush_uuid, retired_handles, Some(&refs), Some(expected)).await;
    Ok(meta)
}

/// The log of what a commit retired: `handles`, and the chunk objects the
/// replaced pointer named that `refs` (the new one's) does not.
#[allow(clippy::too_many_arguments)]
async fn log_retired(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    seq: u64,
    epoch: u64,
    flush_uuid: &str,
    mut handles: Vec<String>,
    refs: Option<&[String]>,
    expected: Option<&ManifestHandle>,
) {
    handles.sort();
    handles.dedup();
    let mut chunks: Vec<String> = vec![];
    if let (Some(refs), Some(h)) = (refs, expected) {
        let now: std::collections::HashSet<&str> = refs.iter().map(|a| a.as_str()).collect();
        chunks = h
            .prev_chunks
            .iter()
            .map(|r| r.addr.clone())
            .chain(h.prev_tombstones.clone())
            .filter(|a| !now.contains(a.as_str()))
            .collect();
        chunks.sort();
        chunks.dedup();
    }
    if !handles.is_empty() || !chunks.is_empty() {
        let at = super::now_unix();
        let log = RetireLog { at_unix: at, handles, chunks };
        let key = format!("{}{at:020}-{flush_uuid}-{}.json", retired_prefix(cfg), uuid::Uuid::new_v4());
        let body = match serde_json::to_vec(&log) {
            Ok(b) => b,
            Err(e) => {
                eprintln!("flint-sync: could not encode what seq {seq} retired ({e}); the sweeps judge it by write age");
                return;
            }
        };
        let crc = crc64_nvme(&body);
        let stamps = GenerationStamps {
            generation: seq,
            epoch,
            flush_uuid: flush_uuid.to_string(),
            boundary_source: None,
            posix: None,
        };
        if let Err(e) = store.put_whole(&key, body.into(), &PutCondition::IfNoneMatchAny, &stamps, crc).await {
            eprintln!("flint-sync: could not log what seq {seq} retired ({e}); the sweeps judge it by write age");
        }
    }
}

/// CAS the pointer. Split out because a takeover rotation writes ONLY
/// this — no entries object, no multi-MB anything.
async fn put_pointer(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    pointer: &Pointer,
    expected: Option<&ManifestHandle>,
    stamps: &GenerationStamps,
) -> LeanResult<ObjectMeta> {
    let body = serde_json::to_vec(pointer)
        .map_err(|e| super::LeanError::State(format!("pointer encode: {e}")))?;
    let crc = crc64_nvme(&body);
    // A LEGACY handle is a migration: the writer read the old single
    // object and the pointer does not exist yet, so the precondition is
    // "no pointer", not "the pointer I read". A concurrent migration
    // therefore loses here and retries against the layout that won,
    // instead of both of them believing they installed it.
    let cond = match expected {
        Some(h) if !h.legacy => PutCondition::IfMatch(h.etag.clone()),
        _ => PutCondition::IfNoneMatchAny,
    };
    let meta = store.put_whole(&cfg.current_key(), body.into(), &cond, stamps, crc).await?;
    if let Some(h) = expected.filter(|h| h.legacy) {
        // The workspace has moved. Poison the legacy key rather than
        // delete it — see REFUSAL_DOC. Conditional on the very object
        // this writer read, so a racing writer's view is never
        // clobbered, and best effort: the refusal doc is decorative to
        // every new reader, and the ONLY reader it speaks to is an old
        // binary that must refuse rather than re-seed.
        let bytes = REFUSAL_DOC.as_bytes().to_vec();
        let crc = crc64_nvme(&bytes);
        if let Err(e) = store
            .put_whole(
                &cfg.manifest_key(),
                bytes.into(),
                &PutCondition::IfMatch(h.etag.clone()),
                stamps,
                crc,
            )
            .await
        {
            eprintln!("flint-sync: could not poison the legacy manifest key after migration: {e}");
        }
    }
    Ok(meta)
}

/// Bump the generation without touching the entries.
///
/// This is what a takeover does, and under the single-object layout it
/// was a multi-MB GET + PUT per claim that also moved the ETag every
/// other syncer's no-change early exit compares against — so one claim
/// cost every follower a full fetch and parse of a document in which
/// nothing had changed. Here it is one small CAS: `seq` moves, so an
/// outstanding pointer handle goes stale exactly as before, and
/// `entries_seq` does NOT, so a follower can see that the entries are
/// the ones it already has.
pub async fn rotate_for_takeover(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    epoch: u64,
) -> LeanResult<Option<(LeanManifest, String)>> {
    for _ in 0..3 {
        // The legacy layout has no pointer to rotate: migrate by
        // rewriting the document once, then rotations are cheap forever.
        let Some(LoadedPointer { pointer: p, etag, .. }) = load_pointer(store, cfg).await? else {
            let Some(loaded) = load(store, cfg).await? else {
                return Ok(None);
            };
            let mut rotated = loaded.manifest.clone();
            rotated.seq += 1;
            let h = loaded.handle();
            match cas_write(store, cfg, &rotated, Some(&h), epoch, "takeover-rotation").await {
                Ok(meta) => return Ok(Some((rotated, meta.etag))),
                Err(super::LeanError::Store(StoreError::PreconditionFailed(_) | StoreError::Conflict(_))) => continue,
                Err(e) => return Err(e),
            }
        };
        let next = Pointer { seq: p.seq + 1, epoch, ..p.clone() };
        let stamps = GenerationStamps {
            generation: next.seq,
            epoch,
            flush_uuid: "takeover-rotation".to_string(),
            boundary_source: next.boundary_source.clone(),
            posix: None,
        };
        let h = ManifestHandle { etag, legacy: false, prev_chunks: Vec::new(), prev_tombstones: None };
        match put_pointer(store, cfg, &next, Some(&h), &stamps).await {
            Ok(meta) => {
                // The document the caller gets back is the standing one
                // with the rotated seq — the entries were not read and
                // did not need to be.
                let mut m = LeanManifest::default();
                m.seq = next.seq;
                m.sole_writer = next.sole_writer;
                m.boundary_source = next.boundary_source.clone();
                return Ok(Some((m, meta.etag)));
            }
            // S3 answers a conditional write racing another on the key with 409,
            // and a takeover whose acquire has LANDED must not fail on a lost race
            // (review 2026-09-18, H2).
            Err(super::LeanError::Store(StoreError::PreconditionFailed(_) | StoreError::Conflict(_))) => continue,
            Err(e) => return Err(e),
        }
    }
    Err(super::LeanError::State(
        "takeover rotation lost 3 CAS races — a live writer is still publishing".into(),
    ))
}

/// How many superseded generations to keep behind the live one.
///
/// Not zero, and not one. A reader GETs the pointer and then GETs the
/// object it names, and those are two requests with a gap between them:
/// reaping the generation a reader has just resolved would turn a
/// perfectly ordinary read into a `NotFound`. A handful is a few hundred
/// KiB at any realistic entry count and buys that gap several publishes.
pub const KEEP_GENERATIONS: usize = 5;

/// How long a generation NEWER than the live one is left alone before it
/// is treated as an orphan.
///
/// An object above the live generation is one of two things: a publish
/// that is in flight right now — its entries written, its pointer CAS
/// not yet landed — or the wreckage of one that died in that window. The
/// two are indistinguishable from a listing, and deleting the first
/// breaks a live publish, so age is the only honest discriminator. Well
/// past any plausible manifest PUT, including a 264 MiB one over a slow
/// link.
pub const ORPHAN_GRACE_SECS: u64 = 3600;

/// Retire-age G's default (`LeanConfig::retire_grace_secs`): how long a
/// handle or chunk stays after the commit that stopped referencing it.
/// Past a large checkout's manifest load and a UI read's two GETs; a
/// reader that lags longer re-resolves the pointer, as it always has. It
/// costs storage: every version a commit replaced, kept this long.
pub const RETIRE_GRACE_SECS: u64 = 600;

/// Delete superseded generation objects, and orphans left by a publish
/// that died between its entries PUT and its pointer CAS.
///
/// Immutable metadata that is never collected is a leak, and this one
/// grows by a whole manifest per publish.
///
/// The two cases are NOT symmetric, which the first cut of this got
/// wrong. BELOW the live generation, everything is superseded by
/// definition and the only question is how much slack to leave a reader
/// mid-resolve — that is `KEEP_GENERATIONS`. ABOVE it, an object may be
/// a publish still in flight, and a keep-window is exactly the wrong
/// tool: an orphan at a high seq sorts to the front and the window
/// protects it forever while costing a real generation its slot. Age
/// decides that case instead.
///
/// Best effort and off the critical path: a failure here costs storage,
/// never correctness, so it warns rather than failing a publish.
pub async fn sweep_generations(store: &dyn ObjectStore, cfg: &LeanConfig) -> LeanResult<usize> {
    let Some(lp) = load_pointer(store, cfg).await? else {
        return Ok(0);
    };
    // A chunked pointer names no generation object at all, so there is
    // nothing here to keep a window of. Chunks are shared across
    // generations and are collected by a different rule entirely
    // (chunked design §8.1) — sweeping them by "not named by the live
    // pointer" would delete history the window exists to preserve.
    let Some(live) = lp.pointer.entries_key else {
        // The workspace has MOVED to the chunk layout, so no pointer
        // names a generation object any more and every one of them is
        // superseded. Collect them past the grace rather than leaking
        // the whole pre-migration history forever — a reader still
        // resolving one revalidates and restarts (§8.2), and the grace
        // covers the window in which it might not have noticed yet.
        let prefix = format!("{}/{}/manifests/", cfg.prefix, super::LEAN_DIR);
        let now = super::now_unix();
        let mut removed = 0;
        for o in store.list(&prefix).await? {
            let age = o.last_modified_unix.map(|t| now.saturating_sub(t));
            if !age.is_some_and(|a| a >= cfg.orphan_grace_secs) {
                continue;
            }
            match store.delete(&o.key).await {
                Ok(()) => removed += 1,
                Err(e) => eprintln!("flint-sync: could not reap migrated generation {}: {e}", o.key),
            }
        }
        return Ok(removed);
    };
    let prefix = format!("{}/{}/manifests/", cfg.prefix, super::LEAN_DIR);
    let mut listed = store.list(&prefix).await?;
    // The key is `<seq:020>-<uuid>`, so lexical order IS generation
    // order and a plain sort is enough — the zero padding is doing this
    // job, which is why it is there.
    listed.sort_by(|a, b| a.key.cmp(&b.key));
    let now = super::now_unix();
    let live_ix = listed.iter().position(|o| o.key == live);

    let mut doomed: Vec<String> = Vec::new();
    for (i, o) in listed.iter().enumerate() {
        if o.key == live {
            continue;
        }
        match live_ix {
            // Newer than the live pointer: in flight, or wreckage.
            Some(ix) if i > ix => {
                let age = o.last_modified_unix.map(|t| now.saturating_sub(t));
                // No timestamp ⇒ leave it. A store that cannot date its
                // objects gets a leak, not a deleted live publish.
                if age.is_some_and(|a| a > ORPHAN_GRACE_SECS) {
                    doomed.push(o.key.clone());
                }
            }
            // Older than the live pointer: superseded. Keep a window — by
            // count, and (M1) until the generation that superseded it is
            // `retire_grace_secs` old: at one UI save a second, the count
            // alone is seconds. No date on the successor ⇒ keep it.
            Some(ix) => {
                let superseded_for = listed[i + 1].last_modified_unix.map(|t| now.saturating_sub(t));
                if ix.saturating_sub(i) > KEEP_GENERATIONS
                    && superseded_for.is_some_and(|a| a >= cfg.retire_grace_secs)
                {
                    doomed.push(o.key.clone());
                }
            }
            // The pointer names an object that is not in the listing.
            // `load` already refuses that workspace; sweep nothing.
            None => return Ok(0),
        }
    }
    let mut removed = 0;
    for k in doomed {
        match store.delete(&k).await {
            Ok(()) => removed += 1,
            Err(e) => eprintln!("flint-sync: could not reap superseded generation {k}: {e}"),
        }
    }
    Ok(removed)
}

/// Collect chunk objects that no live pointer references.
///
/// Built to `lean/formal/LeanChunkGC.tla`, which refuted the ordering
/// rule the design originally prescribed. Four things carry the safety
/// here and each has a mutation config that breaks it alone:
///
/// 1. **The reference set must not predate a CAS the delete follows.**
///    S3 cannot make the read and the delete atomic, so this fences
///    instead: the pointer's ETag is read before the listing and again
///    after, and the pass ABORTS if it moved. A publish landed, and we
///    no longer know whether it references a candidate.
/// 2. **A grace.** A chunk written and not yet referenced would
///    otherwise be collected out from under its own publish.
/// 3. **The grace outlives the longest publish** — a timing
///    assumption, which is why `orphan_grace_secs` is config and why
///    this comment names it rather than leaving it implied.
/// 4. **Adoption rewrites what it adopts** (`cas_write_chunked`), so
///    any chunk a concurrent publish could reference is FRESH and the
///    grace covers it. Without that rule an adopted orphan is old, and
///    the age this reads is a lie about whether it is in use.
///
/// Deliberately no history window, unlike `sweep_generations`. That
/// window existed partly so a reader mid-resolve could finish; readers
/// now revalidate against the pointer and restart onto the current
/// generation (§8.2), which removes the reason. Adding one back means
/// retaining pointer SNAPSHOTS so their chunk sets are enumerable
/// (§9) — chunks are shared, so "what the live pointer does not name"
/// is the only reference set available without them.
pub async fn sweep_chunks(store: &dyn ObjectStore, cfg: &LeanConfig) -> LeanResult<usize> {
    sweep_chunks_sparing(store, cfg, &std::collections::HashSet::new()).await
}

/// `sweep_chunks`, sparing the chunk KEYS a young retire log names (M1):
/// a chunk a commit superseded less than `retire_grace_secs` ago, however
/// long ago it was written. A reader mid-load of the previous pointer
/// finds it.
pub async fn sweep_chunks_sparing(
    store: &dyn ObjectStore,
    cfg: &LeanConfig,
    spared: &std::collections::HashSet<String>,
) -> LeanResult<usize> {
    let Some(before) = load_pointer(store, cfg).await? else {
        return Ok(0);
    };
    let mut refs: std::collections::HashSet<String> = match before.pointer.entries()? {
        Entries::Chunked(c) => c.iter().map(|r| r.addr.clone()).collect(),
        // The single-object layout has no chunks; `sweep_generations`
        // owns it, and running both over one workspace would have each
        // reasoning about objects the other manages.
        Entries::Single { .. } => return Ok(0),
    };
    // The tombstone object lives beside the chunks and is referenced
    // the same way.
    if let Some(t) = &before.pointer.tombstones {
        refs.insert(t.clone());
    }

    let prefix = format!("{}/{}/chunks/", cfg.prefix, super::LEAN_DIR);
    let listed = store.list(&prefix).await?;

    // THE FENCE (rule 1). Nothing published across the reference read
    // and the listing, so the candidates and the references describe
    // the same moment. A publish that lands after this still cannot
    // lose a chunk: everything it references is either already in
    // `refs` or was just written or rewritten, and rule 4 guarantees
    // the latter are fresh.
    let after = load_pointer(store, cfg).await?;
    if after.map(|p| p.etag) != Some(before.etag) {
        return Ok(0);
    }

    let now = super::now_unix();
    let mut removed = 0;
    for o in listed {
        let Some(addr) = o.key.rsplit('/').next() else { continue };
        if refs.contains(addr) || spared.contains(&o.key) {
            continue;
        }
        // A candidate the LISTING already shows inside the grace cannot be
        // past it on a HEAD taken now: a rewrite only makes an object
        // younger, so the listing's age bounds the HEAD's from above.
        // Skipping it keeps a chunk longer — a leak, never a deleted live
        // chunk — and every deletion still rests on the HEAD below. Without
        // this every commit HEADed every chunk superseded within the grace,
        // one at a time inside the fence: 700 ms of a 1 s hold after five
        // minutes of six writers (contention drill), growing for the hour.
        if let Some(listed) = o.last_modified_unix {
            if now.saturating_sub(listed) < cfg.orphan_grace_secs {
                continue;
            }
        }
        // The age must come from a HEAD taken NOW, never from the
        // listing above. That listing predates the fence, and rule 4
        // (adoption REWRITES what it adopts) works by refreshing a
        // chunk's age — so judging by the listing's timestamp makes
        // exactly the refresh that rule exists to produce invisible,
        // and the reaper deletes a chunk an in-flight publish is about
        // to name. The model computed this against the store's state at
        // DELETE time; reading it from a pre-fence snapshot was an
        // implementation that did not match the rule it was built to.
        //
        // One HEAD per CANDIDATE past the grace by the listing, not per
        // chunk: a busy workspace supersedes chunks on every publish.
        //
        // Gone already ⇒ nothing to do. Any other HEAD failure, or no
        // timestamp ⇒ leave it: a store that cannot date its objects
        // gets a leak, not a deleted live chunk, the same rule
        // `sweep_generations` follows and for the same reason.
        let fresh = match store.head(&o.key).await {
            Ok(m) => m.last_modified_unix,
            Err(StoreError::NotFound(_)) => continue,
            Err(e) => {
                eprintln!("flint-sync: could not date orphan chunk {} ({e}); leaving it", o.key);
                continue;
            }
        };
        // `>=`, not `>`: the config value is a DURATION the object must
        // have survived, so `orphan_grace_secs = 0` has to mean "no
        // grace". With `>` it would silently mean "collect nothing
        // written this second", which is most of them — a knob whose
        // zero disables the thing it configures is a trap.
        let age = fresh.map(|t| now.saturating_sub(t));
        if !age.is_some_and(|a| a >= cfg.orphan_grace_secs) {
            continue;
        }
        match store.delete(&o.key).await {
            Ok(()) => removed += 1,
            Err(e) => eprintln!("flint-sync: could not reap orphan chunk {}: {e}", o.key),
        }
    }
    Ok(removed)
}

/// The three-way merge (the model's `inst`). Starts from THEIRS so
/// foreign entries survive by construction; applies my upserts; applies
/// my deletes only where theirs is unchanged since my merge base.
///
/// **`mine_upserts` and `mine_deletes` must be DISJOINT**, and that is
/// the caller's job, not this function's: which one wins depends on
/// which the tree saw LAST, and neither set carries that. The fused
/// barrier gets it free (both are computed from one scan). Resolving an
/// overlap here —
/// in either direction — cites a deleted file half the time and
/// amputates a live one the other half. TLC found both halves.
/// Returns (as `Merged`) the merged document, the foreign entries a consume must
/// integrate next (present in theirs, changed vs base, not mine), each
/// declared delete's outcome, and the entries MINE PUBLISHES OVER: theirs' version at a path this barrier
/// uploaded, changed since the merge base. The slot's 412 used to catch
/// those at the PUT; under handles the merge is the one place they show,
/// and the caller decides which it never integrated and must surface
/// (design 2026-09-19, R7).
pub fn merge(
    base: &BTreeMap<String, String>,
    theirs: &LeanManifest,
    mine_upserts: &BTreeMap<String, LeanEntry>,
    mine_deletes: &BTreeSet<String>,
    parked: &BTreeSet<String>,
) -> Merged {
    let base_of = |p: &str| base.get(p).map(|b| b.as_str());
    let at: BTreeMap<String, LeanEntry> = mine_upserts
        .keys()
        .chain(mine_deletes.iter())
        .filter_map(|p| theirs.entries.get(p).map(|e| (p.clone(), e.clone())))
        .collect();
    let c = merge_changes(&base_of, &at, &theirs.tombstones, theirs.seq, mine_upserts, mine_deletes, parked);
    let mut foreign: Vec<(String, LeanEntry)> = vec![];
    for (p, e) in &theirs.entries {
        let changed = base.get(p).map(|b| b != &e.etag).unwrap_or(true);
        if changed && !mine_upserts.contains_key(p) && !parked.contains(p) {
            foreign.push((p.clone(), e.clone()));
        }
    }
    let mut merged = theirs.clone();
    merged.seq = c.seq;
    // Same discipline, and for a sharper reason: inheriting it would
    // let a workspace that once published as a mirror keep refusing
    // adopts forever, and NOT inheriting it would let a mirror silently
    // stop being one the first time anything else wrote a manifest.
    // The installing pass sets it from its own config.
    merged.sole_writer = false;
    merged.boundary_source = None;
    for (p, v) in &c.overlay {
        match v {
            Some(e) => {
                merged.entries.insert(p.clone(), e.clone());
            }
            None => {
                merged.entries.remove(p);
            }
        }
    }
    merged.tombstones = c.tombstones;
    Merged {
        doc: merged,
        foreign,
        overridden: c.overridden,
        deletes: c.deletes,
        recreated: c.recreated,
        over_theirs: c.over_theirs,
    }
}

/// What `merge` decides for the paths a barrier changes (lever G, latency
/// plan §12): everything but the foreign set, which is about every OTHER
/// path. `merge` is this over the whole document plus that set, so a path
/// cannot be decided two ways.
pub struct MergedChanges {
    /// The merged document's generation.
    pub seq: u64,
    /// Each changed path's entry in the merged document: `Some` cited (an
    /// upsert), `None` no longer cited (a delete that applied to a cited
    /// path). Every other path is theirs, as it stands.
    pub overlay: BTreeMap<String, Option<LeanEntry>>,
    /// The merged document's tombstones, whole.
    pub tombstones: BTreeMap<String, Tombstone>,
    pub overridden: Vec<(String, LeanEntry)>,
    pub deletes: BTreeMap<String, DeleteOutcome>,
    pub recreated: Vec<(String, String)>,
    pub over_theirs: Vec<(String, LeanEntry)>,
}

/// `merge`'s per-path decisions. `base(p)` is the merge base's etag at
/// `p`; `theirs_at` is theirs' entry at every path of `mine_upserts` and
/// `mine_deletes` it cites (a path it lacks is not cited).
pub fn merge_changes<'b>(
    base: &dyn Fn(&str) -> Option<&'b str>,
    theirs_at: &BTreeMap<String, LeanEntry>,
    theirs_tombstones: &BTreeMap<String, Tombstone>,
    theirs_seq: u64,
    mine_upserts: &BTreeMap<String, LeanEntry>,
    mine_deletes: &BTreeSet<String>,
    parked: &BTreeSet<String>,
) -> MergedChanges {
    let seq = theirs_seq + 1;
    let mut tombstones = theirs_tombstones.clone();
    let mut overlay: BTreeMap<String, Option<LeanEntry>> = BTreeMap::new();
    // The DELETES mine publishes over: a path mine uploads that the merge
    // base cited and theirs no longer does. Named by the version the
    // delete removed — theirs' tombstone, else the base's citation.
    let recreated: Vec<(String, String)> = mine_upserts
        .keys()
        .filter(|p| !theirs_at.contains_key(*p))
        .filter_map(|p| {
            let base_etag = base(p)?;
            let deleted = theirs_tombstones.get(p).map(|t| t.etag.clone()).unwrap_or_else(|| base_etag.to_string());
            Some((p.clone(), deleted))
        })
        .collect();
    // Theirs' version at a path mine uploads, changed since the merge base
    // (R7), in path order.
    let overridden: Vec<(String, LeanEntry)> = mine_upserts
        .keys()
        .filter_map(|p| theirs_at.get(p).map(|e| (p, e)))
        .filter(|(p, e)| base(p).map(|b| b != e.etag).unwrap_or(true))
        .map(|(p, e)| (p.clone(), e.clone()))
        .collect();
    for (p, e) in mine_upserts {
        overlay.insert(p.clone(), Some(e.clone()));
        // A path cited again has no delete to remember.
        tombstones.remove(p);
    }
    let mut deletes = BTreeMap::new();
    let mut over_theirs: Vec<(String, LeanEntry)> = vec![];
    for p in mine_deletes {
        let outcome = delete_outcome_with(base(p), theirs_at.get(p), parked.contains(p));
        deletes.insert(p.clone(), outcome);
        if outcome == DeleteOutcome::OverTheirs {
            if let Some(e) = theirs_at.get(p) {
                over_theirs.push((p.clone(), e.clone()));
            }
        }
        if outcome != DeleteOutcome::Parked {
            // What the merged document cites at `p` so far: mine's upsert,
            // else theirs.
            let cited = match overlay.get(p) {
                Some(v) => v.clone(),
                None => theirs_at.get(p).cloned(),
            };
            if let Some(retired) = cited {
                overlay.insert(p.clone(), None);
                // H1e: the document says what this delete retired.
                tombstones.insert(p.clone(), Tombstone { etag: retired.etag, seq });
            }
        }
    }
    tombstones.retain(|_, t| seq.saturating_sub(t.seq) <= TOMBSTONE_KEEP_SEQS);
    MergedChanges { seq, overlay, tombstones, overridden, deletes, recreated, over_theirs }
}

/// What `merge` returns: the document, and what it decided per path.
pub struct Merged {
    pub doc: LeanManifest,
    /// Entries a consume must integrate next: present in theirs, changed
    /// vs base, not mine.
    pub foreign: Vec<(String, LeanEntry)>,
    /// Theirs' version at a path this barrier uploaded, changed since the
    /// merge base (R7).
    pub overridden: Vec<(String, LeanEntry)>,
    /// Every declared delete, and what the merge did with it.
    pub deletes: BTreeMap<String, DeleteOutcome>,
    /// The deletes mine publishes over: `(path, the deleted version's
    /// etag)` for a path mine uploads that the merge base cited and theirs
    /// deleted. Mine wins; the caller records each (the user's rule: a
    /// delete over a dirty path lands, with a record).
    pub recreated: Vec<(String, String)>,
    /// The versions mine DELETES over: theirs at a path this barrier
    /// deletes, changed since the merge base (M3, P1-lite). Mine wins; the
    /// caller preserves and records each before the collector takes it.
    pub over_theirs: Vec<(String, LeanEntry)>,
}

/// What the merge does with one of my deletes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeleteOutcome {
    /// Theirs is unchanged since my merge base, or already gone: the
    /// citation goes.
    Applied,
    /// Theirs moved off my merge base to a version this tree never had
    /// (M3, P1-lite): the delete still applies — mine wins — and theirs is
    /// preserved and recorded (`Merged::over_theirs`). Before P1-lite theirs
    /// stayed cited (outranked), and under a derived owed set that never
    /// settles: the tree lacks the path, the document cites theirs, the
    /// path is dirty so nothing is owed, and every barrier repeats.
    OverTheirs,
    /// Withheld this barrier.
    Parked,
}

/// The ONE place a delete's outcome is decided. `merge` applies it, and
/// a caller that must know whether an install moves a path off its
/// citation asks this with the same base — never re-derives it. Re-deriving
/// it is how the repair rule came to read a different base from the merge
/// (M5, L-124).
pub fn delete_outcome(
    base: &BTreeMap<String, String>,
    theirs: &LeanManifest,
    p: &str,
    parked: &BTreeSet<String>,
) -> DeleteOutcome {
    delete_outcome_with(base.get(p).map(|b| b.as_str()), theirs.entries.get(p), parked.contains(p))
}

/// [`delete_outcome`] from the two things it reads: the merge base's etag
/// at the path, and theirs' entry there.
pub fn delete_outcome_with(base: Option<&str>, theirs: Option<&LeanEntry>, parked: bool) -> DeleteOutcome {
    if parked {
        return DeleteOutcome::Parked;
    }
    match (theirs, base) {
        (Some(e), Some(b)) if e.etag == b => DeleteOutcome::Applied,
        // Already deleted by someone else: nothing is left to cite.
        (None, _) => DeleteOutcome::Applied,
        // Changed since base, or present in theirs and not in base.
        (Some(_), _) => DeleteOutcome::OverTheirs,
    }
}