flint-lean 0.12.0

//! flint-lean: the checkout/publish syncer core (plan of record:
//! docs/plans/flint-lean-plan.md v2).
//!
//! Lean is a SEPARATE front end from the hub: no NFS, no daemon in the
//! data path. An agent pod gets a full local checkout on an emptyDir
//! (real ext4/APFS POSIX), and a syncer publishes snapshots to the
//! bucket through a barrier. Durability is RPO-shaped (the flush floor);
//! coherence is snapshot-level (the subtree manifest CAS is the atomic
//! commit point). This module reuses `tier::store` (ObjectStore,
//! GenerationStamps, conditional puts, the epoch cell) as a library and
//! deliberately does NOT reuse the hub's flush/arbitration machinery —
//! the plan pins that: the hub's 412 arbitration is LOCAL-WINS-overwrite
//! (`tier/flush.rs`), which is exactly wrong under HITL second writers.
//!
//! The protocol here is the shipped shape `lean/PROTOCOL.md` writes out,
//! from its model `lean/formal/LeanP1.tla` (the shape since the
//! simplification of 2026-09-25: P1, P2, M1). PROTOCOL.md has every rule
//! with the invariant it holds up and the model world that fails without
//! it; the load-bearing ones, in brief:
//!
//! - **Every write lands at a fresh immutable handle**
//!   (`files/<path>@<flush>`), and **the pointer CAS is the only commit**
//!   — a writer's and the gateway's alike. Nothing overwrites a handle.
//! - **The gateway commits each UI verb itself (P2)**: a save PUTs a fresh
//!   handle, then ONE pointer CAS that lands only over the version the UI
//!   read (else 412); a rename or a delete is one CAS. It never takes the
//!   writers' lease. The request cell (`inbox`, `.flint/lean/inbox`)
//!   carries only the two verb requests, "please publish" and "please
//!   pull" — no entries, no removals, no window.
//! - **The barrier**: consume (the cheap path: the scan trigger) → scan →
//!   upload → claim the lease → verify the uploads (R4a) → install, a
//!   three-way merge onto the CURRENT document with the baseline as the
//!   merge base (mine wins modify/modify; R7 records theirs; a delete over
//!   theirs applies, theirs kept, M3) → collect → finish (the baseline
//!   follows what landed).
//! - **What a commit stops citing is retired, not deleted**: logged at
//!   the CAS and reaped only once G old (M1), so a reader that loaded
//!   less than G ago can still fetch it. The orphan sweep runs only under
//!   the lease (R4b) and spares a save in flight (G3).
//! - **Takeover rotation**: a successor rewrites the document before
//!   serving, so a deposed holder's pointer CAS 412s; the worker-facing
//!   verbs are epoch-validated per request.
//! - **Restart matrix**: marker present ⇒ never re-materialize (a
//!   re-checkout would resurrect unpublished deletes); reload the
//!   persisted baseline, rescan, self-recognize the lease via the
//!   persisted incarnation id (emptyDir-scoped — a replacement pod gets
//!   a fresh identity and takes the takeover path).
//! - **Deletion basis**: a path is delete-eligible only if absent in
//!   two consecutive scans AND present in our own baseline (the
//!   rename-vs-walk guard).

pub mod barrier;
pub mod checkout;
pub mod chunk;
pub mod chunkcache;
pub mod conformance;
pub mod control;
pub mod docview;
pub mod gauges;
pub use gauges::{status_report, Gauges, StatusReport};
pub mod inbox;
pub mod lease;
pub mod manifest;
pub mod metrics;
pub mod news;
pub mod reader;
pub mod scan;
pub(crate) mod safefs;
pub mod sentinel;
pub mod state;
pub mod sync;
pub mod trace;
pub mod tracked;
pub mod untracked;
pub mod uds;
pub mod verbs;

#[cfg(test)]
mod tests;
#[cfg(test)] mod tests_upload_gate;
#[cfg(test)] mod tests_conformance;
#[cfg(test)] mod tests_bench;
#[cfg(test)] mod tests_steps;
#[cfg(test)] mod tests_cache;
#[cfg(test)] mod tests_baseline;
#[cfg(test)] mod tests_walk;
#[cfg(test)] mod tests_news;
// The binary's allocator (`fastalloc`, bin/flint_sync.rs) never reached
// the test binary: the battery and the latency benches ran the system
// malloc (glibc on the box) while flint-sync ships mimalloc. Under
// `fastalloc` they run the shipped allocator too.
#[cfg(all(test, feature = "fastalloc"))]
#[global_allocator]
static TEST_GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
#[cfg(test)] mod tests_rechunk;
#[cfg(test)] mod tests_docview;
#[cfg(test)] mod tests_rehash;
#[cfg(all(test, target_os = "linux"))] mod tests_fuzz;

use std::path::PathBuf;
use std::sync::Arc;

use flint_store::ObjectStore;

/// Reserved control namespace inside the subtree prefix. Never scanned,
/// never part of a checkout, never GC'd by the barrier.
pub const LEAN_DIR: &str = ".flint/lean";

/// The syncer's durable bookkeeping directory inside the workspace
/// (emptyDir-scoped: survives container restarts, dies with the pod).
pub const STATE_DIR: &str = ".flint-sync";

/// The chunk cache's directory inside the state dir (`chunkcache`).
pub const CHUNK_CACHE_DIR: &str = "chunk-cache";

/// Chunk GETs a manifest load keeps in flight (`LeanConfig::
/// chunk_fetch_parallelism`). A chunk is about 1 MiB, so this many is
/// 16 MiB in flight; the bodies are all held until the document is
/// assembled anyway, as they were when they were fetched one at a time.
pub const CHUNK_FETCH_PARALLELISM: usize = 16;

/// The workspace-local control namespace (boundary-verbs plan D0): the
/// agent's side of the file protocol. Reserved exactly as the gateway
/// reserves `.flint/` on the HTTP side (flint-lean-gateway's `path_ok`) — the scan
/// skips it, `classify` never makes it delete-eligible, and checkout
/// never materializes a `files/.flint/...` citation into it.
pub const CONTROL_DIR: &str = ".flint";

/// Protocol version advertised in `.flint/capabilities.json`.
pub const SENTINEL_PROTOCOL: u32 = 1;

/// This binary's version, echoed to the agent (`capabilities.json`) and
/// to the operator (the lease cell's echo, §2.6). Both mixed-version
/// holes — agent↔syncer and operator↔syncer — are detected by
/// comparing what is RUNNING against what was asked for, and neither
/// comparison exists without a version on the running side.
pub const SYNCER_VERSION: &str = env!("CARGO_PKG_VERSION");

/// Default whole-object publish ceiling; larger files go through the
/// multipart compose path (`tier/flush.rs` uses the same 64 MiB split).
pub const WHOLE_PUT_MAX: u64 = 64 * 1024 * 1024;

#[derive(Debug, thiserror::Error)]
pub enum LeanError {
    #[error("store: {0}")]
    Store(#[from] flint_store::StoreError),
    #[error("io: {0}")]
    Io(#[from] std::io::Error),
    #[error("state: {0}")]
    State(String),
    /// A gateway write's precondition, judged against the version the
    /// workspace tracked, was lost at the inbox CELL's CAS: a different
    /// write is tracked at the path now (design 2026-09-19, class 5 —
    /// L-48's two browsers, decided atomically where the entry lands).
    #[error("{path} is now tracked at {current}; re-read it and try again")]
    Lost { path: String, current: String },
    /// The cell moved under a commit section: this barrier is
    /// abandoned — its manifest is never installed, its uploads stand
    /// as uncited generations the next barrier adopts by `flush_uuid`.
    /// Not a process exit and not a terminal state: the lease is held
    /// per barrier, so the next one claims again.
    #[error("fenced: {0}")]
    Fenced(String),
    /// A budget refused the operation (checkout bytes/file-count).
    #[error("budget: {0}")]
    Budget(String),
    /// A precondition the OPERATOR must change refused the run (the
    /// prefix is claimed by another project). A restart cannot fix it,
    /// so the binary exits `EXIT_REFUSED` — the code the CSI delivery
    /// reads as final. Any other non-zero exit is retried in place by
    /// `restartPolicy: OnFailure`, and a refusal sharing that code was a
    /// crash loop the plugin reported to the tenant as "checkout in
    /// progress" for as long as it lived (audit 2026-09-03, finding 5,
    /// leg S22).
    #[error("refused: {0}")]
    Refused(String),
}

/// Process exit code for `LeanError::Refused` (sysexits `EX_CONFIG`).
/// Mirrored as `SYNCER_EXIT_REFUSED` in the CSI node plugin, which does
/// not depend on this crate; drill leg S22 is where the two are checked
/// against each other.
pub const EXIT_REFUSED: i32 = 78;

impl LeanError {
    /// The store refused our credentials (401/403, `ExpiredToken`,
    /// `RequestTimeTooSkewed`): a broker, token or clock fault.
    ///
    /// A predicate rather than a `LeanError` variant on purpose.
    /// `Store` is built by `#[from]`, so every `?` in the crate
    /// constructs it implicitly; a variant would have to be produced by
    /// hand at each of those sites and would be missed at exactly the
    /// one that mattered. Reading through the wrapper cannot be
    /// bypassed by a conversion nobody remembered to special-case.
    pub fn is_auth(&self) -> bool {
        matches!(self, LeanError::Store(flint_store::StoreError::Auth(_)))
    }
}

pub type LeanResult<T> = Result<T, LeanError>;

/// Sentinel posture knob (D0.4). `auto` = verbs on unless the
/// pre-existing-`.flint/` pre-flight trips; `force` accepts consumption
/// of pre-existing sentinel-named files.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SentinelMode {
    Auto,
    Off,
    Force,
}

impl SentinelMode {
    pub fn parse(s: &str) -> Option<SentinelMode> {
        match s {
            "auto" => Some(SentinelMode::Auto),
            "off" => Some(SentinelMode::Off),
            "force" => Some(SentinelMode::Force),
            _ => None,
        }
    }
    pub fn as_str(&self) -> &'static str {
        match self {
            SentinelMode::Auto => "auto",
            SentinelMode::Off => "off",
            SentinelMode::Force => "force",
        }
    }
}

/// What this syncer may do to the bucket (per-user access design §4.4).
///
/// `Read` is a syncer that follows the workspace and never writes to it:
/// no fence, no barrier, no inbox consume, no garbage collection. Its
/// floor tick pulls — the inbox and the pointer, and a `sync` when
/// either moved — and a `publish` is answered `refused-read-only`. It is
/// what a read-only credential is SUFFICIENT for; the credential, not
/// this flag, is what keeps a reader's bytes out of the bucket.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub enum Access {
    #[default]
    ReadWrite,
    Read,
}

impl Access {
    /// `FLINT_SYNC_ACCESS`: `read` | `readWrite`.
    pub fn parse(s: &str) -> Option<Access> {
        match s {
            "read" => Some(Access::Read),
            "readWrite" => Some(Access::ReadWrite),
            _ => None,
        }
    }
    pub fn as_str(&self) -> &'static str {
        match self {
            Access::ReadWrite => "readWrite",
            Access::Read => "read",
        }
    }
    pub fn is_read(&self) -> bool {
        *self == Access::Read
    }
}

#[derive(Clone)]
pub struct LeanConfig {
    /// Bucket key prefix for this subtree (project- or subtree-scoped;
    /// the proxy's tenancy boundary).
    pub prefix: String,
    /// Workspace root (the agent's tree).
    pub root: PathBuf,
    /// Publish cadence floor, seconds (the durability contract).
    pub floor_secs: u64,
    /// Stamp `sole_writer` on every manifest this syncer installs.
    ///
    /// Set for a workspace that is PUBLISHED rather than worked in —
    /// forge's legible export is the shipped case. It tells every later
    /// reader that an object off its citation was moved by a stranger,
    /// so the reader refuses instead of adopting.
    pub sole_writer: bool,
    /// Read or read-write (`Access`). Default read-write.
    pub access: Access,
    /// Whole-object ceiling; larger files use multipart compose.
    pub whole_put_max: u64,
    /// Entries per manifest chunk, in expectation, and the floor and
    /// ceiling around it (chunked design §3). Config rather than
    /// constants because §4 already anticipates this moving: the
    /// pointer body is O(entries / target), and past ~10M entries the
    /// target is one of the two things that can be changed instead of
    /// adding a second level. Not surfaced on the CRD — nothing has
    /// needed to tune it per workspace yet, and a knob with no caller
    /// is a support burden.
    /// Publish the manifest as a CHUNK LIST rather than one generation
    /// object per citation (chunked design). Default ON since
    /// 2026-09-04.
    ///
    /// The migration is FAIL-CLOSED and one-way per workspace: a
    /// pointer-era binary cannot parse a chunked pointer, which is the
    /// point — every layout change has to make an old reader REFUSE
    /// rather than conclude the project is empty and re-seed over it.
    ///
    /// Small projects are not penalised, which is what made turning it
    /// on unconditional rather than threshold-gated: under the target a
    /// project is ONE chunk, so it costs a pointer plus one object and
    /// two GETs to read — exactly what the single-generation layout
    /// already cost. The indirection was paid for by step one.
    ///
    /// `false` is still a supported READ path forever (a workspace that
    /// has not published since the migration) and is what the
    /// single-layout tests set explicitly.
    pub chunked: bool,
    /// How long an unreferenced chunk must have sat before the reaper
    /// may take it. Config rather than a constant because the model
    /// makes it load-bearing in a workload-dependent way: it must
    /// outlive the LONGEST publish, not the longest plausible one
    /// (`LeanChunkGCRacyGrace`).
    pub orphan_grace_secs: u64,
    /// RETIRE-AGE G (M1 of the 2026-09-24 simplification analysis): a
    /// handle or chunk a commit stopped referencing is collected only once
    /// its RETIREMENT is this old — not its write, which for a long-lived
    /// file or chunk is long past the moment a UI save supersedes it. A
    /// reader that loaded a document less than G ago can fetch everything
    /// it cites. 0 = collect at once (the shape before; the battery's
    /// default). `FLINT_SYNC_RETIRE_GRACE_SECS`.
    pub retire_grace_secs: u64,
    /// THE COLLECTION CADENCE (latency plan 2026-10-23, lever C): how often
    /// a writer's commit section runs the collection pass — read the retire
    /// logs, reap what is due, the orphan sweep when it is due too, the
    /// generation and chunk sweeps. Each pass is a LIST of `retired/`, of
    /// `manifests/` and of `chunks/`, and pointer GETs, all under the
    /// lease; run on every publish they were seven sequential requests
    /// added to every commit. Between passes nothing is deleted, so what a
    /// commit retires waits up to G plus this long: storage, never safety
    /// (the model's reap and sweeps may run at any commit, or never).
    /// 0 = every commit, as before (the battery's setting).
    /// `FLINT_SYNC_COLLECT_EVERY_SECS`.
    pub collect_every_secs: u64,
    /// Manifest chunks kept locally by content address, verified on every
    /// use (`chunkcache`; latency plan lever A). `None` fetches every
    /// chunk of every load, as before. The syncer keeps them under its
    /// state dir, the gateway in memory.
    pub chunk_cache: Option<Arc<dyn chunkcache::ChunkCache>>,
    /// Chunk fetches a load keeps in flight at once: the misses of a
    /// cached load, every chunk of an uncached one. 1 = one at a time,
    /// the shape before. `FLINT_SYNC_CHUNK_FETCH_PARALLELISM`.
    pub chunk_fetch_parallelism: usize,
    /// A commit reads the document chunk by chunk (latency plan lever G,
    /// `docview`): only the chunks its paths fall in, and it re-chunks only
    /// what it changed. `false` loads and merges the whole document, the
    /// shape before. `FLINT_SYNC_CHUNK_VIEW=off`.
    pub chunk_view: bool,
    /// Tests only: every chunk-view commit also runs the whole-document
    /// merge and asserts the same document, chunk list, retire set and
    /// per-path decisions (latency plan §12, "Verification").
    pub g_shadow: bool,
    /// A checkout measures its entries' racy flags when it opens the
    /// agent's gate, not when it wrote each file (latency plan §17;
    /// LeanScanRacy's `MeasureAtGate`), so the first scan hashes only what
    /// it wrote in its last window. `false` leaves every checked-out entry
    /// racy, the shape before. `FLINT_SYNC_RACY_AT_GATE=off`.
    pub racy_at_gate: bool,
    /// The tree's reserved blocks are this syncer's (the s3csi quota
    /// image reserves 1/16 of itself for a group only the syncer's worker
    /// holds): tree CONTENT it writes — a checkout, a consume, a sync — is
    /// held to what its tenant could still write (`f_bavail`), so the
    /// reserve is left for the bookkeeping that must outlive a full tree
    /// (`safefs::leave_the_reserve`). `FLINT_SYNC_TREE_RESERVE=on`, set by
    /// the plugin on a quota'd tree; off, nothing is checked.
    pub tree_reserve: bool,
    /// Tests only: the bytes `leave_the_reserve` reads as the tenant's room
    /// instead of asking `statvfs` (a full filesystem needs root to build).
    pub tree_room_for_tests: Option<u64>,
    /// The fewest files still racy at the gate for which the checkout opens
    /// it the rest of their window later (at most two seconds), so the
    /// first scan hashes none of them: about where hashing them costs more
    /// than the wait. 0 never waits. `FLINT_SYNC_RACY_WAIT_MIN`.
    pub racy_wait_min: usize,
    /// This workspace's chunks' aliases, by address (`docview`): what lets
    /// a commit answer "who else cites this handle" without the document.
    /// Shared by every clone of the config, so a syncer's views share it.
    pub alias_cache: Arc<docview::AliasCache>,
    pub chunk_target: usize,
    pub chunk_min: usize,
    pub chunk_max: usize,
    /// Checkout refusal budgets (0 = unlimited).
    pub max_bytes: u64,
    pub max_files: u64,
    /// Bounded concurrency for uploads and checkout fetches. The 0b
    /// rig measured the sequential loops at 561-854 PUTs/s and
    /// 1,000-2,000 GETs/s; fan-out multiplies directly against those.
    pub fanout: usize,
    /// Concurrent UPLOADS. Deliberately separate from `fanout`, and
    /// measured separately: the read path's width is what the checkout
    /// rig moved, this one is what the publish rig moved. Both paths are
    /// byte-bounded, by different gates — the read side by
    /// `fetch_inflight_max_bytes` here, the upload side by the store's
    /// own gate (`with_upload_inflight_max_bytes`, 512 MiB from the
    /// binary), which charges every whole body and every multipart part
    /// from before its read until its PUT returns. So this number sets
    /// request concurrency, not peak RSS.
    pub upload_fanout: usize,
    /// The project this workspace claims to be, stamped from the CR
    /// (`FLINT_SYNC_PROJECT_ID`). When set, `lease::verify_claim`
    /// refuses to run over a prefix whose claim cell names another
    /// project — the operator's refuse-foreign, enforced on the DATA
    /// plane (audit 2026-09-03, finding 5). `None` = unstamped: no
    /// check, the pre-operator posture.
    pub project_id: Option<String>,
    /// Ceiling on bytes in flight across the checkout fan-out window.
    ///
    /// `fanout` bounds the number of concurrent fetches but NOT their
    /// size, and `get_whole` holds each whole object in RAM before it
    /// reaches disk — so peak RSS was `fanout x largest object`, an
    /// unbounded product of two numbers nobody sets together. A 32-wide
    /// window over 64 MiB objects is 2 GiB in a syncer that ships with
    /// no memory limit at all. Each entry takes permits proportional to
    /// its size, so small-file trees still run the full width.
    pub fetch_inflight_max_bytes: u64,
    /// Driver tasks for the checkout fan-out (see `materialize`). Each
    /// runs `fanout / fetch_drivers` fetches at a time over its own
    /// slice of the tree, so the per-request SDK work spreads over that
    /// many cores instead of one. Default: the machine's cores, at most
    /// 8 (`FLINT_SYNC_FETCH_DRIVERS` / `spec.fetchDrivers`, 0 = auto).
    /// Measured 2026-09-12: one driver is the ~3,300 files/s plateau on
    /// i4i.xlarge whatever `fanout` says; four are 6,188-6,521 at
    /// fanout 256, and fanout 256 beats 128 for the first time.
    pub fetch_drivers: usize,

    /// Materialise an object this size or larger as PARALLEL RANGES
    /// rather than one whole-object GET. `0` disables it.
    ///
    /// Defaults to 8 MiB on the measurement that the repo's rule asked
    /// for — runcr, i4i.large spot, us-west-1, one AZ, n=3, arms
    /// interleaved per rep, RANGES not means:
    ///
    ///   6 x 1 GiB          71.33-71.88 s  ->  21.24-23.96 s
    ///   4 GiB + 2k x 16 K  57.02-57.83 s  ->  15.30-18.30 s
    ///   20k x 8 KiB        16.92-19.44 s  ->  16.40-17.39 s
    ///
    /// The third row is the one that justifies a THRESHOLD rather than
    /// ranging everything: every object there is under it, the ranged
    /// path never fires, and the arms agree. Before this, the shipped
    /// path lost a cold 6 GiB checkout to a plain `aws s3 cp` of the
    /// same prefix on the same node (21.7 s); it now ties it.
    ///
    /// Ranging also LOWERS peak RSS for a large object — see below — so
    /// this default costs memory nothing and returns some.
    ///
    /// NOTE the interaction with `range_get_chunk_bytes`: a single part
    /// is "one whole GET with extra steps" and `fetch_ranged` declines
    /// it, so an object in [`range_get_min_bytes`, `range_get_chunk_bytes`)
    /// passes this threshold and still takes the whole-object path. With
    /// the defaults below the EFFECTIVE threshold is therefore 16 MiB,
    /// not 8.
    ///
    /// Two costs of `get_whole` are paid together and this is what
    /// removes both: one object is one TCP stream, so a multi-GiB
    /// checkpoint runs at single-connection throughput no matter how
    /// wide the fan-out; and the body is held whole in RAM before it
    /// reaches disk, which is the product `fetch_inflight_max_bytes`
    /// exists to bound. Ranges write at their offset as they land, so
    /// peak RSS per object becomes `range_get_chunk_bytes x
    /// range_get_parallelism` instead of the object's size.
    pub range_get_min_bytes: u64,
    /// Bytes per range request. Large enough that per-request overhead
    /// is noise against S3's first-byte latency; small enough that a
    /// retried range is cheap.
    pub range_get_chunk_bytes: u64,
    /// Ranges in flight FOR ONE OBJECT. The checkout's byte semaphore
    /// still bounds the whole window, so this multiplies concurrency
    /// only where the fan-out has run out of objects to work on — which
    /// is exactly the tail this exists to shorten.
    pub range_get_parallelism: usize,

    // --- boundary verbs (plan docs/plans/flint-lean-boundary-verbs-plan.md) ---
    /// Sentinel posture (D0.4 pre-flight override).
    pub sentinel_mode: SentinelMode,
    /// A consumed sentinel arriving sooner than this after the previous
    /// sentinel-honoring barrier waits; touches inside the interval
    /// coalesce into the pending record.
    pub sentinel_min_interval_secs: u64,
    /// Work-metered hourly cap (D3.1): a honor charges
    /// `max(1, ceil(published_bytes / whole_put_max))` units, or 0 for
    /// a no-diff honor. Exhaustion defers honors to the floor tick —
    /// the workspace degrades to exactly cadence behavior.
    pub sentinel_hourly_budget: u64,
    /// Sentinel poll cadence (env-only, not a fleet contract).
    pub sentinel_poll_secs: u64,

    // --- the news poll (latency plan 2026-10-23, lever B; `news.rs`) ---
    /// While the workspace is active, read the pointer (a writer the cell
    /// too) this often, and run the floor's barrier early when a peer or
    /// the gateway moved it. 0 = off: news waits for the floor, as
    /// before. `FLINT_SYNC_NEWS_POLL_SECS`.
    pub news_poll_secs: u64,
    /// How long after the last sign of work (a publish, a consume, an ack,
    /// a served request, news) the workspace counts as active. Past it
    /// the poll stops and only the floor reads the bucket.
    /// `FLINT_SYNC_NEWS_ACTIVE_SECS`.
    pub news_active_secs: u64,
    /// The fewest seconds between two news-triggered barriers. News inside
    /// it waits, so two busy writers cannot answer each other's publishes
    /// faster than this. `FLINT_SYNC_NEWS_MIN_INTERVAL_SECS`.
    pub news_min_interval_secs: u64,

    // --- the publish fence (design 2026-09-13 §4; `lease.rs`) ---
    /// How often a barrier waiting for the cell re-reads it.
    pub claim_poll_secs: u64,
    /// The poll cadence of the waiter at the head of the queue
    /// (`lease::CLAIM_POLL_HEAD_MS`).
    pub claim_poll_head_ms: u64,
    /// Minimum spacing between two observations that COUNT toward the
    /// dead-holder (`QUIET_POLLS`) and abandoned-handoff thresholds.
    /// The tests set it to 0 so a deposal takes six polls, not a minute.
    pub claim_quiet_spacing_secs: u64,
    /// Longest a barrier waits for the cell before it is abandoned and
    /// retried at the next floor.
    pub claim_deadline_secs: u64,
    /// DRILL-ONLY (`FLINT_SYNC_DRILL_HOLD_COMMIT_SECS`, never stamped by
    /// the operator): sleep this long inside the commit section, right
    /// after the claim. The one straggler the per-barrier lease still
    /// has is a holder that stalls INSIDE its commit section, and that
    /// section is milliseconds long — no drill can hit it by timing, so
    /// this is how `run-writers.sh` opens the window. 0 = off.
    pub drill_hold_commit_secs: u64,
    /// DRILL-ONLY (`FLINT_SYNC_DRILL_HOLD_GC_SECS`, never stamped by the
    /// operator): sleep this long before the collector's batched DELETE
    /// of what the install retired, once per barrier. Under immutable
    /// handles the collector is one unconditional `delete_many` with no
    /// HEAD before it, so the window a drill wants is the one between
    /// the CAS and that batch — a peer's sweep, or its commit, running
    /// while this barrier still owes its tidy-up. 0 = off.
    pub drill_hold_gc_secs: u64,
    /// The protocol event trace (`trace.rs`); `None` = off.
    pub event_trace: Option<trace::Sink>,
    /// How often a WRITER sweeps for an upload a lost writer left at a
    /// cited key (`untracked.rs`, finding 10), seconds;
    /// `FLINT_SYNC_UNTRACKED_SWEEP_SECS`, 0 = off. Each sweep is one LIST of
    /// the files prefix plus a manifest and an inbox read.
    pub untracked_sweep_secs: u64,
    /// How long such an object must have sat untracked before the sweep
    /// tracks it — the gateway's `UNTRACKED_GRACE_SECS`, past any live
    /// writer's claim and commit.
    pub untracked_grace_secs: u64,
    /// Threads the scan walks the tree with (`scan::scan_with`, lever E);
    /// `FLINT_SYNC_SCAN_THREADS`. The CPUs this process may use, at most
    /// 8; 1 walks with one worker.
    pub scan_threads: usize,
    /// How old the last cell WRITE (claim or renew) may be before a
    /// delete in the commit section renews first (review 2026-09-18,
    /// C2). A deposal needs the token still for six spaced polls (60 s),
    /// so a write younger than this is proof no deposal has happened;
    /// an older one is renewed — a conditional write that a deposal has
    /// moved, so it fences. Comfortably below the threshold. Tests set 0
    /// to collapse a stall.
    pub renew_within_secs: u64,
}

/// The gateway's flush id for a UI write's handle: `ui-<added_unix>-<uuid>`.
/// The time is the order a consume reads off two UI handles of one path
/// (design 2026-09-19, R7's "later"): a citation that moved to a LATER
/// gateway write supersedes an entry, one that moved to an earlier one
/// leaves it live. One gateway's clock orders its own writes exactly;
/// two replicas' clocks are the residual, and a tie or an inversion
/// between them costs a surfaced conflict, never a silent drop.
pub fn ui_flush(added_unix: u64) -> String {
    format!("ui-{added_unix}-{}", uuid::Uuid::new_v4())
}

/// The longest flush id a handle carries: the gateway's `ui-<secs>-<uuid>`
/// with a `u64` of seconds (a barrier's is a bare uuid; an ingress copy's,
/// `ingress-<etag>`, adopts an object whose bare key already fit).
pub const MAX_FLUSH_LEN: usize = "ui-".len() + 20 + "-".len() + 36;

/// The longest key the stores lean runs on take: 1,024 bytes of UTF-8 on
/// S3 (`KeyTooLongError`) and on GCS.
pub const MAX_KEY_BYTES: usize = 1024;

/// The `added_unix` a UI flush id carries; `None` for any other flush.
pub fn ui_flush_added(flush: &str) -> Option<u64> {
    let rest = flush.strip_prefix("ui-")?;
    let (t, _) = rest.split_once('-')?;
    t.parse().ok()
}

/// Does `flush` read as a flush id this crate or the gateway mints — a
/// uuid, the `ui-…` form, or the gateway's `gateway-…` forms — rather than
/// a tail of a path that happens to follow an `@`?
fn flush_looks_minted(flush: &str) -> bool {
    if flush.starts_with("ui-")
        || flush.starts_with("gateway-")
        || flush.starts_with("ingress-")
        || flush == "conflict-preserve"
    {
        return true;
    }
    uuid::Uuid::parse_str(flush).is_ok()
}

impl LeanConfig {
    pub fn new(prefix: &str, root: impl Into<PathBuf>) -> Self {
        let root: PathBuf = root.into();
        let chunk_cache: Arc<dyn chunkcache::ChunkCache> =
            Arc::new(chunkcache::DiskChunkCache::new(root.join(STATE_DIR).join(CHUNK_CACHE_DIR)));
        LeanConfig {
            prefix: prefix.trim_end_matches('/').to_string(),
            root,
            floor_secs: 60,
            // Off by default: a workspace an agent WORKS IN wants the
            // S3-wins arm, where an object past its citation is a human
            // whose bytes should win. Only a published mirror sets it.
            sole_writer: false,
            access: Access::ReadWrite,
            whole_put_max: WHOLE_PUT_MAX,
            chunked: true,
            orphan_grace_secs: manifest::ORPHAN_GRACE_SECS,
            retire_grace_secs: manifest::RETIRE_GRACE_SECS,
            collect_every_secs: manifest::RETIRE_GRACE_SECS / 2,
            chunk_cache: Some(chunk_cache),
            chunk_fetch_parallelism: CHUNK_FETCH_PARALLELISM,
            chunk_view: true,
            g_shadow: false,
            racy_at_gate: true,
            tree_reserve: false,
            tree_room_for_tests: None,
            racy_wait_min: 10_000,
            alias_cache: Arc::default(),
            chunk_target: chunk::CHUNK_TARGET,
            chunk_min: chunk::CHUNK_MIN,
            chunk_max: chunk::CHUNK_MAX,
            max_bytes: 0,
            max_files: 0,
            fanout: 128,
            upload_fanout: 32,
            project_id: None,
            // 128 MiB (was 512): fits the deployed worker's 1Gi limit on a
            // tree of 1 GiB objects at no cost in wall time (door drill
            // 2026-09-12 §5, finding 4).
            fetch_inflight_max_bytes: 128 * 1024 * 1024,
            fetch_drivers: std::thread::available_parallelism()
                .map(|n| n.get())
                .unwrap_or(1)
                .clamp(1, 8),
            range_get_min_bytes: 8 * 1024 * 1024,
            range_get_chunk_bytes: 16 * 1024 * 1024,
            range_get_parallelism: 4,
            sentinel_mode: SentinelMode::Auto,
            sentinel_min_interval_secs: 5,
            sentinel_hourly_budget: 60,
            sentinel_poll_secs: 1,
            news_poll_secs: 1,
            news_active_secs: 300,
            news_min_interval_secs: 5,
            claim_poll_secs: lease::CLAIM_POLL_SECS,
            claim_poll_head_ms: lease::CLAIM_POLL_HEAD_MS,
            claim_quiet_spacing_secs: lease::QUIET_SPACING_SECS,
            claim_deadline_secs: lease::CLAIM_DEADLINE_SECS,
            drill_hold_commit_secs: 0,
            drill_hold_gc_secs: 0,
            event_trace: None,
            untracked_sweep_secs: 3600,
            untracked_grace_secs: untracked::UNTRACKED_GRACE_SECS,
            scan_threads: scan::default_threads(),
            renew_within_secs: 20,
        }
    }

    /// The INGRESS key of a path: where an outside writer's `aws s3 cp`
    /// lands, and where every object of this workspace lived before the
    /// immutable-handles layout (design 2026-09-19, R5). Nothing this
    /// crate publishes writes it any more; the read door, the consume and
    /// the checkout fetch HANDLES.
    pub fn file_key(&self, path: &str) -> String {
        format!("{}/files/{}", self.prefix, path)
    }
    /// A HANDLE: the one key a write lands at, `files/<path>@<flush>`
    /// (design 2026-09-19, R1). Nobody else ever writes it, the manifest
    /// cites it by name, and it is deleted only once no document cites it
    /// (R3). It sorts beside the bare path, so a listing of a directory
    /// shows every live version of a file together. `flush` is the
    /// writer's flush id: a barrier's uuid, or the gateway's `ui-…` form
    /// ([`ui_flush`]), whose ORDER a consume can read off the name.
    pub fn handle_key(&self, path: &str, flush: &str) -> String {
        format!("{}/files/{}@{}", self.prefix, path, flush)
    }
    /// Whether every handle `path` can have is a key the store takes
    /// (L-145): its handle under the longest flush id anyone mints
    /// ([`MAX_FLUSH_LEN`]) within [`MAX_KEY_BYTES`]. One rule for every
    /// writer, so a path the agent can publish the UI can save too. A
    /// file whose path does not fit is HELD by the barrier (as a name that
    /// is not UTF-8 is), and the gateway refuses to save or move one
    /// there: the PUT would fail on S3 at every attempt.
    pub fn key_fits(&self, path: &str) -> bool {
        self.handle_key(path, "").len() + MAX_FLUSH_LEN <= MAX_KEY_BYTES
    }
    /// The files namespace this workspace's handles and ingress objects
    /// live under, with its trailing slash — what a sweep lists.
    pub fn files_prefix(&self) -> String {
        format!("{}/files/", self.prefix)
    }
    /// `(path, flush)` of a handle under this prefix; `None` for a key
    /// that is not a handle here — the bare path of an ingress object
    /// included, so a sweep can tell the two apart by the name alone. A
    /// path may itself contain `@`: the flush is the LAST `@`-suffix, and
    /// a flush is a uuid or the gateway's `ui-` form, neither of which a
    /// path component would be mistaken for in the same position.
    pub fn handle_parts<'a>(&self, key: &'a str) -> Option<(&'a str, &'a str)> {
        let rest = key.strip_prefix(&self.files_prefix())?;
        let (path, flush) = rest.rsplit_once('@')?;
        if path.is_empty() || !flush_looks_minted(flush) {
            return None;
        }
        Some((path, flush))
    }
    /// The LEGACY single-object manifest key. Still read (a workspace
    /// written before the pointer layout has one) and, once migrated,
    /// overwritten with a document that cannot parse as a manifest —
    /// see `manifest::REFUSAL_DOC` for why deleting it would be a
    /// data-loss shape rather than a tidy-up.
    pub fn manifest_key(&self) -> String {
        format!("{}/{}/manifest", self.prefix, LEAN_DIR)
    }
    /// The pointer: the ONLY mutable metadata object. Small, CAS'd on
    /// every citation and on every takeover rotation.
    pub fn current_key(&self) -> String {
        format!("{}/{}/current", self.prefix, LEAN_DIR)
    }
    /// One immutable generation of the entries.
    ///
    /// Zero-padded seq FIRST, so a plain lexical listing is also
    /// chronological — what the reaper and a human with `mc ls` both
    /// want — and the writer's flush uuid after it, so the key is
    /// unique per WRITE rather than per generation. Two writers that
    /// reach the same seq therefore write two different objects and
    /// race only at the pointer, which is the one place a race should
    /// be decided; and a path that legitimately rewrites without
    /// bumping is not refused by a write-once key it never meant to
    /// collide with.
    pub fn generation_key(&self, seq: u64, flush_uuid: &str) -> String {
        format!("{}/{}/manifests/{seq:020}-{flush_uuid}", self.prefix, LEAN_DIR)
    }

    /// A chunk object, addressed by CONTENT rather than by generation.
    /// Two generations that share a chunk share the object, which is
    /// where the O(changed) publish actually comes from — naming chunks
    /// by seq would rewrite every one of them on every publish even
    /// when their contents were identical.
    pub fn chunk_key(&self, addr: &str) -> String {
        format!("{}/{}/chunks/{addr}", self.prefix, LEAN_DIR)
    }
    pub fn inbox_key(&self) -> String {
        format!("{}/{}/inbox", self.prefix, LEAN_DIR)
    }
    pub fn epoch_key(&self) -> String {
        format!("{}/{}/epoch", self.prefix, LEAN_DIR)
    }
    /// The operator's claim cell for this prefix (`lean_operator::
    /// reconcile::claim_key` writes it; the syncer only READS it).
    pub fn claim_key(&self) -> String {
        format!("{}/{}/claim", self.prefix, LEAN_DIR)
    }
    pub fn conflict_key(&self, uuid: &str, path: &str) -> String {
        format!("{}/{}/conflicts/{}/{}", self.prefix, LEAN_DIR, uuid, path)
    }

    /// A per-user DRAFT: a durable edit that has NOT been published.
    ///
    /// Under `LEAN_DIR`, which is the whole safety argument: that
    /// namespace is never scanned, never checked out, never cited by a
    /// manifest, and never swept (both sweeps are prefix-scoped —
    /// `manifest::sweep_generations` to `manifests/`, `sweep_chunks` to
    /// `chunks/`). So a draft is durable in the bucket while being
    /// invisible to `classify`, to `checkout`, to every other user, and
    /// to the agent — which is exactly what "saved but not published"
    /// has to mean.
    ///
    /// Body and meta live under DISJOINT subtrees rather than sharing
    /// one with a suffix: with `drafts/<u>/<path>.meta` a file legally
    /// named `notes.meta` would have its BODY at the key that is
    /// `notes`'s META, and one user's draft would silently read as
    /// another draft's base etag.
    pub fn draft_body_key(&self, user: &str, path: &str) -> String {
        format!("{}/{}/drafts/{}/body/{}", self.prefix, LEAN_DIR, user, path)
    }
    /// The draft's base: which version of `files/<path>` it was edited
    /// against. Written AFTER the body, so the only crash residue is a
    /// body with no meta — an INCOMPLETE draft, which `promote` refuses
    /// by name rather than publishing against a base it had to guess.
    pub fn draft_meta_key(&self, user: &str, path: &str) -> String {
        format!("{}/{}/drafts/{}/meta/{}", self.prefix, LEAN_DIR, user, path)
    }
    /// Listing prefix for one user's draft metas (the resume view).
    pub fn draft_meta_prefix(&self, user: &str) -> String {
        format!("{}/{}/drafts/{}/meta/", self.prefix, LEAN_DIR, user)
    }
    pub fn state_dir(&self) -> PathBuf {
        self.root.join(STATE_DIR)
    }
    /// The workspace-local control namespace (D0).
    pub fn control_dir(&self) -> PathBuf {
        self.root.join(CONTROL_DIR)
    }
}

/// Everything a syncer operation needs: store + config + durable state.
pub struct Syncer {
    pub store: Arc<dyn ObjectStore>,
    pub cfg: LeanConfig,
    pub state: state::SyncerState,
    /// The lease held for the CURRENT commit section, if any. `None`
    /// between barriers: the cell arbitrates who installs a manifest and
    /// nothing else (design 2026-09-13 §4), so a barrier claims after
    /// its uploads and releases after its baseline rewrite.
    pub lease: Option<flint_store::EpochLease>,
    /// When this process last WROTE the cell (a claim, a renew): the
    /// clock the commit section's deletes fence on (review 2026-09-18,
    /// C2). `None` while no write is known — an adopted claim writes
    /// nothing — so the first delete renews.
    pub cell_written_at: Option<std::time::Instant>,
    /// Standing conditions already written to `conflicts.jsonl` by THIS
    /// process, so a condition that persists across every poll tick is
    /// recorded once rather than per tick. Deliberately in-memory: a
    /// restart is a new incarnation and should re-record what it finds.
    pub noted_not_regular: std::collections::BTreeSet<String>,
}

/// Seconds since the epoch, saturating at 0 on a clock before it.
pub fn now_unix() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}