flint_lean/

barrier.rs

//! The publish barrier (plan §2.1, seven steps) — the machine
//! `lean/formal/LeanSubtree.tla` checks. Order is load-bearing:
//! consume → scan → intent/window → uploads → manifest CAS (merge) →
//! GC deletes LAST → baseline rewrite.

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

use bytes::Bytes;

use flint_store::{
    EpochState,
    crc64_nvme, crc64_to_b64, GenerationStamps, PosixStamps, PutCondition, StoreError,
};

use super::inbox::{self, InboxEntry};
use super::manifest::{self, LeanEntry};
use super::scan;
use super::state::{BaselineEntry, ConflictRecord, IntentJournal};
use super::{now_unix, LeanError, LeanResult, Syncer};

/// The last gateway request this workspace acted on, per verb. A
/// watermark rather than a consumed queue — see `note_verb_requests`.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct VerbWatermark {
    #[serde(default)]
    pub boundary_unix: u64,
    #[serde(default)]
    pub sync_unix: u64,
}

#[derive(Debug, Default)]
pub struct BarrierReport {
    pub seq: Option<u64>,
    pub uploaded: Vec<String>,
    pub deleted: Vec<String>,
    pub parked: Vec<String>,
    pub consumed: usize,
    pub foreign_queued: usize,
    /// Paths whose transfer drifted or was swept mid-compose: nothing
    /// published, nothing advanced; the next scan re-queues them.
    pub deferred: Vec<String>,
    pub no_change: bool,
    /// Bytes this barrier actually published — the input to the
    /// sentinel work meter (boundary-verbs plan D3.1). Metering work
    /// rather than calls is what keeps a sentinel storm from
    /// un-coalescing a hot large file's republish: a counted budget
    /// charges a 2 GiB checkpoint the same one unit as a 4 KiB file.
    pub published_bytes: u64,
    /// The manifest ETag this barrier installed (the ack's CAS token).
    pub manifest_etag: Option<String>,
    /// What the manifest HEAD/install told us the bucket is at, for the
    /// news ticker (D5) — free, off requests the barrier already made.
    pub observed_seq: Option<u64>,
    pub observed_etag: Option<String>,
    /// Paths this barrier saw absent for the FIRST time: their deletes
    /// are withheld to the next scan (the rename-vs-walk guard). On a
    /// declared boundary a non-empty set means the confirmation lstat
    /// found them back on disk — the race the guard exists for.
    pub first_absence: Vec<String>,
    /// First-absence paths a DECLARED boundary confirmed gone by direct
    /// lstat and published as ordinary deletes (`confirm_absences`).
    pub absences_confirmed: usize,
    /// This barrier finished a rescope a crash had left half-applied
    /// before it did anything else. Reported so a run that looks like
    /// an ordinary barrier can be told from one that had to converge a
    /// workspace first.
    pub rescope_replayed: bool,
}

/// Upload waves per chunk. The chunk is `fanout * this`, so a wave
/// still saturates fan-out and the sync point between chunks is rare.
const UPLOAD_CHUNK_WAVES: usize = 16;

/// How stale the lease may get inside a barrier before it is renewed.
/// Deliberately BELOW the run loop's `renew_every` (min(floor, 30) s) so
/// a long barrier keeps the same heartbeat cadence the loop would have,
/// and comfortably below the 6-poll (~60 s) takeover window.
const RENEW_WITHIN_SECS: u64 = 20;

impl Syncer {
    fn lease_epoch(&self) -> LeanResult<u64> {
        self.lease
            .as_ref()
            .map(|l| l.epoch)
            .ok_or_else(|| LeanError::State("barrier without a held lease".into()))
    }

    /// The in-loop honor paths' fence check (boundary-verbs plan D2):
    /// `Syncer::sync` carries no lease/epoch check of its own, so a
    /// straggler consuming a sync sentinel between deposal and its next
    /// cooperative fence would apply the successor's manifest onto its
    /// zombie tree and ack SUCCESS.
    pub async fn verify_not_deposed_pub(&self) -> LeanResult<()> {
        self.verify_not_deposed().await
    }

    /// Verify the cell still names us at OUR epoch; anything else is a
    /// fence. Read-verify before the manifest CAS (the per-request
    /// validation the gateway will also enforce).
    async fn verify_not_deposed(&self) -> LeanResult<()> {
        let lease = self
            .lease
            .as_ref()
            .ok_or_else(|| LeanError::State("no lease".into()))?;
        match self.store.epoch_read(&self.cfg.epoch_key()).await? {
            Some(state) if state.epoch == lease.epoch && state.holder_id == lease.holder_id => {
                Ok(())
            }
            Some(state) => Err(LeanError::Fenced(format!(
                "cell at epoch {} holder {} (we are epoch {})",
                state.epoch, state.holder_id, lease.epoch
            ))),
            None => Err(LeanError::Fenced("epoch cell vanished".into())),
        }
    }

    /// Step 1: consume the inbox. A barrier NEVER runs against an
    /// unconsumed inbox — this is what makes HITL uploads structurally
    /// un-amputatable. Returns the entries integrated (dropped from the
    /// cell at the window-open CAS).
    pub async fn consume_inbox(&mut self) -> LeanResult<Vec<InboxEntry>> {
        let loaded = inbox::load(self.store.as_ref(), &self.cfg).await?;
        // §2.5's layered doors ride the inbox GET this function already
        // pays for: promptness is one tick, and the added request count
        // is ZERO. A failure here must never fail the consume — the
        // request is idempotent state and the next tick re-reads it.
        if let Err(e) = self.note_verb_requests(&loaded.doc) {
            eprintln!("flint-sync: verb request not consumed (retrying next tick): {e}");
        }
        let mut consumed = vec![];
        let mut baseline = self.state.load_baseline()?;
        for entry in &loaded.doc.entries {
            let key = self.cfg.file_key(&entry.path);
            // Containment BEFORE anything else: a path we could never
            // safely materialize must be surfaced and dropped, not
            // routed through the locally-dirty branch — which would
            // "preserve" it with a GET+PUT and leave a baseline entry
            // for a path the scanner can never see (the planted-symlink
            // shape: `inputs -> /root/.aws` reads as locally-present,
            // therefore dirty, therefore a conflict-preserve of someone
            // else's file).
            if let Err(e) = check_contained(&self.cfg.root, &entry.path) {
                self.state.append_conflict(&ConflictRecord {
                    path: entry.path.clone(),
                    foreign_etag: entry.etag.clone(),
                    preserved_key: None,
                    kind: format!("consume-refused-containment: {e}"),
                    at_unix: now_unix(),
                })?;
                consumed.push(entry.clone());
                continue;
            }
            // Already integrated (a crashed earlier consume): idempotent.
            if baseline.entries.get(&entry.path).map(|b| b.etag == entry.etag).unwrap_or(false) {
                consumed.push(entry.clone());
                continue;
            }
            let head = match self.store.head(&key).await {
                Ok(m) => m,
                Err(StoreError::NotFound(_)) => {
                    // The object vanished under a tracked entry: surface
                    // it — the bytes are NOT recoverable from here.
                    self.state.append_conflict(&ConflictRecord {
                        path: entry.path.clone(),
                        foreign_etag: entry.etag.clone(),
                        preserved_key: None,
                        kind: "consume-object-missing".into(),
                        at_unix: now_unix(),
                    })?;
                    consumed.push(entry.clone());
                    continue;
                }
                Err(e) => return Err(e.into()),
            };
            if head.etag != entry.etag {
                // Superseded by a newer write (its own inbox entry
                // follows, or it is the syncer's): drop.
                consumed.push(entry.clone());
                continue;
            }
            let local_path = self.cfg.root.join(&entry.path);
            let dirty = local_dirty(&local_path, baseline.entries.get(&entry.path));
            if dirty {
                // Locally-dirty wins; preserve the FOREIGN bytes first
                // (a conflict record must keep both versions
                // recoverable), then advance the recognized ETag so our
                // eventual publish supersedes it KNOWINGLY.
                let preserved = self.preserve_conflict_copy(&entry.path, &head.etag).await?;
                self.state.append_conflict(&ConflictRecord {
                    path: entry.path.clone(),
                    foreign_etag: entry.etag.clone(),
                    preserved_key: Some(preserved),
                    kind: "consume-dirty".into(),
                    at_unix: now_unix(),
                })?;
                let stamps = GenerationStamps::from_meta(&head.meta);
                baseline.entries.insert(
                    entry.path.clone(),
                    BaselineEntry {
                        etag: head.etag.clone(),
                        generation: stamps.map(|s| s.generation).unwrap_or(0),
                        // Deliberately NOT the local file's stat: the
                        // path must still scan as dirty so the local
                        // version publishes.
                        size: u64::MAX,
                        mtime_unix: 0,
                        version_id: None,
                        // The sentinel's bytes are the LOCAL edit; the
                        // publish that supersedes hashes them itself.
                        crc64_b64: None,
                    },
                );
            } else {
                // Clean: adopt the foreign content into the tree.
                let (meta, body) =
                    self.store.get_whole(&key, Some(&entry.etag)).await.map_err(|e| match e {
                        StoreError::PreconditionFailed(m) => {
                            StoreError::Other(format!("consume raced a newer write: {m}"))
                        }
                        other => other,
                    })?;
                let mode = PosixStamps::from_meta(&meta.meta).map(|p| p.mode);
                // VERIFIED before it is written, as checkout's fresh
                // fetch is: against the writer's CRC when the inbox
                // entry carries one (the gateway hashes what it sent),
                // else the backend's attestation when it offers one.
                // Ozone offers none, so the inbox's is what a HITL
                // upload gets checked against there. What the baseline
                // records — and the next citation repair cites — is
                // OURS, over the bytes actually written.
                let got = crc64_to_b64(crc64_nvme(&body));
                let want = entry.crc64_b64.clone().or_else(|| meta.crc64_b64.clone());
                if let Some(want) = want {
                    if want != got {
                        // NOT consumed: the entry stays in the cell and
                        // the record repeats — a permanent, visible
                        // contradiction is the right failure for
                        // bytes nobody vouches for.
                        self.state.append_conflict(&ConflictRecord {
                            path: entry.path.clone(),
                            foreign_etag: entry.etag.clone(),
                            preserved_key: None,
                            kind: format!(
                                "consume-refused-checksum: etag {} is attested with CRC-64 \
                                 {want} but the bytes fetched under it hash to {got}",
                                entry.etag
                            ),
                            at_unix: now_unix(),
                        })?;
                        continue;
                    }
                }
                // CONTAINMENT and I/O are split, because the answers are
                // opposite. Containment was already decided above by
                // `check_contained`, so a refusal reaching here is a
                // path that only became unsafe once parents were
                // created — still a permanent property of the path, so
                // it is surfaced and DROPPED, exactly as above.
                //
                // Everything else — ENOSPC, EACCES, EIO, a read-only
                // filesystem — is TRANSIENT, and dropping the entry for
                // one is silent permanent loss: the foreign bytes stay
                // in the bucket, the workspace never adopts them, and
                // nothing ever re-offers the entry. It used to be
                // recorded as `consume-refused-containment`, which sent
                // whoever read it looking for a hostile path that was
                // never there. A full disk is not a planted symlink.
                let target = match contained_path(&self.cfg.root, &entry.path) {
                    Ok(t) => t,
                    Err(e) => {
                        self.state.append_conflict(&ConflictRecord {
                            path: entry.path.clone(),
                            foreign_etag: entry.etag.clone(),
                            preserved_key: None,
                            kind: format!("consume-refused-containment: {e}"),
                            at_unix: now_unix(),
                        })?;
                        consumed.push(entry.clone());
                        continue;
                    }
                };
                if let Err(e) = write_file_atomic(&target, &body, mode) {
                    // NOT consumed: the entry stays in the cell so the
                    // next barrier retries it. A visible conflict record
                    // that repeats is a far better failure than one
                    // silent drop.
                    self.state.append_conflict(&ConflictRecord {
                        path: entry.path.clone(),
                        foreign_etag: entry.etag.clone(),
                        preserved_key: None,
                        kind: format!("consume-write-failed (will retry): {e}"),
                        at_unix: now_unix(),
                    })?;
                    continue;
                }
                let st = std::fs::metadata(&local_path)?;
                let stamps = GenerationStamps::from_meta(&meta.meta);
                baseline.entries.insert(
                    entry.path.clone(),
                    BaselineEntry {
                        etag: meta.etag.clone(),
                        generation: stamps.map(|s| s.generation).unwrap_or(0),
                        size: st.len(),
                        mtime_unix: mtime_of(&st),
                        version_id: None,
                        crc64_b64: Some(got),
                    },
                );
                baseline.prev_scan.insert(entry.path.clone());
            }
            consumed.push(entry.clone());
        }
        // The files this baseline vouches for reached the disk before
        // the baseline does (audit 2026-09-03, finding 9).
        self.state.sync_tree()?;
        self.state.save_baseline(&baseline)?;
        Ok(consumed)
    }

    /// Act on the inbox document's two request fields (§2.5, D14).
    ///
    /// The asymmetry is deliberate and is about blast radius, not
    /// principle: a boundary request is PERFORMED (it publishes what is
    /// already on disk and mutates nothing local), a sync request is
    /// CARRIED into the ticker as advisory news and the agent decides.
    ///
    /// Both are watermarked by `requested_unix` rather than consumed by
    /// a clearing CAS. The field is idempotent state, so a repeat is a
    /// no-op and a lost clear cannot double-publish; and not writing
    /// means the doors cost no request at all, which is what §2.5
    /// promises.
    fn note_verb_requests(&mut self, doc: &inbox::InboxDoc) -> LeanResult<()> {
        let mut mark = self.load_verb_watermark()?;
        if let Some(r) = &doc.sync_request {
            if r.requested_unix > mark.sync_unix {
                self.carry_sync_request(r.requested_unix, &r.requestor)?;
                mark.sync_unix = r.requested_unix;
            }
        }
        if let Some(r) = &doc.boundary_request {
            if r.requested_unix > mark.boundary_unix {
                self.request_boundary(
                    &format!("gw:{}:{}", r.requested_unix, r.requestor),
                    Some(format!("boundary requested by {}", r.requestor)),
                )?;
                mark.boundary_unix = r.requested_unix;
            }
        }
        self.save_verb_watermark(&mark)
    }

    fn verb_watermark_path(&self) -> std::path::PathBuf {
        self.cfg.state_dir().join("verb-requests.json")
    }

    pub(crate) fn load_verb_watermark(&self) -> LeanResult<VerbWatermark> {
        let p = self.verb_watermark_path();
        if !p.exists() {
            return Ok(VerbWatermark::default());
        }
        Ok(std::fs::read(&p)
            .ok()
            .and_then(|b| serde_json::from_slice(&b).ok())
            .unwrap_or_default())
    }

    fn save_verb_watermark(&self, m: &VerbWatermark) -> LeanResult<()> {
        let bytes = serde_json::to_vec_pretty(m)
            .map_err(|e| LeanError::State(format!("verb watermark: {e}")))?;
        super::control::write_atomic(&self.verb_watermark_path(), &bytes)
    }

    async fn preserve_conflict_copy(&self, path: &str, etag: &str) -> LeanResult<String> {
        let key = self.cfg.file_key(path);
        let dst = self.cfg.conflict_key(&uuid::Uuid::new_v4().to_string(), path);
        // v1: client-side copy (GET + guarded PUT). The server-side
        // CopyObject lever is the designed v2 optimization for large
        // files.
        let (meta, body) = self.store.get_whole(&key, Some(etag)).await?;
        let crc = crc64_nvme(&body);
        let stamps = GenerationStamps {
            generation: 0,
            epoch: self.lease.as_ref().map(|l| l.epoch).unwrap_or(0),
            flush_uuid: "conflict-preserve".into(),
            boundary_source: None,
            posix: PosixStamps::from_meta(&meta.meta),
        };
        self.store
            .put_whole(&dst, body, &PutCondition::IfNoneMatchAny, &stamps, crc)
            .await?;
        Ok(dst)
    }

    /// Take the SECOND absence observation now, by direct `lstat`.
    ///
    /// `scan::classify` withholds a path's delete until absence has
    /// survived two consecutive scans — the rename-vs-walk race guard
    /// (`lib.rs:37`). The hazard that rule names is the WALK missing a
    /// file renamed under it, not deletion itself. On the cadence path
    /// the second walk arrives at the next floor tick and nobody has
    /// been promised otherwise. On a DECLARED boundary it cannot wait:
    /// the ack would claim a coherent point (D1 — "everything
    /// ordered-before T") while the manifest still cites a file the
    /// agent removed before it touched the sentinel, and it would say
    /// so with `report.deleted: 0` and `status: "ok"`.
    ///
    /// A direct `lstat` is exactly what the rename-vs-walk guard asks
    /// for and is immune to the walk race by construction, so the
    /// second observation costs one syscall per transiently-absent
    /// path — never a second full pass. The cadence path is unchanged
    /// and still waits for the second walk.
    pub(crate) fn confirm_absences(&self, classified: &mut scan::Classified) -> LeanResult<usize> {
        if classified.first_absence.is_empty() {
            return Ok(0);
        }
        let mut confirmed = 0;
        for path in std::mem::take(&mut classified.first_absence) {
            // ONLY NotFound is absence. Every other errno — EACCES on a
            // parent, EIO, EMFILE, ELOOP — used to read as "the agent
            // deleted it", and this is the oracle that PUBLISHES the
            // delete and then DELETES THE OBJECT. An unreadable file is
            // the one case where guessing is unrecoverable, so it fails
            // CLOSED, exactly as the walk does at `scan.rs`'s
            // `symlink_metadata(&path)?`. Two call sites, one syscall:
            // they must not have opposite error policies.
            let gone = match std::fs::symlink_metadata(self.cfg.root.join(&path)) {
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => true,
                Err(e) => {
                    classified.first_absence.insert(path.clone());
                    return Err(LeanError::State(format!(
                        "cannot confirm the absence of {path:?}: {e} — refusing to publish \
                         a deletion on an unreadable path"
                    )));
                }
                Ok(_) => false,
            };
            if gone {
                classified.deletes.insert(path);
                confirmed += 1;
            } else {
                // Back on disk: the rename-vs-walk race, caught exactly
                // as the rule intends. Still only a first absence.
                classified.first_absence.insert(path);
            }
        }
        Ok(confirmed)
    }

    /// Renew the lease if it has gone stale MID-BARRIER, and hand back
    /// the cell as it was read so the CALLER can fence on it.
    ///
    /// Costs nothing on a short barrier: the renewal is skipped unless
    /// it was already due, so the request count is unchanged — this
    /// moves WHEN the heartbeat can happen, not how often.
    ///
    /// What this does NOT do, stated because the previous comment
    /// claimed it did: it never raises `Fenced` for a deposed holder.
    /// `lease::renew` can only 412 when the cell is still at OUR epoch
    /// with a token we do not hold; a cell at a FOREIGN epoch is
    /// skipped here (renewing it would be nonsense), so a deposed
    /// straggler passed straight through. The 2026-09-03 audit found
    /// the cadence barrier relying on exactly that phantom fence
    /// between upload chunks and completing every remaining data PUT
    /// after a takeover. The cell is returned so the caller can compare
    /// it against its lease — the gated lanes already pay a separate
    /// `verify_not_deposed_pub` for the same purpose; the cadence
    /// barrier uses the read it already paid for. `None` when there is
    /// no lease or the read failed: neither is a fence, and the
    /// ordinary renewal arm will try again.
    pub(crate) async fn renew_if_due(&mut self) -> LeanResult<Option<EpochState>> {
        let Some(lease) = self.lease.clone() else { return Ok(None) };
        let state = match self.store.epoch_read(&self.cfg.epoch_key()).await {
            Ok(Some(s)) => s,
            _ => return Ok(None),
        };
        let now = super::now_unix();
        let fresh = state
            .last_renew_unix
            .map(|t| now.saturating_sub(t) < RENEW_WITHIN_SECS)
            .unwrap_or(false);
        if fresh || state.epoch != lease.epoch {
            // Not due, or not ours to renew. The caller decides what a
            // foreign cell means; this function only knows it cannot
            // renew one.
            return Ok(Some(state));
        }
        super::lease::renew(self).await?;
        Ok(Some(state))
    }

    /// The between-chunk fence of the cadence barrier: a cell that no
    /// longer names this holder at this epoch stops the upload set
    /// HERE, before the next chunk's PUTs. Every one of those PUTs
    /// carries If-Match on a baseline etag that still matches — the
    /// successor has published nothing yet — so without this a deposed
    /// straggler overwrites the cited generation of every key it still
    /// had to upload, and every reader's S3-wins arm then adopts the
    /// uncited bytes silently (audit 2026-09-03, finding 1).
    fn fence_on_cell(&self, cell: &EpochState) -> LeanResult<()> {
        let Some(lease) = self.lease.as_ref() else {
            return Err(LeanError::Fenced("lease dropped mid-barrier".into()));
        };
        if cell.epoch != lease.epoch || cell.holder_id != lease.holder_id {
            return Err(LeanError::Fenced(format!(
                "deposed between upload chunks: cell at epoch {} holder {} (we are epoch {})",
                cell.epoch, cell.holder_id, lease.epoch
            )));
        }
        Ok(())
    }

    /// Steps 2–7. `barrier` = one full publish cycle (the cadence arm).
    pub async fn run_barrier(&mut self) -> LeanResult<BarrierReport> {
        self.barrier_inner(false, "manual").await
    }

    /// The floor tick's barrier. Stamped `cadence` so the bucket can
    /// answer "which clock installed this boundary?" without the ack —
    /// the ack is a LOCAL file, and the fleet-visible question is asked
    /// by operators and by the gateway's `/status`, neither of which
    /// can see it.
    pub async fn cadence_barrier(&mut self) -> LeanResult<BarrierReport> {
        self.barrier_inner(false, "cadence").await
    }

    /// A barrier whose result somebody is going to ACK: a sentinel
    /// honor (D1) or the preStop drain (D10). Identical to
    /// `run_barrier` except that it confirms first-absence paths rather
    /// than acking a boundary that withholds them.
    pub async fn declared_barrier(&mut self) -> LeanResult<BarrierReport> {
        self.barrier_inner(true, "sentinel").await
    }

    /// `declared_barrier` under an explicit provenance stamp — the
    /// drain and the budget-deferred honor are the same barrier
    /// reporting a different clock.
    pub async fn declared_barrier_as(&mut self, source: &str) -> LeanResult<BarrierReport> {
        self.barrier_inner(true, source).await
    }

    async fn barrier_inner(&mut self, declared: bool, source: &str) -> LeanResult<BarrierReport> {
        let epoch = self.lease_epoch()?;
        let mut report = BarrierReport::default();

        // STEP 0: finish any rescope a crash left half-applied, before
        // the scan reads the tree (scoped-read design §4.3). This is the
        // gate that makes the narrow verb safe at all: a half-applied
        // narrow is a tree whose files and whose citations disagree, and
        // `classify` reads that disagreement as an upload or as a
        // DELETE depending on which half landed. Replay is idempotent,
        // so the cost on the normal path is one `stat` of a file that
        // is not there.
        if let Some(r) = self.replay_scope_intent().await? {
            report.rescope_replayed = true;
            eprintln!(
                "flint-sync: finished an interrupted rescope before the barrier \
                 (uncited {}, unlinked {}, materialised {})",
                r.uncited, r.unlinked, r.materialized
            );
        }

        // Cooperative deposal check BEFORE any write (a thawed
        // straggler fences here instead of landing data PUTs). This
        // narrows the window; the per-request epoch validation that
        // CLOSES it is the gateway's (P5) — the model's LeanNoEpochCheck
        // mutation is the proof rotation alone does not cover the data
        // path.
        self.verify_not_deposed().await?;

        // Step 1.
        let consumed = self.consume_inbox().await?;
        report.consumed = consumed.len();

        // Step 2: scan-diff against the persisted baseline.
        let mut baseline = self.state.load_baseline()?;
        let scanned = scan::scan(&self.cfg.root)?;
        let mut classified = scan::classify(&scanned, &baseline);
        if declared {
            report.absences_confirmed = self.confirm_absences(&mut classified)?;
        }
        report.first_absence = classified.first_absence.iter().cloned().collect();

        // Skip-on-no-diff: nothing local, nothing consumed, no pending
        // citation repair, and the bucket manifest where we left it.
        let repairs_pending = baseline
            .entries
            .iter()
            .any(|(p, be)| be.size != u64::MAX && baseline.inst_base.get(p) != Some(&be.etag));
        if classified.uploads.is_empty()
            && classified.deletes.is_empty()
            && consumed.is_empty()
            && classified.first_absence.is_empty()
            && !repairs_pending
        {
            // Read the POINTER, never the entries: the 0b rig measured
            // the idle tick at 1M entries as 27 s / 1.3 GiB — dominated
            // by fetching and parsing a 264 MiB document just to read
            // `seq`. The pointer is a few hundred bytes and carries the
            // seq in its body, so a GET of it costs one round trip and
            // answers strictly more than a HEAD of the old object did.
            //
            // Legacy workspaces (no pointer yet) keep the HEAD they had.
            let unchanged = match manifest::load_pointer(self.store.as_ref(), &self.cfg).await? {
                Some(manifest::LoadedPointer { pointer: p, etag, .. }) => {
                    // The news ticker rides this request for free — D5's
                    // "zero added bucket requests" is still literal.
                    report.observed_seq = Some(p.seq);
                    report.observed_etag = Some(etag.clone());
                    baseline.manifest_etag.as_deref() == Some(etag.as_str())
                }
                None => match self.store.head(&self.cfg.manifest_key()).await {
                    Ok(meta) => {
                        report.observed_seq =
                            GenerationStamps::from_meta(&meta.meta).map(|s| s.generation);
                        report.observed_etag = Some(meta.etag.clone());
                        baseline.manifest_etag.as_deref() == Some(meta.etag.as_str())
                    }
                    Err(StoreError::NotFound(_)) => baseline.manifest_etag.is_none(),
                    Err(e) => return Err(e.into()),
                },
            };
            if unchanged {
                report.no_change = true;
                report.seq = Some(baseline.seq);
                // prev_scan still advances (the two-scan rule's clock) —
                // but the baseline document (hundreds of MB at 1M
                // entries) is rewritten only when the scan SET moved.
                // Compare without building the set: both sides iterate
                // sorted, and the allocation this used to make was the
                // size of the very document it was trying not to write.
                if scanned.len() != baseline.prev_scan.len()
                    || !scanned.keys().eq(baseline.prev_scan.iter())
                {
                    baseline.prev_scan = scanned.keys().cloned().collect();
                    self.state.save_baseline(&baseline)?;
                }
                return Ok(report);
            }
        }

        // Step 3: intent journal, then the window (the commitment
        // point: the same CAS drops the consumed entries).
        let flush_uuid = uuid::Uuid::new_v4().to_string();
        let mut intent = self.state.load_intent()?;
        let prior_uuids = {
            let mut v = intent.recent_uuids.clone();
            if !intent.flush_uuid.is_empty() {
                v.push(intent.flush_uuid.clone());
            }
            v
        };
        // Carried forward, not dropped: it stays true until we install
        // again, and the merge below is what reads it.
        let prev_installed = intent.installed_etag.clone();
        intent = IntentJournal {
            flush_uuid: flush_uuid.clone(),
            keys: classified.uploads.iter().map(|p| self.cfg.file_key(p)).collect(),
            recent_uuids: prior_uuids.clone(),
            installed_etag: prev_installed.clone(),
        };
        self.state.save_intent(&intent)?;
        let deadline = now_unix() + self.cfg.window_slack_secs;
        inbox::open_window(self.store.as_ref(), &self.cfg, epoch, deadline).await?;
        if !consumed.is_empty() {
            // Drop consumed entries now that they are durably in the
            // baseline (idempotent if this crashes: re-consume skips).
            inbox::drop_entries(self.store.as_ref(), &self.cfg, epoch, &consumed).await?;
        }

        // Step 4: guarded uploads, fanned out under a bounded window
        // (each key's guard chain is independent; the 412 policy and
        // conflict records are applied to the collected results below,
        // in deterministic path order).
        let mut upserts: BTreeMap<String, LeanEntry> = BTreeMap::new();
        let mut parked: BTreeSet<String> = BTreeSet::new();
        let mut new_baseline_entries: BTreeMap<String, BaselineEntry> = BTreeMap::new();
        let outcomes: Vec<(String, LeanResult<UploadOutcome>)> = {
            use futures::stream::{self, StreamExt};
            // CHUNKED, and the reason is the lease rather than memory.
            //
            // The run loop is one `select!`, so while this call runs the
            // renewal arm CANNOT fire — branches are mutually exclusive.
            // An unchunked upload phase therefore starves the heartbeat
            // for its whole duration, which does two bad things: a
            // deposed straggler cannot learn it was deposed (the CAS
            // that would tell it never runs), and a HEALTHY syncer can
            // outrun the 60 s takeover window and have a standby take
            // the lease off a live writer. The 0b numbers put a 1M-file
            // checkout at 7 m 05 s, so this is reachable, not exotic.
            //
            // Between chunks nothing borrows `self`, so the renewal can
            // run on its ordinary ≤30 s cadence — it costs NO extra
            // requests, because that renewal was already due. The chunk
            // is a multiple of fan-out so a wave still saturates it.
            //
            // Residual, stated rather than hidden: starvation is now
            // bounded by ONE CHUNK's duration instead of the whole
            // barrier. A chunk of very large files can still exceed the
            // window; closing that needs the renewal on its own task,
            // which needs the lease behind a shared cell.
            let mut outcomes: Vec<(String, LeanResult<UploadOutcome>)> = Vec::new();
            // NOT cfg.fanout. That knob was raised to 128 for the READ
            // path, where a byte semaphore (fetch_inflight_max_bytes)
            // bounds what concurrency can hold in memory. THIS path has
            // no such gate — `buffer_unordered` is the only bound — so
            // riding the same number would have quadrupled both the
            // bodies held in memory and `chunk`, widening the
            // lease-fence window from 512 files to 2048 and making the
            // hazard named in the comment above four times likelier.
            // The read-side measurement (2.5x on 20k small files) says
            // nothing about either, so uploads keep the value they were
            // measured at until someone measures them.
            let fanout = self.cfg.upload_fanout.max(1);
            let chunk = fanout.saturating_mul(UPLOAD_CHUNK_WAVES).max(1);
            let all: Vec<&String> = classified.uploads.iter().collect();
            for group in all.chunks(chunk) {
                {
                    let this: &Syncer = &*self;
                    let mut part: Vec<(String, LeanResult<UploadOutcome>)> =
                        stream::iter(group.iter().map(|path| {
                            let path = *path;
                            let scanned_entry = &scanned[path];
                            let base = baseline.entries.get(path);
                            let flush_uuid = &flush_uuid;
                            let prior_uuids = &prior_uuids;
                            async move {
                                let r = this
                                    .upload_one(
                                        path, scanned_entry, base, epoch, flush_uuid,
                                        prior_uuids,
                                    )
                                    .await;
                                (path.clone(), r)
                            }
                        }))
                        .buffer_unordered(fanout)
                        .collect()
                        .await;
                    outcomes.append(&mut part);
                }
                // Now that the borrow is released: keep the lease alive,
                // and FENCE on the cell that read returned. A deposed
                // straggler stops here, before the next chunk's PUTs —
                // the renewal alone never fenced a foreign cell (see
                // `renew_if_due`), which is how a straggler used to run
                // its upload set to completion after a takeover.
                if let Some(cell) = self.renew_if_due().await? {
                    self.fence_on_cell(&cell)?;
                }
            }
            outcomes
        };
        let mut outcomes = outcomes;
        outcomes.sort_by(|a, b| a.0.cmp(&b.0));
        for (path, outcome) in outcomes {
            let path = &path;
            match outcome? {
                UploadOutcome::Published { entry, baseline_entry } => {
                    report.published_bytes += entry.size;
                    upserts.insert(path.clone(), entry);
                    new_baseline_entries.insert(path.clone(), baseline_entry);
                    report.uploaded.push(path.clone());
                }
                UploadOutcome::Parked { foreign_etag } => {
                    parked.insert(path.clone());
                    self.state.append_conflict(&ConflictRecord {
                        path: path.clone(),
                        foreign_etag,
                        preserved_key: None, // parking preserves in place
                        kind: "upload-412-parked".into(),
                        at_unix: now_unix(),
                    })?;
                    report.parked.push(path.clone());
                }
                UploadOutcome::Deferred => {
                    report.deferred.push(path.clone());
                }
            }
        }

        // Citation repairs: paths whose integrated object (the
        // baseline) differs from the manifest's citation — consumed
        // HITL adoptions and checkout's S3-wins arm. No bytes move; the
        // manifest re-cites what this syncer already integrated.
        // Without this, an adopted upload is clean-vs-baseline, never
        // enters the upload set, and the manifest silently drops it —
        // the battery's amputation leg caught exactly that.
        let repair_candidates: Vec<String> = baseline
            .entries
            .iter()
            .filter(|(path, be)| {
                !classified.uploads.contains(*path)
                    && !classified.deletes.contains(*path)
                    && be.size != u64::MAX // consume-dirty sentinel: publishes via upload
                    && baseline.inst_base.get(*path) != Some(&be.etag)
            })
            .map(|(path, _)| path.clone())
            .collect();
        for path in repair_candidates {
            let key = self.cfg.file_key(&path);
            let be = baseline.entries[&path].clone();
            match self.store.head(&key).await {
                Ok(meta) if meta.etag == be.etag => {
                    let crc = repair_crc(&path, &be, &meta)?;
                    let stamps = GenerationStamps::from_meta(&meta.meta);
                    let scan_entry = scanned.get(&path);
                    upserts.insert(
                        path.clone(),
                        LeanEntry {
                            key,
                            etag: meta.etag.clone(),
                            crc64_b64: crc,
                            size: meta.size,
                            mode: stamps
                                .as_ref()
                                .and_then(|s| s.posix)
                                .map(|p| p.mode)
                                .or(scan_entry.map(|s| s.mode))
                                .unwrap_or(0o644),
                            mtime_unix: scan_entry.map(|s| s.mtime_unix).unwrap_or(0),
                            generation: stamps.map(|s| s.generation).unwrap_or(be.generation),
                            epoch,
                            version_id: meta.version_id.clone(),
                        },
                    );
                }
                // Moved again or gone: the next consume reconciles it.
                _ => {}
            }
        }

        // Step 5: the manifest CAS (three-way merge; bounded retries).
        self.verify_not_deposed().await?;
        let mut foreign_entries: Vec<(String, LeanEntry)> = vec![];
        let mut attempt = 0;
        let (installed, installed_etag) = loop {
            attempt += 1;
            if attempt > 4 {
                return Err(LeanError::State(
                    "manifest CAS lost 4 merge races — refusing this barrier".into(),
                ));
            }
            let current = manifest::load(self.store.as_ref(), &self.cfg).await?;
            let (theirs, expected) = match &current {
                // The handle carries the LAYOUT as well as the tag: a
                // workspace still on the legacy single object CASes a
                // pointer that must not exist yet, not one it read.
                Some(l) => (l.manifest.clone(), Some(l.handle())),
                None => (Default::default(), None),
            };
            // If the bucket is still at the document THIS workspace
            // installed, that document IS the merge base — whatever the
            // persisted one says. Step 7 rewrites the merge base after
            // the CAS, so a restart in between leaves it behind a
            // document we wrote, and every entry in it would read as
            // somebody else's change. See `IntentJournal::installed_etag`.
            let own_base;
            let base: &BTreeMap<String, String> =
                if prev_installed.is_some()
                    && prev_installed.as_deref() == expected.as_ref().map(|h| h.etag.as_str())
                {
                    own_base = theirs
                        .entries
                        .iter()
                        .map(|(p, e)| (p.clone(), e.etag.clone()))
                        .collect();
                    &own_base
                } else {
                    &baseline.inst_base
                };
            let (mut merged, foreign) =
                manifest::merge(base, &theirs, &upserts, &classified.deletes, &parked);
            // `merge` clears it; the installing pass owns it. A mirror
            // is a property of how this workspace is DEPLOYED, so it
            // comes from config on every publish rather than being
            // inherited from whatever wrote last.
            merged.sole_writer = self.cfg.sole_writer;
            match manifest::cas_write_stamped(
                self.store.as_ref(),
                &self.cfg,
                &merged,
                expected.as_ref(),
                epoch,
                &flush_uuid,
                Some(source),
            )
            .await
            {
                Ok(meta) => {
                    // Before the deletes and before step 7: this is the
                    // only record that survives a crash in that window.
                    intent.installed_etag = Some(meta.etag.clone());
                    self.state.save_intent(&intent)?;
                    foreign_entries = foreign;
                    break (merged, meta.etag);
                }
                Err(LeanError::Store(StoreError::PreconditionFailed(_)))
                | Err(LeanError::Store(StoreError::Conflict(_))) => {
                    // Re-verify the cell before retrying: a rotation is
                    // exactly this 412, and re-merging past it would be
                    // the straggler install.
                    self.verify_not_deposed().await?;
                    continue;
                }
                Err(e) => return Err(e),
            }
        };
        report.seq = Some(installed.seq);
        // A fused install IS a coherent point, and cadence/hybrid have
        // exactly one source. The gauges must not report "no boundary
        // ever" on a workspace that publishes every minute.
        self.note_boundary(super::gated::CitationSource::Cadence.as_str(), installed.seq)?;
        report.manifest_etag = Some(installed_etag.clone());
        report.observed_seq = Some(installed.seq);
        report.observed_etag = Some(installed_etag.clone());
        report.foreign_queued = foreign_entries.len();

        // Step 6: deletes LAST — GC of keys the NEW manifest no longer
        // references, HEAD-guarded on the recognized ETag.
        for path in &classified.deletes {
            if installed.entries.contains_key(path) {
                continue; // delete/modify resolved foreign-wins: not garbage
            }
            let key = self.cfg.file_key(path);
            let recognized = baseline.entries.get(path).map(|b| b.etag.clone());
            match self.store.head(&key).await {
                Err(StoreError::NotFound(_)) => {
                    report.deleted.push(path.clone());
                }
                Ok(meta) if Some(&meta.etag) == recognized.as_ref() => {
                    self.store.delete(&key).await?;
                    report.deleted.push(path.clone());
                }
                Ok(meta) => {
                    // An ETag this syncer does not recognize is NEVER
                    // deleted (a HITL re-create landed after our CAS).
                    self.state.append_conflict(&ConflictRecord {
                        path: path.clone(),
                        foreign_etag: meta.etag,
                        preserved_key: None,
                        kind: "gc-skip".into(),
                        at_unix: now_unix(),
                    })?;
                }
                Err(e) => return Err(e.into()),
            }
        }

        // Step 7: baseline rewrite, intent clear, window clear (+ queue
        // the merge-preserved foreign entries for the next consume).
        for (path, be) in new_baseline_entries {
            baseline.entries.insert(path, be);
        }
        for path in &report.deleted {
            baseline.entries.remove(path);
        }
        baseline.inst_base =
            installed.entries.iter().map(|(p, e)| (p.clone(), e.etag.clone())).collect();
        baseline.seq = installed.seq;
        baseline.manifest_etag = Some(installed_etag);
        baseline.prev_scan = scanned.keys().cloned().collect();
        self.state.save_baseline(&baseline)?;
        self.state.clear_intent_keys()?;
        let queue: Vec<InboxEntry> = foreign_entries
            .into_iter()
            .map(|(path, e)| InboxEntry {
                path,
                etag: e.etag,
                author: "merge-preserved".into(),
                added_unix: now_unix(),
                crc64_b64: Some(e.crc64_b64),
            })
            .collect();
        inbox::clear_window(self.store.as_ref(), &self.cfg, epoch, &queue).await?;
        // Reap superseded generations. Immutable metadata that is never
        // collected is a leak that grows by a whole manifest per
        // publish, and this also collects the orphan a crash between the
        // entries PUT and the pointer CAS leaves behind. Best effort by
        // design: a publish that succeeded is not un-done by a failure
        // to tidy up after it.
        match manifest::sweep_generations(self.store.as_ref(), &self.cfg).await {
            Ok(0) => {}
            Ok(n) => eprintln!("flint-sync: reaped {n} superseded manifest generation(s)"),
            Err(e) => eprintln!("flint-sync: generation sweep: {e}"),
        }
        // And the CHUNK reaper, same reason, same best-effort terms.
        // `sweep_chunks` returns 0 without a request on a workspace that
        // is not chunked, so this costs nothing until the layout moves.
        //
        // It was written to four model-established rules, guarded by six
        // mutation configs, and called from nothing but its own tests —
        // so with chunking on, every publish left its superseded chunks
        // in the bucket forever. The rules ran in the model and in the
        // suite and never once in production. `the_barrier_reaps...`
        // below is the test that asks whether this line exists at all.
        match manifest::sweep_chunks(self.store.as_ref(), &self.cfg).await {
            Ok(0) => {}
            Ok(n) => eprintln!("flint-sync: reaped {n} unreferenced manifest chunk(s)"),
            Err(e) => eprintln!("flint-sync: chunk sweep: {e}"),
        }
        Ok(report)
    }

    /// The > whole_put_max path: contiguous `PartSource::Local` chunks
    /// through `compose_generation` (streaming multipart; the store
    /// aborts its partial assembly on every failure path). The CRC is
    /// computed by a streaming pass first; a writer racing the compose
    /// fails server-side validation and the path DEFERS to the next
    /// barrier — publish-possibly-torn is put_whole's documented
    /// dilemma, not this path's.
    #[allow(clippy::too_many_arguments)]
    async fn upload_compose(
        &self,
        path: &str,
        key: &str,
        local_path: &Path,
        size: u64,
        scanned: &scan::ScanEntry,
        condition: PutCondition,
        stamps: GenerationStamps,
        generation: u64,
        epoch: u64,
    ) -> LeanResult<UploadOutcome> {
        // NO PRE-PASS. The store accumulates the full-object CRC from
        // the parts as it reads them for upload, so this file is read
        // ONCE. It used to be read TWICE — a streaming hash of the whole
        // file, and then again part by part — which on the 2026-09-10
        // door drill was roughly 11 s of the `mixed` workload's 24.3 s,
        // all of it at the disk's read ceiling and none of it on the
        // wire.
        //
        // That pre-pass was not laziness: asking the store to accumulate
        // used to pin its part uploads to a sequential loop, because a
        // CRC-64 taken out of order is a different number. `crc64_combine`
        // removed the choice — parts are hashed independently and folded
        // in part order — so the single read and the parallel upload are
        // no longer alternatives.

        // Part grid: within [min_part_size, ...], at most max_parts,
        // contiguous from 0.
        let min_part = self.store.min_part_size().max(1);
        let max_parts = self.store.max_parts().max(1) as u64;
        let mut chunk = self.cfg.whole_put_max.max(min_part);
        let need = size.div_ceil(chunk);
        if need > max_parts {
            chunk = size.div_ceil(max_parts).div_ceil(min_part) * min_part;
        }
        let mut parts = vec![];
        let mut off = 0u64;
        while off < size {
            let len = chunk.min(size - off);
            parts.push(flint_store::PartSource::Local { offset: off, len });
            off += len;
        }

        let spec = flint_store::ComposeSpec {
            progress: None,
            key,
            local_path,
            parts,
            base_key: None,
            base_etag: None,
            condition,
            stamps: stamps.clone(),
            crc64: None,
        };
        match self.store.compose_generation(&spec).await {
            Ok(meta) => {
                // The store computed it; refuse to cite bytes nothing
                // vouched for rather than invent a number for the
                // manifest. A backend that returns no checksum has not
                // validated the publish server-side either.
                let crc = meta
                    .crc64_b64
                    .as_deref()
                    .and_then(flint_store::crc64_from_b64)
                    .ok_or_else(|| {
                        LeanError::State(format!(
                            "compose of {key} returned no full-object checksum — refusing to \
                             cite bytes nothing vouched for"
                        ))
                    })?;
                Ok(UploadOutcome::published(
                    path, key.to_string(), meta.etag, crc, scanned, generation, epoch,
                    meta.version_id,
                ))
            }
            Err(StoreError::ChecksumMismatch(_)) | Err(StoreError::NoSuchUpload(_)) => {
                // Drift mid-compose, or the operator sweep aborted us:
                // nothing published; the next barrier re-queues.
                Ok(UploadOutcome::Deferred)
            }
            Err(StoreError::PreconditionFailed(_)) => {
                // The AdoptOwn recognizer compares OUR bytes' checksum
                // against what landed, and without the pre-pass we do
                // not have one — so the file is read HERE, on the rare
                // recovery path, instead of on every publish.
                let crc = file_crc(local_path)?;
                let head = self.store.head(key).await?;
                let head_stamps = GenerationStamps::from_meta(&head.meta);
                let own = head_stamps
                    .as_ref()
                    .map(|s| s.flush_uuid == stamps.flush_uuid)
                    .unwrap_or(false);
                if own || head.crc64_b64.as_deref() == Some(crc64_to_b64(crc).as_str()) {
                    // Our own torn earlier Complete with these bytes:
                    // cite it.
                    let g = head_stamps.map(|s| s.generation).unwrap_or(generation);
                    return Ok(UploadOutcome::published(
                        path, key.to_string(), head.etag, crc, scanned, g, epoch,
                        head.version_id,
                    ));
                }
                Ok(UploadOutcome::Parked { foreign_etag: head.etag })
            }
            Err(e) => Err(e.into()),
        }
    }

    /// The gated staging lane's entry point into the shipped guard
    /// chain — same 412 policy, same AdoptOwn recognizer, same park;
    /// only the caller's use of the response differs. `None` = parked
    /// on a foreign etag.
    #[allow(clippy::too_many_arguments)]
    pub(super) async fn upload_one_pub(
        &self,
        path: &str,
        scanned: &scan::ScanEntry,
        base: Option<&BaselineEntry>,
        epoch: u64,
        flush_uuid: &str,
        prior_uuids: &[String],
    ) -> LeanResult<Option<(LeanEntry, BaselineEntry)>> {
        match self.upload_one(path, scanned, base, epoch, flush_uuid, prior_uuids).await? {
            UploadOutcome::Published { entry, baseline_entry } => {
                Ok(Some((entry, baseline_entry)))
            }
            UploadOutcome::Parked { .. } | UploadOutcome::Deferred => Ok(None),
        }
    }

    async fn upload_one(
        &self,
        path: &str,
        scanned: &scan::ScanEntry,
        base: Option<&BaselineEntry>,
        epoch: u64,
        flush_uuid: &str,
        prior_uuids: &[String],
    ) -> LeanResult<UploadOutcome> {
        let key = self.cfg.file_key(path);
        let local_path = self.cfg.root.join(path);
        let generation = base.map(|b| b.generation + 1).unwrap_or(1);
        let posix = std::fs::metadata(&local_path).ok().map(|m| PosixStamps::from_metadata(&m));
        let stamps = GenerationStamps {
            generation,
            epoch,
            flush_uuid: flush_uuid.to_string(),
            boundary_source: None,
            posix,
        };
        let condition = match base {
            Some(b) => PutCondition::IfMatch(b.etag.clone()),
            None => PutCondition::IfNoneMatchAny,
        };
        let size = std::fs::metadata(&local_path)
            .map_err(|e| LeanError::State(format!("stat {}: {e}", local_path.display())))?
            .len();
        if size > self.cfg.whole_put_max {
            // Streaming multipart compose: put_whole is never fed past
            // whole_put_max (unbounded memory + S3's 5 GiB wall).
            return self
                .upload_compose(path, &key, &local_path, size, scanned, condition, stamps, generation, epoch)
                .await;
        }
        // `Bytes::from(Vec<u8>)` TAKES OWNERSHIP without copying, so the
        // old `Bytes::from(body.clone())` was a full memcpy of every
        // published file body — bought solely to leave `body` intact for
        // the 412 retry below, which is reached almost never. Build the
        // `Bytes` once; the retry gets a refcount clone.
        let body = Bytes::from(std::fs::read(&local_path)?);
        let crc = crc64_nvme(&body);
        match self
            .store
            .put_whole(&key, body.clone(), &condition, &stamps, crc)
            .await
        {
            Ok(meta) => Ok(UploadOutcome::published(
                path, key, meta.etag, crc, scanned, generation, epoch, meta.version_id,
            )),
            Err(StoreError::PreconditionFailed(_)) => {
                // The 412 policy: my own crashed/torn PUT ⇒ adopt;
                // foreign ⇒ park. NEVER the inherited LOCAL-WINS
                // overwrite.
                let head = self.store.head(&key).await?;
                let head_stamps = GenerationStamps::from_meta(&head.meta);
                let own = head_stamps
                    .as_ref()
                    .map(|s| s.flush_uuid == flush_uuid || prior_uuids.contains(&s.flush_uuid))
                    .unwrap_or(false);
                if !own {
                    return Ok(UploadOutcome::Parked { foreign_etag: head.etag });
                }
                if head.crc64_b64.as_deref() == Some(crc64_to_b64(crc).as_str()) {
                    // Bytes already there (torn response): cite it.
                    let g = head_stamps.map(|s| s.generation).unwrap_or(generation);
                    return Ok(UploadOutcome::published(
                        path, key, head.etag, crc, scanned, g, epoch, head.version_id,
                    ));
                }
                // Our earlier PUT, older content: supersede it knowingly.
                let meta = self
                    .store
                    .put_whole(&key, body, &PutCondition::IfMatch(head.etag), &stamps, crc)
                    .await?;
                Ok(UploadOutcome::published(
                    path, key, meta.etag, crc, scanned, generation, epoch, meta.version_id,
                ))
            }
            Err(e) => Err(e.into()),
        }
    }
}

enum UploadOutcome {
    Published { entry: LeanEntry, baseline_entry: BaselineEntry },
    Parked { foreign_etag: String },
    /// The source drifted mid-transfer (checksum refused server-side)
    /// or the assembly was swept: publish nothing, advance nothing —
    /// the next scan re-queues the path.
    Deferred,
}

impl UploadOutcome {
    #[allow(clippy::too_many_arguments)]
    fn published(
        path: &str,
        key: String,
        etag: String,
        crc: u64,
        scanned: &scan::ScanEntry,
        generation: u64,
        epoch: u64,
        version_id: Option<String>,
    ) -> UploadOutcome {
        let _ = path;
        UploadOutcome::Published {
            entry: LeanEntry {
                key,
                etag: etag.clone(),
                crc64_b64: crc64_to_b64(crc),
                size: scanned.size,
                mode: scanned.mode,
                mtime_unix: scanned.mtime_unix,
                generation,
                epoch,
                version_id: version_id.clone(),
            },
            baseline_entry: BaselineEntry {
                etag,
                generation,
                version_id,
                // The PRE-read stat: if the agent wrote during our read,
                // the next scan sees the drift and re-queues (the
                // re-stat/re-queue valve).
                size: scanned.size,
                mtime_unix: scanned.mtime_unix,
                crc64_b64: Some(crc64_to_b64(crc)),
            },
        }
    }
}

fn local_dirty(local: &Path, base: Option<&BaselineEntry>) -> bool {
    match (std::fs::metadata(local), base) {
        (Err(_), None) => false,                   // both absent: clean
        (Err(_), Some(_)) => true,                 // locally deleted vs baseline
        (Ok(_), None) => true,                     // local exists, never published
        (Ok(m), Some(b)) => m.len() != b.size || mtime_of(&m) != b.mtime_unix,
    }
}

/// Full-object CRC-64/NVME of a local file, streamed.
///
/// Reached only by the 412 recovery path: the ordinary publish gets its
/// checksum from the store, which computes one while uploading.
/// The CRC a citation repair cites: the bytes this syncer integrated,
/// as the baseline recorded them when it wrote or uploaded them. The
/// HEAD's own checksum is a cross-check when the backend offers one,
/// never the source — Ozone offers none, a HITL uploader may have sent
/// none, and the manifest must carry a CRC either way, because every
/// reader now verifies a fresh fetch against it.
///
/// A baseline entry with no CRC is refused loudly rather than cited
/// bare: the only entry built without one is the consume-dirty sentinel,
/// which the candidate filters exclude, so reaching this is a defect.
/// A HEAD that attests a DIFFERENT value is refused the same way — the
/// workspace holds bytes the object does not, and citing either would
/// make one reader or another refuse the path forever.
pub(crate) fn repair_crc(
    path: &str,
    be: &BaselineEntry,
    head: &flint_store::ObjectMeta,
) -> LeanResult<String> {
    let Some(ours) = be.crc64_b64.clone() else {
        return Err(LeanError::State(format!(
            "citation repair of {path}: the baseline records no CRC for the bytes it \
             integrated at etag {} — refusing to cite bytes nothing hashed",
            be.etag
        )));
    };
    if let Some(theirs) = head.crc64_b64.as_deref() {
        if theirs != ours {
            return Err(LeanError::State(format!(
                "citation repair of {path}: the store attests CRC-64 {theirs} for etag {} but \
                 the bytes this workspace integrated under that etag hash to {ours} — the \
                 workspace holds bytes the object does not; refusing to cite either",
                be.etag
            )));
        }
    }
    Ok(ours)
}

fn file_crc(local_path: &Path) -> LeanResult<u64> {
    use std::io::Read;
    let mut crc = flint_store::Crc64Nvme::new();
    let mut f = std::fs::File::open(local_path)?;
    let mut buf = vec![0u8; 4 << 20];
    loop {
        let n = f.read(&mut buf)?;
        if n == 0 {
            break;
        }
        crc.update(&buf[..n]);
    }
    Ok(crc.finalize())
}

pub(super) fn mtime_of(m: &std::fs::Metadata) -> i64 {
    m.modified()
        .ok()
        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0)
}

/// Containment-safe workspace write (boundary-verbs plan §2.2 security
/// gate). `rel` is a workspace-relative path; the target may not escape
/// the root, traverse a symlink component, or land in the reserved
/// control namespace.
///
/// The hazard this closes is pre-existing and reachable from three
/// callers (checkout, inbox consume, sync): `write_file_atomic` did
/// `create_dir_all(parent)` + write with no `O_NOFOLLOW` and no
/// root-containment check, while the scanner SKIPS symlinks — so a
/// planted symlink is invisible to the syncer. An unprivileged app
/// that plants `inputs -> /root/.aws`, lands an object at
/// `inputs/<path>` and drops a scoped sync turns the credential-holding
/// syncer into an on-demand arbitrary-file-write primitive outside the
/// workspace.
pub(super) fn write_file_atomic_in(
    root: &Path,
    rel: &str,
    bytes: &[u8],
    mode: Option<u32>,
) -> LeanResult<()> {
    let target = contained_path(root, rel)?;
    write_file_atomic(&target, bytes, mode).map(|_| ())
}

/// Validate `rel` under `root` WITHOUT creating anything — for callers
/// that must decide whether a path is writable before doing any work
/// (inbox consume refuses a refused path outright rather than routing it
/// through conflict-preserve, which would cost a GET+PUT and leave the
/// path in the baseline).
pub(super) fn check_contained(root: &Path, rel: &str) -> LeanResult<()> {
    resolve_contained(root, rel, false).map(|_| ())
}

/// Resolve `rel` under `root`, refusing escapes, creating missing parent
/// directories. Walks the parent chain component by component: a
/// component that EXISTS as a symlink is a refusal (never followed).
pub(super) fn contained_path(root: &Path, rel: &str) -> LeanResult<std::path::PathBuf> {
    resolve_contained(root, rel, true).map(|(p, _)| p)
}

/// As `contained_path`, and also what the walk already learned about
/// the FINAL component: its `lstat` if it exists, `None` if it does
/// not. The walk stats every component to refuse a symlink; a caller
/// that then asked `exists()` and `metadata()` paid two more stats for
/// an answer it was already holding (checkout's resume check did, on
/// every one of 20,000 files).
pub(super) fn contained_path_stat(
    root: &Path,
    rel: &str,
) -> LeanResult<(std::path::PathBuf, Option<std::fs::Metadata>)> {
    resolve_contained(root, rel, true)
}

fn resolve_contained(
    root: &Path,
    rel: &str,
    create_dirs: bool,
) -> LeanResult<(std::path::PathBuf, Option<std::fs::Metadata>)> {
    use std::path::Component;
    let refuse = |why: &str| {
        Err(LeanError::State(format!(
            "refusing write to {rel:?}: {why} (containment)"
        )))
    };
    if rel.is_empty() {
        return refuse("empty path");
    }
    if super::scan::is_control_path(rel) {
        // The reserved namespace is the syncer's own; a citation or
        // inbox entry naming it is surfaced, never materialized (D0.3).
        return refuse("reserved control namespace");
    }
    let relp = Path::new(rel);
    let mut cur = root.to_path_buf();
    let mut comps: Vec<&std::ffi::OsStr> = vec![];
    for c in relp.components() {
        match c {
            Component::Normal(n) => comps.push(n),
            Component::CurDir => {}
            Component::ParentDir => return refuse("`..` component"),
            Component::RootDir | Component::Prefix(_) => return refuse("absolute path"),
        }
    }
    if comps.is_empty() {
        return refuse("no path components");
    }
    let last = comps.len() - 1;
    let mut final_meta = None;
    for (i, c) in comps.iter().enumerate() {
        cur.push(c);
        match std::fs::symlink_metadata(&cur) {
            Ok(m) if m.file_type().is_symlink() => {
                return refuse("path traverses a symlink");
            }
            Ok(m) if i < last && !m.is_dir() => {
                return refuse("parent component is not a directory");
            }
            Ok(m) => {
                if i == last {
                    final_meta = Some(m);
                }
            }
            Err(_) if i < last => {
                if create_dirs {
                    if let Err(e) = std::fs::create_dir(&cur) {
                        // A SIBLING may have created it between the stat
                        // above and this mkdir: fetches run on their own
                        // tasks, so the first files of one directory
                        // race here, and the loser's EEXIST is not a
                        // containment refusal. It is fine only if what
                        // exists now is a plain directory — a symlink
                        // that appeared in the window is refused exactly
                        // as one the stat found.
                        match std::fs::symlink_metadata(&cur) {
                            Ok(m) if m.file_type().is_symlink() => {
                                return refuse("path traverses a symlink");
                            }
                            Ok(m) if m.is_dir() => {}
                            _ => {
                                return Err(LeanError::State(format!(
                                    "mkdir {}: {e}",
                                    cur.display()
                                )))
                            }
                        }
                    }
                }
            }
            Err(_) => {}
        }
    }
    Ok((cur, final_meta))
}

/// Write `bytes` to `path` through an exclusive temp sibling and a
/// rename. The parent already exists — containment created it — and
/// one that vanished since is recreated on the retry path in `safefs`,
/// so the common path pays no `mkdir` and no `stat` for it. Returns
/// the written file's metadata.
pub(super) fn write_file_atomic(
    path: &Path,
    bytes: &[u8],
    mode: Option<u32>,
) -> LeanResult<std::fs::Metadata> {
    super::safefs::check_parent(path)?;
    // NOT with_extension(): that REPLACES the final extension, so
    // "a.txt" and "a.md" would collide on one tmp name.
    let tmp = path.with_file_name(format!(
        "{}.flint-sync-tmp",
        path.file_name().map(|n| n.to_string_lossy()).unwrap_or_default()
    ));
    // The temp sibling is computed AFTER containment ran, so the walk
    // never saw it: it needs its own refusal, not a plain write.
    // FAST (no per-file fsync): materialisations are made durable by
    // `sync_tree` before the marker or baseline that vouches for them.
    super::safefs::write_via_tmp_fast(path, &tmp, bytes, mode)
}